fix: reduce audio jitter buffer latency (#673)

* fix: reduce audio jitter buffer latency

- Release listenersMu before dispatching samples to listeners
- Buffer Track.sample channel (size 2) with non-blocking WriteSample
- Buffer pipeline.sample channel (size 4) to unblock GStreamer thread
- Add leaky=downstream max-size-buffers=5 to audio queue to prevent burst accumulation
- Set sync=false on appsink to deliver buffers immediately without clock wait

These changes break the backpressure chain that caused the GStreamer
streaming thread to stall. When the Go scheduler has a hiccup, buffered
frames were delivered in a burst, causing the browser audio jitter buffer
to grow to ~10 seconds over time.

Fixes: #598

* fix: log message when dropping samples due to full track channel

server/internal/capture/manager.go

@@ -135,7 +135,7 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
 				"pulsesrc device=%s "+
 					"! audio/x-raw,channels=2 "+
 					"! audioconvert "+
-					"! queue "+
+					"! queue max-size-buffers=5 leaky=downstream "+
 					"! %s "+
 					"! appsink name=appsink", config.AudioDevice, config.AudioCodec.Pipeline,
 			), nil

server/internal/capture/streamsink.go

@@ -369,7 +369,6 @@ func (manager *StreamSinkManagerCtx) saveSampleBitrate(timestamp time.Time, delt
 
 func (manager *StreamSinkManagerCtx) onSample(sample types.Sample) {
 	manager.listenersMu.Lock()
-	defer manager.listenersMu.Unlock()
 
 	// save to metrics
 	length := float64(sample.Length)
@@ -386,7 +385,14 @@ func (manager *StreamSinkManagerCtx) onSample(sample types.Sample) {
 		manager.listenersKf = make(map[uintptr]types.SampleListener)
 	}
 
+	// copy listeners before releasing lock to avoid holding it during dispatch
+	listeners := make([]types.SampleListener, 0, len(manager.listeners))
 	for _, l := range manager.listeners {
+		listeners = append(listeners, l)
+	}
+	manager.listenersMu.Unlock()
+
+	for _, l := range listeners {
 		l.WriteSample(sample)
 	}
 }

server/internal/webrtc/track.go

@@ -45,7 +45,7 @@ func NewTrack(logger zerolog.Logger, codec codec.RTPCodec, connection *webrtc.Pe
 		logger: logger.With().Str("id", id).Logger(),
 		track:  track,
 		rtcpCh: nil,
-		sample: make(chan types.Sample),
+		sample: make(chan types.Sample, 2),
 	}
 
 	for _, opt := range opts {
@@ -110,7 +110,11 @@ func (t *Track) sampleReader() {
 }
 
 func (t *Track) WriteSample(sample types.Sample) {
-	t.sample <- sample
+	select {
+	case t.sample <- sample:
+	default:
+		t.logger.Trace().Msg("dropping sample: track channel full")
+	}
 }
 
 // --- stream ---

server/pkg/gst/gst.c

@@ -108,7 +108,7 @@ static GstFlowReturn gstreamer_send_new_sample_handler(GstElement *object, gpoin
 
 void gstreamer_pipeline_attach_appsink(GstPipelineCtx *ctx, char *sinkName) {
   ctx->appsink = gst_bin_get_by_name(GST_BIN(ctx->pipeline), sinkName);
-  g_object_set(ctx->appsink, "emit-signals", TRUE, NULL);
+  g_object_set(ctx->appsink, "emit-signals", TRUE, "sync", FALSE, NULL);
   g_signal_connect(ctx->appsink, "new-sample", G_CALLBACK(gstreamer_send_new_sample_handler), ctx);
 }
 

server/pkg/gst/gst.go

@@ -83,7 +83,7 @@ func CreatePipeline(pipelineStr string) (Pipeline, error) {
 			Int("pipeline_id", int(id)).Logger(),
 		src:    pipelineStr,
 		ctx:    ctx,
-		sample: make(chan types.Sample),
+		sample: make(chan types.Sample, 4),
 	}
 
 	pipelines[p.id] = p