0516c4d fix(screencast): wait without a deadline for frames in flight when the bus is lost · neferwl

M internal/adapters/screencast/harness_test.go +9 -2


@@ -112,14 +112,17 @@ type svcHarness struct {
112	112		log *logBuf
113	113		done chan struct{}
114	114
115		-	killBus func() // loses the bus: the daemon dies
116		-	closed chan struct{} // a PipeWire connection was closed
	115	+	killBus func() // loses the bus: the daemon dies
	116	+	cancel context.CancelFunc // stops the service, as the process exit does
	117	+	closed chan struct{} // a PipeWire connection was closed
117	118		}
118	119
119	120		type harnessOpts struct {
120	121		allow *captureallow.Policy
121	122		noPW bool
122	123		holdOp bool
	124	+	// drain overrides how long the stopping service waits for frames.
	125	+	drain time.Duration
123	126		// iterate runs inside every PipeWire loop iteration, on the owner.
124	127		iterate func()
125	128		}

@@ -138,6 +141,9 @@ func newSvcHarness(t testing.T, o harnessOpts) svcHarness {
138	141		}
139	142		h.svc = New(func() *captureallow.Policy { return allow }, h.cmds, testLog(h.log))
140	143		h.svc.address = h.addr
	144	+	if o.drain > 0 {
	145	+	h.svc.drain = o.drain
	146	+	}
141	147
142	148		res := newMockpeerResolver(t)
143	149		res.EXPECT().Open(mock.Anything).RunAndReturn(func(pid int) (*os.File, error) {

@@ -201,6 +207,7 @@ func newSvcHarness(t testing.T, o harnessOpts) svcHarness {
201	207		h.svc.pw = pwOpen
202	208
203	209		ctx, cancel := context.WithCancel(context.Background())
	210	+	h.cancel = cancel
204	211		go func() { defer close(h.done); h.svc.Run(ctx, 10*time.Millisecond) }()
205	212		t.Cleanup(func() {
206	213		cancel()

 	log   *logBuf
 	done  chan struct{}
-	killBus func()        // loses the bus: the daemon dies
+	killBus func()             // loses the bus: the daemon dies
-	closed  chan struct{} // a PipeWire connection was closed
+	cancel  context.CancelFunc // stops the service, as the process exit does
+	closed  chan struct{}      // a PipeWire connection was closed
 }
 type harnessOpts struct {
 	allow  *captureallow.Policy
 	noPW   bool
 	holdOp bool
+	// drain overrides how long the stopping service waits for frames.
+	drain time.Duration
 	// iterate runs inside every PipeWire loop iteration, on the owner.
 	iterate func()
 }
 	}
 	h.svc = New(func() *captureallow.Policy { return allow }, h.cmds, testLog(h.log))
 	h.svc.address = h.addr
+	if o.drain > 0 {
+		h.svc.drain = o.drain
+	}
 	res := newMockpeerResolver(t)
 	res.EXPECT().Open(mock.Anything).RunAndReturn(func(pid int) (*os.File, error) {
 	h.svc.pw = pwOpen
 	ctx, cancel := context.WithCancel(context.Background())
+	h.cancel = cancel
 	go func() { defer close(h.done); h.svc.Run(ctx, 10*time.Millisecond) }()
 	t.Cleanup(func() {
 		cancel()

M internal/adapters/screencast/producer_test.go +39 -19

@@ -1,7 +1,9 @@
1	1		package screencast
2	2
3	3		import (
	4	+	"context"
4	5		"errors"
	6	+	"sync/atomic"
5	7		"testing"
6	8		"time"
7	9

@@ -368,7 +370,7 @@ func TestShutdownClosesStreams(t *testing.T) {
368	370		var order []string
369	371		p.src.EXPECT().Destroy().Run(func() { order = append(order, "destroy") }).Once()
370	372		p.core.EXPECT().Close().Run(func() { order = append(order, "close") }).Once()
371		-	p.s.shutdown()
	373	+	p.s.shutdown(context.Background())
372	374		if c := p.cmd(); c.Kind != ports.CastClose \|\| c.ID != 1 {
373	375		t.Fatalf("%+v", c)
374	376		}

@@ -388,7 +390,7 @@ func TestShutdownClosesStreamWithoutSource(t *testing.T) {
388	390		t.Fatalf("%+v", c)
389	391		}
390	392		p.core.EXPECT().Close().Return().Once()
391		-	p.s.shutdown()
	393	+	p.s.shutdown(context.Background())
392	394		if c := p.cmd(); c.Kind != ports.CastClose \|\| c.ID != 1 {
393	395		t.Fatalf("%+v", c)
394	396		}

@@ -397,36 +399,51 @@ func TestShutdownClosesStreamWithoutSource(t *testing.T) {
397	399		}
398	400		}
399	401
400		-	// A frame in flight at shutdown is waited for: nothing is destroyed before
401		-	// its CastFrameDone, then the buffer is released and the node destroyed.
	402	+	// A frame in flight at shutdown is waited for: nothing is released or
	403	+	// destroyed before its CastFrameDone, then the buffer is released, the node
	404	+	// destroyed and PipeWire closed, in that order. The wait has no deadline
	405	+	// while the service runs on (drain is far shorter than the wait here).
402	406		func TestShutdownWaitsForFrameInFlight(t *testing.T) {
403	407		p := newProducer(t)
	408	+	p.s.drain = time.Nanosecond
404	409		p.streaming()
405	410		p.src.EXPECT().Dequeue().Return(p.buf).Once()
406	411		p.s.pump(p.st, p.now)
407	412		p.cmd()
	413	+	var delivered atomic.Bool
408	414		var order []string
409		-	p.src.EXPECT().Release(p.buf).Run(func(*pipewire.VideoBuffer) { order = append(order, "release") }).Once()
410		-	p.src.EXPECT().Destroy().Run(func() { order = append(order, "destroy") }).Once()
411		-	p.core.EXPECT().Close().Run(func() { order = append(order, "close") }).Once()
412		-	go func() {
413		-	time.Sleep(30 * time.Millisecond)
414		-	p.s.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
415		-	}()
416		-	start := time.Now()
417		-	p.s.shutdown()
418		-	if time.Since(start) < 30*time.Millisecond {
419		-	t.Fatal("shutdown did not wait for the frame")
	415	+	step := func(name string) {
	416	+	if !delivered.Load() {
	417	+	t.Errorf("%s before the CastFrameDone", name)
	418	+	}
	419	+	order = append(order, name)
420	420		}
421		-	if c := p.cmd(); c.Kind != ports.CastClose {
422		-	t.Fatalf("%+v", c)
	421	+	p.src.EXPECT().Release(p.buf).Run(func(*pipewire.VideoBuffer) { step("release") }).Once()
	422	+	p.src.EXPECT().Destroy().Run(func() { step("destroy") }).Once()
	423	+	p.core.EXPECT().Close().Run(func() { step("close") }).Once()
	424	+	done := make(chan struct{})
	425	+	go func() { defer close(done); p.s.shutdown(context.Background()) }()
	426	+	select {
	427	+	case c := <-p.cmds:
	428	+	if c.Kind != ports.CastClose {
	429	+	t.Errorf("%+v", c)
	430	+	}
	431	+	case <-time.After(5 * time.Second):
	432	+	t.Fatal("no CastClose")
	433	+	}
	434	+	delivered.Store(true)
	435	+	p.s.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
	436	+	select {
	437	+	case <-done:
	438	+	case <-time.After(5 * time.Second):
	439	+	t.Fatal("shutdown did not return")
423	440		}
424	441		if len(order) != 3 \|\| order[0] != "release" \|\| order[1] != "destroy" \|\| order[2] != "close" {
425	442		t.Fatalf("order %v", order)
426	443		}
427	444		}
428	445
429		-	// When the frame never comes back, nothing is destroyed or closed (the
	446	+	// When the service itself is stopping and the frame never comes back, nothing is destroyed or closed (the
430	447		// source's handler and its buffer are still in use), the storage the
431	448		// handler points into is left untouched and abandoned, and the service has
432	449		// a fresh table: repeated reconnects do not exhaust it.

@@ -438,7 +455,10 @@ func TestShutdownTimeoutKeepsHandlerStateAndFreesTable(t *testing.T) {
438	455		p.s.pump(p.st, p.now)
439	456		p.cmd()
440	457		old := p.s.sessions
441		-	p.s.shutdown() // no Destroy, Release or Close is expected
	458	+	// The service itself is stopping: the wait is bounded.
	459	+	stopped, stop := context.WithCancel(context.Background())
	460	+	stop()
	461	+	p.s.shutdown(stopped) // no Destroy, Release or Close is expected
442	462		if c := p.cmd(); c.Kind != ports.CastClose {
443	463		t.Fatalf("%+v", c)
444	464		}

 package screencast
 import (
+	"context"
 	"errors"
+	"sync/atomic"
 	"testing"
 	"time"
 	var order []string
 	p.src.EXPECT().Destroy().Run(func() { order = append(order, "destroy") }).Once()
 	p.core.EXPECT().Close().Run(func() { order = append(order, "close") }).Once()
-	p.s.shutdown()
+	p.s.shutdown(context.Background())
 	if c := p.cmd(); c.Kind != ports.CastClose || c.ID != 1 {
 		t.Fatalf("%+v", c)
 	}
 		t.Fatalf("%+v", c)
 	}
 	p.core.EXPECT().Close().Return().Once()
-	p.s.shutdown()
+	p.s.shutdown(context.Background())
 	if c := p.cmd(); c.Kind != ports.CastClose || c.ID != 1 {
 		t.Fatalf("%+v", c)
 	}
 	}
 }
-// A frame in flight at shutdown is waited for: nothing is destroyed before
+// A frame in flight at shutdown is waited for: nothing is released or
-// its CastFrameDone, then the buffer is released and the node destroyed.
+// destroyed before its CastFrameDone, then the buffer is released, the node
+// destroyed and PipeWire closed, in that order. The wait has no deadline
+// while the service runs on (drain is far shorter than the wait here).
 func TestShutdownWaitsForFrameInFlight(t *testing.T) {
 	p := newProducer(t)
+	p.s.drain = time.Nanosecond
 	p.streaming()
 	p.src.EXPECT().Dequeue().Return(p.buf).Once()
 	p.s.pump(p.st, p.now)
 	p.cmd()
+	var delivered atomic.Bool
 	var order []string
-	p.src.EXPECT().Release(p.buf).Run(func(*pipewire.VideoBuffer) { order = append(order, "release") }).Once()
+	step := func(name string) {
-	p.src.EXPECT().Destroy().Run(func() { order = append(order, "destroy") }).Once()
+		if !delivered.Load() {
-	p.core.EXPECT().Close().Run(func() { order = append(order, "close") }).Once()
+			t.Errorf("%s before the CastFrameDone", name)
-	go func() {
+		}
-		time.Sleep(30 * time.Millisecond)
+		order = append(order, name)
-		p.s.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
-	}()
-	start := time.Now()
-	p.s.shutdown()
-	if time.Since(start) < 30*time.Millisecond {
-		t.Fatal("shutdown did not wait for the frame")
 	}
-	if c := p.cmd(); c.Kind != ports.CastClose {
+	p.src.EXPECT().Release(p.buf).Run(func(*pipewire.VideoBuffer) { step("release") }).Once()
-		t.Fatalf("%+v", c)
+	p.src.EXPECT().Destroy().Run(func() { step("destroy") }).Once()
+	p.core.EXPECT().Close().Run(func() { step("close") }).Once()
+	done := make(chan struct{})
+	go func() { defer close(done); p.s.shutdown(context.Background()) }()
+	select {
+	case c := <-p.cmds:
+		if c.Kind != ports.CastClose {
+			t.Errorf("%+v", c)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("no CastClose")
+	}
+	delivered.Store(true)
+	p.s.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("shutdown did not return")
 	}
 	if len(order) != 3 || order[0] != "release" || order[1] != "destroy" || order[2] != "close" {
 		t.Fatalf("order %v", order)
 	}
 }
-// When the frame never comes back, nothing is destroyed or closed (the
+// When the service itself is stopping and the frame never comes back, nothing is destroyed or closed (the
 // source's handler and its buffer are still in use), the storage the
 // handler points into is left untouched and abandoned, and the service has
 // a fresh table: repeated reconnects do not exhaust it.
 	p.s.pump(p.st, p.now)
 	p.cmd()
 	old := p.s.sessions
-	p.s.shutdown() // no Destroy, Release or Close is expected
+	// The service itself is stopping: the wait is bounded.
+	stopped, stop := context.WithCancel(context.Background())
+	stop()
+	p.s.shutdown(stopped) // no Destroy, Release or Close is expected
 	if c := p.cmd(); c.Kind != ports.CastClose {
 		t.Fatalf("%+v", c)
 	}

M internal/adapters/screencast/screencast.go +29 -16


@@ -52,7 +52,8 @@ const (
52	52		// shotTimeout bounds a nefercap run.
53	53		shotTimeout = 60 * time.Second
54	54
55		-	// shutdownWait bounds the wait for frames still in flight at shutdown.
	55	+	// shutdownWait bounds the wait for frames still in flight at shutdown,
	56	+	// once the service itself is stopping.
56	57		shutdownWait = 2 * time.Second
57	58
58	59		// shutdownPoll is how often shutdown looks again for the events that

@@ -99,9 +100,9 @@ type Service struct {
99	100		haveInv bool
100	101		callers [maxCallers]caller
101	102		pending [maxPending]pending
102		-	// sessions is a pointer so that a shutdown that had to leave a PipeWire
103		-	// source behind can abandon the table: the source's handler still
104		-	// points into it, so it is never reset or reused.
	103	+	// sessions is a pointer so that a shutdown at process exit that had to
	104	+	// leave a PipeWire source behind can abandon the table: the source's
	105	+	// handler still points into it, so it is never reset or reused.
105	106		sessions *[maxSessions]session
106	107		nextSess uint32
107	108		nextStrm uint32

@@ -111,7 +112,7 @@ type Service struct {
111	112		shotSeq uint64 // ID of the last screenshot job; never reset
112	113		fatal error
113	114		jobs sync.WaitGroup
114		-	drain time.Duration // how long shutdown waits for frames in flight
	115	+	drain time.Duration // how long shutdown waits for frames once the service stops
115	116		}
116	117
117	118		// New creates the service. policy returns the capture allowlist in force;

@@ -207,7 +208,7 @@ func (s *Service) Run(ctx context.Context, retry time.Duration) {
207	208		}
208	209
209	210		// attempt is one connection of the service to the buses.
210		-	func (s *Service) attempt(ctx context.Context) (err error) {
	211	+	func (s *Service) attempt(parent context.Context) (err error) {
211	212		// The PipeWire loop belongs to this OS thread for its whole life.
212	213		runtime.LockOSThread()
213	214		defer runtime.UnlockOSThread()

@@ -217,14 +218,14 @@ func (s *Service) attempt(ctx context.Context) (err error) {
217	218		}
218	219		defer func() { _ = conn.Close() }()
219	220		// The screenshot job of this attempt ends with it.
220		-	ctx, cancel := context.WithCancel(ctx)
	221	+	ctx, cancel := context.WithCancel(parent)
221	222		defer cancel()
222	223		s.reset(ctx, conn)
223	224		if s.pwc, err = s.pw.Open(); err != nil {
224	225		s.pwc = nil
225	226		s.log.Warn().Err(err).Msg("PipeWire unavailable: screen casting is not offered")
226	227		}
227		-	defer s.shutdown()
	228	+	defer s.shutdown(parent)
228	229		if err = s.acquire(); err != nil {
229	230		return err
230	231		}

@@ -346,16 +347,25 @@ func (s *Service) service() {
346	347		// resources: CastClose goes to the wayland server, a frame still in flight
347	348		// is waited for (its buffer is lent to the wayland server), then the source
348	349		// is destroyed and its buffers released. Only then is PipeWire closed.
	350	+	// parent is the context the service runs under; the attempt's own context
	351	+	// may have ended for another reason.
349	352		//
350		-	// When a frame does not come back in time, the source cannot be destroyed
351		-	// safely. It is left as it is, with its loop: the sessions table that its
352		-	// handler points into is abandoned (never reset or reused) and a new one
353		-	// takes its place, so later attempts neither touch that storage nor run out
354		-	// of slots.
355		-	func (s *Service) shutdown() {
	353	+	// While parent is alive the attempt ended by itself (the bus was lost) and a
	354	+	// new one follows. The wayland server is alive and answers every CastFrame
	355	+	// with a CastFrameDone, so the wait for the frames in flight has no
	356	+	// deadline: a source left behind would keep its slot in the process-wide
	357	+	// table of PipeWire video sources for good, and repeated reconnects would
	358	+	// exhaust it. If parent ends meanwhile, the wait is bounded as below.
	359	+	//
	360	+	// Once parent has ended, the process is going away and the wait for the
	361	+	// frames in flight is bounded by s.drain. When a frame does not come back in
	362	+	// time, the source cannot be destroyed safely. It is left as it is, with its
	363	+	// loop, to the process exit. The sessions table that its handler points into
	364	+	// is abandoned (never reset or reused) and a new one takes its place.
	365	+	func (s *Service) shutdown(parent context.Context) {
356	366		s.stopShot()
357	367		s.forEachSession(func(sess *session) { s.closeSession(sess, true) })
358		-	deadline := time.Now().Add(s.drain)
	368	+	var deadline time.Time // set once parent ended
359	369		for {
360	370		evs, _, _, _ := s.take()
361	371		for i := range evs {

@@ -366,7 +376,10 @@ func (s *Service) shutdown() {
366	376		if !s.streamsLive() {
367	377		break
368	378		}
369		-	if !time.Now().Before(deadline) {
	379	+	if deadline.IsZero() && parent.Err() != nil {
	380	+	deadline = time.Now().Add(s.drain)
	381	+	}
	382	+	if !deadline.IsZero() && !time.Now().Before(deadline) {
370	383		s.log.Warn().Int("streams", s.streamCount()).Msg("frames still in flight at shutdown; PipeWire resources are left to the process exit")
371	384		s.sessions = new([maxSessions]session)
372	385		s.pwc = nil

 	// shotTimeout bounds a nefercap run.
 	shotTimeout = 60 * time.Second
-	// shutdownWait bounds the wait for frames still in flight at shutdown.
+	// shutdownWait bounds the wait for frames still in flight at shutdown,
+	// once the service itself is stopping.
 	shutdownWait = 2 * time.Second
 	// shutdownPoll is how often shutdown looks again for the events that
 	haveInv bool
 	callers [maxCallers]caller
 	pending [maxPending]pending
-	// sessions is a pointer so that a shutdown that had to leave a PipeWire
+	// sessions is a pointer so that a shutdown at process exit that had to
-	// source behind can abandon the table: the source's handler still
+	// leave a PipeWire source behind can abandon the table: the source's
-	// points into it, so it is never reset or reused.
+	// handler still points into it, so it is never reset or reused.
 	sessions *[maxSessions]session
 	nextSess uint32
 	nextStrm uint32
 	shotSeq  uint64 // ID of the last screenshot job; never reset
 	fatal    error
 	jobs     sync.WaitGroup
-	drain    time.Duration // how long shutdown waits for frames in flight
+	drain    time.Duration // how long shutdown waits for frames once the service stops
 }
 // New creates the service. policy returns the capture allowlist in force;
 }
 // attempt is one connection of the service to the buses.
-func (s *Service) attempt(ctx context.Context) (err error) {
+func (s *Service) attempt(parent context.Context) (err error) {
 	// The PipeWire loop belongs to this OS thread for its whole life.
 	runtime.LockOSThread()
 	defer runtime.UnlockOSThread()
 	}
 	defer func() { _ = conn.Close() }()
 	// The screenshot job of this attempt ends with it.
-	ctx, cancel := context.WithCancel(ctx)
+	ctx, cancel := context.WithCancel(parent)
 	defer cancel()
 	s.reset(ctx, conn)
 	if s.pwc, err = s.pw.Open(); err != nil {
 		s.pwc = nil
 		s.log.Warn().Err(err).Msg("PipeWire unavailable: screen casting is not offered")
 	}
-	defer s.shutdown()
+	defer s.shutdown(parent)
 	if err = s.acquire(); err != nil {
 		return err
 	}
 // resources: CastClose goes to the wayland server, a frame still in flight
 // is waited for (its buffer is lent to the wayland server), then the source
 // is destroyed and its buffers released. Only then is PipeWire closed.
+// parent is the context the service runs under; the attempt's own context
+// may have ended for another reason.
 //
-// When a frame does not come back in time, the source cannot be destroyed
+// While parent is alive the attempt ended by itself (the bus was lost) and a
-// safely. It is left as it is, with its loop: the sessions table that its
+// new one follows. The wayland server is alive and answers every CastFrame
-// handler points into is abandoned (never reset or reused) and a new one
+// with a CastFrameDone, so the wait for the frames in flight has no
-// takes its place, so later attempts neither touch that storage nor run out
+// deadline: a source left behind would keep its slot in the process-wide
-// of slots.
+// table of PipeWire video sources for good, and repeated reconnects would
-func (s *Service) shutdown() {
+// exhaust it. If parent ends meanwhile, the wait is bounded as below.
+//
+// Once parent has ended, the process is going away and the wait for the
+// frames in flight is bounded by s.drain. When a frame does not come back in
+// time, the source cannot be destroyed safely. It is left as it is, with its
+// loop, to the process exit. The sessions table that its handler points into
+// is abandoned (never reset or reused) and a new one takes its place.
+func (s *Service) shutdown(parent context.Context) {
 	s.stopShot()
 	s.forEachSession(func(sess *session) { s.closeSession(sess, true) })
-	deadline := time.Now().Add(s.drain)
+	var deadline time.Time // set once parent ended
 	for {
 		evs, _, _, _ := s.take()
 		for i := range evs {
 		if !s.streamsLive() {
 			break
 		}
-		if !time.Now().Before(deadline) {
+		if deadline.IsZero() && parent.Err() != nil {
+			deadline = time.Now().Add(s.drain)
+		}
+		if !deadline.IsZero() && !time.Now().Before(deadline) {
 			s.log.Warn().Int("streams", s.streamCount()).Msg("frames still in flight at shutdown; PipeWire resources are left to the process exit")
 			s.sessions = new([maxSessions]session)
 			s.pwc = nil

M internal/adapters/screencast/shutdown_test.go +41 -15

@@ -1,6 +1,7 @@
1	1		package screencast
2	2
3	3		import (
	4	+	"sync/atomic"
4	5		"testing"
5	6		"time"
6	7

@@ -26,10 +27,10 @@ func (l *liveCast) loop(f func()) {
26	27		l.h.svc.Push(ports.CastEvent{Kind: ports.CastDamage, ID: 1 << 40}) // wakes the owner
27	28		}
28	29
29		-	func startLiveCast(t testing.T, withFrame bool) liveCast {
	30	+	func startLiveCast(t testing.T, withFrame bool, drain time.Duration) liveCast {
30	31		t.Helper()
31	32		l := &liveCast{onLoop: make(chan func(), 8), hdl: make(chan pipewire.VideoHandler, 1)}
32		-	l.h, _ = ready(t, harnessOpts{iterate: func() {
	33	+	l.h, _ = ready(t, harnessOpts{drain: drain, iterate: func() {
33	34		select {
34	35		case f := <-l.onLoop:
35	36		f()

@@ -104,7 +105,7 @@ func waitClosed(t testing.T, h svcHarness) {
104	105		// Losing the bus with a live stream and no frame in flight: the cast is
105	106		// closed on the wayland side, the source destroyed, PipeWire closed.
106	107		func TestBusLossTearsDownStream(t *testing.T) {
107		-	l := startLiveCast(t, false)
	108	+	l := startLiveCast(t, false, 0)
108	109		l.src.EXPECT().Destroy().Return().Once()
109	110		l.h.killBus()
110	111		if cmd := recvCmd(t, l.h); cmd.Kind != ports.CastClose \|\| cmd.ID != 1 {

@@ -114,25 +115,50 @@ func TestBusLossTearsDownStream(t *testing.T) {
114	115		}
115	116
116	117		// With a frame in flight: CastClose goes out at once, but the source is
117		-	// destroyed only after the frame's CastFrameDone, and its buffer released.
	118	+	// destroyed only after the frame's CastFrameDone, even when that comes later
	119	+	// than drain: the service runs on, so nothing is left behind. The order is
	120	+	// synchronized by channels: the buffer must be released before the node is
	121	+	// destroyed, and PipeWire closed after both.
118	122		func TestBusLossWaitsForFrameInFlight(t *testing.T) {
119		-	l := startLiveCast(t, true)
120		-	if cmd := func() ports.CastCommand { l.h.killBus(); return recvCmd(t, l.h) }(); cmd.Kind != ports.CastClose \|\| cmd.ID != 1 {
	123	+	l := startLiveCast(t, true, time.Nanosecond)
	124	+	l.h.killBus()
	125	+	if cmd := recvCmd(t, l.h); cmd.Kind != ports.CastClose \|\| cmd.ID != 1 {
121	126		t.Fatalf("%+v", cmd)
122	127		}
123		-	select {
124		-	case <-l.h.closed:
125		-	t.Fatal("PipeWire closed with a frame in flight")
126		-	case <-time.After(200 * time.Millisecond):
	128	+	var delivered atomic.Bool
	129	+	steps := make(chan string, 3)
	130	+	step := func(name string) {
	131	+	if !delivered.Load() {
	132	+	t.Errorf("%s before the CastFrameDone", name)
	133	+	}
	134	+	steps <- name
127	135		}
128		-	released := make(chan struct{})
129		-	l.src.EXPECT().Release(l.buf).Run(func(*pipewire.VideoBuffer) { close(released) }).Once()
130		-	l.src.EXPECT().Destroy().Return().Once()
	136	+	l.src.EXPECT().Release(l.buf).Run(func(*pipewire.VideoBuffer) { step("release") }).Once()
	137	+	l.src.EXPECT().Destroy().Run(func() { step("destroy") }).Once()
	138	+	delivered.Store(true)
131	139		l.h.svc.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 2})
132	140		waitClosed(t, l.h)
	141	+	if len(steps) != 2 \|\| <-steps != "release" \|\| <-steps != "destroy" {
	142	+	t.Fatal("release and destroy did not run in order before PipeWire closed")
	143	+	}
	144	+	}
	145	+
	146	+	// The service itself stopping with a frame that is never answered returns
	147	+	// within the bounded wait, and leaves the source and PipeWire alone.
	148	+	func TestStopWithUnansweredFrameIsBounded(t *testing.T) {
	149	+	l := startLiveCast(t, true, 20*time.Millisecond)
	150	+	l.h.cancel()
	151	+	if cmd := recvCmd(t, l.h); cmd.Kind != ports.CastClose \|\| cmd.ID != 1 {
	152	+	t.Fatalf("%+v", cmd)
	153	+	}
133	154		select {
134		-	case <-released:
	155	+	case <-l.h.done:
	156	+	case <-time.After(5 * time.Second):
	157	+	t.Fatal("Run did not return")
	158	+	}
	159	+	select {
	160	+	case <-l.h.closed:
	161	+	t.Fatal("PipeWire closed under a source with a frame in flight")
135	162		default:
136		-	t.Fatal("buffer not released")
137	163		}
138	164		}

 package screencast
 import (
+	"sync/atomic"
 	"testing"
 	"time"
 	l.h.svc.Push(ports.CastEvent{Kind: ports.CastDamage, ID: 1 << 40}) // wakes the owner
 }
-func startLiveCast(t *testing.T, withFrame bool) *liveCast {
+func startLiveCast(t *testing.T, withFrame bool, drain time.Duration) *liveCast {
 	t.Helper()
 	l := &liveCast{onLoop: make(chan func(), 8), hdl: make(chan pipewire.VideoHandler, 1)}
-	l.h, _ = ready(t, harnessOpts{iterate: func() {
+	l.h, _ = ready(t, harnessOpts{drain: drain, iterate: func() {
 		select {
 		case f := <-l.onLoop:
 			f()
 // Losing the bus with a live stream and no frame in flight: the cast is
 // closed on the wayland side, the source destroyed, PipeWire closed.
 func TestBusLossTearsDownStream(t *testing.T) {
-	l := startLiveCast(t, false)
+	l := startLiveCast(t, false, 0)
 	l.src.EXPECT().Destroy().Return().Once()
 	l.h.killBus()
 	if cmd := recvCmd(t, l.h); cmd.Kind != ports.CastClose || cmd.ID != 1 {
 }
 // With a frame in flight: CastClose goes out at once, but the source is
-// destroyed only after the frame's CastFrameDone, and its buffer released.
+// destroyed only after the frame's CastFrameDone, even when that comes later
+// than drain: the service runs on, so nothing is left behind. The order is
+// synchronized by channels: the buffer must be released before the node is
+// destroyed, and PipeWire closed after both.
 func TestBusLossWaitsForFrameInFlight(t *testing.T) {
-	l := startLiveCast(t, true)
+	l := startLiveCast(t, true, time.Nanosecond)
-	if cmd := func() ports.CastCommand { l.h.killBus(); return recvCmd(t, l.h) }(); cmd.Kind != ports.CastClose || cmd.ID != 1 {
+	l.h.killBus()
+	if cmd := recvCmd(t, l.h); cmd.Kind != ports.CastClose || cmd.ID != 1 {
 		t.Fatalf("%+v", cmd)
 	}
-	select {
+	var delivered atomic.Bool
-	case <-l.h.closed:
+	steps := make(chan string, 3)
-		t.Fatal("PipeWire closed with a frame in flight")
+	step := func(name string) {
-	case <-time.After(200 * time.Millisecond):
+		if !delivered.Load() {
+			t.Errorf("%s before the CastFrameDone", name)
+		}
+		steps <- name
 	}
-	released := make(chan struct{})
+	l.src.EXPECT().Release(l.buf).Run(func(*pipewire.VideoBuffer) { step("release") }).Once()
-	l.src.EXPECT().Release(l.buf).Run(func(*pipewire.VideoBuffer) { close(released) }).Once()
+	l.src.EXPECT().Destroy().Run(func() { step("destroy") }).Once()
-	l.src.EXPECT().Destroy().Return().Once()
+	delivered.Store(true)
 	l.h.svc.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 2})
 	waitClosed(t, l.h)
+	if len(steps) != 2 || <-steps != "release" || <-steps != "destroy" {
+		t.Fatal("release and destroy did not run in order before PipeWire closed")
+	}
+}
+// The service itself stopping with a frame that is never answered returns
+// within the bounded wait, and leaves the source and PipeWire alone.
+func TestStopWithUnansweredFrameIsBounded(t *testing.T) {
+	l := startLiveCast(t, true, 20*time.Millisecond)
+	l.h.cancel()
+	if cmd := recvCmd(t, l.h); cmd.Kind != ports.CastClose || cmd.ID != 1 {
+		t.Fatalf("%+v", cmd)
+	}
 	select {
-	case <-released:
+	case <-l.h.done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Run did not return")
+	}
+	select {
+	case <-l.h.closed:
+		t.Fatal("PipeWire closed under a source with a frame in flight")
 	default:
-		t.Fatal("buffer not released")
 	}
 }