e11014e test(screencast): make cast and pending-table tests deterministic (!6) · neferwl

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