fix(atspi): own the source from the hints worker and bound connect attempts

commit: 73f97723d0a734209d0b820a8b7e8eebd1eab362
author: brice · 3d ago · 2026-10-19 15:57 UTC
parent: 88f7989
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +232 -42

M internal/adapters/atspi/source.go +90 -31


@@ -32,26 +32,39 @@ type connector interface {
32	32		Connect(ctx context.Context) (transport, error)
33	33		}
34	34
	35	+	// stallWarnAfter is how long an attempt may be in flight before it is
	36	+	// reported as a stalled accessibility bus.
	37	+	const stallWarnAfter = 5 * time.Second
	38	+
35	39		// dialResult is the outcome of a connection attempt.
36	40		type dialResult struct {
37	41		tr transport
38	42		err error
	43	+	at time.Time // when the attempt finished
39	44		}
40	45
41	46		// Source reads hint targets from the accessibility tree. It owns one
42		-	// connection to the accessibility bus and is used by one goroutine at a
43		-	// time (the hints worker). The connection is made on the first Targets call
	47	+	// connection to the accessibility bus. It is NOT safe for concurrent use:
	48	+	// it is owned by the hints worker goroutine, which calls Targets and,
	49	+	// when it returns, Close. The connection is made on the first Targets call
44	50		// that needs it, never at startup, and made again after a failure, at most
45	51		// once every retryEvery; meanwhile Targets finds nothing and OCR takes over.
	52	+	//
	53	+	// At most one connection attempt (one goroutine and one socket) exists at
	54	+	// any time: an attempt that outlives the wait of ensure stays in flight, is
	55	+	// polled by the later Targets calls and is never doubled.
46	56		type Source struct {
47	57		tr transport
48	58		dial connector
49		-	// attempt is the outcome of a connection attempt that may still run: a
50		-	// silent service is abandoned, not waited for, and its late result is
51		-	// closed.
52		-	attempt chan dialResult
53		-	attemptDue time.Time
54		-	retryAt time.Time
	59	+	// attempt delivers the outcome of the attempt in flight (buffered, so
	60	+	// the attempt goroutine never blocks); nil when none runs.
	61	+	attempt chan dialResult
	62	+	attemptStarted time.Time
	63	+	warnedStall bool
	64	+	retryAt time.Time
	65	+	closed bool
	66	+	// wait is how long ensure waits for a new attempt; connectBudget if 0.
	67	+	wait time.Duration
55	68		}
56	69
57	70		// New returns a Source. It does not touch the bus: the connection is made by

@@ -64,6 +77,11 @@ type busConnector struct{}
64	77
65	78		func (busConnector) Connect(ctx context.Context) (transport, error) {
66	79		log := logging.For(ctx, "atspi")
	80	+	// zerobus has no deadline or context API: SessionBus and Dial run the
	81	+	// socket dial, SASL auth and Hello without a bound, and the watchdog
	82	+	// below cannot exist before the connection does. A stall there is
	83	+	// bounded by the single-in-flight rule of Source.ensure, not by a
	84	+	// deadline. Follow-up: a deadline-capable zerobus API.
67	85		sess, err := zerobus.SessionBus()
68	86		if err != nil {
69	87		return nil, fmt.Errorf("atspi: session bus: %w", err)

@@ -111,8 +129,13 @@ func (busConnector) Connect(ctx context.Context) (transport, error) {
111	129		}
112	130
113	131		// ensure returns the connection, making one if none exists and the last
114		-	// attempt is old enough. It waits at most connectBudget for an attempt.
	132	+	// attempt is old enough. It waits at most the budget for a new attempt; an
	133	+	// attempt that does not answer in time stays in flight (abandoned, never
	134	+	// doubled) and is polled without blocking by the next calls.
115	135		func (s *Source) ensure(ctx context.Context) transport {
	136	+	if s.closed {
	137	+	return nil
	138	+	}
116	139		if s.tr != nil {
117	140		return s.tr
118	141		}

@@ -120,47 +143,82 @@ func (s *Source) ensure(ctx context.Context) transport {
120	143		return nil
121	144		}
122	145		now := time.Now()
123		-	if s.attempt == nil {
124		-	if now.Before(s.retryAt) {
125		-	return nil
	146	+	if s.attempt != nil {
	147	+	select {
	148	+	case r := <-s.attempt:
	149	+	s.finish(ctx, r)
	150	+	default:
	151	+	if !s.warnedStall && now.Sub(s.attemptStarted) > stallWarnAfter {
	152	+	s.warnedStall = true
	153	+	log := logging.For(ctx, "atspi")
	154	+	log.Warn().Dur("for", now.Sub(s.attemptStarted)).Msg("accessibility bus not answering")
	155	+	}
126	156		}
127		-	s.retryAt = now.Add(retryEvery)
128		-	s.attemptDue = now.Add(connectBudget)
129		-	ch := make(chan dialResult, 1)
130		-	s.attempt = ch
131		-	go func() {
132		-	tr, err := s.dial.Connect(ctx)
133		-	ch <- dialResult{tr, err}
134		-	}()
	157	+	return s.tr
	158	+	}
	159	+	if now.Before(s.retryAt) {
	160	+	return nil
135	161		}
136		-	timer := time.NewTimer(time.Until(s.attemptDue))
	162	+	ch := make(chan dialResult, 1)
	163	+	s.attempt = ch
	164	+	s.attemptStarted = now
	165	+	s.warnedStall = false
	166	+	dial := s.dial
	167	+	go func() {
	168	+	tr, err := dial.Connect(ctx)
	169	+	ch <- dialResult{tr: tr, err: err, at: time.Now()}
	170	+	}()
	171	+	wait := s.wait
	172	+	if wait <= 0 {
	173	+	wait = connectBudget
	174	+	}
	175	+	timer := time.NewTimer(wait)
137	176		defer timer.Stop()
138	177		select {
139		-	case r := <-s.attempt:
140		-	s.attempt = nil
141		-	if r.err != nil {
142		-	log := logging.For(ctx, "atspi")
143		-	log.Warn().Err(r.err).Msg("accessibility hint targets unavailable, using OCR only")
144		-	return nil
145		-	}
146		-	s.tr = r.tr
	178	+	case r := <-ch:
	179	+	s.finish(ctx, r)
147	180		case <-timer.C:
	181	+	log := logging.For(ctx, "atspi")
	182	+	log.Debug().Dur("after", wait).Msg("accessibility connect attempt abandoned in flight")
148	183		case <-ctx.Done():
149	184		}
150	185		return s.tr
151	186		}
152	187
	188	+	// finish takes the outcome of the attempt in flight: a transport is kept,
	189	+	// an error starts the backoff from the time the attempt finished.
	190	+	func (s *Source) finish(ctx context.Context, r dialResult) {
	191	+	s.attempt = nil
	192	+	if r.err != nil \|\| r.tr == nil {
	193	+	s.retryAt = r.at.Add(retryEvery)
	194	+	err := r.err
	195	+	if err == nil {
	196	+	err = errors.New("atspi: no transport")
	197	+	}
	198	+	log := logging.For(ctx, "atspi")
	199	+	log.Warn().Err(err).Msg("accessibility hint targets unavailable, using OCR only")
	200	+	return
	201	+	}
	202	+	s.tr = r.tr
	203	+	}
	204	+
153	205		// drop closes a broken connection; the next attempt reconnects.
154	206		func (s *Source) drop() {
	207	+	s.retryAt = time.Now().Add(retryEvery)
155	208		if s.tr != nil {
156	209		_ = s.tr.Close()
157	210		s.tr = nil
158	211		}
159	212		}
160	213
161		-	// Close closes the accessibility bus connection, and the one an abandoned
162		-	// attempt may still deliver.
	214	+	// Close closes the accessibility bus connection, and the one an attempt
	215	+	// still in flight may deliver late. After Close, Targets finds nothing and
	216	+	// never connects. Like Targets it is called by the owner goroutine only.
163	217		func (s *Source) Close() error {
	218	+	if s.closed {
	219	+	return nil
	220	+	}
	221	+	s.closed = true
164	222		var err error
165	223		if s.tr != nil {
166	224		err = s.tr.Close()

@@ -168,6 +226,7 @@ func (s *Source) Close() error {
168	226		}
169	227		if ch := s.attempt; ch != nil {
170	228		s.attempt = nil
	229	+	// One reaper at most: Close runs once, and one attempt exists.
171	230		go func() {
172	231		if r := <-ch; r.tr != nil {
173	232		_ = r.tr.Close()

 	Connect(ctx context.Context) (transport, error)
 }
+// stallWarnAfter is how long an attempt may be in flight before it is
+// reported as a stalled accessibility bus.
+const stallWarnAfter = 5 * time.Second
 // dialResult is the outcome of a connection attempt.
 type dialResult struct {
 	tr  transport
 	err error
+	at  time.Time // when the attempt finished
 }
 // Source reads hint targets from the accessibility tree. It owns one
-// connection to the accessibility bus and is used by one goroutine at a
+// connection to the accessibility bus. It is NOT safe for concurrent use:
-// time (the hints worker). The connection is made on the first Targets call
+// it is owned by the hints worker goroutine, which calls Targets and,
+// when it returns, Close. The connection is made on the first Targets call
 // that needs it, never at startup, and made again after a failure, at most
 // once every retryEvery; meanwhile Targets finds nothing and OCR takes over.
+//
+// At most one connection attempt (one goroutine and one socket) exists at
+// any time: an attempt that outlives the wait of ensure stays in flight, is
+// polled by the later Targets calls and is never doubled.
 type Source struct {
 	tr   transport
 	dial connector
-	// attempt is the outcome of a connection attempt that may still run: a
+	// attempt delivers the outcome of the attempt in flight (buffered, so
-	// silent service is abandoned, not waited for, and its late result is
+	// the attempt goroutine never blocks); nil when none runs.
-	// closed.
+	attempt        chan dialResult
-	attempt    chan dialResult
+	attemptStarted time.Time
-	attemptDue time.Time
+	warnedStall    bool
-	retryAt    time.Time
+	retryAt        time.Time
+	closed         bool
+	// wait is how long ensure waits for a new attempt; connectBudget if 0.
+	wait time.Duration
 }
 // New returns a Source. It does not touch the bus: the connection is made by
 func (busConnector) Connect(ctx context.Context) (transport, error) {
 	log := logging.For(ctx, "atspi")
+	// zerobus has no deadline or context API: SessionBus and Dial run the
+	// socket dial, SASL auth and Hello without a bound, and the watchdog
+	// below cannot exist before the connection does. A stall there is
+	// bounded by the single-in-flight rule of Source.ensure, not by a
+	// deadline. Follow-up: a deadline-capable zerobus API.
 	sess, err := zerobus.SessionBus()
 	if err != nil {
 		return nil, fmt.Errorf("atspi: session bus: %w", err)
 }
 // ensure returns the connection, making one if none exists and the last
-// attempt is old enough. It waits at most connectBudget for an attempt.
+// attempt is old enough. It waits at most the budget for a new attempt; an
+// attempt that does not answer in time stays in flight (abandoned, never
+// doubled) and is polled without blocking by the next calls.
 func (s *Source) ensure(ctx context.Context) transport {
+	if s.closed {
+		return nil
+	}
 	if s.tr != nil {
 		return s.tr
 	}
 		return nil
 	}
 	now := time.Now()
-	if s.attempt == nil {
+	if s.attempt != nil {
-		if now.Before(s.retryAt) {
+		select {
-			return nil
+		case r := <-s.attempt:
+			s.finish(ctx, r)
+		default:
+			if !s.warnedStall && now.Sub(s.attemptStarted) > stallWarnAfter {
+				s.warnedStall = true
+				log := logging.For(ctx, "atspi")
+				log.Warn().Dur("for", now.Sub(s.attemptStarted)).Msg("accessibility bus not answering")
+			}
 		}
-		s.retryAt = now.Add(retryEvery)
+		return s.tr
-		s.attemptDue = now.Add(connectBudget)
+	}
-		ch := make(chan dialResult, 1)
+	if now.Before(s.retryAt) {
-		s.attempt = ch
+		return nil
-		go func() {
-			tr, err := s.dial.Connect(ctx)
-			ch <- dialResult{tr, err}
-		}()
 	}
-	timer := time.NewTimer(time.Until(s.attemptDue))
+	ch := make(chan dialResult, 1)
+	s.attempt = ch
+	s.attemptStarted = now
+	s.warnedStall = false
+	dial := s.dial
+	go func() {
+		tr, err := dial.Connect(ctx)
+		ch <- dialResult{tr: tr, err: err, at: time.Now()}
+	}()
+	wait := s.wait
+	if wait <= 0 {
+		wait = connectBudget
+	}
+	timer := time.NewTimer(wait)
 	defer timer.Stop()
 	select {
-	case r := <-s.attempt:
+	case r := <-ch:
-		s.attempt = nil
+		s.finish(ctx, r)
-		if r.err != nil {
-			log := logging.For(ctx, "atspi")
-			log.Warn().Err(r.err).Msg("accessibility hint targets unavailable, using OCR only")
-			return nil
-		}
-		s.tr = r.tr
 	case <-timer.C:
+		log := logging.For(ctx, "atspi")
+		log.Debug().Dur("after", wait).Msg("accessibility connect attempt abandoned in flight")
 	case <-ctx.Done():
 	}
 	return s.tr
 }
+// finish takes the outcome of the attempt in flight: a transport is kept,
+// an error starts the backoff from the time the attempt finished.
+func (s *Source) finish(ctx context.Context, r dialResult) {
+	s.attempt = nil
+	if r.err != nil || r.tr == nil {
+		s.retryAt = r.at.Add(retryEvery)
+		err := r.err
+		if err == nil {
+			err = errors.New("atspi: no transport")
+		}
+		log := logging.For(ctx, "atspi")
+		log.Warn().Err(err).Msg("accessibility hint targets unavailable, using OCR only")
+		return
+	}
+	s.tr = r.tr
+}
 // drop closes a broken connection; the next attempt reconnects.
 func (s *Source) drop() {
+	s.retryAt = time.Now().Add(retryEvery)
 	if s.tr != nil {
 		_ = s.tr.Close()
 		s.tr = nil
 	}
 }
-// Close closes the accessibility bus connection, and the one an abandoned
+// Close closes the accessibility bus connection, and the one an attempt
-// attempt may still deliver.
+// still in flight may deliver late. After Close, Targets finds nothing and
+// never connects. Like Targets it is called by the owner goroutine only.
 func (s *Source) Close() error {
+	if s.closed {
+		return nil
+	}
+	s.closed = true
 	var err error
 	if s.tr != nil {
 		err = s.tr.Close()
 	}
 	if ch := s.attempt; ch != nil {
 		s.attempt = nil
+		// One reaper at most: Close runs once, and one attempt exists.
 		go func() {
 			if r := <-ch; r.tr != nil {
 				_ = r.tr.Close()

M internal/adapters/atspi/source_test.go +86 -9


@@ -215,24 +215,101 @@ func TestBrokenConnectionIsDropped(t *testing.T) {
215	215		require.Empty(t, targets) // backoff: no new attempt yet
216	216		}
217	217
218		-	// A silent service cannot hold Targets beyond the connect budget.
219		-	func TestConnectIsBounded(t *testing.T) {
	218	+	const testWait = 20 * time.Millisecond
	219	+
	220	+	// blockedConnector returns a connector whose Connect blocks until release is
	221	+	func blockedConnector(t testing.T, release <-chan struct{}, tr transport, err error, times int) mockconnector {
	222	+	t.Helper()
	223	+	c := newMockconnector(t)
	224	+	c.EXPECT().Connect(mock.Anything).RunAndReturn(func(context.Context) (transport, error) {
	225	+	<-release
	226	+	return tr, err
	227	+	}).Times(times)
	228	+	return c
	229	+	}
	230	+
	231	+	// A silent service cannot hold Targets beyond the wait, is not doubled while
	232	+	// in flight, and its late transport is used.
	233	+	func TestConnectIsBoundedAndNotDoubled(t *testing.T) {
220	234		release := make(chan struct{})
221	235		tr := newMocktransport(t)
222		-	tr.EXPECT().Close().Return(nil).Once()
	236	+	tr.EXPECT().Do(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(answerTree(
	237	+	[4]int32{0, 0, 800, 600}, []ref{{":1.5", "/a"}}, map[string][4]int32{"/a": {1, 1, 9, 9}}, map[string]uint32{"/a": roleLink}, false))
	238	+	c := blockedConnector(t, release, tr, nil, 1)
	239	+	s := &Source{dial: c, wait: testWait}
	240	+	t0 := time.Now()
	241	+	targets, _, err := s.Targets(context.Background(), oneWindow, false)
	242	+	require.NoError(t, err)
	243	+	require.Empty(t, targets)
	244	+	require.Less(t, time.Since(t0), testWait+200*time.Millisecond)
	245	+	require.NotNil(t, s.attempt)
	246	+	// Still in flight: no second Connect (Times(1)), and no blocking.
	247	+	targets, _, err = s.Targets(context.Background(), oneWindow, false)
	248	+	require.NoError(t, err)
	249	+	require.Empty(t, targets)
	250	+	// It connects late: the next call uses it.
	251	+	close(release)
	252	+	require.Eventually(t, func() bool { return len(s.attempt) == 1 }, time.Second, time.Millisecond)
	253	+	targets, _, err = s.Targets(context.Background(), oneWindow, false)
	254	+	require.NoError(t, err)
	255	+	require.Len(t, targets, 1)
	256	+	require.Nil(t, s.attempt)
	257	+	}
	258	+
	259	+	// An abandoned attempt that fails applies the backoff, then one new attempt.
	260	+	func TestAbandonedAttemptErrorBacksOff(t *testing.T) {
	261	+	release := make(chan struct{})
	262	+	tr := newMocktransport(t)
	263	+	tr.EXPECT().Do(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(answerTree(
	264	+	[4]int32{0, 0, 800, 600}, []ref{{":1.5", "/a"}}, map[string][4]int32{"/a": {1, 1, 9, 9}}, map[string]uint32{"/a": roleLink}, false))
223	265		c := newMockconnector(t)
224	266		c.EXPECT().Connect(mock.Anything).RunAndReturn(func(context.Context) (transport, error) {
225	267		<-release
226		-	return tr, nil
	268	+	return nil, errors.New("down")
227	269		}).Once()
228		-	s := &Source{dial: c}
229		-	t0 := time.Now()
	270	+	c.EXPECT().Connect(mock.Anything).Return(tr, nil).Once()
	271	+	s := &Source{dial: c, wait: testWait}
230	272		targets, _, err := s.Targets(context.Background(), oneWindow, false)
231	273		require.NoError(t, err)
232	274		require.Empty(t, targets)
233		-	require.Less(t, time.Since(t0), connectBudget+200*time.Millisecond)
234		-	// The late result is closed with the source.
	275	+	close(release)
	276	+	require.Eventually(t, func() bool { return len(s.attempt) == 1 }, time.Second, time.Millisecond)
	277	+	// The error is collected; the backoff runs from its finish time.
	278	+	targets, _, err = s.Targets(context.Background(), oneWindow, false)
	279	+	require.NoError(t, err)
	280	+	require.Empty(t, targets)
	281	+	require.Nil(t, s.attempt)
	282	+	require.WithinDuration(t, time.Now().Add(retryEvery), s.retryAt, time.Second)
	283	+	targets, _, err = s.Targets(context.Background(), oneWindow, false) // backoff: no Connect
	284	+	require.NoError(t, err)
	285	+	require.Empty(t, targets)
	286	+	s.retryAt = time.Now().Add(-time.Second)
	287	+	targets, _, err = s.Targets(context.Background(), oneWindow, false)
	288	+	require.NoError(t, err)
	289	+	require.Len(t, targets, 1)
	290	+	}
	291	+
	292	+	// Close with an attempt in flight closes the late transport; afterwards
	293	+	// Targets finds nothing and never connects.
	294	+	func TestCloseReapsLateTransport(t *testing.T) {
	295	+	release := make(chan struct{})
	296	+	closed := make(chan struct{})
	297	+	tr := newMocktransport(t)
	298	+	tr.EXPECT().Close().RunAndReturn(func() error { close(closed); return nil }).Once()
	299	+	c := blockedConnector(t, release, tr, nil, 1)
	300	+	s := &Source{dial: c, wait: testWait}
	301	+	_, _, err := s.Targets(context.Background(), oneWindow, false)
	302	+	require.NoError(t, err)
235	303		require.NoError(t, s.Close())
236	304		close(release)
237		-	require.Eventually(t, func() bool { return tr.AssertExpectations(&testing.T{}) }, time.Second, 5*time.Millisecond)
	305	+	select {
	306	+	case <-closed:
	307	+	case <-time.After(5 * time.Second):
	308	+	t.Fatal("late transport not closed")
	309	+	}
	310	+	targets, covered, err := s.Targets(context.Background(), oneWindow, false)
	311	+	require.NoError(t, err)
	312	+	require.Empty(t, targets)
	313	+	require.Empty(t, covered)
	314	+	require.NoError(t, s.Close()) // idempotent
238	315		}


@@ -215,24 +215,101 @@ func TestBrokenConnectionIsDropped(t *testing.T) {
215		require.Empty(t, targets) // backoff: no new attempt yet	215		require.Empty(t, targets) // backoff: no new attempt yet
216		}	216		}
217			217
218	-	// A silent service cannot hold Targets beyond the connect budget.	218	+	const testWait = 20 * time.Millisecond
219	-	func TestConnectIsBounded(t *testing.T) {	219	+
			220	+	// blockedConnector returns a connector whose Connect blocks until release is
			221	+	func blockedConnector(t testing.T, release <-chan struct{}, tr transport, err error, times int) mockconnector {
			222	+	t.Helper()
			223	+	c := newMockconnector(t)
			224	+	c.EXPECT().Connect(mock.Anything).RunAndReturn(func(context.Context) (transport, error) {
			225	+	<-release
			226	+	return tr, err
			227	+	}).Times(times)
			228	+	return c
			229	+	}
			230	+
			231	+	// A silent service cannot hold Targets beyond the wait, is not doubled while
			232	+	// in flight, and its late transport is used.
			233	+	func TestConnectIsBoundedAndNotDoubled(t *testing.T) {
220		release := make(chan struct{})	234		release := make(chan struct{})
221		tr := newMocktransport(t)	235		tr := newMocktransport(t)
222	-	tr.EXPECT().Close().Return(nil).Once()	236	+	tr.EXPECT().Do(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(answerTree(
			237	+	[4]int32{0, 0, 800, 600}, []ref{{":1.5", "/a"}}, map[string][4]int32{"/a": {1, 1, 9, 9}}, map[string]uint32{"/a": roleLink}, false))
			238	+	c := blockedConnector(t, release, tr, nil, 1)
			239	+	s := &Source{dial: c, wait: testWait}
			240	+	t0 := time.Now()
			241	+	targets, _, err := s.Targets(context.Background(), oneWindow, false)
			242	+	require.NoError(t, err)
			243	+	require.Empty(t, targets)
			244	+	require.Less(t, time.Since(t0), testWait+200*time.Millisecond)
			245	+	require.NotNil(t, s.attempt)
			246	+	// Still in flight: no second Connect (Times(1)), and no blocking.
			247	+	targets, _, err = s.Targets(context.Background(), oneWindow, false)
			248	+	require.NoError(t, err)
			249	+	require.Empty(t, targets)
			250	+	// It connects late: the next call uses it.
			251	+	close(release)
			252	+	require.Eventually(t, func() bool { return len(s.attempt) == 1 }, time.Second, time.Millisecond)
			253	+	targets, _, err = s.Targets(context.Background(), oneWindow, false)
			254	+	require.NoError(t, err)
			255	+	require.Len(t, targets, 1)
			256	+	require.Nil(t, s.attempt)
			257	+	}
			258	+
			259	+	// An abandoned attempt that fails applies the backoff, then one new attempt.
			260	+	func TestAbandonedAttemptErrorBacksOff(t *testing.T) {
			261	+	release := make(chan struct{})
			262	+	tr := newMocktransport(t)
			263	+	tr.EXPECT().Do(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(answerTree(
			264	+	[4]int32{0, 0, 800, 600}, []ref{{":1.5", "/a"}}, map[string][4]int32{"/a": {1, 1, 9, 9}}, map[string]uint32{"/a": roleLink}, false))
223		c := newMockconnector(t)	265		c := newMockconnector(t)
224		c.EXPECT().Connect(mock.Anything).RunAndReturn(func(context.Context) (transport, error) {	266		c.EXPECT().Connect(mock.Anything).RunAndReturn(func(context.Context) (transport, error) {
225		<-release	267		<-release
226	-	return tr, nil	268	+	return nil, errors.New("down")
227		}).Once()	269		}).Once()
228	-	s := &Source{dial: c}	270	+	c.EXPECT().Connect(mock.Anything).Return(tr, nil).Once()
229	-	t0 := time.Now()	271	+	s := &Source{dial: c, wait: testWait}
230		targets, _, err := s.Targets(context.Background(), oneWindow, false)	272		targets, _, err := s.Targets(context.Background(), oneWindow, false)
231		require.NoError(t, err)	273		require.NoError(t, err)
232		require.Empty(t, targets)	274		require.Empty(t, targets)
233	-	require.Less(t, time.Since(t0), connectBudget+200*time.Millisecond)	275	+	close(release)
234	-	// The late result is closed with the source.	276	+	require.Eventually(t, func() bool { return len(s.attempt) == 1 }, time.Second, time.Millisecond)
			277	+	// The error is collected; the backoff runs from its finish time.
			278	+	targets, _, err = s.Targets(context.Background(), oneWindow, false)
			279	+	require.NoError(t, err)
			280	+	require.Empty(t, targets)
			281	+	require.Nil(t, s.attempt)
			282	+	require.WithinDuration(t, time.Now().Add(retryEvery), s.retryAt, time.Second)
			283	+	targets, _, err = s.Targets(context.Background(), oneWindow, false) // backoff: no Connect
			284	+	require.NoError(t, err)
			285	+	require.Empty(t, targets)
			286	+	s.retryAt = time.Now().Add(-time.Second)
			287	+	targets, _, err = s.Targets(context.Background(), oneWindow, false)
			288	+	require.NoError(t, err)
			289	+	require.Len(t, targets, 1)
			290	+	}
			291	+
			292	+	// Close with an attempt in flight closes the late transport; afterwards
			293	+	// Targets finds nothing and never connects.
			294	+	func TestCloseReapsLateTransport(t *testing.T) {
			295	+	release := make(chan struct{})
			296	+	closed := make(chan struct{})
			297	+	tr := newMocktransport(t)
			298	+	tr.EXPECT().Close().RunAndReturn(func() error { close(closed); return nil }).Once()
			299	+	c := blockedConnector(t, release, tr, nil, 1)
			300	+	s := &Source{dial: c, wait: testWait}
			301	+	_, _, err := s.Targets(context.Background(), oneWindow, false)
			302	+	require.NoError(t, err)
235		require.NoError(t, s.Close())	303		require.NoError(t, s.Close())
236		close(release)	304		close(release)
237	-	require.Eventually(t, func() bool { return tr.AssertExpectations(&testing.T{}) }, time.Second, 5*time.Millisecond)	305	+	select {
			306	+	case <-closed:
			307	+	case <-time.After(5 * time.Second):
			308	+	t.Fatal("late transport not closed")
			309	+	}
			310	+	targets, covered, err := s.Targets(context.Background(), oneWindow, false)
			311	+	require.NoError(t, err)
			312	+	require.Empty(t, targets)
			313	+	require.Empty(t, covered)
			314	+	require.NoError(t, s.Close()) // idempotent
238		}	315		}

M internal/adapters/hints/hints.go +7 -0


@@ -43,6 +43,9 @@ func Run(ctx context.Context, requests <-chan ports.HintScanRequest, captures ch
43	43		// returns the targets (logical output-local) and the windows they cover.
44	44		type targetSource interface {
45	45		Targets(ctx context.Context, windows []ports.HintWindow, linksOnly bool) ([]ports.HintTarget, map[ports.WindowID]bool, error)
	46	+	// Close releases the source. It is called once, by the worker goroutine,
	47	+	// when the worker returns; the source is not safe for concurrent use.
	48	+	Close() error
46	49		}
47	50
48	51		// atspiResult is what the source found for one request.

@@ -77,6 +80,10 @@ type frame struct {
77	80
78	81		func run(ctx context.Context, requests <-chan ports.HintScanRequest, captures chan<- ports.CaptureRequest, captured <-chan ports.CaptureDone, results chan<- ports.HintScanResult, timeout time.Duration, opts Options) error {
79	82		w := &worker{dumpDir: opts.DumpDir, source: opts.Source}
	83	+	if opts.Source != nil {
	84	+	// The worker goroutine is the only one to touch the source.
	85	+	defer func() { _ = opts.Source.Close() }()
	86	+	}
80	87		for {
81	88		select {
82	89		case <-ctx.Done():


@@ -43,6 +43,9 @@ func Run(ctx context.Context, requests <-chan ports.HintScanRequest, captures ch
43	// returns the targets (logical output-local) and the windows they cover.	43		// returns the targets (logical output-local) and the windows they cover.
44	type targetSource interface {	44		type targetSource interface {
45	Targets(ctx context.Context, windows []ports.HintWindow, linksOnly bool) ([]ports.HintTarget, map[ports.WindowID]bool, error)	45		Targets(ctx context.Context, windows []ports.HintWindow, linksOnly bool) ([]ports.HintTarget, map[ports.WindowID]bool, error)
		46	+	// Close releases the source. It is called once, by the worker goroutine,
		47	+	// when the worker returns; the source is not safe for concurrent use.
		48	+	Close() error
46	}	49		}
47		50
48	// atspiResult is what the source found for one request.	51		// atspiResult is what the source found for one request.

@@ -77,6 +80,10 @@ type frame struct {
77		80
78	func run(ctx context.Context, requests <-chan ports.HintScanRequest, captures chan<- ports.CaptureRequest, captured <-chan ports.CaptureDone, results chan<- ports.HintScanResult, timeout time.Duration, opts Options) error {	81		func run(ctx context.Context, requests <-chan ports.HintScanRequest, captures chan<- ports.CaptureRequest, captured <-chan ports.CaptureDone, results chan<- ports.HintScanResult, timeout time.Duration, opts Options) error {
79	w := &worker{dumpDir: opts.DumpDir, source: opts.Source}	82		w := &worker{dumpDir: opts.DumpDir, source: opts.Source}
		83	+	if opts.Source != nil {
		84	+	// The worker goroutine is the only one to touch the source.
		85	+	defer func() { _ = opts.Source.Close() }()
		86	+	}
80	for {	87		for {
81	select {	88		select {
82	case <-ctx.Done():	89		case <-ctx.Done():

M internal/adapters/hints/hints_test.go +3 -0


@@ -305,6 +305,7 @@ func TestHintDumpRetention(t *testing.T) {
305	305		func TestScanMergesSource(t *testing.T) {
306	306		pix, w, hgt := loadBGRA(t)
307	307		src := newMocktargetSource(t)
	308	+	src.EXPECT().Close().Return(nil).Once() // the worker closes its source on return
308	309		wins := []ports.HintWindow{{ID: 4, PID: 9, Rect: ports.Rect{X: 0, Y: 0, W: w, H: hgt}}}
309	310		at := []ports.HintTarget{{Rect: ports.Rect{X: 5, Y: 5, W: 30, H: 10}, Kind: ports.HintLink}}
310	311		src.EXPECT().Targets(mock.Anything, wins, false).Return(at, map[ports.WindowID]bool{4: true}, nil).Once()

@@ -327,6 +328,7 @@ func TestScanSourceErrorUsesOCR(t *testing.T) {
327	328		}
328	329		pix, w, hgt := loadBGRA(t)
329	330		src := newMocktargetSource(t)
	331	+	src.EXPECT().Close().Return(nil).Once() // the worker closes its source on return
330	332		wins := []ports.HintWindow{{ID: 4, PID: 9, Rect: ports.Rect{W: w, H: hgt}}}
331	333		src.EXPECT().Targets(mock.Anything, wins, false).Return(nil, nil, errors.New("bus gone")).Once()
332	334		h := startWith(t, 5*time.Second, Options{Source: src})

@@ -394,6 +396,7 @@ func TestNoTesseractSkipsCapture(t *testing.T) {
394	396		covered := ports.HintTarget{Kind: ports.HintLink, Rect: ports.Rect{X: 30, W: 5, H: 5}}
395	397		wins := []ports.HintWindow{{ID: 1, Rect: ports.Rect{W: 20, H: 20}}, {ID: 2, Rect: ports.Rect{X: 20, W: 20, H: 20}}}
396	398		src := newMocktargetSource(t)
	399	+	src.EXPECT().Close().Return(nil).Once() // the worker closes its source on return
397	400		src.EXPECT().Targets(mock.Anything, wins, false).Return([]ports.HintTarget{button, covered}, map[ports.WindowID]bool{2: true}, nil).Times(2)
398	401		h := startWith(t, time.Second, Options{Source: src})
399	402		for seq := uint64(1); seq <= 2; seq++ {


@@ -305,6 +305,7 @@ func TestHintDumpRetention(t *testing.T) {
305	func TestScanMergesSource(t *testing.T) {	305		func TestScanMergesSource(t *testing.T) {
306	pix, w, hgt := loadBGRA(t)	306		pix, w, hgt := loadBGRA(t)
307	src := newMocktargetSource(t)	307		src := newMocktargetSource(t)
		308	+	src.EXPECT().Close().Return(nil).Once() // the worker closes its source on return
308	wins := []ports.HintWindow{{ID: 4, PID: 9, Rect: ports.Rect{X: 0, Y: 0, W: w, H: hgt}}}	309		wins := []ports.HintWindow{{ID: 4, PID: 9, Rect: ports.Rect{X: 0, Y: 0, W: w, H: hgt}}}
309	at := []ports.HintTarget{{Rect: ports.Rect{X: 5, Y: 5, W: 30, H: 10}, Kind: ports.HintLink}}	310		at := []ports.HintTarget{{Rect: ports.Rect{X: 5, Y: 5, W: 30, H: 10}, Kind: ports.HintLink}}
310	src.EXPECT().Targets(mock.Anything, wins, false).Return(at, map[ports.WindowID]bool{4: true}, nil).Once()	311		src.EXPECT().Targets(mock.Anything, wins, false).Return(at, map[ports.WindowID]bool{4: true}, nil).Once()

@@ -327,6 +328,7 @@ func TestScanSourceErrorUsesOCR(t *testing.T) {
327	}	328		}
328	pix, w, hgt := loadBGRA(t)	329		pix, w, hgt := loadBGRA(t)
329	src := newMocktargetSource(t)	330		src := newMocktargetSource(t)
		331	+	src.EXPECT().Close().Return(nil).Once() // the worker closes its source on return
330	wins := []ports.HintWindow{{ID: 4, PID: 9, Rect: ports.Rect{W: w, H: hgt}}}	332		wins := []ports.HintWindow{{ID: 4, PID: 9, Rect: ports.Rect{W: w, H: hgt}}}
331	src.EXPECT().Targets(mock.Anything, wins, false).Return(nil, nil, errors.New("bus gone")).Once()	333		src.EXPECT().Targets(mock.Anything, wins, false).Return(nil, nil, errors.New("bus gone")).Once()
332	h := startWith(t, 5*time.Second, Options{Source: src})	334		h := startWith(t, 5*time.Second, Options{Source: src})

@@ -394,6 +396,7 @@ func TestNoTesseractSkipsCapture(t *testing.T) {
394	covered := ports.HintTarget{Kind: ports.HintLink, Rect: ports.Rect{X: 30, W: 5, H: 5}}	396		covered := ports.HintTarget{Kind: ports.HintLink, Rect: ports.Rect{X: 30, W: 5, H: 5}}
395	wins := []ports.HintWindow{{ID: 1, Rect: ports.Rect{W: 20, H: 20}}, {ID: 2, Rect: ports.Rect{X: 20, W: 20, H: 20}}}	397		wins := []ports.HintWindow{{ID: 1, Rect: ports.Rect{W: 20, H: 20}}, {ID: 2, Rect: ports.Rect{X: 20, W: 20, H: 20}}}
396	src := newMocktargetSource(t)	398		src := newMocktargetSource(t)
		399	+	src.EXPECT().Close().Return(nil).Once() // the worker closes its source on return
397	src.EXPECT().Targets(mock.Anything, wins, false).Return([]ports.HintTarget{button, covered}, map[ports.WindowID]bool{2: true}, nil).Times(2)	400		src.EXPECT().Targets(mock.Anything, wins, false).Return([]ports.HintTarget{button, covered}, map[ports.WindowID]bool{2: true}, nil).Times(2)
398	h := startWith(t, time.Second, Options{Source: src})	401		h := startWith(t, time.Second, Options{Source: src})
399	for seq := uint64(1); seq <= 2; seq++ {	402		for seq := uint64(1); seq <= 2; seq++ {

M internal/adapters/hints/target_source_mock_test.go +44 -0


@@ -47,6 +47,50 @@ func (_m mocktargetSource) EXPECT() mocktargetSource_Expecter {
47	47		return &mocktargetSource_Expecter{mock: &_m.Mock}
48	48		}
49	49
	50	+	// Close provides a mock function for the type mocktargetSource
	51	+	func (_mock *mocktargetSource) Close() error {
	52	+	ret := _mock.Called()
	53	+
	54	+	if len(ret) == 0 {
	55	+	panic("no return value specified for Close")
	56	+	}
	57	+
	58	+	var r0 error
	59	+	if returnFunc, ok := ret.Get(0).(func() error); ok {
	60	+	r0 = returnFunc()
	61	+	} else {
	62	+	r0 = ret.Error(0)
	63	+	}
	64	+	return r0
	65	+	}
	66	+
	67	+	// mocktargetSource_Close_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Close'
	68	+	type mocktargetSource_Close_Call struct {
	69	+	*mock.Call
	70	+	}
	71	+
	72	+	// Close is a helper method to define mock.On call
	73	+	func (_e mocktargetSource_Expecter) Close() mocktargetSource_Close_Call {
	74	+	return &mocktargetSource_Close_Call{Call: _e.mock.On("Close")}
	75	+	}
	76	+
	77	+	func (_c mocktargetSource_Close_Call) Run(run func()) mocktargetSource_Close_Call {
	78	+	_c.Call.Run(func(args mock.Arguments) {
	79	+	run()
	80	+	})
	81	+	return _c
	82	+	}
	83	+
	84	+	func (_c mocktargetSource_Close_Call) Return(err error) mocktargetSource_Close_Call {
	85	+	_c.Call.Return(err)
	86	+	return _c
	87	+	}
	88	+
	89	+	func (_c mocktargetSource_Close_Call) RunAndReturn(run func() error) mocktargetSource_Close_Call {
	90	+	_c.Call.Return(run)
	91	+	return _c
	92	+	}
	93	+
50	94		// Targets provides a mock function for the type mocktargetSource
51	95		func (_mock *mocktargetSource) Targets(ctx context.Context, windows []ports.HintWindow, linksOnly bool) ([]ports.HintTarget, map[ports.WindowID]bool, error) {
52	96		ret := _mock.Called(ctx, windows, linksOnly)


@@ -47,6 +47,50 @@ func (_m mocktargetSource) EXPECT() mocktargetSource_Expecter {
47	return &mocktargetSource_Expecter{mock: &_m.Mock}	47		return &mocktargetSource_Expecter{mock: &_m.Mock}
48	}	48		}
49		49
		50	+	// Close provides a mock function for the type mocktargetSource
		51	+	func (_mock *mocktargetSource) Close() error {
		52	+	ret := _mock.Called()
		53	+
		54	+	if len(ret) == 0 {
		55	+	panic("no return value specified for Close")
		56	+	}
		57	+
		58	+	var r0 error
		59	+	if returnFunc, ok := ret.Get(0).(func() error); ok {
		60	+	r0 = returnFunc()
		61	+	} else {
		62	+	r0 = ret.Error(0)
		63	+	}
		64	+	return r0
		65	+	}
		66	+
		67	+	// mocktargetSource_Close_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Close'
		68	+	type mocktargetSource_Close_Call struct {
		69	+	*mock.Call
		70	+	}
		71	+
		72	+	// Close is a helper method to define mock.On call
		73	+	func (_e mocktargetSource_Expecter) Close() mocktargetSource_Close_Call {
		74	+	return &mocktargetSource_Close_Call{Call: _e.mock.On("Close")}
		75	+	}
		76	+
		77	+	func (_c mocktargetSource_Close_Call) Run(run func()) mocktargetSource_Close_Call {
		78	+	_c.Call.Run(func(args mock.Arguments) {
		79	+	run()
		80	+	})
		81	+	return _c
		82	+	}
		83	+
		84	+	func (_c mocktargetSource_Close_Call) Return(err error) mocktargetSource_Close_Call {
		85	+	_c.Call.Return(err)
		86	+	return _c
		87	+	}
		88	+
		89	+	func (_c mocktargetSource_Close_Call) RunAndReturn(run func() error) mocktargetSource_Close_Call {
		90	+	_c.Call.Return(run)
		91	+	return _c
		92	+	}
		93	+
50	// Targets provides a mock function for the type mocktargetSource	94		// Targets provides a mock function for the type mocktargetSource
51	func (_mock *mocktargetSource) Targets(ctx context.Context, windows []ports.HintWindow, linksOnly bool) ([]ports.HintTarget, map[ports.WindowID]bool, error) {	95		func (_mock *mocktargetSource) Targets(ctx context.Context, windows []ports.HintWindow, linksOnly bool) ([]ports.HintTarget, map[ports.WindowID]bool, error) {
52	ret := _mock.Called(ctx, windows, linksOnly)	96		ret := _mock.Called(ctx, windows, linksOnly)

M internal/app/hintdebug.go +2 -2


@@ -16,9 +16,9 @@ import (
16	16		// by the first scan; without it the worker reads windows with OCR only.
17	17		func hintOptions(ctx context.Context) hints.Options {
18	18		opts := hintDump(ctx)
19		-	// The source connects on its first scan, and again after a failure.
	19	+	// The source connects on its first scan, and again after a failure. The
	20	+	// hints worker owns it and closes it when it returns.
20	21		src := atspi.New()
21		-	context.AfterFunc(ctx, func() { _ = src.Close() })
22	22		opts.Source = src
23	23		return opts
24	24		}

 // by the first scan; without it the worker reads windows with OCR only.
 func hintOptions(ctx context.Context) hints.Options {
 	opts := hintDump(ctx)
-	// The source connects on its first scan, and again after a failure.
+	// The source connects on its first scan, and again after a failure. The
+	// hints worker owns it and closes it when it returns.
 	src := atspi.New()
-	context.AfterFunc(ctx, func() { _ = src.Close() })
 	opts.Source = src
 	return opts
 }