test(screencast): saturate pending table without blocking owner

Drive admission and caller completion deterministically with real private-bus replies. Blocking resolver Open on the owner could prevent the overflow request from being processed. Remove the blocking hook and bound reply reads by closing the client on timeout.

commit: 3592453f463d671ba6231196bb16ec235441c911
author: brice · 3d ago · 2026-10-19 18:41 UTC
parent: 7edbc98
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
checks: 1/1

2 files · +35 -25

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
 		}