From db135acea9628c47ce638ecab4639ccf4ffdc62c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9dric=20Verstraeten?= Date: Mon, 10 Aug 2026 12:47:10 +0000 Subject: [PATCH] Implement backchannel reconnection logic and enhance test coverage for write failures --- machinery/src/components/backchannel_test.go | 149 +++++++++++++++++++ machinery/src/routers/mqtt/main.go | 25 +++- machinery/src/routers/mqtt/main_test.go | 38 +++++ 3 files changed, 210 insertions(+), 2 deletions(-) create mode 100644 machinery/src/routers/mqtt/main_test.go diff --git a/machinery/src/components/backchannel_test.go b/machinery/src/components/backchannel_test.go index 9c5d58f..e0644fc 100644 --- a/machinery/src/components/backchannel_test.go +++ b/machinery/src/components/backchannel_test.go @@ -1,12 +1,86 @@ package components import ( + "context" + "errors" + "sync" "testing" "time" "github.com/kerberos-io/agent/machinery/src/models" + "github.com/kerberos-io/agent/machinery/src/packets" ) +type fakeBackchannelClient struct { + mutex sync.Mutex + startErrors []error + connectError error + writeErrors []error + startCalls int + connectCalls int + closeCalls int + writeCalls int + connectAttempt chan struct{} + successfulWrite chan packets.Packet +} + +func (f *fakeBackchannelClient) ConnectBackChannel(context.Context, context.Context) error { + f.mutex.Lock() + f.connectCalls++ + err := f.connectError + f.mutex.Unlock() + + select { + case f.connectAttempt <- struct{}{}: + default: + } + return err +} + +func (f *fakeBackchannelClient) StartBackChannel(context.Context, context.Context) error { + f.mutex.Lock() + defer f.mutex.Unlock() + f.startCalls++ + if len(f.startErrors) == 0 { + return nil + } + err := f.startErrors[0] + f.startErrors = f.startErrors[1:] + return err +} + +func (f *fakeBackchannelClient) WritePacket(pkt packets.Packet) error { + f.mutex.Lock() + f.writeCalls++ + var err error + if len(f.writeErrors) != 0 { + err = f.writeErrors[0] + f.writeErrors = f.writeErrors[1:] + } + f.mutex.Unlock() + + if err == nil { + select { + case f.successfulWrite <- pkt: + default: + } + } + return err +} + +func (f *fakeBackchannelClient) Close(context.Context) error { + f.mutex.Lock() + f.closeCalls++ + f.mutex.Unlock() + return nil +} + +func (f *fakeBackchannelClient) callCounts() (start, connect, close, write int) { + f.mutex.Lock() + defer f.mutex.Unlock() + return f.startCalls, f.connectCalls, f.closeCalls, f.writeCalls +} + func TestBackchannelPacketizerUsesFullRTPClock(t *testing.T) { packetizer := backchannelPacketizer{ssrc: 1} audio := models.AudioDataPartial{Data: make([]int16, 1024)} @@ -57,4 +131,79 @@ func TestBackchannelPacketizerMarksTalkspurtStart(t *testing.T) { if !afterGap.Packet.Marker { t.Fatal("packet after an audio gap must mark a new talkspurt") } +} + +func TestWriteAudioToBackchannelReconnectsAfterWriteFailure(t *testing.T) { + writeFailure := errors.New("EOF") + client := &fakeBackchannelClient{ + writeErrors: []error{writeFailure, nil}, + connectAttempt: make(chan struct{}, 1), + successfulWrite: make(chan packets.Packet, 1), + } + audioChannel := make(chan models.AudioDataPartial, 2) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + writeAudioToBackchannel(ctx, ctx, audioChannel, client) + close(done) + }() + + audioChannel <- models.AudioDataPartial{Data: make([]int16, 1024)} + select { + case <-client.connectAttempt: + case <-time.After(time.Second): + t.Fatal("backchannel was not reconnected after the write failure") + } + + audioChannel <- models.AudioDataPartial{Data: make([]int16, 1024)} + select { + case pkt := <-client.successfulWrite: + if !pkt.Packet.Marker { + t.Fatal("first packet after reconnect must mark a new talkspurt") + } + case <-time.After(time.Second): + t.Fatal("fresh audio was not written after reconnect") + } + + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("backchannel writer did not stop after cancellation") + } + + startCalls, connectCalls, closeCalls, writeCalls := client.callCounts() + if startCalls != 2 || connectCalls != 1 || closeCalls != 1 || writeCalls != 2 { + t.Fatalf("calls (start, connect, close, write) = (%d, %d, %d, %d), want (2, 1, 1, 2)", startCalls, connectCalls, closeCalls, writeCalls) + } +} + +func TestWriteAudioToBackchannelCancellationStopsReconnect(t *testing.T) { + client := &fakeBackchannelClient{ + connectError: errors.New("camera unavailable"), + writeErrors: []error{errors.New("EOF")}, + connectAttempt: make(chan struct{}, 1), + successfulWrite: make(chan packets.Packet, 1), + } + audioChannel := make(chan models.AudioDataPartial, 1) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + writeAudioToBackchannel(ctx, ctx, audioChannel, client) + close(done) + }() + + audioChannel <- models.AudioDataPartial{Data: make([]int16, 1024)} + select { + case <-client.connectAttempt: + case <-time.After(time.Second): + t.Fatal("expected a reconnect attempt") + } + cancel() + + select { + case <-done: + case <-time.After(250 * time.Millisecond): + t.Fatal("cancellation did not interrupt reconnect backoff") + } } \ No newline at end of file diff --git a/machinery/src/routers/mqtt/main.go b/machinery/src/routers/mqtt/main.go index 694039f..af38a2b 100644 --- a/machinery/src/routers/mqtt/main.go +++ b/machinery/src/routers/mqtt/main.go @@ -332,7 +332,7 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory case "record": go HandleRecording(mqttClient, hubKey, payload, configuration, communication) case "get-audio-backchannel": - go HandleAudio(mqttClient, hubKey, payload, configuration, communication) + HandleAudio(mqttClient, hubKey, payload, configuration, communication) case "get-ptz-position": go HandleGetPTZPosition(mqttClient, hubKey, payload, configuration, communication) case "update-ptz-position": @@ -442,10 +442,31 @@ func HandleAudio(mqttClient mqtt.Client, hubKey string, payload models.Payload, Timestamp: audioPayload.Timestamp, Data: audioPayload.Data, } - communication.HandleAudio <- audioDataPartial + if enqueueLatestAudio(communication.HandleAudio, audioDataPartial) { + log.Log.Debug("routers.mqtt.main.HandleAudio(): dropped stale audio because the backchannel queue was full") + } } } +func enqueueLatestAudio(audioChannel chan models.AudioDataPartial, audio models.AudioDataPartial) bool { + select { + case audioChannel <- audio: + return false + default: + } + + select { + case <-audioChannel: + default: + } + + select { + case audioChannel <- audio: + default: + } + return true +} + func HandleGetPTZPosition(mqttClient mqtt.Client, hubKey string, payload models.Payload, configuration *models.Configuration, communication *models.Communication) { value := payload.Value diff --git a/machinery/src/routers/mqtt/main_test.go b/machinery/src/routers/mqtt/main_test.go new file mode 100644 index 0000000..1773fb9 --- /dev/null +++ b/machinery/src/routers/mqtt/main_test.go @@ -0,0 +1,38 @@ +package mqtt + +import ( + "testing" + + "github.com/kerberos-io/agent/machinery/src/models" +) + +func TestEnqueueLatestAudioReplacesOldestFrameWhenFull(t *testing.T) { + audioChannel := make(chan models.AudioDataPartial, 2) + audioChannel <- models.AudioDataPartial{Timestamp: 1} + audioChannel <- models.AudioDataPartial{Timestamp: 2} + + dropped := enqueueLatestAudio(audioChannel, models.AudioDataPartial{Timestamp: 3}) + if !dropped { + t.Fatal("enqueueLatestAudio() dropped = false, want true") + } + + first := <-audioChannel + second := <-audioChannel + if first.Timestamp != 2 || second.Timestamp != 3 { + t.Fatalf("queued timestamps = (%d, %d), want (2, 3)", first.Timestamp, second.Timestamp) + } +} + +func TestEnqueueLatestAudioDoesNotBlockNilChannel(t *testing.T) { + done := make(chan struct{}) + go func() { + enqueueLatestAudio(nil, models.AudioDataPartial{Timestamp: 1}) + close(done) + }() + + select { + case <-done: + default: + t.Fatal("enqueueLatestAudio() blocked on a nil channel") + } +} \ No newline at end of file