feat(screencast): produce paced PipeWire frames from cast damage

commit: f6fd269473665efc97327725ade18b79a75f2253
author: brice · 8d ago · 2026-10-14 20:29 UTC
committer: brice · 8d ago · 2026-10-14 21:39 UTC
parent: b413735
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +391 -25

M Makefile +1 -1


@@ -69,7 +69,7 @@ adapter-imports:
69	69		[ -z "$$bad" ] \|\| { echo "$$bad"; echo 'adapter imports another adapter: use ports or a shared library (Makefile ADAPTER_LIBS)' >&2; exit 1; }
70	70		# Fast, mandatory guards for tiled content publications and SHM copies.
71	71		perf-check:
72		-	CGO_ENABLED=0 go test ./internal/core ./internal/app ./internal/adapters/drm ./internal/adapters/headless ./internal/adapters/vulkan ./internal/adapters/wayland ./internal/adapters/outputkit/capture -run '^(TestAnimateAllocations\|TestPublishAllocations\|TestPublishSubscribersAllocations\|TestPublishStashSlideAllocations\|TestPublishOverviewAllocations\|TestPublishLeavingAllocations\|TestOutputRoutingAllocations\|TestReportSeenAllocations\|TestFrameLifecycleTransitionsAllocations\|TestFullscreenShownAllocations\|TestFrameDecisionAllocations\|TestCommitFrameAllocations\|TestColorPipelineAllocations\|TestShownBySnapshotAllocations\|TestFlipShowsAllocations\|TestMRUNoteFocusDoesNotAllocateWhenSettled\|TestDueFramesAllocations\|TestRenderSteadyStateAllocations\|TestRenderSteadyStateDMABufAllocations\|TestSceneOpaqueRegionAllocations\|TestSceneDeltaAllocations\|TestAccountFlipAllocations\|TestFlipDoneAllocations\|TestSceneWalkUnchangedTiledSHM\|TestTiledCommitPublishAllocations\|TestCapturedCommitApplyAllocations\|TestCapturedViewportApplyAllocations\|TestEffectiveInputEmptyTreeAllocations\|TestHeadlessTiledSHMCallbackAndCopyBudget\|TestReleaseHeldAllocations\|TestCommitDamageAllocations\|TestEmitContentAllocations\|TestPresentationPushAllocations\|TestAnimatingFrameSendAllocations\|TestContentBatchCycleAllocations\|TestWriteMappedAllocations\|TestCapturedForwardAllocations\|TestCastFrameAllocations)$$' -count=1
	72	+	CGO_ENABLED=0 go test ./internal/core ./internal/app ./internal/adapters/drm ./internal/adapters/headless ./internal/adapters/vulkan ./internal/adapters/wayland ./internal/adapters/outputkit/capture ./internal/adapters/screencast -run '^(TestAnimateAllocations\|TestPublishAllocations\|TestPublishSubscribersAllocations\|TestPublishStashSlideAllocations\|TestPublishOverviewAllocations\|TestPublishLeavingAllocations\|TestOutputRoutingAllocations\|TestReportSeenAllocations\|TestFrameLifecycleTransitionsAllocations\|TestFullscreenShownAllocations\|TestFrameDecisionAllocations\|TestCommitFrameAllocations\|TestColorPipelineAllocations\|TestShownBySnapshotAllocations\|TestFlipShowsAllocations\|TestMRUNoteFocusDoesNotAllocateWhenSettled\|TestDueFramesAllocations\|TestRenderSteadyStateAllocations\|TestRenderSteadyStateDMABufAllocations\|TestSceneOpaqueRegionAllocations\|TestSceneDeltaAllocations\|TestAccountFlipAllocations\|TestFlipDoneAllocations\|TestSceneWalkUnchangedTiledSHM\|TestTiledCommitPublishAllocations\|TestCapturedCommitApplyAllocations\|TestCapturedViewportApplyAllocations\|TestEffectiveInputEmptyTreeAllocations\|TestHeadlessTiledSHMCallbackAndCopyBudget\|TestReleaseHeldAllocations\|TestCommitDamageAllocations\|TestEmitContentAllocations\|TestPresentationPushAllocations\|TestAnimatingFrameSendAllocations\|TestContentBatchCycleAllocations\|TestWriteMappedAllocations\|TestCapturedForwardAllocations\|TestCastFrameAllocations\|TestCastProducerAllocations)$$' -count=1
73	73		check: vet test arch fakes-check log-check perf-check
74	74
75	75		# On-demand SDR/HDR colour check with examples/testpattern on a headless


@@ -69,7 +69,7 @@ adapter-imports:
69		[ -z "$$bad" ] \|\| { echo "$$bad"; echo 'adapter imports another adapter: use ports or a shared library (Makefile ADAPTER_LIBS)' >&2; exit 1; }	69		[ -z "$$bad" ] \|\| { echo "$$bad"; echo 'adapter imports another adapter: use ports or a shared library (Makefile ADAPTER_LIBS)' >&2; exit 1; }
70		# Fast, mandatory guards for tiled content publications and SHM copies.	70		# Fast, mandatory guards for tiled content publications and SHM copies.
71		perf-check:	71		perf-check:
72	-	CGO_ENABLED=0 go test ./internal/core ./internal/app ./internal/adapters/drm ./internal/adapters/headless ./internal/adapters/vulkan ./internal/adapters/wayland ./internal/adapters/outputkit/capture -run '^(TestAnimateAllocations\|TestPublishAllocations\|TestPublishSubscribersAllocations\|TestPublishStashSlideAllocations\|TestPublishOverviewAllocations\|TestPublishLeavingAllocations\|TestOutputRoutingAllocations\|TestReportSeenAllocations\|TestFrameLifecycleTransitionsAllocations\|TestFullscreenShownAllocations\|TestFrameDecisionAllocations\|TestCommitFrameAllocations\|TestColorPipelineAllocations\|TestShownBySnapshotAllocations\|TestFlipShowsAllocations\|TestMRUNoteFocusDoesNotAllocateWhenSettled\|TestDueFramesAllocations\|TestRenderSteadyStateAllocations\|TestRenderSteadyStateDMABufAllocations\|TestSceneOpaqueRegionAllocations\|TestSceneDeltaAllocations\|TestAccountFlipAllocations\|TestFlipDoneAllocations\|TestSceneWalkUnchangedTiledSHM\|TestTiledCommitPublishAllocations\|TestCapturedCommitApplyAllocations\|TestCapturedViewportApplyAllocations\|TestEffectiveInputEmptyTreeAllocations\|TestHeadlessTiledSHMCallbackAndCopyBudget\|TestReleaseHeldAllocations\|TestCommitDamageAllocations\|TestEmitContentAllocations\|TestPresentationPushAllocations\|TestAnimatingFrameSendAllocations\|TestContentBatchCycleAllocations\|TestWriteMappedAllocations\|TestCapturedForwardAllocations\|TestCastFrameAllocations)$$' -count=1	72	+	CGO_ENABLED=0 go test ./internal/core ./internal/app ./internal/adapters/drm ./internal/adapters/headless ./internal/adapters/vulkan ./internal/adapters/wayland ./internal/adapters/outputkit/capture ./internal/adapters/screencast -run '^(TestAnimateAllocations\|TestPublishAllocations\|TestPublishSubscribersAllocations\|TestPublishStashSlideAllocations\|TestPublishOverviewAllocations\|TestPublishLeavingAllocations\|TestOutputRoutingAllocations\|TestReportSeenAllocations\|TestFrameLifecycleTransitionsAllocations\|TestFullscreenShownAllocations\|TestFrameDecisionAllocations\|TestCommitFrameAllocations\|TestColorPipelineAllocations\|TestShownBySnapshotAllocations\|TestFlipShowsAllocations\|TestMRUNoteFocusDoesNotAllocateWhenSettled\|TestDueFramesAllocations\|TestRenderSteadyStateAllocations\|TestRenderSteadyStateDMABufAllocations\|TestSceneOpaqueRegionAllocations\|TestSceneDeltaAllocations\|TestAccountFlipAllocations\|TestFlipDoneAllocations\|TestSceneWalkUnchangedTiledSHM\|TestTiledCommitPublishAllocations\|TestCapturedCommitApplyAllocations\|TestCapturedViewportApplyAllocations\|TestEffectiveInputEmptyTreeAllocations\|TestHeadlessTiledSHMCallbackAndCopyBudget\|TestReleaseHeldAllocations\|TestCommitDamageAllocations\|TestEmitContentAllocations\|TestPresentationPushAllocations\|TestAnimatingFrameSendAllocations\|TestContentBatchCycleAllocations\|TestWriteMappedAllocations\|TestCapturedForwardAllocations\|TestCastFrameAllocations\|TestCastProducerAllocations)$$' -count=1
73		check: vet test arch fakes-check log-check perf-check	73		check: vet test arch fakes-check log-check perf-check
74			74
75		# On-demand SDR/HDR colour check with examples/testpattern on a headless	75		# On-demand SDR/HDR colour check with examples/testpattern on a headless

A internal/adapters/screencast/norace_test.go +5 -0

@@ -0,0 +1,5 @@
	1	+	//go:build !race
	2	+
	3	+	package screencast
	4	+
	5	+	const raceEnabled = false

@@ -0,0 +1,5 @@
	1	+	//go:build !race
	2	+
	3	+	package screencast
	4	+
	5	+	const raceEnabled = false

M internal/adapters/screencast/producer.go +55 -24


@@ -104,15 +104,25 @@ func (s Service) giveBack(st stream) {
104	104		// frameDone settles the frame in flight: a complete image is queued and
105	105		// triggered, anything else gives the buffer back.
106	106		func (s Service) frameDone(st stream, ev *ports.CastEvent) {
	107	+	if b := s.settleFrame(st, ev); b != nil {
	108	+	st.src.Queue(b, st.seq, st.pts, false)
	109	+	st.src.Trigger()
	110	+	}
	111	+	}
	112	+
	113	+	// settleFrame updates the stream for the CastFrameDone of its frame and
	114	+	// returns the buffer to queue, nil when there is none. It touches no
	115	+	// PipeWire object but Release on the failure path.
	116	+	func (s Service) settleFrame(st stream, ev ports.CastEvent) pipewire.VideoBuffer {
107	117		if !st.inflight \|\| st.buf == nil \|\| st.buf.Index != ev.Buf {
108	118		s.log.Warn().Str("stream", st.path).Int("buf", ev.Buf).Msg("frame done for no frame in flight")
109		-	return
	119	+	return nil
110	120		}
111	121		st.inflight = false
112	122		if ev.Err != nil \|\| st.closing {
113	123		s.giveBack(st)
114	124		if ev.Err == nil {
115		-	return
	125	+	return nil
116	126		}
117	127		st.errs++
118	128		st.dirty = true

@@ -122,18 +132,17 @@ func (s Service) frameDone(st stream, ev *ports.CastEvent) {
122	132		s.closeSession(st.sess, true)
123	133		}
124	134		}
125		-	return
	135	+	return nil
126	136		}
127	137		st.errs = 0
128	138		st.seq++
129		-	var pts time.Duration
	139	+	st.pts = 0
130	140		if !ev.Time.IsZero() {
131		-	pts = time.Duration(ev.Time.UnixNano())
	141	+	st.pts = time.Duration(ev.Time.UnixNano())
132	142		}
133	143		b := st.buf
134	144		st.buf = nil
135		-	st.src.Queue(b, st.seq, pts, false)
136		-	st.src.Trigger()
	145	+	return b
137	146		}
138	147
139	148		// pumpAll runs every stream: commands, negotiation and frames.

@@ -234,37 +243,59 @@ func (s Service) openSource(st stream) {
234	243
235	244		// pumpFrame requests the next frame when one is due.
236	245		func (s Service) pumpFrame(st stream, now time.Time) {
237		-	if st.inflight \|\| st.pwState != pipewire.StreamStateStreaming \|\| st.needResize \|\|
238		-	st.fw != st.w \|\| st.fh != st.h \|\| st.fw == 0 {
239		-	return
240		-	}
241		-	if !st.dirty && !st.forced {
242		-	return
243		-	}
244		-	interval := time.Second / time.Duration(max(st.maxFPS, 1))
245		-	if since := now.Sub(st.lastReq); !st.lastReq.IsZero() && since < interval {
246		-	s.wait(interval - since)
	246	+	if !s.frameDue(st, now) {
247	247		return
248	248		}
249	249		b := st.src.Dequeue()
250	250		if b == nil {
251	251		// Every buffer is with the consumer: look again shortly.
252		-	s.wait(interval)
	252	+	s.wait(st.interval())
253	253		return
254	254		}
255		-	if b.Width != st.w \|\| b.Height != st.h \|\| len(b.Data) < b.Stride*b.Height {
	255	+	if !s.issueFrame(st, b, now) {
256	256		st.src.Release(b)
257		-	s.wait(interval)
258		-	return
	257	+	}
	258	+	}
	259	+
	260	+	// interval is the shortest time between two frames.
	261	+	func (st *stream) interval() time.Duration {
	262	+	return time.Second / time.Duration(max(st.maxFPS, 1))
	263	+	}
	264	+
	265	+	// frameDue reports whether a frame may be requested now: streaming, the
	266	+	// negotiated size is the target's, something changed (or it is the first
	267	+	// frame), none in flight and the interval has passed. When only the interval
	268	+	// is missing, it asks for a poll timeout.
	269	+	func (s Service) frameDue(st stream, now time.Time) bool {
	270	+	if st.inflight \|\| st.pwState != pipewire.StreamStateStreaming \|\| st.needResize \|\|
	271	+	st.fw != st.w \|\| st.fh != st.h \|\| st.fw == 0 {
	272	+	return false
	273	+	}
	274	+	if !st.dirty && !st.forced {
	275	+	return false
	276	+	}
	277	+	if since := now.Sub(st.lastReq); !st.lastReq.IsZero() && since < st.interval() {
	278	+	s.wait(st.interval() - since)
	279	+	return false
	280	+	}
	281	+	return true
	282	+	}
	283	+
	284	+	// issueFrame lends buffer b to the wayland server. False: the buffer is not
	285	+	// taken and stays the caller's.
	286	+	func (s Service) issueFrame(st stream, b *pipewire.VideoBuffer, now time.Time) bool {
	287	+	if b.Width != st.w \|\| b.Height != st.h \|\| len(b.Data) < b.Stride*b.Height {
	288	+	s.wait(st.interval())
	289	+	return false
259	290		}
260	291		cmd := ports.CastCommand{Kind: ports.CastFrame, ID: st.cast, Buf: b.Index, Data: b.Data, Stride: b.Stride, Width: b.Width, Height: b.Height, Format: shmFormat(b.Format)}
261	292		if !s.sendCmd(cmd) {
262		-	st.src.Release(b)
263		-	s.wait(interval)
264		-	return
	293	+	s.wait(st.interval())
	294	+	return false
265	295		}
266	296		st.inflight, st.buf = true, b
267	297		st.lastReq, st.dirty, st.forced = now, false, false
	298	+	return true
268	299		}
269	300
270	301		// pumpClosing finishes a stream of an ended session: CastClose, then the

 // frameDone settles the frame in flight: a complete image is queued and
 // triggered, anything else gives the buffer back.
 func (s *Service) frameDone(st *stream, ev *ports.CastEvent) {
+	if b := s.settleFrame(st, ev); b != nil {
+		st.src.Queue(b, st.seq, st.pts, false)
+		st.src.Trigger()
+	}
+}
+// settleFrame updates the stream for the CastFrameDone of its frame and
+// returns the buffer to queue, nil when there is none. It touches no
+// PipeWire object but Release on the failure path.
+func (s *Service) settleFrame(st *stream, ev *ports.CastEvent) *pipewire.VideoBuffer {
 	if !st.inflight || st.buf == nil || st.buf.Index != ev.Buf {
 		s.log.Warn().Str("stream", st.path).Int("buf", ev.Buf).Msg("frame done for no frame in flight")
-		return
+		return nil
 	}
 	st.inflight = false
 	if ev.Err != nil || st.closing {
 		s.giveBack(st)
 		if ev.Err == nil {
-			return
+			return nil
 		}
 		st.errs++
 		st.dirty = true
 				s.closeSession(st.sess, true)
 			}
 		}
-		return
+		return nil
 	}
 	st.errs = 0
 	st.seq++
-	var pts time.Duration
+	st.pts = 0
 	if !ev.Time.IsZero() {
-		pts = time.Duration(ev.Time.UnixNano())
+		st.pts = time.Duration(ev.Time.UnixNano())
 	}
 	b := st.buf
 	st.buf = nil
-	st.src.Queue(b, st.seq, pts, false)
+	return b
-	st.src.Trigger()
 }
 // pumpAll runs every stream: commands, negotiation and frames.
 // pumpFrame requests the next frame when one is due.
 func (s *Service) pumpFrame(st *stream, now time.Time) {
-	if st.inflight || st.pwState != pipewire.StreamStateStreaming || st.needResize ||
+	if !s.frameDue(st, now) {
-		st.fw != st.w || st.fh != st.h || st.fw == 0 {
-		return
-	}
-	if !st.dirty && !st.forced {
-		return
-	}
-	interval := time.Second / time.Duration(max(st.maxFPS, 1))
-	if since := now.Sub(st.lastReq); !st.lastReq.IsZero() && since < interval {
-		s.wait(interval - since)
 		return
 	}
 	b := st.src.Dequeue()
 	if b == nil {
 		// Every buffer is with the consumer: look again shortly.
-		s.wait(interval)
+		s.wait(st.interval())
 		return
 	}
-	if b.Width != st.w || b.Height != st.h || len(b.Data) < b.Stride*b.Height {
+	if !s.issueFrame(st, b, now) {
 		st.src.Release(b)
-		s.wait(interval)
+	}
-		return
+}
+// interval is the shortest time between two frames.
+func (st *stream) interval() time.Duration {
+	return time.Second / time.Duration(max(st.maxFPS, 1))
+}
+// frameDue reports whether a frame may be requested now: streaming, the
+// negotiated size is the target's, something changed (or it is the first
+// frame), none in flight and the interval has passed. When only the interval
+// is missing, it asks for a poll timeout.
+func (s *Service) frameDue(st *stream, now time.Time) bool {
+	if st.inflight || st.pwState != pipewire.StreamStateStreaming || st.needResize ||
+		st.fw != st.w || st.fh != st.h || st.fw == 0 {
+		return false
+	}
+	if !st.dirty && !st.forced {
+		return false
+	}
+	if since := now.Sub(st.lastReq); !st.lastReq.IsZero() && since < st.interval() {
+		s.wait(st.interval() - since)
+		return false
+	}
+	return true
+}
+// issueFrame lends buffer b to the wayland server. False: the buffer is not
+// taken and stays the caller's.
+func (s *Service) issueFrame(st *stream, b *pipewire.VideoBuffer, now time.Time) bool {
+	if b.Width != st.w || b.Height != st.h || len(b.Data) < b.Stride*b.Height {
+		s.wait(st.interval())
+		return false
 	}
 	cmd := ports.CastCommand{Kind: ports.CastFrame, ID: st.cast, Buf: b.Index, Data: b.Data, Stride: b.Stride, Width: b.Width, Height: b.Height, Format: shmFormat(b.Format)}
 	if !s.sendCmd(cmd) {
-		st.src.Release(b)
+		s.wait(st.interval())
-		s.wait(interval)
+		return false
-		return
 	}
 	st.inflight, st.buf = true, b
 	st.lastReq, st.dirty, st.forced = now, false, false
+	return true
 }
 // pumpClosing finishes a stream of an ended session: CastClose, then the

A internal/adapters/screencast/producer_test.go +324 -0

@@ -0,0 +1,324 @@
	1	+	package screencast
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"testing"
	6	+	"time"
	7	+
	8	+	"github.com/bnema/neferwl/internal/adapters/captureallow"
	9	+	"github.com/bnema/neferwl/internal/ports"
	10	+	"github.com/bnema/purego-pipewire/pipewire"
	11	+	"github.com/bnema/zerobus"
	12	+	"github.com/stretchr/testify/mock"
	13	+	)
	14	+
	15	+	// producer is a service whose owner state is driven directly by the test, on
	16	+	// the test goroutine (the owner): a generated mock bus, PipeWire core and
	17	+	// video source stand where the real ones would be.
	18	+	type producer struct {
	19	+	t *testing.T
	20	+	s *Service
	21	+	st *stream
	22	+	src *mockvideoSource
	23	+	core *mockvideoCore
	24	+	bus *mockbus
	25	+	now time.Time
	26	+	buf *pipewire.VideoBuffer
	27	+	cmds chan ports.CastCommand
	28	+	}
	29	+
	30	+	func newProducer(t testing.T) producer {
	31	+	t.Helper()
	32	+	cmds := make(chan ports.CastCommand, 16)
	33	+	p := &producer{t: t, now: time.Unix(1000, 0), cmds: cmds}
	34	+	p.s = New(func() *captureallow.Policy { return captureallow.Disabled() }, cmds, testLog(&logBuf{}))
	35	+	t.Cleanup(func() { _ = p.s.poller.Close() })
	36	+	p.bus = newMockbus(t)
	37	+	p.bus.EXPECT().NewSignalTo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
	38	+	func(string, string, string, string, string) *zerobus.Encoder { return &zerobus.Encoder{} }).Maybe()
	39	+	p.bus.EXPECT().Send().Return(1, nil).Maybe()
	40	+	p.s.conn = p.bus
	41	+	p.src = newMockvideoSource(t)
	42	+	p.core = newMockvideoCore(t)
	43	+	p.s.pwc = p.core
	44	+	p.buf = &pipewire.VideoBuffer{Index: 3, Data: make([]byte, 64324), Stride: 64 * 4, Width: 64, Height: 32, Format: pipewire.VideoFormatBGRx}
	45	+
	46	+	sess := &p.s.sessions[0]
	47	+	*sess = session{used: true, id: 1, owner: ":1.7", path: prefixSession + "1", started: true}
	48	+	st := &sess.streams[0]
	49	+	*st = stream{used: true, sess: sess, id: 1, cast: 1, path: prefixStream + "1", output: "HEADLESS-1", maxFPS: 60, pwState: -2, needOpen: true}
	50	+	st.h2 = streamHandler{st: st}
	51	+	p.st = st
	52	+	return p
	53	+	}
	54	+
	55	+	func (p *producer) cmd() ports.CastCommand {
	56	+	p.t.Helper()
	57	+	select {
	58	+	case c := <-p.cmds:
	59	+	return c
	60	+	default:
	61	+	p.t.Fatal("no command")
	62	+	return ports.CastCommand{}
	63	+	}
	64	+	}
	65	+
	66	+	func (p *producer) none() {
	67	+	p.t.Helper()
	68	+	select {
	69	+	case c := <-p.cmds:
	70	+	p.t.Fatalf("unexpected command %+v", c)
	71	+	default:
	72	+	}
	73	+	}
	74	+
	75	+	// streaming opens the cast, sizes it, creates the source and brings it to
	76	+	// STREAMING with the negotiated format.
	77	+	func (p *producer) streaming() {
	78	+	p.t.Helper()
	79	+	p.s.pump(p.st, p.now)
	80	+	if c := p.cmd(); c.Kind != ports.CastOpen \|\| c.ID != 1 \|\| c.Output != "HEADLESS-1" {
	81	+	p.t.Fatalf("open %+v", c)
	82	+	}
	83	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastSize, ID: 1, Width: 64, Height: 32})
	84	+	p.core.EXPECT().NewVideoSource(mock.Anything, mock.Anything).RunAndReturn(
	85	+	func(cfg pipewire.VideoConfig, h pipewire.VideoHandler) (videoSource, error) {
	86	+	if cfg.Width != 64 \|\| cfg.Height != 32 \|\| cfg.MaxFPS != 60 {
	87	+	p.t.Errorf("config %+v", cfg)
	88	+	}
	89	+	return p.src, nil
	90	+	}).Once()
	91	+	p.s.pump(p.st, p.now)
	92	+	p.st.h2.StateChanged(pipewire.StreamStatePaused, 42, "")
	93	+	p.st.h2.FormatChanged(pipewire.VideoFormatBGRx, 64, 32, 256)
	94	+	p.st.h2.StateChanged(pipewire.StreamStateStreaming, 42, "")
	95	+	}
	96	+
	97	+	// The node id goes to the session owner alone, once, as the stream's signal.
	98	+	func TestAnnouncesNodeToOwnerOnce(t *testing.T) {
	99	+	p := newProducer(t)
	100	+	var dest, path, member string
	101	+	p.bus.ExpectedCalls = nil
	102	+	p.bus.EXPECT().NewSignalTo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
	103	+	func(d, pa, iface, m, sig string) *zerobus.Encoder {
	104	+	dest, path, member = d, pa, m
	105	+	if iface != ifaceStream \|\| sig != "u" {
	106	+	t.Errorf("signal %s %s", iface, sig)
	107	+	}
	108	+	return &zerobus.Encoder{}
	109	+	}).Once()
	110	+	p.bus.EXPECT().Send().Return(1, nil).Once()
	111	+	p.streaming()
	112	+	p.src.EXPECT().Dequeue().Return(nil).Maybe()
	113	+	p.s.pump(p.st, p.now)
	114	+	p.s.pump(p.st, p.now)
	115	+	if dest != ":1.7" \|\| path != prefixStream+"1" \|\| member != "PipeWireStreamAdded" {
	116	+	t.Fatalf("signal to %q %q %q", dest, path, member)
	117	+	}
	118	+	}
	119	+
	120	+	// Nothing is requested without damage, except the first frame.
	121	+	func TestPacingForcesFirstFrameThenWaitsForDamage(t *testing.T) {
	122	+	p := newProducer(t)
	123	+	p.streaming()
	124	+	p.src.EXPECT().Dequeue().Return(p.buf).Once()
	125	+	p.s.pump(p.st, p.now)
	126	+	c := p.cmd()
	127	+	if c.Kind != ports.CastFrame \|\| c.Buf != 3 \|\| c.Stride != 256 \|\| c.Width != 64 \|\| c.Height != 32 \|\| c.Format != shmXRGB8888 \|\| len(c.Data) != 64324 \|\| &c.Data[0] != &p.buf.Data[0] {
	128	+	t.Fatalf("frame %+v", c)
	129	+	}
	130	+	// Done: queued, then triggered, in that order.
	131	+	var order []string
	132	+	p.src.EXPECT().Queue(p.buf, uint64(1), mock.Anything, false).Run(func(*pipewire.VideoBuffer, uint64, time.Duration, bool) { order = append(order, "queue") }).Once()
	133	+	p.src.EXPECT().Trigger().Run(func() { order = append(order, "trigger") }).Once()
	134	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Time: time.Unix(5, 0)})
	135	+	if len(order) != 2 \|\| order[0] != "queue" \|\| order[1] != "trigger" {
	136	+	t.Fatalf("order %v", order)
	137	+	}
	138	+	// No damage: no request however much time passes (Dequeue is not expected).
	139	+	p.now = p.now.Add(time.Second)
	140	+	p.s.pump(p.st, p.now)
	141	+	p.none()
	142	+	}
	143	+
	144	+	// At most one frame per interval, damage or not; the poll timeout is the
	145	+	// time left.
	146	+	func TestPacingOnePerInterval(t *testing.T) {
	147	+	p := newProducer(t)
	148	+	p.streaming()
	149	+	p.src.EXPECT().Dequeue().Return(p.buf).Twice()
	150	+	p.s.pump(p.st, p.now)
	151	+	p.cmd()
	152	+	p.src.EXPECT().Queue(mock.Anything, mock.Anything, mock.Anything, false).Return().Once()
	153	+	p.src.EXPECT().Trigger().Return().Once()
	154	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
	155	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastDamage, ID: 1})
	156	+	p.now = p.now.Add(5 * time.Millisecond)
	157	+	p.s.pumpAll(p.now)
	158	+	p.none()
	159	+	if got := p.s.timeout(); got < 11 \|\| got > 12 {
	160	+	t.Fatalf("timeout %d ms, want the ~11.7 ms left of 16.7", got)
	161	+	}
	162	+	p.now = p.now.Add(12 * time.Millisecond)
	163	+	p.s.pumpAll(p.now)
	164	+	if c := p.cmd(); c.Kind != ports.CastFrame {
	165	+	t.Fatalf("%+v", c)
	166	+	}
	167	+	if got := p.s.timeout(); got != -1 {
	168	+	t.Fatalf("idle timeout %d", got)
	169	+	}
	170	+	}
	171	+
	172	+	// A new size waits for the frame in flight, then renegotiates.
	173	+	func TestResizeWaitsForInflightFrame(t *testing.T) {
	174	+	p := newProducer(t)
	175	+	p.streaming()
	176	+	p.src.EXPECT().Dequeue().Return(p.buf).Once()
	177	+	p.s.pump(p.st, p.now)
	178	+	p.cmd()
	179	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastSize, ID: 1, Width: 32, Height: 16})
	180	+	p.s.pump(p.st, p.now) // Resize is not expected while the frame is in flight
	181	+	p.src.EXPECT().Resize(32, 16).Return(nil).Once()
	182	+	p.src.EXPECT().Release(p.buf).Return().Once()
	183	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Err: ports.ErrCaptureTransient})
	184	+	p.s.pump(p.st, p.now)
	185	+	// No frame until the new format is negotiated.
	186	+	p.now = p.now.Add(time.Second)
	187	+	p.s.pump(p.st, p.now)
	188	+	p.none()
	189	+	}
	190	+
	191	+	// A failed frame gives the buffer back; three in a row stop the stream.
	192	+	func TestFrameErrorsReleaseAndStop(t *testing.T) {
	193	+	p := newProducer(t)
	194	+	p.streaming()
	195	+	p.src.EXPECT().Dequeue().Return(p.buf)
	196	+	p.src.EXPECT().Release(p.buf).Return()
	197	+	for i := range maxFrameErrors {
	198	+	p.s.pump(p.st, p.now)
	199	+	if c := p.cmd(); c.Kind != ports.CastFrame {
	200	+	t.Fatalf("frame %d: %+v", i, c)
	201	+	}
	202	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Err: errTest})
	203	+	p.now = p.now.Add(time.Second)
	204	+	}
	205	+	if !p.st.closing \|\| !p.s.sessions[0].dying {
	206	+	t.Fatal("stream not stopped after three errors")
	207	+	}
	208	+	p.src.EXPECT().Destroy().Return().Once()
	209	+	p.s.pump(p.st, p.now)
	210	+	if c := p.cmd(); c.Kind != ports.CastClose \|\| c.ID != 1 {
	211	+	t.Fatalf("%+v", c)
	212	+	}
	213	+	if p.st.used {
	214	+	t.Fatal("stream slot not freed")
	215	+	}
	216	+	}
	217	+
	218	+	// Ending the session with a frame in flight closes the cast at once but
	219	+	// keeps the PipeWire node until the frame's CastFrameDone.
	220	+	func TestTeardownWaitsForFrameDone(t *testing.T) {
	221	+	p := newProducer(t)
	222	+	p.streaming()
	223	+	p.src.EXPECT().Dequeue().Return(p.buf).Once()
	224	+	p.s.pump(p.st, p.now)
	225	+	p.cmd()
	226	+	p.s.closeSession(p.st.sess, true)
	227	+	p.s.pump(p.st, p.now)
	228	+	if c := p.cmd(); c.Kind != ports.CastClose {
	229	+	t.Fatalf("%+v", c)
	230	+	}
	231	+	p.s.pump(p.st, p.now)
	232	+	p.s.reap()
	233	+	if !p.st.used \|\| !p.s.sessions[0].used {
	234	+	t.Fatal("released while a frame is in flight") // Destroy is not expected yet
	235	+	}
	236	+	p.src.EXPECT().Release(p.buf).Return().Once()
	237	+	p.src.EXPECT().Destroy().Return().Once()
	238	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
	239	+	p.s.pump(p.st, p.now)
	240	+	p.s.reap()
	241	+	if p.st.used \|\| p.s.sessions[0].used {
	242	+	t.Fatal("stream and session not freed after the frame")
	243	+	}
	244	+	}
	245	+
	246	+	// A cast that wayland ended closes the session and signals Closed.
	247	+	func TestCastStoppedClosesSession(t *testing.T) {
	248	+	p := newProducer(t)
	249	+	p.streaming()
	250	+	p.bus.ExpectedCalls = nil
	251	+	var member string
	252	+	p.bus.EXPECT().NewSignalTo(":1.7", prefixSession+"1", ifaceSession, "Closed", "").RunAndReturn(
	253	+	func(_, _, _, m, _ string) *zerobus.Encoder {
	254	+	member = m
	255	+	return &zerobus.Encoder{}
	256	+	}).Once()
	257	+	p.bus.EXPECT().Send().Return(1, nil).Once()
	258	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastStopped, ID: 1, Reason: ports.CaptureReasonWindowGone})
	259	+	if member != "Closed" {
	260	+	t.Fatal("Closed not signalled")
	261	+	}
	262	+	p.src.EXPECT().Destroy().Return().Once()
	263	+	p.s.pump(p.st, p.now)
	264	+	// Wayland ended it itself: no CastClose.
	265	+	p.none()
	266	+	}
	267	+
	268	+	// A PipeWire stream that went to ERROR ends the session.
	269	+	func TestPipeWireErrorEndsSession(t *testing.T) {
	270	+	p := newProducer(t)
	271	+	p.streaming()
	272	+	p.st.h2.StateChanged(pipewire.StreamStateError, 42, "gone")
	273	+	p.s.pump(p.st, p.now)
	274	+	if !p.s.sessions[0].dying {
	275	+	t.Fatal("session still open")
	276	+	}
	277	+	}
	278	+
	279	+	var errTest = errors.New("frame failed")
	280	+
	281	+	// The steady damage -> frame -> done cycle allocates nothing. Measured with
	282	+	// real values: the sink, the pacing and the settle logic. The PipeWire calls
	283	+	// of the cycle (Dequeue, Queue, Trigger) are the VideoSource's own,
	284	+	// measured in purego-pipewire (TestVideoSourceFrameAllocations); a mock in
	285	+	// the closure would record every call and grow.
	286	+	func TestCastProducerAllocations(t *testing.T) {
	287	+	if raceEnabled {
	288	+	t.Skip("the race detector makes the wake callback allocate")
	289	+	}
	290	+	p := newProducer(t)
	291	+	p.cmds = make(chan ports.CastCommand, 4)
	292	+	p.s.cmds = p.cmds
	293	+	st := p.st
	294	+	st.pwState, st.w, st.h, st.fw, st.fh, st.haveSize = pipewire.StreamStateStreaming, 64, 32, 64, 32, true
	295	+	cmds := p.cmds
	296	+	b := p.buf
	297	+	damage := ports.CastEvent{Kind: ports.CastDamage, ID: 1}
	298	+	done := ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: b.Index, Time: time.Unix(5, 0)}
	299	+	now := p.now
	300	+	cycle := func() {
	301	+	p.s.Push(damage)
	302	+	evs, _, _, _ := p.s.take()
	303	+	for i := range evs {
	304	+	p.s.onCastEvent(&evs[i])
	305	+	}
	306	+	now = now.Add(time.Second)
	307	+	if !p.s.frameDue(st, now) \|\| !p.s.issueFrame(st, b, now) {
	308	+	t.Fatal("no frame")
	309	+	}
	310	+	<-cmds
	311	+	p.s.Push(done)
	312	+	evs, _, _, _ = p.s.take()
	313	+	if p.s.settleFrame(st, &evs[0]) != b {
	314	+	t.Fatal("frame not settled")
	315	+	}
	316	+	st.buf = nil
	317	+	}
	318	+	for range 8 {
	319	+	cycle()
	320	+	}
	321	+	if n := testing.AllocsPerRun(200, cycle); n != 0 {
	322	+	t.Fatalf("%v allocations per frame, want 0", n)
	323	+	}
	324	+	}

@@ -0,0 +1,324 @@
	1	+	package screencast
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"testing"
	6	+	"time"
	7	+
	8	+	"github.com/bnema/neferwl/internal/adapters/captureallow"
	9	+	"github.com/bnema/neferwl/internal/ports"
	10	+	"github.com/bnema/purego-pipewire/pipewire"
	11	+	"github.com/bnema/zerobus"
	12	+	"github.com/stretchr/testify/mock"
	13	+	)
	14	+
	15	+	// producer is a service whose owner state is driven directly by the test, on
	16	+	// the test goroutine (the owner): a generated mock bus, PipeWire core and
	17	+	// video source stand where the real ones would be.
	18	+	type producer struct {
	19	+	t *testing.T
	20	+	s *Service
	21	+	st *stream
	22	+	src *mockvideoSource
	23	+	core *mockvideoCore
	24	+	bus *mockbus
	25	+	now time.Time
	26	+	buf *pipewire.VideoBuffer
	27	+	cmds chan ports.CastCommand
	28	+	}
	29	+
	30	+	func newProducer(t testing.T) producer {
	31	+	t.Helper()
	32	+	cmds := make(chan ports.CastCommand, 16)
	33	+	p := &producer{t: t, now: time.Unix(1000, 0), cmds: cmds}
	34	+	p.s = New(func() *captureallow.Policy { return captureallow.Disabled() }, cmds, testLog(&logBuf{}))
	35	+	t.Cleanup(func() { _ = p.s.poller.Close() })
	36	+	p.bus = newMockbus(t)
	37	+	p.bus.EXPECT().NewSignalTo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
	38	+	func(string, string, string, string, string) *zerobus.Encoder { return &zerobus.Encoder{} }).Maybe()
	39	+	p.bus.EXPECT().Send().Return(1, nil).Maybe()
	40	+	p.s.conn = p.bus
	41	+	p.src = newMockvideoSource(t)
	42	+	p.core = newMockvideoCore(t)
	43	+	p.s.pwc = p.core
	44	+	p.buf = &pipewire.VideoBuffer{Index: 3, Data: make([]byte, 64324), Stride: 64 * 4, Width: 64, Height: 32, Format: pipewire.VideoFormatBGRx}
	45	+
	46	+	sess := &p.s.sessions[0]
	47	+	*sess = session{used: true, id: 1, owner: ":1.7", path: prefixSession + "1", started: true}
	48	+	st := &sess.streams[0]
	49	+	*st = stream{used: true, sess: sess, id: 1, cast: 1, path: prefixStream + "1", output: "HEADLESS-1", maxFPS: 60, pwState: -2, needOpen: true}
	50	+	st.h2 = streamHandler{st: st}
	51	+	p.st = st
	52	+	return p
	53	+	}
	54	+
	55	+	func (p *producer) cmd() ports.CastCommand {
	56	+	p.t.Helper()
	57	+	select {
	58	+	case c := <-p.cmds:
	59	+	return c
	60	+	default:
	61	+	p.t.Fatal("no command")
	62	+	return ports.CastCommand{}
	63	+	}
	64	+	}
	65	+
	66	+	func (p *producer) none() {
	67	+	p.t.Helper()
	68	+	select {
	69	+	case c := <-p.cmds:
	70	+	p.t.Fatalf("unexpected command %+v", c)
	71	+	default:
	72	+	}
	73	+	}
	74	+
	75	+	// streaming opens the cast, sizes it, creates the source and brings it to
	76	+	// STREAMING with the negotiated format.
	77	+	func (p *producer) streaming() {
	78	+	p.t.Helper()
	79	+	p.s.pump(p.st, p.now)
	80	+	if c := p.cmd(); c.Kind != ports.CastOpen \|\| c.ID != 1 \|\| c.Output != "HEADLESS-1" {
	81	+	p.t.Fatalf("open %+v", c)
	82	+	}
	83	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastSize, ID: 1, Width: 64, Height: 32})
	84	+	p.core.EXPECT().NewVideoSource(mock.Anything, mock.Anything).RunAndReturn(
	85	+	func(cfg pipewire.VideoConfig, h pipewire.VideoHandler) (videoSource, error) {
	86	+	if cfg.Width != 64 \|\| cfg.Height != 32 \|\| cfg.MaxFPS != 60 {
	87	+	p.t.Errorf("config %+v", cfg)
	88	+	}
	89	+	return p.src, nil
	90	+	}).Once()
	91	+	p.s.pump(p.st, p.now)
	92	+	p.st.h2.StateChanged(pipewire.StreamStatePaused, 42, "")
	93	+	p.st.h2.FormatChanged(pipewire.VideoFormatBGRx, 64, 32, 256)
	94	+	p.st.h2.StateChanged(pipewire.StreamStateStreaming, 42, "")
	95	+	}
	96	+
	97	+	// The node id goes to the session owner alone, once, as the stream's signal.
	98	+	func TestAnnouncesNodeToOwnerOnce(t *testing.T) {
	99	+	p := newProducer(t)
	100	+	var dest, path, member string
	101	+	p.bus.ExpectedCalls = nil
	102	+	p.bus.EXPECT().NewSignalTo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
	103	+	func(d, pa, iface, m, sig string) *zerobus.Encoder {
	104	+	dest, path, member = d, pa, m
	105	+	if iface != ifaceStream \|\| sig != "u" {
	106	+	t.Errorf("signal %s %s", iface, sig)
	107	+	}
	108	+	return &zerobus.Encoder{}
	109	+	}).Once()
	110	+	p.bus.EXPECT().Send().Return(1, nil).Once()
	111	+	p.streaming()
	112	+	p.src.EXPECT().Dequeue().Return(nil).Maybe()
	113	+	p.s.pump(p.st, p.now)
	114	+	p.s.pump(p.st, p.now)
	115	+	if dest != ":1.7" \|\| path != prefixStream+"1" \|\| member != "PipeWireStreamAdded" {
	116	+	t.Fatalf("signal to %q %q %q", dest, path, member)
	117	+	}
	118	+	}
	119	+
	120	+	// Nothing is requested without damage, except the first frame.
	121	+	func TestPacingForcesFirstFrameThenWaitsForDamage(t *testing.T) {
	122	+	p := newProducer(t)
	123	+	p.streaming()
	124	+	p.src.EXPECT().Dequeue().Return(p.buf).Once()
	125	+	p.s.pump(p.st, p.now)
	126	+	c := p.cmd()
	127	+	if c.Kind != ports.CastFrame \|\| c.Buf != 3 \|\| c.Stride != 256 \|\| c.Width != 64 \|\| c.Height != 32 \|\| c.Format != shmXRGB8888 \|\| len(c.Data) != 64324 \|\| &c.Data[0] != &p.buf.Data[0] {
	128	+	t.Fatalf("frame %+v", c)
	129	+	}
	130	+	// Done: queued, then triggered, in that order.
	131	+	var order []string
	132	+	p.src.EXPECT().Queue(p.buf, uint64(1), mock.Anything, false).Run(func(*pipewire.VideoBuffer, uint64, time.Duration, bool) { order = append(order, "queue") }).Once()
	133	+	p.src.EXPECT().Trigger().Run(func() { order = append(order, "trigger") }).Once()
	134	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Time: time.Unix(5, 0)})
	135	+	if len(order) != 2 \|\| order[0] != "queue" \|\| order[1] != "trigger" {
	136	+	t.Fatalf("order %v", order)
	137	+	}
	138	+	// No damage: no request however much time passes (Dequeue is not expected).
	139	+	p.now = p.now.Add(time.Second)
	140	+	p.s.pump(p.st, p.now)
	141	+	p.none()
	142	+	}
	143	+
	144	+	// At most one frame per interval, damage or not; the poll timeout is the
	145	+	// time left.
	146	+	func TestPacingOnePerInterval(t *testing.T) {
	147	+	p := newProducer(t)
	148	+	p.streaming()
	149	+	p.src.EXPECT().Dequeue().Return(p.buf).Twice()
	150	+	p.s.pump(p.st, p.now)
	151	+	p.cmd()
	152	+	p.src.EXPECT().Queue(mock.Anything, mock.Anything, mock.Anything, false).Return().Once()
	153	+	p.src.EXPECT().Trigger().Return().Once()
	154	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
	155	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastDamage, ID: 1})
	156	+	p.now = p.now.Add(5 * time.Millisecond)
	157	+	p.s.pumpAll(p.now)
	158	+	p.none()
	159	+	if got := p.s.timeout(); got < 11 \|\| got > 12 {
	160	+	t.Fatalf("timeout %d ms, want the ~11.7 ms left of 16.7", got)
	161	+	}
	162	+	p.now = p.now.Add(12 * time.Millisecond)
	163	+	p.s.pumpAll(p.now)
	164	+	if c := p.cmd(); c.Kind != ports.CastFrame {
	165	+	t.Fatalf("%+v", c)
	166	+	}
	167	+	if got := p.s.timeout(); got != -1 {
	168	+	t.Fatalf("idle timeout %d", got)
	169	+	}
	170	+	}
	171	+
	172	+	// A new size waits for the frame in flight, then renegotiates.
	173	+	func TestResizeWaitsForInflightFrame(t *testing.T) {
	174	+	p := newProducer(t)
	175	+	p.streaming()
	176	+	p.src.EXPECT().Dequeue().Return(p.buf).Once()
	177	+	p.s.pump(p.st, p.now)
	178	+	p.cmd()
	179	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastSize, ID: 1, Width: 32, Height: 16})
	180	+	p.s.pump(p.st, p.now) // Resize is not expected while the frame is in flight
	181	+	p.src.EXPECT().Resize(32, 16).Return(nil).Once()
	182	+	p.src.EXPECT().Release(p.buf).Return().Once()
	183	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Err: ports.ErrCaptureTransient})
	184	+	p.s.pump(p.st, p.now)
	185	+	// No frame until the new format is negotiated.
	186	+	p.now = p.now.Add(time.Second)
	187	+	p.s.pump(p.st, p.now)
	188	+	p.none()
	189	+	}
	190	+
	191	+	// A failed frame gives the buffer back; three in a row stop the stream.
	192	+	func TestFrameErrorsReleaseAndStop(t *testing.T) {
	193	+	p := newProducer(t)
	194	+	p.streaming()
	195	+	p.src.EXPECT().Dequeue().Return(p.buf)
	196	+	p.src.EXPECT().Release(p.buf).Return()
	197	+	for i := range maxFrameErrors {
	198	+	p.s.pump(p.st, p.now)
	199	+	if c := p.cmd(); c.Kind != ports.CastFrame {
	200	+	t.Fatalf("frame %d: %+v", i, c)
	201	+	}
	202	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Err: errTest})
	203	+	p.now = p.now.Add(time.Second)
	204	+	}
	205	+	if !p.st.closing \|\| !p.s.sessions[0].dying {
	206	+	t.Fatal("stream not stopped after three errors")
	207	+	}
	208	+	p.src.EXPECT().Destroy().Return().Once()
	209	+	p.s.pump(p.st, p.now)
	210	+	if c := p.cmd(); c.Kind != ports.CastClose \|\| c.ID != 1 {
	211	+	t.Fatalf("%+v", c)
	212	+	}
	213	+	if p.st.used {
	214	+	t.Fatal("stream slot not freed")
	215	+	}
	216	+	}
	217	+
	218	+	// Ending the session with a frame in flight closes the cast at once but
	219	+	// keeps the PipeWire node until the frame's CastFrameDone.
	220	+	func TestTeardownWaitsForFrameDone(t *testing.T) {
	221	+	p := newProducer(t)
	222	+	p.streaming()
	223	+	p.src.EXPECT().Dequeue().Return(p.buf).Once()
	224	+	p.s.pump(p.st, p.now)
	225	+	p.cmd()
	226	+	p.s.closeSession(p.st.sess, true)
	227	+	p.s.pump(p.st, p.now)
	228	+	if c := p.cmd(); c.Kind != ports.CastClose {
	229	+	t.Fatalf("%+v", c)
	230	+	}
	231	+	p.s.pump(p.st, p.now)
	232	+	p.s.reap()
	233	+	if !p.st.used \|\| !p.s.sessions[0].used {
	234	+	t.Fatal("released while a frame is in flight") // Destroy is not expected yet
	235	+	}
	236	+	p.src.EXPECT().Release(p.buf).Return().Once()
	237	+	p.src.EXPECT().Destroy().Return().Once()
	238	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
	239	+	p.s.pump(p.st, p.now)
	240	+	p.s.reap()
	241	+	if p.st.used \|\| p.s.sessions[0].used {
	242	+	t.Fatal("stream and session not freed after the frame")
	243	+	}
	244	+	}
	245	+
	246	+	// A cast that wayland ended closes the session and signals Closed.
	247	+	func TestCastStoppedClosesSession(t *testing.T) {
	248	+	p := newProducer(t)
	249	+	p.streaming()
	250	+	p.bus.ExpectedCalls = nil
	251	+	var member string
	252	+	p.bus.EXPECT().NewSignalTo(":1.7", prefixSession+"1", ifaceSession, "Closed", "").RunAndReturn(
	253	+	func(_, _, _, m, _ string) *zerobus.Encoder {
	254	+	member = m
	255	+	return &zerobus.Encoder{}
	256	+	}).Once()
	257	+	p.bus.EXPECT().Send().Return(1, nil).Once()
	258	+	p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastStopped, ID: 1, Reason: ports.CaptureReasonWindowGone})
	259	+	if member != "Closed" {
	260	+	t.Fatal("Closed not signalled")
	261	+	}
	262	+	p.src.EXPECT().Destroy().Return().Once()
	263	+	p.s.pump(p.st, p.now)
	264	+	// Wayland ended it itself: no CastClose.
	265	+	p.none()
	266	+	}
	267	+
	268	+	// A PipeWire stream that went to ERROR ends the session.
	269	+	func TestPipeWireErrorEndsSession(t *testing.T) {
	270	+	p := newProducer(t)
	271	+	p.streaming()
	272	+	p.st.h2.StateChanged(pipewire.StreamStateError, 42, "gone")
	273	+	p.s.pump(p.st, p.now)
	274	+	if !p.s.sessions[0].dying {
	275	+	t.Fatal("session still open")
	276	+	}
	277	+	}
	278	+
	279	+	var errTest = errors.New("frame failed")
	280	+
	281	+	// The steady damage -> frame -> done cycle allocates nothing. Measured with
	282	+	// real values: the sink, the pacing and the settle logic. The PipeWire calls
	283	+	// of the cycle (Dequeue, Queue, Trigger) are the VideoSource's own,
	284	+	// measured in purego-pipewire (TestVideoSourceFrameAllocations); a mock in
	285	+	// the closure would record every call and grow.
	286	+	func TestCastProducerAllocations(t *testing.T) {
	287	+	if raceEnabled {
	288	+	t.Skip("the race detector makes the wake callback allocate")
	289	+	}
	290	+	p := newProducer(t)
	291	+	p.cmds = make(chan ports.CastCommand, 4)
	292	+	p.s.cmds = p.cmds
	293	+	st := p.st
	294	+	st.pwState, st.w, st.h, st.fw, st.fh, st.haveSize = pipewire.StreamStateStreaming, 64, 32, 64, 32, true
	295	+	cmds := p.cmds
	296	+	b := p.buf
	297	+	damage := ports.CastEvent{Kind: ports.CastDamage, ID: 1}
	298	+	done := ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: b.Index, Time: time.Unix(5, 0)}
	299	+	now := p.now
	300	+	cycle := func() {
	301	+	p.s.Push(damage)
	302	+	evs, _, _, _ := p.s.take()
	303	+	for i := range evs {
	304	+	p.s.onCastEvent(&evs[i])
	305	+	}
	306	+	now = now.Add(time.Second)
	307	+	if !p.s.frameDue(st, now) \|\| !p.s.issueFrame(st, b, now) {
	308	+	t.Fatal("no frame")
	309	+	}
	310	+	<-cmds
	311	+	p.s.Push(done)
	312	+	evs, _, _, _ = p.s.take()
	313	+	if p.s.settleFrame(st, &evs[0]) != b {
	314	+	t.Fatal("frame not settled")
	315	+	}
	316	+	st.buf = nil
	317	+	}
	318	+	for range 8 {
	319	+	cycle()
	320	+	}
	321	+	if n := testing.AllocsPerRun(200, cycle); n != 0 {
	322	+	t.Fatalf("%v allocations per frame, want 0", n)
	323	+	}
	324	+	}

A internal/adapters/screencast/race_test.go +5 -0

@@ -0,0 +1,5 @@
	1	+	//go:build race
	2	+
	3	+	package screencast
	4	+
	5	+	const raceEnabled = true

@@ -0,0 +1,5 @@
	1	+	//go:build race
	2	+
	3	+	package screencast
	4	+
	5	+	const raceEnabled = true

M internal/adapters/screencast/stream.go +1 -0


@@ -53,6 +53,7 @@ type stream struct {
53	53		buf *pipewire.VideoBuffer
54	54		lastReq time.Time
55	55		seq uint64
	56	+	pts time.Duration
56	57		errs int // consecutive failed frames
57	58		}
58	59


@@ -53,6 +53,7 @@ type stream struct {
53	buf *pipewire.VideoBuffer	53		buf *pipewire.VideoBuffer
54	lastReq time.Time	54		lastReq time.Time
55	seq uint64	55		seq uint64
		56	+	pts time.Duration
56	errs int // consecutive failed frames	57		errs int // consecutive failed frames
57	}	58		}
58		59