fix(core): bound repeat backlog under encoder backpressure

commit: 8a7c43d693467d9692139d96a5f6ad0b764d5ed5
author: brice · 22d ago · 2026-09-30 08:22 UTC
committer: brice · 22d ago · 2026-09-30 14:48 UTC
parent: 1a6bd84
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +103 -15

M internal/adapters/cli/cli.go +2 -2


@@ -49,7 +49,7 @@ Commands:
49	49		gui open the capture control panel (default)
50	50		outputs list outputs
51	51		screenshot save a PNG screenshot
52		-	record record silent video until interrupted or -duration elapses
	52	+	record record silent video until interrupted or target video length is reached
53	53
54	54		Flags:
55	55		-debug enable debug logging (all commands)

@@ -58,7 +58,7 @@ Flags:
58	58		-file PATH new file to write, required (screenshot, record)
59	59		-fps N frames per second, 1..%d (record, default %d)
60	60		-size WxH scaled video size, both even, at most %d each (record, default source size)
61		-	-duration D recording length such as 30s; 0 records until interrupted (record)
	61	+	-duration D encoded video length such as 30s; 0 records until interrupted (record)
62	62		`, ports.MaxFPS, DefaultFPS, ports.MaxDimension)
63	63		}
64	64


@@ -49,7 +49,7 @@ Commands:
49		gui open the capture control panel (default)	49		gui open the capture control panel (default)
50		outputs list outputs	50		outputs list outputs
51		screenshot save a PNG screenshot	51		screenshot save a PNG screenshot
52	-	record record silent video until interrupted or -duration elapses	52	+	record record silent video until interrupted or target video length is reached
53			53
54		Flags:	54		Flags:
55		-debug enable debug logging (all commands)	55		-debug enable debug logging (all commands)

@@ -58,7 +58,7 @@ Flags:
58		-file PATH new file to write, required (screenshot, record)	58		-file PATH new file to write, required (screenshot, record)
59		-fps N frames per second, 1..%d (record, default %d)	59		-fps N frames per second, 1..%d (record, default %d)
60		-size WxH scaled video size, both even, at most %d each (record, default source size)	60		-size WxH scaled video size, both even, at most %d each (record, default source size)
61	-	-duration D recording length such as 30s; 0 records until interrupted (record)	61	+	-duration D encoded video length such as 30s; 0 records until interrupted (record)
62		`, ports.MaxFPS, DefaultFPS, ports.MaxDimension)	62		`, ports.MaxFPS, DefaultFPS, ports.MaxDimension)
63		}	63		}
64			64

M internal/core/service.go +41 -13


@@ -28,9 +28,22 @@ func New(source ports.Source, png ports.ScreenshotWriter, video ports.VideoWrite
28	28		}
29	29
30	30		// Run validates selection and then takes one screenshot or records silent
31		-	// fixed-rate video. Recording ends at the selection duration or when ctx is
32		-	// cancelled; both finalize the file through VideoWriter.Close. Captures and
33		-	// writes are sequential, so at most one is in flight.
	31	+	// fixed-rate video. Captures and writes are sequential, so at most one is in
	32	+	// flight, and frame storage is reused without copying.
	33	+	//
	34	+	// A timed recording (Duration > 0) encodes exactly ceil(Duration*FPS) frames,
	35	+	// so its encoded length is Duration rounded up to a frame. Wall time may run
	36	+	// longer when Capture or Write is slow: missed slots repeat the current frame
	37	+	// (at most two per fresh Capture) and the schedule then moves
	38	+	// forward, so a slow encoder lengthens the run instead of starving Capture.
	39	+	// The clock starts after Start and the first Write succeed.
	40	+	//
	41	+	// Cancelling ctx stops recording at the last frame already written and
	42	+	// finalizes through VideoWriter.Close. Nothing is written after cancellation.
	43	+	// Source and VideoWriter must report cancellation as ctx.Err() (or an error
	44	+	// wrapping only it); a failure joined with a context error, or a custom
	45	+	// cancellation cause, is treated as a real failure and returned.
	46	+	// A failure before the first frame is written aborts the file instead.
34	47		func (s *Service) Run(ctx context.Context, selection ports.Selection) error {
35	48		if err := validateSelection(selection); err != nil {
36	49		return err

@@ -90,6 +103,11 @@ func (s *Service) abort(err error) error {
90	103		return err
91	104		}
92	105
	106	+	// maxRepeatSlots bounds how many missed slots repeat the current frame before
	107	+	// the next fresh Capture. A larger lag moves the schedule forward instead, so
	108	+	// encoder backpressure becomes wall-time overrun rather than a stale-frame spiral.
	109	+	const maxRepeatSlots = 2
	110	+
93	111		// schedule maps elapsed time since the first frame to fixed-rate frame slots.
94	112		// Slot n is due at n/fps. total is the slot count of a timed recording, or 0
95	113		// when recording until cancellation.

@@ -116,22 +134,29 @@ func (sc schedule) due(n int64) time.Duration {
116	134		return time.Duration(q)time.Second + time.Duration(r)time.Second/time.Duration(sc.fps)
117	135		}
118	136
119		-	// current returns the slot containing now, limited to the last slot of a
120		-	// timed recording. past reports that a timed recording's window has ended.
121		-	func (sc schedule) current() (slot int64, past bool) {
	137	+	// slot returns the slot containing now, unclamped.
	138	+	func (sc schedule) slot() int64 {
122	139		elapsed := time.Since(sc.start)
123	140		secs, rem := int64(elapsed/time.Second), int64(elapsed%time.Second)
124		-	slot = secssc.fps + remsc.fps/int64(time.Second)
125		-	if sc.total > 0 && slot >= sc.total {
126		-	return sc.total - 1, true
	141	+	return secssc.fps + remsc.fps/int64(time.Second)
	142	+	}
	143	+
	144	+	// catchUp limits the lag of now behind slot next to maxRepeatSlots by moving
	145	+	// the schedule start forward. It returns the resulting current slot (>= next).
	146	+	func (sc *schedule) catchUp(next int64) int64 {
	147	+	cur := max(sc.slot(), next)
	148	+	if cur-next > maxRepeatSlots {
	149	+	sc.start = sc.start.Add(time.Since(sc.start) - sc.due(next+maxRepeatSlots))
	150	+	cur = max(sc.slot(), next)
127	151		}
128		-	return slot, false
	152	+	return cur
129	153		}
130	154
131	155		// recordLoop fills slots 1.. after the already written first frame. A slot
132	156		// missed because Capture or Write ran long reuses the current frame, which is
133	157		// still valid until the next Capture, so the encoded duration stays exact
134		-	// even when wall time overruns. A timed recording ends after ceil(duration*fps)
	158	+	// even when wall time overruns. At most maxRepeatSlots slots repeat per fresh
	159	+	// Capture; see catchUp. A timed recording ends after ceil(duration*fps)
135	160		// frames, so a duration shorter than one frame interval encodes one frame and
136	161		// returns without waiting. Cancellation ends the recording at the last frame
137	162		// already written: nothing is written afterwards, because the writer is

@@ -158,8 +183,11 @@ func (s *Service) recordLoop(ctx context.Context, frame ports.Frame, sel ports.S
158	183		return nil
159	184		}
160	185
161		-	cur, past := sc.current()
162		-	cur = max(cur, next)
	186	+	cur := sc.catchUp(next)
	187	+	past := sc.total > 0 && cur >= sc.total
	188	+	if past {
	189	+	cur = sc.total - 1
	190	+	}
163	191		// Slots already missed repeat the frame before it is invalidated.
164	192		last := cur - 1
165	193		if past {


@@ -28,9 +28,22 @@ func New(source ports.Source, png ports.ScreenshotWriter, video ports.VideoWrite
28		}	28		}
29			29
30		// Run validates selection and then takes one screenshot or records silent	30		// Run validates selection and then takes one screenshot or records silent
31	-	// fixed-rate video. Recording ends at the selection duration or when ctx is	31	+	// fixed-rate video. Captures and writes are sequential, so at most one is in
32	-	// cancelled; both finalize the file through VideoWriter.Close. Captures and	32	+	// flight, and frame storage is reused without copying.
33	-	// writes are sequential, so at most one is in flight.	33	+	//
			34	+	// A timed recording (Duration > 0) encodes exactly ceil(Duration*FPS) frames,
			35	+	// so its encoded length is Duration rounded up to a frame. Wall time may run
			36	+	// longer when Capture or Write is slow: missed slots repeat the current frame
			37	+	// (at most two per fresh Capture) and the schedule then moves
			38	+	// forward, so a slow encoder lengthens the run instead of starving Capture.
			39	+	// The clock starts after Start and the first Write succeed.
			40	+	//
			41	+	// Cancelling ctx stops recording at the last frame already written and
			42	+	// finalizes through VideoWriter.Close. Nothing is written after cancellation.
			43	+	// Source and VideoWriter must report cancellation as ctx.Err() (or an error
			44	+	// wrapping only it); a failure joined with a context error, or a custom
			45	+	// cancellation cause, is treated as a real failure and returned.
			46	+	// A failure before the first frame is written aborts the file instead.
34		func (s *Service) Run(ctx context.Context, selection ports.Selection) error {	47		func (s *Service) Run(ctx context.Context, selection ports.Selection) error {
35		if err := validateSelection(selection); err != nil {	48		if err := validateSelection(selection); err != nil {
36		return err	49		return err

@@ -90,6 +103,11 @@ func (s *Service) abort(err error) error {
90		return err	103		return err
91		}	104		}
92			105
			106	+	// maxRepeatSlots bounds how many missed slots repeat the current frame before
			107	+	// the next fresh Capture. A larger lag moves the schedule forward instead, so
			108	+	// encoder backpressure becomes wall-time overrun rather than a stale-frame spiral.
			109	+	const maxRepeatSlots = 2
			110	+
93		// schedule maps elapsed time since the first frame to fixed-rate frame slots.	111		// schedule maps elapsed time since the first frame to fixed-rate frame slots.
94		// Slot n is due at n/fps. total is the slot count of a timed recording, or 0	112		// Slot n is due at n/fps. total is the slot count of a timed recording, or 0
95		// when recording until cancellation.	113		// when recording until cancellation.

@@ -116,22 +134,29 @@ func (sc schedule) due(n int64) time.Duration {
116		return time.Duration(q)time.Second + time.Duration(r)time.Second/time.Duration(sc.fps)	134		return time.Duration(q)time.Second + time.Duration(r)time.Second/time.Duration(sc.fps)
117		}	135		}
118			136
119	-	// current returns the slot containing now, limited to the last slot of a	137	+	// slot returns the slot containing now, unclamped.
120	-	// timed recording. past reports that a timed recording's window has ended.	138	+	func (sc schedule) slot() int64 {
121	-	func (sc schedule) current() (slot int64, past bool) {
122		elapsed := time.Since(sc.start)	139		elapsed := time.Since(sc.start)
123		secs, rem := int64(elapsed/time.Second), int64(elapsed%time.Second)	140		secs, rem := int64(elapsed/time.Second), int64(elapsed%time.Second)
124	-	slot = secssc.fps + remsc.fps/int64(time.Second)	141	+	return secssc.fps + remsc.fps/int64(time.Second)
125	-	if sc.total > 0 && slot >= sc.total {	142	+	}
126	-	return sc.total - 1, true	143	+
			144	+	// catchUp limits the lag of now behind slot next to maxRepeatSlots by moving
			145	+	// the schedule start forward. It returns the resulting current slot (>= next).
			146	+	func (sc *schedule) catchUp(next int64) int64 {
			147	+	cur := max(sc.slot(), next)
			148	+	if cur-next > maxRepeatSlots {
			149	+	sc.start = sc.start.Add(time.Since(sc.start) - sc.due(next+maxRepeatSlots))
			150	+	cur = max(sc.slot(), next)
127		}	151		}
128	-	return slot, false	152	+	return cur
129		}	153		}
130			154
131		// recordLoop fills slots 1.. after the already written first frame. A slot	155		// recordLoop fills slots 1.. after the already written first frame. A slot
132		// missed because Capture or Write ran long reuses the current frame, which is	156		// missed because Capture or Write ran long reuses the current frame, which is
133		// still valid until the next Capture, so the encoded duration stays exact	157		// still valid until the next Capture, so the encoded duration stays exact
134	-	// even when wall time overruns. A timed recording ends after ceil(duration*fps)	158	+	// even when wall time overruns. At most maxRepeatSlots slots repeat per fresh
			159	+	// Capture; see catchUp. A timed recording ends after ceil(duration*fps)
135		// frames, so a duration shorter than one frame interval encodes one frame and	160		// frames, so a duration shorter than one frame interval encodes one frame and
136		// returns without waiting. Cancellation ends the recording at the last frame	161		// returns without waiting. Cancellation ends the recording at the last frame
137		// already written: nothing is written afterwards, because the writer is	162		// already written: nothing is written afterwards, because the writer is

@@ -158,8 +183,11 @@ func (s *Service) recordLoop(ctx context.Context, frame ports.Frame, sel ports.S
158		return nil	183		return nil
159		}	184		}
160			185
161	-	cur, past := sc.current()	186	+	cur := sc.catchUp(next)
162	-	cur = max(cur, next)	187	+	past := sc.total > 0 && cur >= sc.total
			188	+	if past {
			189	+	cur = sc.total - 1
			190	+	}
163		// Slots already missed repeat the frame before it is invalidated.	191		// Slots already missed repeat the frame before it is invalidated.
164		last := cur - 1	192		last := cur - 1
165		if past {	193		if past {

M internal/core/service_test.go +60 -0


@@ -404,6 +404,66 @@ func TestRecordExactFrameCountWithSlowStartAndCapture(t *testing.T) {
404	404		}
405	405		}
406	406
	407	+	// backpressure wires a slow Write and distinct capture storage, and reports
	408	+	// fresh captures, total writes and the longest run of repeated frames.
	409	+	type backpressure struct {
	410	+	captures atomic.Int32
	411	+	writes atomic.Int32
	412	+	maxRun int
	413	+	run int
	414	+	last *byte
	415	+	}
	416	+
	417	+	func (bp backpressure) wire(h harness, start mock.Call, slow time.Duration) {
	418	+	h.src.EXPECT().Capture(mock.Anything, target).RunAndReturn(func(context.Context, ports.Target) (ports.Frame, error) {
	419	+	bp.captures.Add(1)
	420	+	return newFrame(), nil
	421	+	}).Maybe()
	422	+	h.vid.EXPECT().Write(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, f ports.Frame) error {
	423	+	time.Sleep(slow)
	424	+	bp.writes.Add(1)
	425	+	if &f.Pixels[0] == bp.last {
	426	+	bp.run++
	427	+	} else {
	428	+	bp.run = 0
	429	+	}
	430	+	bp.maxRun = max(bp.maxRun, bp.run)
	431	+	bp.last = &f.Pixels[0]
	432	+	return nil
	433	+	}).NotBefore(start).Maybe()
	434	+	}
	435	+
	436	+	func TestRecordEncoderBackpressureKeepsCapturesFresh(t *testing.T) {
	437	+	h := newHarness(t)
	438	+	sel := rec(2 * time.Second)
	439	+	sel.Video.FPS = 60
	440	+	start := recordOrder(h, sel)
	441	+	var bp backpressure
	442	+	bp.wire(h, start, 20*time.Millisecond)
	443	+	h.vid.EXPECT().Close().Return(nil).Once()
	444	+
	445	+	require.NoError(t, h.svc.Run(context.Background(), sel))
	446	+	assert.Equal(t, int32(120), bp.writes.Load(), "exact encoded length")
	447	+	assert.GreaterOrEqual(t, bp.captures.Load(), int32(40), "captures must not starve")
	448	+	assert.LessOrEqual(t, bp.maxRun, 2, "repeat backlog is bounded")
	449	+	}
	450	+
	451	+	func TestRecordUntimedBackpressureBoundsRepeats(t *testing.T) {
	452	+	h := newHarness(t)
	453	+	sel := rec(0)
	454	+	sel.Video.FPS = 60
	455	+	ctx, cancel := context.WithCancel(context.Background())
	456	+	start := recordOrder(h, sel)
	457	+	var bp backpressure
	458	+	bp.wire(h, start, 40*time.Millisecond) // slower than two slots per frame
	459	+	h.vid.EXPECT().Close().Return(nil).Once()
	460	+
	461	+	time.AfterFunc(time.Second, cancel)
	462	+	require.NoError(t, h.svc.Run(ctx, sel))
	463	+	assert.Greater(t, bp.captures.Load(), int32(5))
	464	+	assert.LessOrEqual(t, bp.maxRun, 3, "repeated writes between fresh captures")
	465	+	}
	466	+
407	467		func TestRecordCancelWritesNothingAfterStop(t *testing.T) {
408	468		h := newHarness(t)
409	469		sel := rec(0)


@@ -404,6 +404,66 @@ func TestRecordExactFrameCountWithSlowStartAndCapture(t *testing.T) {
404	}	404		}
405	}	405		}
406		406
		407	+	// backpressure wires a slow Write and distinct capture storage, and reports
		408	+	// fresh captures, total writes and the longest run of repeated frames.
		409	+	type backpressure struct {
		410	+	captures atomic.Int32
		411	+	writes atomic.Int32
		412	+	maxRun int
		413	+	run int
		414	+	last *byte
		415	+	}
		416	+
		417	+	func (bp backpressure) wire(h harness, start mock.Call, slow time.Duration) {
		418	+	h.src.EXPECT().Capture(mock.Anything, target).RunAndReturn(func(context.Context, ports.Target) (ports.Frame, error) {
		419	+	bp.captures.Add(1)
		420	+	return newFrame(), nil
		421	+	}).Maybe()
		422	+	h.vid.EXPECT().Write(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, f ports.Frame) error {
		423	+	time.Sleep(slow)
		424	+	bp.writes.Add(1)
		425	+	if &f.Pixels[0] == bp.last {
		426	+	bp.run++
		427	+	} else {
		428	+	bp.run = 0
		429	+	}
		430	+	bp.maxRun = max(bp.maxRun, bp.run)
		431	+	bp.last = &f.Pixels[0]
		432	+	return nil
		433	+	}).NotBefore(start).Maybe()
		434	+	}
		435	+
		436	+	func TestRecordEncoderBackpressureKeepsCapturesFresh(t *testing.T) {
		437	+	h := newHarness(t)
		438	+	sel := rec(2 * time.Second)
		439	+	sel.Video.FPS = 60
		440	+	start := recordOrder(h, sel)
		441	+	var bp backpressure
		442	+	bp.wire(h, start, 20*time.Millisecond)
		443	+	h.vid.EXPECT().Close().Return(nil).Once()
		444	+
		445	+	require.NoError(t, h.svc.Run(context.Background(), sel))
		446	+	assert.Equal(t, int32(120), bp.writes.Load(), "exact encoded length")
		447	+	assert.GreaterOrEqual(t, bp.captures.Load(), int32(40), "captures must not starve")
		448	+	assert.LessOrEqual(t, bp.maxRun, 2, "repeat backlog is bounded")
		449	+	}
		450	+
		451	+	func TestRecordUntimedBackpressureBoundsRepeats(t *testing.T) {
		452	+	h := newHarness(t)
		453	+	sel := rec(0)
		454	+	sel.Video.FPS = 60
		455	+	ctx, cancel := context.WithCancel(context.Background())
		456	+	start := recordOrder(h, sel)
		457	+	var bp backpressure
		458	+	bp.wire(h, start, 40*time.Millisecond) // slower than two slots per frame
		459	+	h.vid.EXPECT().Close().Return(nil).Once()
		460	+
		461	+	time.AfterFunc(time.Second, cancel)
		462	+	require.NoError(t, h.svc.Run(ctx, sel))
		463	+	assert.Greater(t, bp.captures.Load(), int32(5))
		464	+	assert.LessOrEqual(t, bp.maxRun, 3, "repeated writes between fresh captures")
		465	+	}
		466	+
407	func TestRecordCancelWritesNothingAfterStop(t *testing.T) {	467		func TestRecordCancelWritesNothingAfterStop(t *testing.T) {
408	h := newHarness(t)	468		h := newHarness(t)
409	sel := rec(0)	469		sel := rec(0)