fix(core): stop recording at last written frame on cancellation

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

2 files · +29 -35

M internal/core/service.go +16 -25


@@ -90,10 +90,6 @@ func (s *Service) abort(err error) error {
90	90		return err
91	91		}
92	92
93		-	// stopWriteTimeout bounds the trailing duplicate writes made after a stop
94		-	// request, which must not be interrupted by the cancellation itself.
95		-	const stopWriteTimeout = 2 * time.Second
96		-
97	93		// schedule maps elapsed time since the first frame to fixed-rate frame slots.
98	94		// Slot n is due at n/fps. total is the slot count of a timed recording, or 0
99	95		// when recording until cancellation.

@@ -136,7 +132,12 @@ func (sc schedule) current() (slot int64, past bool) {
136	132		// missed because Capture or Write ran long reuses the current frame, which is
137	133		// still valid until the next Capture, so the encoded duration stays exact
138	134		// even when wall time overruns. A timed recording ends after ceil(duration*fps)
139		-	// frames; otherwise it ends when ctx is done.
	135	+	// 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
	137	+	// already written: nothing is written afterwards, because the writer is
	138	+	// cancelled with ctx and storage may be invalidated by an interrupted Capture.
	139	+	// Exact frame counts therefore hold only for timed recordings that are not
	140	+	// cancelled. Slot n < total is due before the duration, so due cannot overflow.
140	141		func (s *Service) recordLoop(ctx context.Context, frame ports.Frame, sel ports.Selection) error {
141	142		sc := newSchedule(time.Now(), sel)
142	143		timer := time.NewTimer(time.Hour)

@@ -149,11 +150,12 @@ func (s *Service) recordLoop(ctx context.Context, frame ports.Frame, sel ports.S
149	150		timer.Reset(wait)
150	151		select {
151	152		case <-ctx.Done():
152		-	return s.fillElapsed(ctx, frame, next, sc)
	153	+	return nil
153	154		case <-timer.C:
154	155		}
155		-	} else if ctx.Err() != nil {
156		-	return s.fillElapsed(ctx, frame, next, sc)
	156	+	}
	157	+	if ctx.Err() != nil {
	158	+	return nil
157	159		}
158	160
159	161		cur, past := sc.current()

@@ -164,6 +166,9 @@ func (s *Service) recordLoop(ctx context.Context, frame ports.Frame, sel ports.S
164	166		last = cur // window over: no capture for the final slot
165	167		}
166	168		for ; next <= last; next++ {
	169	+	if ctx.Err() != nil {
	170	+	return nil
	171	+	}
167	172		if err := s.video.Write(ctx, frame); err != nil {
168	173		return writeResult(ctx, err)
169	174		}

@@ -182,6 +187,9 @@ func (s *Service) recordLoop(ctx context.Context, frame ports.Frame, sel ports.S
182	187		if err := frame.Validate(); err != nil {
183	188		return fmt.Errorf("captured frame: %w", err)
184	189		}
	190	+	if ctx.Err() != nil {
	191	+	return nil
	192	+	}
185	193		if err := s.video.Write(ctx, frame); err != nil {
186	194		return writeResult(ctx, err)
187	195		}

@@ -190,23 +198,6 @@ func (s *Service) recordLoop(ctx context.Context, frame ports.Frame, sel ports.S
190	198		return nil
191	199		}
192	200
193		-	// fillElapsed repeats the current frame for slots that began before the stop
194		-	// request so the file covers the time actually recorded.
195		-	func (s *Service) fillElapsed(ctx context.Context, frame ports.Frame, next int64, sc schedule) error {
196		-	cur, _ := sc.current()
197		-	if next > cur {
198		-	return nil
199		-	}
200		-	wctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), stopWriteTimeout)
201		-	defer cancel()
202		-	for ; next <= cur; next++ {
203		-	if err := s.video.Write(wctx, frame); err != nil {
204		-	return fmt.Errorf("write trailing frame: %w", err)
205		-	}
206		-	}
207		-	return nil
208		-	}
209		-
210	201		func writeResult(ctx context.Context, err error) error {
211	202		if stopped(ctx, err) {
212	203		return nil

 	return err
 }
-// stopWriteTimeout bounds the trailing duplicate writes made after a stop
-// request, which must not be interrupted by the cancellation itself.
-const stopWriteTimeout = 2 * time.Second
 // schedule maps elapsed time since the first frame to fixed-rate frame slots.
 // Slot n is due at n/fps. total is the slot count of a timed recording, or 0
 // when recording until cancellation.
 // missed because Capture or Write ran long reuses the current frame, which is
 // still valid until the next Capture, so the encoded duration stays exact
 // even when wall time overruns. A timed recording ends after ceil(duration*fps)
-// frames; otherwise it ends when ctx is done.
+// frames, so a duration shorter than one frame interval encodes one frame and
+// returns without waiting. Cancellation ends the recording at the last frame
+// already written: nothing is written afterwards, because the writer is
+// cancelled with ctx and storage may be invalidated by an interrupted Capture.
+// Exact frame counts therefore hold only for timed recordings that are not
+// cancelled. Slot n < total is due before the duration, so due cannot overflow.
 func (s *Service) recordLoop(ctx context.Context, frame ports.Frame, sel ports.Selection) error {
 	sc := newSchedule(time.Now(), sel)
 	timer := time.NewTimer(time.Hour)
 			timer.Reset(wait)
 			select {
 			case <-ctx.Done():
-				return s.fillElapsed(ctx, frame, next, sc)
+				return nil
 			case <-timer.C:
 			}
-		} else if ctx.Err() != nil {
+		}
-			return s.fillElapsed(ctx, frame, next, sc)
+		if ctx.Err() != nil {
+			return nil
 		}
 		cur, past := sc.current()
 			last = cur // window over: no capture for the final slot
 		}
 		for ; next <= last; next++ {
+			if ctx.Err() != nil {
+				return nil
+			}
 			if err := s.video.Write(ctx, frame); err != nil {
 				return writeResult(ctx, err)
 			}
 		if err := frame.Validate(); err != nil {
 			return fmt.Errorf("captured frame: %w", err)
 		}
+		if ctx.Err() != nil {
+			return nil
+		}
 		if err := s.video.Write(ctx, frame); err != nil {
 			return writeResult(ctx, err)
 		}
 	return nil
 }
-// fillElapsed repeats the current frame for slots that began before the stop
-// request so the file covers the time actually recorded.
-func (s *Service) fillElapsed(ctx context.Context, frame ports.Frame, next int64, sc schedule) error {
-	cur, _ := sc.current()
-	if next > cur {
-		return nil
-	}
-	wctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), stopWriteTimeout)
-	defer cancel()
-	for ; next <= cur; next++ {
-		if err := s.video.Write(wctx, frame); err != nil {
-			return fmt.Errorf("write trailing frame: %w", err)
-		}
-	}
-	return nil
-}
 func writeResult(ctx context.Context, err error) error {
 	if stopped(ctx, err) {
 		return nil

M internal/core/service_test.go +13 -10


@@ -404,7 +404,7 @@ func TestRecordExactFrameCountWithSlowStartAndCapture(t *testing.T) {
404	404		}
405	405		}
406	406
407		-	func TestRecordStopWritesElapsedSlots(t *testing.T) {
	407	+	func TestRecordCancelWritesNothingAfterStop(t *testing.T) {
408	408		h := newHarness(t)
409	409		sel := rec(0)
410	410		sel.Video.FPS = 50

@@ -412,19 +412,22 @@ func TestRecordStopWritesElapsedSlots(t *testing.T) {
412	412		start := recordOrder(h, sel)
413	413		h.src.EXPECT().Capture(mock.Anything, target).Return(frame, nil).Maybe()
414	414		var writes atomic.Int32
415		-	h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(context.Context, ports.Frame) error {
416		-	writes.Add(1)
	415	+	var closed atomic.Bool
	416	+	h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(ctx context.Context, _ ports.Frame) error {
	417	+	require.NoError(t, ctx.Err(), "write after stop")
	418	+	if writes.Add(1) == 4 {
	419	+	cancel() // stop arrives while this frame is being written
	420	+	}
417	421		return nil
418	422		}).NotBefore(start).Maybe()
419		-	h.vid.EXPECT().Close().Return(nil).Once()
	423	+	h.vid.EXPECT().Close().RunAndReturn(func() error {
	424	+	closed.Store(true)
	425	+	return nil
	426	+	}).Once()
420	427
421		-	stop := time.AfterFunc(200*time.Millisecond, cancel)
422		-	defer stop.Stop()
423		-	began := time.Now()
424	428		require.NoError(t, h.svc.Run(ctx, sel))
425		-	slots := int32(time.Since(began) * 50 / time.Second)
426		-	assert.GreaterOrEqual(t, writes.Load(), int32(10))
427		-	assert.LessOrEqual(t, slots-writes.Load(), int32(1), "elapsed slots must be written")
	429	+	assert.Equal(t, int32(4), writes.Load(), "recording ends at the last frame written before the stop")
	430	+	assert.True(t, closed.Load())
428	431		}
429	432
430	433		func TestRecordHugeDurationDoesNotOverflow(t *testing.T) {

 	}
 }
-func TestRecordStopWritesElapsedSlots(t *testing.T) {
+func TestRecordCancelWritesNothingAfterStop(t *testing.T) {
 	h := newHarness(t)
 	sel := rec(0)
 	sel.Video.FPS = 50
 	start := recordOrder(h, sel)
 	h.src.EXPECT().Capture(mock.Anything, target).Return(frame, nil).Maybe()
 	var writes atomic.Int32
-	h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(context.Context, ports.Frame) error {
+	var closed atomic.Bool
-		writes.Add(1)
+	h.vid.EXPECT().Write(mock.Anything, frame).RunAndReturn(func(ctx context.Context, _ ports.Frame) error {
+		require.NoError(t, ctx.Err(), "write after stop")
+		if writes.Add(1) == 4 {
+			cancel() // stop arrives while this frame is being written
+		}
 		return nil
 	}).NotBefore(start).Maybe()
-	h.vid.EXPECT().Close().Return(nil).Once()
+	h.vid.EXPECT().Close().RunAndReturn(func() error {
+		closed.Store(true)
+		return nil
+	}).Once()
-	stop := time.AfterFunc(200*time.Millisecond, cancel)
-	defer stop.Stop()
-	began := time.Now()
 	require.NoError(t, h.svc.Run(ctx, sel))
-	slots := int32(time.Since(began) * 50 / time.Second)
+	assert.Equal(t, int32(4), writes.Load(), "recording ends at the last frame written before the stop")
-	assert.GreaterOrEqual(t, writes.Load(), int32(10))
+	assert.True(t, closed.Load())
-	assert.LessOrEqual(t, slots-writes.Load(), int32(1), "elapsed slots must be written")
 }
 func TestRecordHugeDurationDoesNotOverflow(t *testing.T) {