Harden RTSP backchannel streaming

Add paced, randomized RTP packetization with talkspurt markers and rollover-safe timestamps. Reconnect failed backchannel sessions with cancellable exponential backoff, initialize audio channels during bootstrap, and cover packetizer behavior with tests.

machinery/src/components/backchannel.go

@@ -2,7 +2,9 @@ package components
 
 import (
 	"bufio"
+	"context"
 	"fmt"
+	"math/rand"
 	"os"
 	"time"
 
@@ -15,6 +17,62 @@ import (
 	"github.com/zaf/g711"
 )
 
+const (
+	backchannelSampleRate       = 8000
+	backchannelTalkspurtGap     = 500 * time.Millisecond
+	backchannelReconnectInitial = time.Second
+	backchannelReconnectMax     = 30 * time.Second
+)
+
+type backchannelClient interface {
+	ConnectBackChannel(ctx context.Context, otelContext context.Context) error
+	StartBackChannel(ctx context.Context, otelContext context.Context) error
+	WritePacket(pkt packets.Packet) error
+	Close(otelContext context.Context) error
+}
+
+type backchannelPacketizer struct {
+	sequenceNumber uint16
+	timestamp      uint32
+	ssrc           uint32
+	lastPacketAt   time.Time
+}
+
+func newBackchannelPacketizer() backchannelPacketizer {
+	return backchannelPacketizer{
+		sequenceNumber: uint16(rand.Uint32()),
+		timestamp:      rand.Uint32(),
+		ssrc:           rand.Uint32(),
+	}
+}
+
+func (p *backchannelPacketizer) packet(audio models.AudioDataPartial, now time.Time) packets.Packet {
+	bufferUlaw := make([]byte, len(audio.Data))
+	for index, sample := range audio.Data {
+		bufferUlaw[index] = g711.EncodeUlawFrame(sample)
+	}
+
+	pkt := packets.Packet{
+		Packet: &rtp.Packet{
+			Header: rtp.Header{
+				Version:        2,
+				Marker:         p.lastPacketAt.IsZero() || now.Sub(p.lastPacketAt) >= backchannelTalkspurtGap,
+				PayloadType:    0,
+				SequenceNumber: p.sequenceNumber,
+				Timestamp:      p.timestamp,
+				SSRC:           p.ssrc,
+			},
+			Payload: bufferUlaw,
+		},
+	}
+
+	p.timestamp += uint32(len(bufferUlaw))
+	p.sequenceNumber++
+	p.lastPacketAt = now
+
+	return pkt
+}
+
 func GetBackChannelAudioCodec(streams []av.CodecData, communication *models.Communication) av.AudioCodecData {
 	for _, stream := range streams {
 		if stream.Type().IsAudio() {
@@ -31,41 +89,115 @@ func GetBackChannelAudioCodec(streams []av.CodecData, communication *models.Comm
 }
 
 func WriteAudioToBackchannel(communication *models.Communication, rtspClient capture.RTSPClient) {
-	log.Log.Info("Audio.WriteAudioToBackchannel(): writing to backchannel audio codec")
-	length := uint32(0)
-	sequenceNumber := uint16(0)
-	for audio := range communication.HandleAudio {
-		// Encode PCM to MULAW
-		var bufferUlaw []byte
-		for _, v := range audio.Data {
-			b := g711.EncodeUlawFrame(v)
-			bufferUlaw = append(bufferUlaw, b)
-		}
-
-		pkt := packets.Packet{
-			Packet: &rtp.Packet{
-				Header: rtp.Header{
-					Version:        2,
-					Marker:         true, // should be true
-					PayloadType:    0,    //packet.PayloadType, // will be owerwriten
-					SequenceNumber: sequenceNumber,
-					Timestamp:      uint32(length),
-					SSRC:           1293847657,
-				},
-				Payload: bufferUlaw,
-			},
-		}
-		err := rtspClient.WritePacket(pkt)
-		if err != nil {
-			log.Log.Error("Audio.WriteAudioToBackchannel(): error writing packet to backchannel")
-		}
-
-		length = (length + uint32(len(bufferUlaw))) % 65536
-		sequenceNumber = (sequenceNumber + 1) % 65535
-		time.Sleep(128 * time.Millisecond)
+	ctx := context.Background()
+	if communication.Context != nil {
+		ctx = *communication.Context
 	}
-	log.Log.Info("Audio.WriteAudioToBackchannel(): finished")
 
+	writeAudioToBackchannel(ctx, ctx, communication.HandleAudio, rtspClient)
+}
+
+func writeAudioToBackchannel(ctx context.Context, otelContext context.Context, audioChannel <-chan models.AudioDataPartial, rtspClient backchannelClient) {
+	log.Log.Info("Audio.WriteAudioToBackchannel(): writing to backchannel audio codec")
+
+	if err := rtspClient.StartBackChannel(ctx, otelContext); err != nil {
+		log.Log.Error("Audio.WriteAudioToBackchannel(): error starting backchannel: " + err.Error())
+		if !reconnectBackchannel(ctx, otelContext, rtspClient) {
+			log.Log.Info("Audio.WriteAudioToBackchannel(): stopped while reconnecting")
+			return
+		}
+	}
+
+	packetizer := newBackchannelPacketizer()
+	for {
+		select {
+		case <-ctx.Done():
+			log.Log.Info("Audio.WriteAudioToBackchannel(): stopped")
+			return
+		case audio, ok := <-audioChannel:
+			if !ok {
+				log.Log.Info("Audio.WriteAudioToBackchannel(): finished")
+				return
+			}
+
+			audio = latestBackchannelAudio(audio, audioChannel)
+			if len(audio.Data) == 0 {
+				continue
+			}
+
+			pkt := packetizer.packet(audio, time.Now())
+			if err := rtspClient.WritePacket(pkt); err != nil {
+				log.Log.Error("Audio.WriteAudioToBackchannel(): error writing packet to backchannel: " + err.Error())
+				if !reconnectBackchannel(ctx, otelContext, rtspClient) {
+					log.Log.Info("Audio.WriteAudioToBackchannel(): stopped while reconnecting")
+					return
+				}
+				packetizer = newBackchannelPacketizer()
+				continue
+			}
+
+			if !waitForBackchannel(ctx, time.Duration(len(audio.Data))*time.Second/backchannelSampleRate) {
+				log.Log.Info("Audio.WriteAudioToBackchannel(): stopped")
+				return
+			}
+		}
+	}
+}
+
+func latestBackchannelAudio(audio models.AudioDataPartial, audioChannel <-chan models.AudioDataPartial) models.AudioDataPartial {
+	for {
+		select {
+		case next, ok := <-audioChannel:
+			if !ok {
+				return audio
+			}
+			audio = next
+		default:
+			return audio
+		}
+	}
+}
+
+func reconnectBackchannel(ctx context.Context, otelContext context.Context, rtspClient backchannelClient) bool {
+	backoff := backchannelReconnectInitial
+	for {
+		if err := rtspClient.Close(otelContext); err != nil {
+			log.Log.Error("Audio.WriteAudioToBackchannel(): error closing failed backchannel: " + err.Error())
+		}
+		if ctx.Err() != nil {
+			return false
+		}
+
+		err := rtspClient.ConnectBackChannel(ctx, otelContext)
+		if err == nil {
+			err = rtspClient.StartBackChannel(ctx, otelContext)
+		}
+		if err == nil {
+			log.Log.Info("Audio.WriteAudioToBackchannel(): reconnected backchannel")
+			return true
+		}
+
+		log.Log.Error("Audio.WriteAudioToBackchannel(): error reconnecting backchannel: " + err.Error())
+		if !waitForBackchannel(ctx, backoff) {
+			return false
+		}
+		backoff *= 2
+		if backoff > backchannelReconnectMax {
+			backoff = backchannelReconnectMax
+		}
+	}
+}
+
+func waitForBackchannel(ctx context.Context, duration time.Duration) bool {
+	timer := time.NewTimer(duration)
+	defer timer.Stop()
+
+	select {
+	case <-ctx.Done():
+		return false
+	case <-timer.C:
+		return true
+	}
 }
 
 func WriteFileToBackChannel(infile av.DemuxCloser) {

machinery/src/components/backchannel_test.go

@@ -0,0 +1,60 @@
+package components
+
+import (
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+func TestBackchannelPacketizerUsesFullRTPClock(t *testing.T) {
+	packetizer := backchannelPacketizer{ssrc: 1}
+	audio := models.AudioDataPartial{Data: make([]int16, 1024)}
+	startedAt := time.Unix(1, 0)
+
+	var timestamp uint32
+	for index := 0; index <= 64; index++ {
+		pkt := packetizer.packet(audio, startedAt.Add(time.Duration(index)*128*time.Millisecond))
+		timestamp = pkt.Packet.Timestamp
+	}
+
+	if timestamp != 65536 {
+		t.Fatalf("timestamp after 64 frames = %d, want 65536", timestamp)
+	}
+}
+
+func TestBackchannelPacketizerUsesNaturalSequenceRollover(t *testing.T) {
+	packetizer := backchannelPacketizer{sequenceNumber: ^uint16(0), ssrc: 1}
+	audio := models.AudioDataPartial{Data: []int16{0}}
+	startedAt := time.Unix(1, 0)
+
+	last := packetizer.packet(audio, startedAt)
+	firstAfterRollover := packetizer.packet(audio, startedAt.Add(time.Millisecond))
+
+	if last.Packet.SequenceNumber != ^uint16(0) {
+		t.Fatalf("last sequence number = %d, want %d", last.Packet.SequenceNumber, ^uint16(0))
+	}
+	if firstAfterRollover.Packet.SequenceNumber != 0 {
+		t.Fatalf("first sequence number after rollover = %d, want 0", firstAfterRollover.Packet.SequenceNumber)
+	}
+}
+
+func TestBackchannelPacketizerMarksTalkspurtStart(t *testing.T) {
+	packetizer := backchannelPacketizer{ssrc: 1}
+	audio := models.AudioDataPartial{Data: []int16{0}}
+	startedAt := time.Unix(1, 0)
+
+	first := packetizer.packet(audio, startedAt)
+	continuous := packetizer.packet(audio, startedAt.Add(128*time.Millisecond))
+	afterGap := packetizer.packet(audio, startedAt.Add(backchannelTalkspurtGap+128*time.Millisecond))
+
+	if !first.Packet.Marker {
+		t.Fatal("first packet must mark the start of a talkspurt")
+	}
+	if continuous.Packet.Marker {
+		t.Fatal("continuous packet must not carry the marker bit")
+	}
+	if !afterGap.Packet.Marker {
+		t.Fatal("packet after an audio gap must mark a new talkspurt")
+	}
+}

machinery/src/components/kerberos.go

@@ -74,6 +74,7 @@ func Bootstrap(ctx context.Context, configDirectory string, configuration *model
 	communication.HandleLiveHDKeepalive = make(chan string, 1)
 	communication.HandleLiveHDPeers = make(chan string, 1)
 	communication.HandleLiveHLS = make(chan string, 1)
+	communication.HandleAudio = make(chan models.AudioDataPartial, 10)
 	communication.IsConfiguring = abool.New()
 	communication.IsRecordingManual = abool.New()
 	communication.RecordingManualHeartbeat = &atomic.Int64{}
@@ -269,11 +270,11 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	communication.MainStreamConnected = true
 
 	// Try to create backchannel
+	communication.HasBackChannel = false
 	rtspBackChannelClient := captureDevice.SetBackChannelClient(rtspUrl)
 	err = rtspBackChannelClient.ConnectBackChannel(ctx, ctxRunAgent)
 	if err == nil {
 		log.Log.Info("components.Kerberos.RunAgent(): opened RTSP backchannel stream: " + rtspUrl)
-		go rtspBackChannelClient.StartBackChannel(ctx, ctxRunAgent)
 	}
 
 	rtspSubClient := captureDevice.RTSPSubClient
@@ -350,7 +351,6 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	// is a no-op if ONVIFMotion is not enabled.
 	go onvif.HandleONVIFEventStream(*communication.Context, configuration, communication)
 
-	communication.HandleAudio = make(chan models.AudioDataPartial, 10)
 	if rtspBackChannelClient.HasBackChannel {
 		communication.HasBackChannel = true
 		go WriteAudioToBackchannel(communication, rtspBackChannelClient)
@@ -441,9 +441,6 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu
 	close(communication.HandleMotion)
 	communication.HandleMotion = nil
 
-	close(communication.HandleAudio)
-	communication.HandleAudio = nil
-
 	close(communication.HandleONVIF)
 	communication.HandleONVIF = nil