Implement backchannel reconnection logic and enhance test coverage for write failures

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")
+	}
 }

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
 

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")
+	}
+}