Raise minimum image upload queue capacity to absorb frame bursts (#1338)

* Raise minimum image upload queue capacity to absorb frame bursts

A capture interval of 3600s or more yields a queue capacity of at most
one slot, which the startup frame burst (videorate skip-to-first plus
the first rate-aligned frame) overflows while the first upload is in
flight, failing the egress with 'upload queue full' at startup.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Address review: use max builtin, reword test comment

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: milos-lk

pkg/pipeline/sink/image.go

@@ -54,6 +54,14 @@ type imageUpdate struct {
 	filename  string
 }
 
+// covers frame bursts that outpace the capture interval: videorate's
+// skip-to-first plus the first rate-aligned frame at startup, and the EOS flush
+const minPendingUploads = 4
+
+func imageQueueCapacity(maxUploadQueue int, captureInterval uint32) int {
+	return max((maxUploadQueue*60)/int(captureInterval), minPendingUploads)
+}
+
 func newImageSink(
 	p *gstreamer.Pipeline,
 	conf *config.PipelineConfig,
@@ -74,10 +82,6 @@ func newImageSink(
 		return nil, err
 	}
 
-	maxPendingUploads := (conf.MaxUploadQueue * 60) / int(o.CaptureInterval)
-	if maxPendingUploads < 1 {
-		maxPendingUploads = 1
-	}
 	return &ImageSink{
 		base: &base{
 			bin: imageBin,
@@ -88,7 +92,7 @@ func newImageSink(
 
 		conf:          conf,
 		callbacks:     callbacks,
-		createdImages: make(chan *imageUpdate, maxPendingUploads),
+		createdImages: make(chan *imageUpdate, imageQueueCapacity(conf.MaxUploadQueue, o.CaptureInterval)),
 	}, nil
 }
 

pkg/pipeline/sink/image_test.go

@@ -162,3 +162,13 @@ func TestImageSinkCloseDrainsPendingUploads(t *testing.T) {
 	default:
 	}
 }
+
+// a capture interval of 3600s or more must not yield a queue too small to
+// absorb the startup frame burst
+func TestImageQueueCapacity(t *testing.T) {
+	require.Equal(t, 360, imageQueueCapacity(60, 10))
+	require.Equal(t, 60, imageQueueCapacity(60, 60))
+	require.Equal(t, minPendingUploads, imageQueueCapacity(60, 3600))
+	require.Equal(t, minPendingUploads, imageQueueCapacity(60, 7200))
+	require.Equal(t, minPendingUploads, imageQueueCapacity(0, 10))
+}