#6 test(screencast): make cast and pending-table tests deterministic

state: merged · e11014e · squash
branch: fix/screencast-test-flakes → main
opened: bnema · 3d ago
updated: 3d ago

Three tests failed at random under make race, on main as well as on merge requests. Only test files change; compositor behaviour is unchanged.

TestCastOutputUnplugStops (wayland): the capture could complete before the output removal reached the display owner. The test now applies the removal before completing the capture, and still requires an error for the in-flight frame.
TestCastInventoryPublishedOncePerChange (wayland): the next frame was lent before the previous one completed, so the owner rejected it as busy. The test now waits for each CastFrameDone.
TestPendingTableFull (screencast): the mocked Open blocked the service owner and could hang the run for 10 minutes. The test now fills the pending table without blocking the owner, and reply reads are bounded.

Blocking receives in the harnesses fail after a few seconds instead of hanging.

Checks: each test 600 times with -race -cpu=1,2,4; the whole TestCast set and the screencast package 30 times; make check, make race, make mocks-check, staticcheck ./... pass.

Commit	Subject	Author	Date	Signature	Status
3592453	test(screencast): saturate pending table without blocking owner	brice	3d ago	GPG Verified
7edbc98	test(wayland): synchronize screencast lifecycle assertions	brice	3d ago	GPG Verified

Checks 1

check success

3 files · +75 -27

1/1

M internal/adapters/screencast/bus_test.go +31 -13


@@ -374,26 +374,44 @@ func TestNonOwnerIsDenied(t *testing.T) {
374	374		// full, the next one gets LimitsExceeded, and the parked ones are answered
375	375		// once the check ends.
376	376		func TestPendingTableFull(t *testing.T) {
377		-	h := newSvcHarness(t, harnessOpts{holdOp: true})
378		-	c := h.client()
379		-	h.waitName(c, nameIntrospect)
380		-	serials := make([]uint32, 0, maxPending+1)
381		-	for range maxPending + 1 {
382		-	c.NewCall(nameIntrospect, pathIntrospect, ifaceIntrospect, "GetWindows", "")
383		-	serial, err := c.Send()
384		-	if err != nil {
385		-	t.Fatal(err)
386		-	}
387		-	serials = append(serials, serial)
	377	+	// Drive admission on this goroutine, delaying the caller-check transition
	378	+	// rather than blocking Open on the service owner. A PID reply may arrive
	379	+	// before all calls do; blocking Open then prevents the owner filling the
	380	+	// table or answering the overflow at all.
	381	+	addr, _ := startBus(t)
	382	+	c, err := zerobus.Dial(addr)
	383	+	if err != nil {
	384	+	t.Fatal(err)
388	385		}
	386	+	t.Cleanup(func() { _ = c.Close() })
	387	+	server, err := zerobus.Dial(addr)
	388	+	if err != nil {
	389	+	t.Fatal(err)
	390	+	}
	391	+	t.Cleanup(func() { _ = server.Close() })
	392	+	s := New(captureallow.Builtin, nil, testLog(&logBuf{}))
	393	+	s.conn = server
	394	+	peer := &s.callers[0]
	395	+	*peer = caller{state: callerPid, name: c.UniqueName()}
	396	+	for i := range maxPending + 1 {
	397	+	s.admit(&request{sender: c.UniqueName(), serial: uint32(i + 1), obj: objIntrospect, method: mGetWindows})
	398	+	}
	399	+	// Closing the real connection bounds reads without concurrently using its
	400	+	// encoder/decoder. No reply means a failed test, not the CI-wide timeout.
	401	+	timer := time.AfterFunc(5*time.Second, func() { _ = c.Close() })
	402	+	defer timer.Stop()
389	403		m, err := c.ReadMessage()
	404	+	for err == nil && m.Type == zerobus.TypeSignal {
	405	+	m, err = c.ReadMessage()
	406	+	}
390	407		if err != nil {
391	408		t.Fatal(err)
392	409		}
393		-	if m.Type != zerobus.TypeError \|\| m.ReplySerial != serials[maxPending] \|\| m.ErrorName != errLimits {
	410	+	if m.Type != zerobus.TypeError \|\| m.ReplySerial != maxPending+1 \|\| m.ErrorName != errLimits {
394	411		t.Fatalf("17th request: type %d error %q reply %d", m.Type, m.ErrorName, m.ReplySerial)
395	412		}
396		-	close(h.peers.openGo)
	413	+	peer.state, peer.exe = callerKnown, "/usr/bin/grim"
	414	+	s.flush(peer)
397	415		for range maxPending {
398	416		m, err := c.ReadMessage()
399	417		if err != nil {

 // full, the next one gets LimitsExceeded, and the parked ones are answered
 // once the check ends.
 func TestPendingTableFull(t *testing.T) {
-	h := newSvcHarness(t, harnessOpts{holdOp: true})
+	// Drive admission on this goroutine, delaying the caller-check transition
-	c := h.client()
+	// rather than blocking Open on the service owner. A PID reply may arrive
-	h.waitName(c, nameIntrospect)
+	// before all calls do; blocking Open then prevents the owner filling the
-	serials := make([]uint32, 0, maxPending+1)
+	// table or answering the overflow at all.
-	for range maxPending + 1 {
+	addr, _ := startBus(t)
-		c.NewCall(nameIntrospect, pathIntrospect, ifaceIntrospect, "GetWindows", "")
+	c, err := zerobus.Dial(addr)
-		serial, err := c.Send()
+	if err != nil {
-		if err != nil {
+		t.Fatal(err)
-			t.Fatal(err)
-		}
-		serials = append(serials, serial)
 	}
+	t.Cleanup(func() { _ = c.Close() })
+	server, err := zerobus.Dial(addr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = server.Close() })
+	s := New(captureallow.Builtin, nil, testLog(&logBuf{}))
+	s.conn = server
+	peer := &s.callers[0]
+	*peer = caller{state: callerPid, name: c.UniqueName()}
+	for i := range maxPending + 1 {
+		s.admit(&request{sender: c.UniqueName(), serial: uint32(i + 1), obj: objIntrospect, method: mGetWindows})
+	}
+	// Closing the real connection bounds reads without concurrently using its
+	// encoder/decoder. No reply means a failed test, not the CI-wide timeout.
+	timer := time.AfterFunc(5*time.Second, func() { _ = c.Close() })
+	defer timer.Stop()
 	m, err := c.ReadMessage()
+	for err == nil && m.Type == zerobus.TypeSignal {
+		m, err = c.ReadMessage()
+	}
 	if err != nil {
 		t.Fatal(err)
 	}
-	if m.Type != zerobus.TypeError || m.ReplySerial != serials[maxPending] || m.ErrorName != errLimits {
+	if m.Type != zerobus.TypeError || m.ReplySerial != maxPending+1 || m.ErrorName != errLimits {
 		t.Fatalf("17th request: type %d error %q reply %d", m.Type, m.ErrorName, m.ReplySerial)
 	}
-	close(h.peers.openGo)
+	peer.state, peer.exe = callerKnown, "/usr/bin/grim"
+	s.flush(peer)
 	for range maxPending {
 		m, err := c.ReadMessage()
 		if err != nil {

M internal/adapters/screencast/harness_test.go +4 -12


@@ -71,12 +71,11 @@ func testLog(out *logBuf) zerowrap.Logger {
71	71		return zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: out}).WithField("component", "screencast")
72	72		}
73	73
74		-	// peers is what the mocked peerResolver answers: the executable of every
75		-	// peer, and a hook to hold Open.
	74	+	// peers is what the mocked peerResolver answers: the executable and pid
	75	+	// of every peer.
76	76		type peers struct {
77	77		exe atomic.Value // string
78	78		exeNow func() string
79		-	openGo chan struct{} // when set, Open waits for it
80	79		pid atomic.Int64
81	80		openErr error
82	81		exePid func(open int) int // pid Exe reports; the opened one when nil

@@ -118,9 +117,8 @@ type svcHarness struct {
118	117		}
119	118
120	119		type harnessOpts struct {
121		-	allow *captureallow.Policy
122		-	noPW bool
123		-	holdOp bool
	120	+	allow *captureallow.Policy
	121	+	noPW bool
124	122		// drain overrides how long the stopping service waits for frames.
125	123		drain time.Duration
126	124		// iterate runs inside every PipeWire loop iteration, on the owner.

@@ -132,9 +130,6 @@ func newSvcHarness(t testing.T, o harnessOpts) svcHarness {
132	130		h := &svcHarness{t: t, cmds: make(chan ports.CastCommand, 64), peers: &peers{}, log: &logBuf{}, done: make(chan struct{}), closed: make(chan struct{}, 8)}
133	131		h.addr, h.killBus = startBus(t)
134	132		h.peers.exe.Store("/usr/bin/grim")
135		-	if o.holdOp {
136		-	h.peers.openGo = make(chan struct{})
137		-	}
138	133		allow := o.allow
139	134		if allow == nil {
140	135		allow = captureallow.Builtin()

@@ -147,9 +142,6 @@ func newSvcHarness(t testing.T, o harnessOpts) svcHarness {
147	142
148	143		res := newMockpeerResolver(t)
149	144		res.EXPECT().Open(mock.Anything).RunAndReturn(func(pid int) (*os.File, error) {
150		-	if h.peers.openGo != nil {
151		-	<-h.peers.openGo
152		-	}
153	145		if h.peers.openErr != nil {
154	146		return nil, h.peers.openErr
155	147		}

 	return zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: out}).WithField("component", "screencast")
 }
-// peers is what the mocked peerResolver answers: the executable of every
+// peers is what the mocked peerResolver answers: the executable and pid
-// peer, and a hook to hold Open.
+// of every peer.
 type peers struct {
 	exe     atomic.Value // string
 	exeNow  func() string
-	openGo  chan struct{} // when set, Open waits for it
 	pid     atomic.Int64
 	openErr error
 	exePid  func(open int) int // pid Exe reports; the opened one when nil
 }
 type harnessOpts struct {
-	allow  *captureallow.Policy
+	allow *captureallow.Policy
-	noPW   bool
+	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.
 	h := &svcHarness{t: t, cmds: make(chan ports.CastCommand, 64), peers: &peers{}, log: &logBuf{}, done: make(chan struct{}), closed: make(chan struct{}, 8)}
 	h.addr, h.killBus = startBus(t)
 	h.peers.exe.Store("/usr/bin/grim")
-	if o.holdOp {
-		h.peers.openGo = make(chan struct{})
-	}
 	allow := o.allow
 	if allow == nil {
 		allow = captureallow.Builtin()
 	res := newMockpeerResolver(t)
 	res.EXPECT().Open(mock.Anything).RunAndReturn(func(pid int) (*os.File, error) {
-		if h.peers.openGo != nil {
-			<-h.peers.openGo
-		}
 		if h.peers.openErr != nil {
 			return nil, h.peers.openErr
 		}

M internal/adapters/wayland/screencast_test.go +40 -2


@@ -76,7 +76,14 @@ func newCastHarness(t testing.T, withSink bool) castHarness {
76	76		ctx, cancel := context.WithCancel(context.Background())
77	77		done := make(chan error, 1)
78	78		go func() { done <- s.Run(ctx) }()
79		-	t.Cleanup(func() { cancel(); <-done })
	79	+	t.Cleanup(func() {
	80	+	cancel()
	81	+	select {
	82	+	case <-done:
	83	+	case <-time.After(3 * time.Second):
	84	+	t.Fatal("server did not stop")
	85	+	}
	86	+	})
80	87		return h
81	88		}
82	89

@@ -107,6 +114,26 @@ func (h *castHarness) next(kind ports.CastEventKind) ports.CastEvent {
107	114		}
108	115		}
109	116
	117	+	func (h *castHarness) nextCastFrameDone() ports.CastEvent {
	118	+	h.t.Helper()
	119	+	timer := time.NewTimer(3 * time.Second)
	120	+	defer timer.Stop()
	121	+	for {
	122	+	select {
	123	+	case ev := <-h.got:
	124	+	if ev.Kind == ports.CastFrameDone {
	125	+	return ev
	126	+	}
	127	+	if ev.Kind != ports.CastDamage {
	128	+	h.t.Fatalf("unexpected event %+v", ev)
	129	+	}
	130	+	case <-timer.C:
	131	+	h.t.Fatal("no frame completion")
	132	+	return ports.CastEvent{}
	133	+	}
	134	+	}
	135	+	}
	136	+
110	137		// none says that no event is pending once the display goroutine caught up.
111	138		func (h *castHarness) none() {
112	139		h.t.Helper()

@@ -331,7 +358,13 @@ func TestCastOutputUnplugStops(t *testing.T) {
331	358		data := make([]byte, 64324)
332	359		h.casts <- castFrame(1, 9, data, 64, 32)
333	360		req := h.request()
334		-	h.commands <- ports.SetOutputs{Outputs: castLayout[1:], Focused: "HEADLESS-2"}
	361	+	// Apply on the owner before completing the capture. Posting a barrier
	362	+	// does not fence the independent Commands forwarding goroutine.
	363	+	if !h.s.display.Do(func() {
	364	+	h.s.apply(ports.SetOutputs{Outputs: castLayout[1:], Focused: "HEADLESS-2"})
	365	+	}) {
	366	+	t.Fatal("display stopped")
	367	+	}
335	368		h.none()
336	369		h.complete(req, nil)
337	370		if done := h.next(ports.CastFrameDone); done.ID != 1 \|\| done.Buf != 9 \|\| done.Err == nil {

@@ -615,6 +648,11 @@ func TestCastInventoryPublishedOncePerChange(t *testing.T) {
615	648		h.flip("HEADLESS-1")
616	649		h.casts <- castFrame(1, i%3, data, 64, 32)
617	650		h.complete(h.request(), nil)
	651	+	// Flips may also publish damage; settle this buffer before lending
	652	+	// the next one, independently of the reply-forwarding goroutine.
	653	+	if ev := h.nextCastFrameDone(); ev.Err != nil {
	654	+	t.Fatalf("frame %d: %+v", i, ev)
	655	+	}
618	656		}
619	657		noInv()
620	658


@@ -76,7 +76,14 @@ func newCastHarness(t testing.T, withSink bool) castHarness {
76		ctx, cancel := context.WithCancel(context.Background())	76		ctx, cancel := context.WithCancel(context.Background())
77		done := make(chan error, 1)	77		done := make(chan error, 1)
78		go func() { done <- s.Run(ctx) }()	78		go func() { done <- s.Run(ctx) }()
79	-	t.Cleanup(func() { cancel(); <-done })	79	+	t.Cleanup(func() {
			80	+	cancel()
			81	+	select {
			82	+	case <-done:
			83	+	case <-time.After(3 * time.Second):
			84	+	t.Fatal("server did not stop")
			85	+	}
			86	+	})
80		return h	87		return h
81		}	88		}
82			89

@@ -107,6 +114,26 @@ func (h *castHarness) next(kind ports.CastEventKind) ports.CastEvent {
107		}	114		}
108		}	115		}
109			116
			117	+	func (h *castHarness) nextCastFrameDone() ports.CastEvent {
			118	+	h.t.Helper()
			119	+	timer := time.NewTimer(3 * time.Second)
			120	+	defer timer.Stop()
			121	+	for {
			122	+	select {
			123	+	case ev := <-h.got:
			124	+	if ev.Kind == ports.CastFrameDone {
			125	+	return ev
			126	+	}
			127	+	if ev.Kind != ports.CastDamage {
			128	+	h.t.Fatalf("unexpected event %+v", ev)
			129	+	}
			130	+	case <-timer.C:
			131	+	h.t.Fatal("no frame completion")
			132	+	return ports.CastEvent{}
			133	+	}
			134	+	}
			135	+	}
			136	+
110		// none says that no event is pending once the display goroutine caught up.	137		// none says that no event is pending once the display goroutine caught up.
111		func (h *castHarness) none() {	138		func (h *castHarness) none() {
112		h.t.Helper()	139		h.t.Helper()

@@ -331,7 +358,13 @@ func TestCastOutputUnplugStops(t *testing.T) {
331		data := make([]byte, 64324)	358		data := make([]byte, 64324)
332		h.casts <- castFrame(1, 9, data, 64, 32)	359		h.casts <- castFrame(1, 9, data, 64, 32)
333		req := h.request()	360		req := h.request()
334	-	h.commands <- ports.SetOutputs{Outputs: castLayout[1:], Focused: "HEADLESS-2"}	361	+	// Apply on the owner before completing the capture. Posting a barrier
			362	+	// does not fence the independent Commands forwarding goroutine.
			363	+	if !h.s.display.Do(func() {
			364	+	h.s.apply(ports.SetOutputs{Outputs: castLayout[1:], Focused: "HEADLESS-2"})
			365	+	}) {
			366	+	t.Fatal("display stopped")
			367	+	}
335		h.none()	368		h.none()
336		h.complete(req, nil)	369		h.complete(req, nil)
337		if done := h.next(ports.CastFrameDone); done.ID != 1 \|\| done.Buf != 9 \|\| done.Err == nil {	370		if done := h.next(ports.CastFrameDone); done.ID != 1 \|\| done.Buf != 9 \|\| done.Err == nil {

@@ -615,6 +648,11 @@ func TestCastInventoryPublishedOncePerChange(t *testing.T) {
615		h.flip("HEADLESS-1")	648		h.flip("HEADLESS-1")
616		h.casts <- castFrame(1, i%3, data, 64, 32)	649		h.casts <- castFrame(1, i%3, data, 64, 32)
617		h.complete(h.request(), nil)	650		h.complete(h.request(), nil)
			651	+	// Flips may also publish damage; settle this buffer before lending
			652	+	// the next one, independently of the reply-forwarding goroutine.
			653	+	if ev := h.nextCastFrameDone(); ev.Err != nil {
			654	+	t.Fatalf("frame %d: %+v", i, ev)
			655	+	}
618		}	656		}
619		noInv()	657		noInv()
620			658

Comments0 open / 0

no comments yet.

Merge settings

Comments0 open / 0