fix(atspi): connect lazily with bounded deadlines and reconnect
- commit
- 3bb35a758880bcfee1836ef9955aa56ab78d3655
- author
- brice · · 2026-10-19 15:34 UTC
- parent
- 3e96bd1
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
7 files · +446 -41
M .mockery.yml +5 -0
| @@ -125,6 +125,11 @@ packages: | |||
| 125 | 125 | dir: internal/adapters/atspi | |
| 126 | 126 | pkgname: atspi | |
| 127 | 127 | filename: transport_mock_test.go | |
| 128 | + | connector: | |
| 129 | + | config: | |
| 130 | + | dir: internal/adapters/atspi | |
| 131 | + | pkgname: atspi | |
| 132 | + | filename: connector_mock_test.go | |
| 128 | 133 | github.com/bnema/neferwl/internal/adapters/screencast: | |
| 129 | 134 | interfaces: | |
| 130 | 135 | bus: | |
| @@ -125,6 +125,11 @@ packages: | |||||
| 125 | dir: internal/adapters/atspi | 125 | dir: internal/adapters/atspi | ||
| 126 | pkgname: atspi | 126 | pkgname: atspi | ||
| 127 | filename: transport_mock_test.go | 127 | filename: transport_mock_test.go | ||
| 128 | + | connector: | |||
| 129 | + | config: | |||
| 130 | + | dir: internal/adapters/atspi | |||
| 131 | + | pkgname: atspi | |||
| 132 | + | filename: connector_mock_test.go | |||
| 128 | github.com/bnema/neferwl/internal/adapters/screencast: | 133 | github.com/bnema/neferwl/internal/adapters/screencast: | ||
| 129 | interfaces: | 134 | interfaces: | ||
| 130 | bus: | 135 | bus: | ||
M internal/adapters/atspi/bus_test.go +31 -0
| @@ -164,3 +164,34 @@ func TestDoTimesOutOnSilentApp(t *testing.T) { | |||
| 164 | 164 | require.ErrorIs(t, rep[0].err, ErrTimeout) | |
| 165 | 165 | require.Less(t, time.Since(t0), 500*time.Millisecond) | |
| 166 | 166 | } | |
| 167 | + | ||
| 168 | + | // A call left unanswered for idle gets ErrTimeout without breaking the | |
| 169 | + | // connection; a batch that hits its deadline breaks it, and Do returns near | |
| 170 | + | // the deadline even though the idle timeout is longer. | |
| 171 | + | func TestDoDeadlineBreaksConnection(t *testing.T) { | |
| 172 | + | addr := startBus(t) | |
| 173 | + | app, err := zerobus.Dial(addr) | |
| 174 | + | require.NoError(t, err) | |
| 175 | + | t.Cleanup(func() { _ = app.Close() }) | |
| 176 | + | _, err = app.RequestName("org.test.Mute", 0) | |
| 177 | + | require.NoError(t, err) | |
| 178 | + | ||
| 179 | + | cc, err := zerobus.Dial(addr) | |
| 180 | + | require.NoError(t, err) | |
| 181 | + | b, err := newConn(cc) | |
| 182 | + | require.NoError(t, err) | |
| 183 | + | t.Cleanup(func() { _ = b.Close() }) | |
| 184 | + | ||
| 185 | + | mute := []call{{op: opRole, name: "org.test.Mute", path: "/x"}} | |
| 186 | + | rep := b.Do(time.Now().Add(time.Second), 40*time.Millisecond, mute) | |
| 187 | + | require.ErrorIs(t, rep[0].err, ErrTimeout) | |
| 188 | + | require.NotErrorIs(t, rep[0].err, ErrBroken) | |
| 189 | + | ||
| 190 | + | t0 := time.Now() | |
| 191 | + | rep = b.Do(time.Now().Add(60*time.Millisecond), time.Second, mute) | |
| 192 | + | require.ErrorIs(t, rep[0].err, ErrBroken) | |
| 193 | + | require.Less(t, time.Since(t0), 300*time.Millisecond) | |
| 194 | + | // The socket is closed: the next batch fails at once. | |
| 195 | + | rep = b.Do(time.Now().Add(time.Second), time.Second, mute) | |
| 196 | + | require.ErrorIs(t, rep[0].err, ErrBroken) | |
| 197 | + | } | |
| @@ -164,3 +164,34 @@ func TestDoTimesOutOnSilentApp(t *testing.T) { | |||||
| 164 | require.ErrorIs(t, rep[0].err, ErrTimeout) | 164 | require.ErrorIs(t, rep[0].err, ErrTimeout) | ||
| 165 | require.Less(t, time.Since(t0), 500*time.Millisecond) | 165 | require.Less(t, time.Since(t0), 500*time.Millisecond) | ||
| 166 | } | 166 | } | ||
| 167 | + | ||||
| 168 | + | // A call left unanswered for idle gets ErrTimeout without breaking the | |||
| 169 | + | // connection; a batch that hits its deadline breaks it, and Do returns near | |||
| 170 | + | // the deadline even though the idle timeout is longer. | |||
| 171 | + | func TestDoDeadlineBreaksConnection(t *testing.T) { | |||
| 172 | + | addr := startBus(t) | |||
| 173 | + | app, err := zerobus.Dial(addr) | |||
| 174 | + | require.NoError(t, err) | |||
| 175 | + | t.Cleanup(func() { _ = app.Close() }) | |||
| 176 | + | _, err = app.RequestName("org.test.Mute", 0) | |||
| 177 | + | require.NoError(t, err) | |||
| 178 | + | ||||
| 179 | + | cc, err := zerobus.Dial(addr) | |||
| 180 | + | require.NoError(t, err) | |||
| 181 | + | b, err := newConn(cc) | |||
| 182 | + | require.NoError(t, err) | |||
| 183 | + | t.Cleanup(func() { _ = b.Close() }) | |||
| 184 | + | ||||
| 185 | + | mute := []call{{op: opRole, name: "org.test.Mute", path: "/x"}} | |||
| 186 | + | rep := b.Do(time.Now().Add(time.Second), 40*time.Millisecond, mute) | |||
| 187 | + | require.ErrorIs(t, rep[0].err, ErrTimeout) | |||
| 188 | + | require.NotErrorIs(t, rep[0].err, ErrBroken) | |||
| 189 | + | ||||
| 190 | + | t0 := time.Now() | |||
| 191 | + | rep = b.Do(time.Now().Add(60*time.Millisecond), time.Second, mute) | |||
| 192 | + | require.ErrorIs(t, rep[0].err, ErrBroken) | |||
| 193 | + | require.Less(t, time.Since(t0), 300*time.Millisecond) | |||
| 194 | + | // The socket is closed: the next batch fails at once. | |||
| 195 | + | rep = b.Do(time.Now().Add(time.Second), time.Second, mute) | |||
| 196 | + | require.ErrorIs(t, rep[0].err, ErrBroken) | |||
| 197 | + | } | |||
A internal/adapters/atspi/connector_mock_test.go +109 -0
| @@ -0,0 +1,109 @@ | |||
| 1 | + | // Code generated by mockery; DO NOT EDIT. | |
| 2 | + | // github.com/vektra/mockery | |
| 3 | + | // template: testify | |
| 4 | + | ||
| 5 | + | package atspi | |
| 6 | + | ||
| 7 | + | import ( | |
| 8 | + | "context" | |
| 9 | + | ||
| 10 | + | mock "github.com/stretchr/testify/mock" | |
| 11 | + | ) | |
| 12 | + | ||
| 13 | + | // newMockconnector creates a new instance of mockconnector. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. | |
| 14 | + | // The first argument is typically a *testing.T value. | |
| 15 | + | func newMockconnector(t interface { | |
| 16 | + | mock.TestingT | |
| 17 | + | Cleanup(func()) | |
| 18 | + | }) *mockconnector { | |
| 19 | + | if helper, ok := t.(interface{ Helper() }); ok { | |
| 20 | + | helper.Helper() | |
| 21 | + | } | |
| 22 | + | ||
| 23 | + | mock := &mockconnector{} | |
| 24 | + | mock.Mock.Test(t) | |
| 25 | + | ||
| 26 | + | t.Cleanup(func() { | |
| 27 | + | if helper, ok := t.(interface{ Helper() }); ok { | |
| 28 | + | helper.Helper() | |
| 29 | + | } | |
| 30 | + | mock.AssertExpectations(t) | |
| 31 | + | }) | |
| 32 | + | ||
| 33 | + | return mock | |
| 34 | + | } | |
| 35 | + | ||
| 36 | + | // mockconnector is an autogenerated mock type for the connector type | |
| 37 | + | type mockconnector struct { | |
| 38 | + | mock.Mock | |
| 39 | + | } | |
| 40 | + | ||
| 41 | + | type mockconnector_Expecter struct { | |
| 42 | + | mock *mock.Mock | |
| 43 | + | } | |
| 44 | + | ||
| 45 | + | func (_m *mockconnector) EXPECT() *mockconnector_Expecter { | |
| 46 | + | return &mockconnector_Expecter{mock: &_m.Mock} | |
| 47 | + | } | |
| 48 | + | ||
| 49 | + | // Connect provides a mock function for the type mockconnector | |
| 50 | + | func (_mock *mockconnector) Connect(ctx context.Context) (transport, error) { | |
| 51 | + | ret := _mock.Called(ctx) | |
| 52 | + | ||
| 53 | + | if len(ret) == 0 { | |
| 54 | + | panic("no return value specified for Connect") | |
| 55 | + | } | |
| 56 | + | ||
| 57 | + | var r0 transport | |
| 58 | + | var r1 error | |
| 59 | + | if returnFunc, ok := ret.Get(0).(func(context.Context) (transport, error)); ok { | |
| 60 | + | return returnFunc(ctx) | |
| 61 | + | } | |
| 62 | + | if returnFunc, ok := ret.Get(0).(func(context.Context) transport); ok { | |
| 63 | + | r0 = returnFunc(ctx) | |
| 64 | + | } else { | |
| 65 | + | if ret.Get(0) != nil { | |
| 66 | + | r0 = ret.Get(0).(transport) | |
| 67 | + | } | |
| 68 | + | } | |
| 69 | + | if returnFunc, ok := ret.Get(1).(func(context.Context) error); ok { | |
| 70 | + | r1 = returnFunc(ctx) | |
| 71 | + | } else { | |
| 72 | + | r1 = ret.Error(1) | |
| 73 | + | } | |
| 74 | + | return r0, r1 | |
| 75 | + | } | |
| 76 | + | ||
| 77 | + | // mockconnector_Connect_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Connect' | |
| 78 | + | type mockconnector_Connect_Call struct { | |
| 79 | + | *mock.Call | |
| 80 | + | } | |
| 81 | + | ||
| 82 | + | // Connect is a helper method to define mock.On call | |
| 83 | + | // - ctx context.Context | |
| 84 | + | func (_e *mockconnector_Expecter) Connect(ctx any) *mockconnector_Connect_Call { | |
| 85 | + | return &mockconnector_Connect_Call{Call: _e.mock.On("Connect", ctx)} | |
| 86 | + | } | |
| 87 | + | ||
| 88 | + | func (_c *mockconnector_Connect_Call) Run(run func(ctx context.Context)) *mockconnector_Connect_Call { | |
| 89 | + | _c.Call.Run(func(args mock.Arguments) { | |
| 90 | + | var arg0 context.Context | |
| 91 | + | if args[0] != nil { | |
| 92 | + | arg0 = args[0].(context.Context) | |
| 93 | + | } | |
| 94 | + | run( | |
| 95 | + | arg0, | |
| 96 | + | ) | |
| 97 | + | }) | |
| 98 | + | return _c | |
| 99 | + | } | |
| 100 | + | ||
| 101 | + | func (_c *mockconnector_Connect_Call) Return(transportMoqParam transport, err error) *mockconnector_Connect_Call { | |
| 102 | + | _c.Call.Return(transportMoqParam, err) | |
| 103 | + | return _c | |
| 104 | + | } | |
| 105 | + | ||
| 106 | + | func (_c *mockconnector_Connect_Call) RunAndReturn(run func(ctx context.Context) (transport, error)) *mockconnector_Connect_Call { | |
| 107 | + | _c.Call.Return(run) | |
| 108 | + | return _c | |
| 109 | + | } | |
| @@ -0,0 +1,109 @@ | |||||
| 1 | + | // Code generated by mockery; DO NOT EDIT. | |||
| 2 | + | // github.com/vektra/mockery | |||
| 3 | + | // template: testify | |||
| 4 | + | ||||
| 5 | + | package atspi | |||
| 6 | + | ||||
| 7 | + | import ( | |||
| 8 | + | "context" | |||
| 9 | + | ||||
| 10 | + | mock "github.com/stretchr/testify/mock" | |||
| 11 | + | ) | |||
| 12 | + | ||||
| 13 | + | // newMockconnector creates a new instance of mockconnector. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. | |||
| 14 | + | // The first argument is typically a *testing.T value. | |||
| 15 | + | func newMockconnector(t interface { | |||
| 16 | + | mock.TestingT | |||
| 17 | + | Cleanup(func()) | |||
| 18 | + | }) *mockconnector { | |||
| 19 | + | if helper, ok := t.(interface{ Helper() }); ok { | |||
| 20 | + | helper.Helper() | |||
| 21 | + | } | |||
| 22 | + | ||||
| 23 | + | mock := &mockconnector{} | |||
| 24 | + | mock.Mock.Test(t) | |||
| 25 | + | ||||
| 26 | + | t.Cleanup(func() { | |||
| 27 | + | if helper, ok := t.(interface{ Helper() }); ok { | |||
| 28 | + | helper.Helper() | |||
| 29 | + | } | |||
| 30 | + | mock.AssertExpectations(t) | |||
| 31 | + | }) | |||
| 32 | + | ||||
| 33 | + | return mock | |||
| 34 | + | } | |||
| 35 | + | ||||
| 36 | + | // mockconnector is an autogenerated mock type for the connector type | |||
| 37 | + | type mockconnector struct { | |||
| 38 | + | mock.Mock | |||
| 39 | + | } | |||
| 40 | + | ||||
| 41 | + | type mockconnector_Expecter struct { | |||
| 42 | + | mock *mock.Mock | |||
| 43 | + | } | |||
| 44 | + | ||||
| 45 | + | func (_m *mockconnector) EXPECT() *mockconnector_Expecter { | |||
| 46 | + | return &mockconnector_Expecter{mock: &_m.Mock} | |||
| 47 | + | } | |||
| 48 | + | ||||
| 49 | + | // Connect provides a mock function for the type mockconnector | |||
| 50 | + | func (_mock *mockconnector) Connect(ctx context.Context) (transport, error) { | |||
| 51 | + | ret := _mock.Called(ctx) | |||
| 52 | + | ||||
| 53 | + | if len(ret) == 0 { | |||
| 54 | + | panic("no return value specified for Connect") | |||
| 55 | + | } | |||
| 56 | + | ||||
| 57 | + | var r0 transport | |||
| 58 | + | var r1 error | |||
| 59 | + | if returnFunc, ok := ret.Get(0).(func(context.Context) (transport, error)); ok { | |||
| 60 | + | return returnFunc(ctx) | |||
| 61 | + | } | |||
| 62 | + | if returnFunc, ok := ret.Get(0).(func(context.Context) transport); ok { | |||
| 63 | + | r0 = returnFunc(ctx) | |||
| 64 | + | } else { | |||
| 65 | + | if ret.Get(0) != nil { | |||
| 66 | + | r0 = ret.Get(0).(transport) | |||
| 67 | + | } | |||
| 68 | + | } | |||
| 69 | + | if returnFunc, ok := ret.Get(1).(func(context.Context) error); ok { | |||
| 70 | + | r1 = returnFunc(ctx) | |||
| 71 | + | } else { | |||
| 72 | + | r1 = ret.Error(1) | |||
| 73 | + | } | |||
| 74 | + | return r0, r1 | |||
| 75 | + | } | |||
| 76 | + | ||||
| 77 | + | // mockconnector_Connect_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Connect' | |||
| 78 | + | type mockconnector_Connect_Call struct { | |||
| 79 | + | *mock.Call | |||
| 80 | + | } | |||
| 81 | + | ||||
| 82 | + | // Connect is a helper method to define mock.On call | |||
| 83 | + | // - ctx context.Context | |||
| 84 | + | func (_e *mockconnector_Expecter) Connect(ctx any) *mockconnector_Connect_Call { | |||
| 85 | + | return &mockconnector_Connect_Call{Call: _e.mock.On("Connect", ctx)} | |||
| 86 | + | } | |||
| 87 | + | ||||
| 88 | + | func (_c *mockconnector_Connect_Call) Run(run func(ctx context.Context)) *mockconnector_Connect_Call { | |||
| 89 | + | _c.Call.Run(func(args mock.Arguments) { | |||
| 90 | + | var arg0 context.Context | |||
| 91 | + | if args[0] != nil { | |||
| 92 | + | arg0 = args[0].(context.Context) | |||
| 93 | + | } | |||
| 94 | + | run( | |||
| 95 | + | arg0, | |||
| 96 | + | ) | |||
| 97 | + | }) | |||
| 98 | + | return _c | |||
| 99 | + | } | |||
| 100 | + | ||||
| 101 | + | func (_c *mockconnector_Connect_Call) Return(transportMoqParam transport, err error) *mockconnector_Connect_Call { | |||
| 102 | + | _c.Call.Return(transportMoqParam, err) | |||
| 103 | + | return _c | |||
| 104 | + | } | |||
| 105 | + | ||||
| 106 | + | func (_c *mockconnector_Connect_Call) RunAndReturn(run func(ctx context.Context) (transport, error)) *mockconnector_Connect_Call { | |||
| 107 | + | _c.Call.Return(run) | |||
| 108 | + | return _c | |||
| 109 | + | } | |||
M internal/adapters/atspi/source.go +135 -15
| @@ -20,22 +20,59 @@ const ( | |||
| 20 | 20 | budget = 300 * time.Millisecond | |
| 21 | 21 | ) | |
| 22 | 22 | ||
| 23 | + | // connectBudget bounds the whole connection setup (session bus, | |
| 24 | + | // GetAddress, IsEnabled, dial of the accessibility bus) of one attempt. | |
| 25 | + | const connectBudget = 300 * time.Millisecond | |
| 26 | + | ||
| 27 | + | // retryEvery is the minimum time between two connection attempts. | |
| 28 | + | const retryEvery = 5 * time.Second | |
| 29 | + | ||
| 30 | + | // connector opens a transport on the accessibility bus. | |
| 31 | + | type connector interface { | |
| 32 | + | Connect(ctx context.Context) (transport, error) | |
| 33 | + | } | |
| 34 | + | ||
| 35 | + | // dialResult is the outcome of a connection attempt. | |
| 36 | + | type dialResult struct { | |
| 37 | + | tr transport | |
| 38 | + | err error | |
| 39 | + | } | |
| 40 | + | ||
| 23 | 41 | // Source reads hint targets from the accessibility tree. It owns one | |
| 24 | 42 | // connection to the accessibility bus and is used by one goroutine at a | |
| 25 | - | // time (the hints worker). | |
| 43 | + | // time (the hints worker). The connection is made on the first Targets call | |
| 44 | + | // that needs it, never at startup, and made again after a failure, at most | |
| 45 | + | // once every retryEvery; meanwhile Targets finds nothing and OCR takes over. | |
| 26 | 46 | type Source struct { | |
| 27 | - | tr transport | |
| 47 | + | tr transport | |
| 48 | + | 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 | |
| 28 | 55 | } | |
| 29 | 56 | ||
| 30 | - | // New enables accessibility on the session and connects to the | |
| 31 | - | // accessibility bus. | |
| 32 | - | func New(ctx context.Context) (*Source, error) { | |
| 57 | + | // New returns a Source. It does not touch the bus: the connection is made by | |
| 58 | + | // the first Targets call. | |
| 59 | + | func New() *Source { return &Source{dial: busConnector{}} } | |
| 60 | + | ||
| 61 | + | // busConnector connects to the session bus, enables accessibility and dials | |
| 62 | + | // the accessibility bus. | |
| 63 | + | type busConnector struct{} | |
| 64 | + | ||
| 65 | + | func (busConnector) Connect(ctx context.Context) (transport, error) { | |
| 33 | 66 | log := logging.For(ctx, "atspi") | |
| 34 | 67 | sess, err := zerobus.SessionBus() | |
| 35 | 68 | if err != nil { | |
| 36 | 69 | return nil, fmt.Errorf("atspi: session bus: %w", err) | |
| 37 | 70 | } | |
| 38 | 71 | defer func() { _ = sess.Close() }() | |
| 72 | + | // Call has no deadline: closing the connection ends the wait for a | |
| 73 | + | // silent service. | |
| 74 | + | stop := time.AfterFunc(connectBudget, func() { _ = sess.Close() }) | |
| 75 | + | defer stop.Stop() | |
| 39 | 76 | ||
| 40 | 77 | sess.NewCall(a11yBusName, a11yBusPath, a11yBusName, "GetAddress", "") | |
| 41 | 78 | m, err := sess.Call() | |
| @@ -70,11 +107,75 @@ func New(ctx context.Context) (*Source, error) { | |||
| 70 | 107 | return nil, err | |
| 71 | 108 | } | |
| 72 | 109 | log.Info().Msg("accessibility bus connected") | |
| 73 | - | return &Source{tr: tr}, nil | |
| 110 | + | return tr, nil | |
| 74 | 111 | } | |
| 75 | 112 | ||
| 76 | - | // Close closes the accessibility bus connection. | |
| 77 | - | func (s *Source) Close() error { return s.tr.Close() } | |
| 113 | + | // 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. | |
| 115 | + | func (s *Source) ensure(ctx context.Context) transport { | |
| 116 | + | if s.tr != nil { | |
| 117 | + | return s.tr | |
| 118 | + | } | |
| 119 | + | if s.dial == nil { | |
| 120 | + | return nil | |
| 121 | + | } | |
| 122 | + | now := time.Now() | |
| 123 | + | if s.attempt == nil { | |
| 124 | + | if now.Before(s.retryAt) { | |
| 125 | + | return nil | |
| 126 | + | } | |
| 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 | + | }() | |
| 135 | + | } | |
| 136 | + | timer := time.NewTimer(time.Until(s.attemptDue)) | |
| 137 | + | defer timer.Stop() | |
| 138 | + | 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 | |
| 147 | + | case <-timer.C: | |
| 148 | + | case <-ctx.Done(): | |
| 149 | + | } | |
| 150 | + | return s.tr | |
| 151 | + | } | |
| 152 | + | ||
| 153 | + | // drop closes a broken connection; the next attempt reconnects. | |
| 154 | + | func (s *Source) drop() { | |
| 155 | + | if s.tr != nil { | |
| 156 | + | _ = s.tr.Close() | |
| 157 | + | s.tr = nil | |
| 158 | + | } | |
| 159 | + | } | |
| 160 | + | ||
| 161 | + | // Close closes the accessibility bus connection, and the one an abandoned | |
| 162 | + | // attempt may still deliver. | |
| 163 | + | func (s *Source) Close() error { | |
| 164 | + | var err error | |
| 165 | + | if s.tr != nil { | |
| 166 | + | err = s.tr.Close() | |
| 167 | + | s.tr = nil | |
| 168 | + | } | |
| 169 | + | if ch := s.attempt; ch != nil { | |
| 170 | + | s.attempt = nil | |
| 171 | + | go func() { | |
| 172 | + | if r := <-ch; r.tr != nil { | |
| 173 | + | _ = r.tr.Close() | |
| 174 | + | } | |
| 175 | + | }() | |
| 176 | + | } | |
| 177 | + | return err | |
| 178 | + | } | |
| 78 | 179 | ||
| 79 | 180 | // appInfo is an application of the registry whose process shows a window. | |
| 80 | 181 | type appInfo struct { | |
| @@ -101,13 +202,6 @@ func (s *Source) Targets(ctx context.Context, windows []ports.HintWindow, linksO | |||
| 101 | 202 | if d, ok := ctx.Deadline(); ok && d.Before(deadline) { | |
| 102 | 203 | deadline = d | |
| 103 | 204 | } | |
| 104 | - | do := func(calls []call) []reply { | |
| 105 | - | if len(calls) == 0 || ctx.Err() != nil { | |
| 106 | - | return make([]reply, len(calls)) | |
| 107 | - | } | |
| 108 | - | return s.tr.Do(deadline, callTimeout, calls) | |
| 109 | - | } | |
| 110 | - | ||
| 111 | 205 | wanted := map[int]bool{} | |
| 112 | 206 | for _, w := range windows { | |
| 113 | 207 | if w.PID > 0 { | |
| @@ -117,6 +211,32 @@ func (s *Source) Targets(ctx context.Context, windows []ports.HintWindow, linksO | |||
| 117 | 211 | if len(wanted) == 0 { | |
| 118 | 212 | return nil, covered, nil | |
| 119 | 213 | } | |
| 214 | + | tr := s.ensure(ctx) | |
| 215 | + | if tr == nil { | |
| 216 | + | return nil, covered, nil | |
| 217 | + | } | |
| 218 | + | broken := false | |
| 219 | + | do := func(calls []call) []reply { | |
| 220 | + | if len(calls) == 0 || ctx.Err() != nil { | |
| 221 | + | return make([]reply, len(calls)) | |
| 222 | + | } | |
| 223 | + | if broken { | |
| 224 | + | out := make([]reply, len(calls)) | |
| 225 | + | for i := range out { | |
| 226 | + | out[i].err = ErrBroken | |
| 227 | + | } | |
| 228 | + | return out | |
| 229 | + | } | |
| 230 | + | out := tr.Do(deadline, callTimeout, calls) | |
| 231 | + | for _, r := range out { | |
| 232 | + | if errors.Is(r.err, ErrBroken) { | |
| 233 | + | broken = true | |
| 234 | + | s.drop() | |
| 235 | + | break | |
| 236 | + | } | |
| 237 | + | } | |
| 238 | + | return out | |
| 239 | + | } | |
| 120 | 240 | ||
| 121 | 241 | // Applications of the registry, kept when their process shows a window. | |
| 122 | 242 | rep := do([]call{{op: opChildren, name: registryName, path: registryRoot}}) | |
| @@ -20,22 +20,59 @@ const ( | |||||
| 20 | budget = 300 * time.Millisecond | 20 | budget = 300 * time.Millisecond | ||
| 21 | ) | 21 | ) | ||
| 22 | 22 | ||||
| 23 | + | // connectBudget bounds the whole connection setup (session bus, | |||
| 24 | + | // GetAddress, IsEnabled, dial of the accessibility bus) of one attempt. | |||
| 25 | + | const connectBudget = 300 * time.Millisecond | |||
| 26 | + | ||||
| 27 | + | // retryEvery is the minimum time between two connection attempts. | |||
| 28 | + | const retryEvery = 5 * time.Second | |||
| 29 | + | ||||
| 30 | + | // connector opens a transport on the accessibility bus. | |||
| 31 | + | type connector interface { | |||
| 32 | + | Connect(ctx context.Context) (transport, error) | |||
| 33 | + | } | |||
| 34 | + | ||||
| 35 | + | // dialResult is the outcome of a connection attempt. | |||
| 36 | + | type dialResult struct { | |||
| 37 | + | tr transport | |||
| 38 | + | err error | |||
| 39 | + | } | |||
| 40 | + | ||||
| 23 | // Source reads hint targets from the accessibility tree. It owns one | 41 | // Source reads hint targets from the accessibility tree. It owns one | ||
| 24 | // connection to the accessibility bus and is used by one goroutine at a | 42 | // connection to the accessibility bus and is used by one goroutine at a | ||
| 25 | - | // time (the hints worker). | 43 | + | // time (the hints worker). The connection is made on the first Targets call |
| 44 | + | // that needs it, never at startup, and made again after a failure, at most | |||
| 45 | + | // once every retryEvery; meanwhile Targets finds nothing and OCR takes over. | |||
| 26 | type Source struct { | 46 | type Source struct { | ||
| 27 | - | tr transport | 47 | + | tr transport |
| 48 | + | 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 | |||
| 28 | } | 55 | } | ||
| 29 | 56 | ||||
| 30 | - | // New enables accessibility on the session and connects to the | 57 | + | // New returns a Source. It does not touch the bus: the connection is made by |
| 31 | - | // accessibility bus. | 58 | + | // the first Targets call. |
| 32 | - | func New(ctx context.Context) (*Source, error) { | 59 | + | func New() *Source { return &Source{dial: busConnector{}} } |
| 60 | + | ||||
| 61 | + | // busConnector connects to the session bus, enables accessibility and dials | |||
| 62 | + | // the accessibility bus. | |||
| 63 | + | type busConnector struct{} | |||
| 64 | + | ||||
| 65 | + | func (busConnector) Connect(ctx context.Context) (transport, error) { | |||
| 33 | log := logging.For(ctx, "atspi") | 66 | log := logging.For(ctx, "atspi") | ||
| 34 | sess, err := zerobus.SessionBus() | 67 | sess, err := zerobus.SessionBus() | ||
| 35 | if err != nil { | 68 | if err != nil { | ||
| 36 | return nil, fmt.Errorf("atspi: session bus: %w", err) | 69 | return nil, fmt.Errorf("atspi: session bus: %w", err) | ||
| 37 | } | 70 | } | ||
| 38 | defer func() { _ = sess.Close() }() | 71 | defer func() { _ = sess.Close() }() | ||
| 72 | + | // Call has no deadline: closing the connection ends the wait for a | |||
| 73 | + | // silent service. | |||
| 74 | + | stop := time.AfterFunc(connectBudget, func() { _ = sess.Close() }) | |||
| 75 | + | defer stop.Stop() | |||
| 39 | 76 | ||||
| 40 | sess.NewCall(a11yBusName, a11yBusPath, a11yBusName, "GetAddress", "") | 77 | sess.NewCall(a11yBusName, a11yBusPath, a11yBusName, "GetAddress", "") | ||
| 41 | m, err := sess.Call() | 78 | m, err := sess.Call() | ||
| @@ -70,11 +107,75 @@ func New(ctx context.Context) (*Source, error) { | |||||
| 70 | return nil, err | 107 | return nil, err | ||
| 71 | } | 108 | } | ||
| 72 | log.Info().Msg("accessibility bus connected") | 109 | log.Info().Msg("accessibility bus connected") | ||
| 73 | - | return &Source{tr: tr}, nil | 110 | + | return tr, nil |
| 74 | } | 111 | } | ||
| 75 | 112 | ||||
| 76 | - | // Close closes the accessibility bus connection. | 113 | + | // ensure returns the connection, making one if none exists and the last |
| 77 | - | func (s *Source) Close() error { return s.tr.Close() } | 114 | + | // attempt is old enough. It waits at most connectBudget for an attempt. |
| 115 | + | func (s *Source) ensure(ctx context.Context) transport { | |||
| 116 | + | if s.tr != nil { | |||
| 117 | + | return s.tr | |||
| 118 | + | } | |||
| 119 | + | if s.dial == nil { | |||
| 120 | + | return nil | |||
| 121 | + | } | |||
| 122 | + | now := time.Now() | |||
| 123 | + | if s.attempt == nil { | |||
| 124 | + | if now.Before(s.retryAt) { | |||
| 125 | + | return nil | |||
| 126 | + | } | |||
| 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 | + | }() | |||
| 135 | + | } | |||
| 136 | + | timer := time.NewTimer(time.Until(s.attemptDue)) | |||
| 137 | + | defer timer.Stop() | |||
| 138 | + | 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 | |||
| 147 | + | case <-timer.C: | |||
| 148 | + | case <-ctx.Done(): | |||
| 149 | + | } | |||
| 150 | + | return s.tr | |||
| 151 | + | } | |||
| 152 | + | ||||
| 153 | + | // drop closes a broken connection; the next attempt reconnects. | |||
| 154 | + | func (s *Source) drop() { | |||
| 155 | + | if s.tr != nil { | |||
| 156 | + | _ = s.tr.Close() | |||
| 157 | + | s.tr = nil | |||
| 158 | + | } | |||
| 159 | + | } | |||
| 160 | + | ||||
| 161 | + | // Close closes the accessibility bus connection, and the one an abandoned | |||
| 162 | + | // attempt may still deliver. | |||
| 163 | + | func (s *Source) Close() error { | |||
| 164 | + | var err error | |||
| 165 | + | if s.tr != nil { | |||
| 166 | + | err = s.tr.Close() | |||
| 167 | + | s.tr = nil | |||
| 168 | + | } | |||
| 169 | + | if ch := s.attempt; ch != nil { | |||
| 170 | + | s.attempt = nil | |||
| 171 | + | go func() { | |||
| 172 | + | if r := <-ch; r.tr != nil { | |||
| 173 | + | _ = r.tr.Close() | |||
| 174 | + | } | |||
| 175 | + | }() | |||
| 176 | + | } | |||
| 177 | + | return err | |||
| 178 | + | } | |||
| 78 | 179 | ||||
| 79 | // appInfo is an application of the registry whose process shows a window. | 180 | // appInfo is an application of the registry whose process shows a window. | ||
| 80 | type appInfo struct { | 181 | type appInfo struct { | ||
| @@ -101,13 +202,6 @@ func (s *Source) Targets(ctx context.Context, windows []ports.HintWindow, linksO | |||||
| 101 | if d, ok := ctx.Deadline(); ok && d.Before(deadline) { | 202 | if d, ok := ctx.Deadline(); ok && d.Before(deadline) { | ||
| 102 | deadline = d | 203 | deadline = d | ||
| 103 | } | 204 | } | ||
| 104 | - | do := func(calls []call) []reply { | |||
| 105 | - | if len(calls) == 0 || ctx.Err() != nil { | |||
| 106 | - | return make([]reply, len(calls)) | |||
| 107 | - | } | |||
| 108 | - | return s.tr.Do(deadline, callTimeout, calls) | |||
| 109 | - | } | |||
| 110 | - | ||||
| 111 | wanted := map[int]bool{} | 205 | wanted := map[int]bool{} | ||
| 112 | for _, w := range windows { | 206 | for _, w := range windows { | ||
| 113 | if w.PID > 0 { | 207 | if w.PID > 0 { | ||
| @@ -117,6 +211,32 @@ func (s *Source) Targets(ctx context.Context, windows []ports.HintWindow, linksO | |||||
| 117 | if len(wanted) == 0 { | 211 | if len(wanted) == 0 { | ||
| 118 | return nil, covered, nil | 212 | return nil, covered, nil | ||
| 119 | } | 213 | } | ||
| 214 | + | tr := s.ensure(ctx) | |||
| 215 | + | if tr == nil { | |||
| 216 | + | return nil, covered, nil | |||
| 217 | + | } | |||
| 218 | + | broken := false | |||
| 219 | + | do := func(calls []call) []reply { | |||
| 220 | + | if len(calls) == 0 || ctx.Err() != nil { | |||
| 221 | + | return make([]reply, len(calls)) | |||
| 222 | + | } | |||
| 223 | + | if broken { | |||
| 224 | + | out := make([]reply, len(calls)) | |||
| 225 | + | for i := range out { | |||
| 226 | + | out[i].err = ErrBroken | |||
| 227 | + | } | |||
| 228 | + | return out | |||
| 229 | + | } | |||
| 230 | + | out := tr.Do(deadline, callTimeout, calls) | |||
| 231 | + | for _, r := range out { | |||
| 232 | + | if errors.Is(r.err, ErrBroken) { | |||
| 233 | + | broken = true | |||
| 234 | + | s.drop() | |||
| 235 | + | break | |||
| 236 | + | } | |||
| 237 | + | } | |||
| 238 | + | return out | |||
| 239 | + | } | |||
| 120 | 240 | ||||
| 121 | // Applications of the registry, kept when their process shows a window. | 241 | // Applications of the registry, kept when their process shows a window. | ||
| 122 | rep := do([]call{{op: opChildren, name: registryName, path: registryRoot}}) | 242 | rep := do([]call{{op: opChildren, name: registryName, path: registryRoot}}) | ||
M internal/adapters/atspi/source_test.go +92 -0
| @@ -2,6 +2,7 @@ package atspi | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | + | "errors" | |
| 5 | 6 | "fmt" | |
| 6 | 7 | "testing" | |
| 7 | 8 | "time" | |
| @@ -144,3 +145,94 @@ func TestCoverageRequiresLinkOrDocument(t *testing.T) { | |||
| 144 | 145 | } | |
| 145 | 146 | } | |
| 146 | 147 | } | |
| 148 | + | ||
| 149 | + | var oneWindow = []ports.HintWindow{{ID: 1, PID: 77, Rect: ports.Rect{W: 800, H: 600}}} | |
| 150 | + | ||
| 151 | + | // New does not touch the bus; the first Targets call connects, once. | |
| 152 | + | func TestConnectsLazilyOnFirstTargets(t *testing.T) { | |
| 153 | + | tr := newMocktransport(t) | |
| 154 | + | tr.EXPECT().Do(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(answerTree( | |
| 155 | + | [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)) | |
| 156 | + | c := newMockconnector(t) | |
| 157 | + | c.EXPECT().Connect(mock.Anything).Return(tr, nil).Once() | |
| 158 | + | s := &Source{dial: c} | |
| 159 | + | for range 2 { | |
| 160 | + | targets, _, err := s.Targets(context.Background(), oneWindow, false) | |
| 161 | + | require.NoError(t, err) | |
| 162 | + | require.Len(t, targets, 1) | |
| 163 | + | } | |
| 164 | + | } | |
| 165 | + | ||
| 166 | + | // Without PIDs nothing is asked and nothing is connected. | |
| 167 | + | func TestNoConnectWithoutWindows(t *testing.T) { | |
| 168 | + | s := &Source{dial: newMockconnector(t)} | |
| 169 | + | _, _, err := s.Targets(context.Background(), []ports.HintWindow{{ID: 1}}, false) | |
| 170 | + | require.NoError(t, err) | |
| 171 | + | } | |
| 172 | + | ||
| 173 | + | // A failed connection is retried no sooner than retryEvery; meanwhile no | |
| 174 | + | // targets are returned. | |
| 175 | + | func TestConnectRetriesWithBackoff(t *testing.T) { | |
| 176 | + | c := newMockconnector(t) | |
| 177 | + | c.EXPECT().Connect(mock.Anything).Return(nil, errors.New("down")).Once() | |
| 178 | + | s := &Source{dial: c} | |
| 179 | + | for range 3 { | |
| 180 | + | targets, covered, err := s.Targets(context.Background(), oneWindow, false) | |
| 181 | + | require.NoError(t, err) | |
| 182 | + | require.Empty(t, targets) | |
| 183 | + | require.Empty(t, covered) | |
| 184 | + | } | |
| 185 | + | tr := newMocktransport(t) | |
| 186 | + | tr.EXPECT().Do(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(answerTree( | |
| 187 | + | [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)) | |
| 188 | + | c.EXPECT().Connect(mock.Anything).Return(tr, nil).Once() | |
| 189 | + | s.retryAt = time.Now().Add(-time.Second) | |
| 190 | + | targets, _, err := s.Targets(context.Background(), oneWindow, false) | |
| 191 | + | require.NoError(t, err) | |
| 192 | + | require.Len(t, targets, 1) | |
| 193 | + | } | |
| 194 | + | ||
| 195 | + | // A broken connection is closed and dropped; the next attempt reconnects. | |
| 196 | + | func TestBrokenConnectionIsDropped(t *testing.T) { | |
| 197 | + | tr := newMocktransport(t) | |
| 198 | + | tr.EXPECT().Do(mock.Anything, mock.Anything, mock.Anything).RunAndReturn( | |
| 199 | + | func(_ time.Time, _ time.Duration, calls []call) []reply { | |
| 200 | + | out := make([]reply, len(calls)) | |
| 201 | + | for i := range out { | |
| 202 | + | out[i].err = ErrBroken | |
| 203 | + | } | |
| 204 | + | return out | |
| 205 | + | }).Once() | |
| 206 | + | tr.EXPECT().Close().Return(nil).Once() | |
| 207 | + | c := newMockconnector(t) | |
| 208 | + | c.EXPECT().Connect(mock.Anything).Return(tr, nil).Once() | |
| 209 | + | s := &Source{dial: c} | |
| 210 | + | _, _, err := s.Targets(context.Background(), oneWindow, false) | |
| 211 | + | require.Error(t, err) | |
| 212 | + | require.Nil(t, s.tr) | |
| 213 | + | targets, _, err := s.Targets(context.Background(), oneWindow, false) | |
| 214 | + | require.NoError(t, err) | |
| 215 | + | require.Empty(t, targets) // backoff: no new attempt yet | |
| 216 | + | } | |
| 217 | + | ||
| 218 | + | // A silent service cannot hold Targets beyond the connect budget. | |
| 219 | + | func TestConnectIsBounded(t *testing.T) { | |
| 220 | + | release := make(chan struct{}) | |
| 221 | + | tr := newMocktransport(t) | |
| 222 | + | tr.EXPECT().Close().Return(nil).Once() | |
| 223 | + | c := newMockconnector(t) | |
| 224 | + | c.EXPECT().Connect(mock.Anything).RunAndReturn(func(context.Context) (transport, error) { | |
| 225 | + | <-release | |
| 226 | + | return tr, nil | |
| 227 | + | }).Once() | |
| 228 | + | s := &Source{dial: c} | |
| 229 | + | t0 := time.Now() | |
| 230 | + | targets, _, err := s.Targets(context.Background(), oneWindow, false) | |
| 231 | + | require.NoError(t, err) | |
| 232 | + | require.Empty(t, targets) | |
| 233 | + | require.Less(t, time.Since(t0), connectBudget+200*time.Millisecond) | |
| 234 | + | // The late result is closed with the source. | |
| 235 | + | require.NoError(t, s.Close()) | |
| 236 | + | close(release) | |
| 237 | + | require.Eventually(t, func() bool { return tr.AssertExpectations(&testing.T{}) }, time.Second, 5*time.Millisecond) | |
| 238 | + | } | |
| @@ -2,6 +2,7 @@ package atspi | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | + | "errors" | |||
| 5 | "fmt" | 6 | "fmt" | ||
| 6 | "testing" | 7 | "testing" | ||
| 7 | "time" | 8 | "time" | ||
| @@ -144,3 +145,94 @@ func TestCoverageRequiresLinkOrDocument(t *testing.T) { | |||||
| 144 | } | 145 | } | ||
| 145 | } | 146 | } | ||
| 146 | } | 147 | } | ||
| 148 | + | ||||
| 149 | + | var oneWindow = []ports.HintWindow{{ID: 1, PID: 77, Rect: ports.Rect{W: 800, H: 600}}} | |||
| 150 | + | ||||
| 151 | + | // New does not touch the bus; the first Targets call connects, once. | |||
| 152 | + | func TestConnectsLazilyOnFirstTargets(t *testing.T) { | |||
| 153 | + | tr := newMocktransport(t) | |||
| 154 | + | tr.EXPECT().Do(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(answerTree( | |||
| 155 | + | [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)) | |||
| 156 | + | c := newMockconnector(t) | |||
| 157 | + | c.EXPECT().Connect(mock.Anything).Return(tr, nil).Once() | |||
| 158 | + | s := &Source{dial: c} | |||
| 159 | + | for range 2 { | |||
| 160 | + | targets, _, err := s.Targets(context.Background(), oneWindow, false) | |||
| 161 | + | require.NoError(t, err) | |||
| 162 | + | require.Len(t, targets, 1) | |||
| 163 | + | } | |||
| 164 | + | } | |||
| 165 | + | ||||
| 166 | + | // Without PIDs nothing is asked and nothing is connected. | |||
| 167 | + | func TestNoConnectWithoutWindows(t *testing.T) { | |||
| 168 | + | s := &Source{dial: newMockconnector(t)} | |||
| 169 | + | _, _, err := s.Targets(context.Background(), []ports.HintWindow{{ID: 1}}, false) | |||
| 170 | + | require.NoError(t, err) | |||
| 171 | + | } | |||
| 172 | + | ||||
| 173 | + | // A failed connection is retried no sooner than retryEvery; meanwhile no | |||
| 174 | + | // targets are returned. | |||
| 175 | + | func TestConnectRetriesWithBackoff(t *testing.T) { | |||
| 176 | + | c := newMockconnector(t) | |||
| 177 | + | c.EXPECT().Connect(mock.Anything).Return(nil, errors.New("down")).Once() | |||
| 178 | + | s := &Source{dial: c} | |||
| 179 | + | for range 3 { | |||
| 180 | + | targets, covered, err := s.Targets(context.Background(), oneWindow, false) | |||
| 181 | + | require.NoError(t, err) | |||
| 182 | + | require.Empty(t, targets) | |||
| 183 | + | require.Empty(t, covered) | |||
| 184 | + | } | |||
| 185 | + | tr := newMocktransport(t) | |||
| 186 | + | tr.EXPECT().Do(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(answerTree( | |||
| 187 | + | [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)) | |||
| 188 | + | c.EXPECT().Connect(mock.Anything).Return(tr, nil).Once() | |||
| 189 | + | s.retryAt = time.Now().Add(-time.Second) | |||
| 190 | + | targets, _, err := s.Targets(context.Background(), oneWindow, false) | |||
| 191 | + | require.NoError(t, err) | |||
| 192 | + | require.Len(t, targets, 1) | |||
| 193 | + | } | |||
| 194 | + | ||||
| 195 | + | // A broken connection is closed and dropped; the next attempt reconnects. | |||
| 196 | + | func TestBrokenConnectionIsDropped(t *testing.T) { | |||
| 197 | + | tr := newMocktransport(t) | |||
| 198 | + | tr.EXPECT().Do(mock.Anything, mock.Anything, mock.Anything).RunAndReturn( | |||
| 199 | + | func(_ time.Time, _ time.Duration, calls []call) []reply { | |||
| 200 | + | out := make([]reply, len(calls)) | |||
| 201 | + | for i := range out { | |||
| 202 | + | out[i].err = ErrBroken | |||
| 203 | + | } | |||
| 204 | + | return out | |||
| 205 | + | }).Once() | |||
| 206 | + | tr.EXPECT().Close().Return(nil).Once() | |||
| 207 | + | c := newMockconnector(t) | |||
| 208 | + | c.EXPECT().Connect(mock.Anything).Return(tr, nil).Once() | |||
| 209 | + | s := &Source{dial: c} | |||
| 210 | + | _, _, err := s.Targets(context.Background(), oneWindow, false) | |||
| 211 | + | require.Error(t, err) | |||
| 212 | + | require.Nil(t, s.tr) | |||
| 213 | + | targets, _, err := s.Targets(context.Background(), oneWindow, false) | |||
| 214 | + | require.NoError(t, err) | |||
| 215 | + | require.Empty(t, targets) // backoff: no new attempt yet | |||
| 216 | + | } | |||
| 217 | + | ||||
| 218 | + | // A silent service cannot hold Targets beyond the connect budget. | |||
| 219 | + | func TestConnectIsBounded(t *testing.T) { | |||
| 220 | + | release := make(chan struct{}) | |||
| 221 | + | tr := newMocktransport(t) | |||
| 222 | + | tr.EXPECT().Close().Return(nil).Once() | |||
| 223 | + | c := newMockconnector(t) | |||
| 224 | + | c.EXPECT().Connect(mock.Anything).RunAndReturn(func(context.Context) (transport, error) { | |||
| 225 | + | <-release | |||
| 226 | + | return tr, nil | |||
| 227 | + | }).Once() | |||
| 228 | + | s := &Source{dial: c} | |||
| 229 | + | t0 := time.Now() | |||
| 230 | + | targets, _, err := s.Targets(context.Background(), oneWindow, false) | |||
| 231 | + | require.NoError(t, err) | |||
| 232 | + | require.Empty(t, targets) | |||
| 233 | + | require.Less(t, time.Since(t0), connectBudget+200*time.Millisecond) | |||
| 234 | + | // The late result is closed with the source. | |||
| 235 | + | require.NoError(t, s.Close()) | |||
| 236 | + | close(release) | |||
| 237 | + | require.Eventually(t, func() bool { return tr.AssertExpectations(&testing.T{}) }, time.Second, 5*time.Millisecond) | |||
| 238 | + | } | |||
M internal/adapters/atspi/transport.go +70 -18
| @@ -4,6 +4,7 @@ import ( | |||
| 4 | 4 | "errors" | |
| 5 | 5 | "fmt" | |
| 6 | 6 | "strings" | |
| 7 | + | "sync/atomic" | |
| 7 | 8 | "syscall" | |
| 8 | 9 | "time" | |
| 9 | 10 | ||
| @@ -75,8 +76,10 @@ type reply struct { | |||
| 75 | 76 | // goroutine at a time. | |
| 76 | 77 | type transport interface { | |
| 77 | 78 | // Do sends calls pipelined and returns one reply per call, in order. | |
| 78 | - | // The batch ends at deadline; it also ends when no reply arrives for | |
| 79 | - | // idle. Calls still unanswered then get ErrTimeout. | |
| 79 | + | // The batch ends at deadline. A call not answered idle after it was | |
| 80 | + | // sent gets ErrTimeout. When the connection fails or the deadline | |
| 81 | + | // finds calls unanswered, those calls get an error wrapping | |
| 82 | + | // ErrBroken: the connection must be dropped. | |
| 80 | 83 | Do(deadline time.Time, idle time.Duration, calls []call) []reply | |
| 81 | 84 | Close() error | |
| 82 | 85 | } | |
| @@ -136,13 +139,46 @@ func (b *conn) readable(d time.Duration) (bool, error) { | |||
| 136 | 139 | return n > 0, perr | |
| 137 | 140 | } | |
| 138 | 141 | ||
| 142 | + | // ErrBroken is the error of calls on a connection that failed or was closed | |
| 143 | + | // at the deadline; the owner must drop it and connect again. | |
| 144 | + | var ErrBroken = errors.New("atspi: connection broken") | |
| 145 | + | ||
| 146 | + | // errDeadline is the error of calls still unanswered at the batch deadline: | |
| 147 | + | // late replies and half-read messages leave the connection unusable. | |
| 148 | + | var errDeadline = fmt.Errorf("%w: %w", ErrBroken, ErrTimeout) | |
| 149 | + | ||
| 150 | + | // watchdogGrace is how long after the deadline the watchdog closes a socket | |
| 151 | + | // that the batch loop could not leave by itself (a stuck read or write). | |
| 152 | + | const watchdogGrace = 20 * time.Millisecond | |
| 153 | + | ||
| 154 | + | // Do sends the calls pipelined. A watchdog closes the socket shortly after | |
| 155 | + | // deadline, so neither a blocked write nor a read stuck in the middle of a | |
| 156 | + | // message can outlive the batch; the connection is then broken. Each call | |
| 157 | + | // has its own idle timeout, counted from its send: traffic for other calls | |
| 158 | + | // does not extend it. | |
| 139 | 159 | func (b *conn) Do(deadline time.Time, idle time.Duration, calls []call) []reply { | |
| 140 | 160 | out := make([]reply, len(calls)) | |
| 141 | 161 | answered := make([]bool, len(calls)) | |
| 142 | - | pending := make(map[uint32]int, min(len(calls), maxInflight)) | |
| 162 | + | type inflight struct { | |
| 163 | + | idx int | |
| 164 | + | sent time.Time | |
| 165 | + | } | |
| 166 | + | pending := make(map[uint32]inflight, min(len(calls), maxInflight)) | |
| 143 | 167 | next, done := 0, 0 | |
| 168 | + | var fired atomic.Bool | |
| 169 | + | watchdog := time.AfterFunc(time.Until(deadline)+watchdogGrace, func() { | |
| 170 | + | fired.Store(true) | |
| 171 | + | _ = b.c.Close() | |
| 172 | + | }) | |
| 173 | + | defer watchdog.Stop() | |
| 144 | 174 | // finish gives err to every call without an answer. | |
| 145 | 175 | finish := func(err error) []reply { | |
| 176 | + | if !errors.Is(err, ErrBroken) && (fired.Load() || !errors.Is(err, ErrTimeout)) { | |
| 177 | + | err = fmt.Errorf("%w: %w", ErrBroken, err) | |
| 178 | + | } | |
| 179 | + | if errors.Is(err, ErrBroken) { | |
| 180 | + | _ = b.c.Close() // late replies would desynchronize the next batch | |
| 181 | + | } | |
| 146 | 182 | for i := range out { | |
| 147 | 183 | if !answered[i] { | |
| 148 | 184 | out[i].err = err | |
| @@ -152,48 +188,64 @@ func (b *conn) Do(deadline time.Time, idle time.Duration, calls []call) []reply | |||
| 152 | 188 | } | |
| 153 | 189 | for done < len(calls) { | |
| 154 | 190 | for next < len(calls) && len(pending) < maxInflight { | |
| 191 | + | if !time.Now().Before(deadline) { | |
| 192 | + | return finish(errDeadline) | |
| 193 | + | } | |
| 155 | 194 | serial, err := b.send(calls[next]) | |
| 156 | 195 | if err != nil { | |
| 157 | - | if errors.Is(err, zerobus.ErrClosed) { | |
| 158 | - | return finish(err) | |
| 159 | - | } | |
| 160 | - | out[next].err = err | |
| 161 | - | answered[next] = true | |
| 162 | - | done++ | |
| 163 | - | } else { | |
| 164 | - | pending[serial] = next | |
| 196 | + | return finish(err) | |
| 165 | 197 | } | |
| 198 | + | pending[serial] = inflight{idx: next, sent: time.Now()} | |
| 166 | 199 | next++ | |
| 167 | 200 | } | |
| 201 | + | // Calls that waited longer than idle get no reply. | |
| 202 | + | now := time.Now() | |
| 203 | + | var oldest time.Time | |
| 204 | + | for serial, p := range pending { | |
| 205 | + | if now.Sub(p.sent) >= idle { | |
| 206 | + | out[p.idx].err = ErrTimeout | |
| 207 | + | answered[p.idx] = true | |
| 208 | + | done++ | |
| 209 | + | delete(pending, serial) | |
| 210 | + | } else if oldest.IsZero() || p.sent.Before(oldest) { | |
| 211 | + | oldest = p.sent | |
| 212 | + | } | |
| 213 | + | } | |
| 168 | 214 | if len(pending) == 0 { | |
| 169 | 215 | continue | |
| 170 | 216 | } | |
| 171 | - | wait := min(idle, time.Until(deadline)) | |
| 217 | + | wait := min(time.Until(oldest.Add(idle)), time.Until(deadline)) | |
| 172 | 218 | if wait <= 0 { | |
| 173 | - | break | |
| 219 | + | if !time.Now().Before(deadline) { | |
| 220 | + | break | |
| 221 | + | } | |
| 222 | + | continue | |
| 174 | 223 | } | |
| 175 | 224 | ok, err := b.readable(wait) | |
| 176 | 225 | if err != nil { | |
| 177 | 226 | return finish(err) | |
| 178 | 227 | } | |
| 179 | 228 | if !ok { | |
| 180 | - | break | |
| 229 | + | continue | |
| 181 | 230 | } | |
| 182 | 231 | m, err := b.c.ReadMessage() | |
| 183 | 232 | if err != nil { | |
| 184 | 233 | return finish(err) | |
| 185 | 234 | } | |
| 186 | - | i, found := pending[m.ReplySerial] | |
| 235 | + | p, found := pending[m.ReplySerial] | |
| 187 | 236 | if !found || (m.Type != zerobus.TypeMethodReturn && m.Type != zerobus.TypeError) { | |
| 188 | 237 | // A signal, or the late reply of an earlier batch. | |
| 189 | 238 | continue | |
| 190 | 239 | } | |
| 191 | 240 | delete(pending, m.ReplySerial) | |
| 192 | - | decode(m, calls[i], &out[i]) | |
| 193 | - | answered[i] = true | |
| 241 | + | decode(m, calls[p.idx], &out[p.idx]) | |
| 242 | + | answered[p.idx] = true | |
| 194 | 243 | done++ | |
| 195 | 244 | } | |
| 196 | - | return finish(ErrTimeout) | |
| 245 | + | if done == len(calls) { | |
| 246 | + | return out | |
| 247 | + | } | |
| 248 | + | return finish(errDeadline) | |
| 197 | 249 | } | |
| 198 | 250 | ||
| 199 | 251 | // send builds and sends one call and returns its serial. | |
| @@ -4,6 +4,7 @@ import ( | |||||
| 4 | "errors" | 4 | "errors" | ||
| 5 | "fmt" | 5 | "fmt" | ||
| 6 | "strings" | 6 | "strings" | ||
| 7 | + | "sync/atomic" | |||
| 7 | "syscall" | 8 | "syscall" | ||
| 8 | "time" | 9 | "time" | ||
| 9 | 10 | ||||
| @@ -75,8 +76,10 @@ type reply struct { | |||||
| 75 | // goroutine at a time. | 76 | // goroutine at a time. | ||
| 76 | type transport interface { | 77 | type transport interface { | ||
| 77 | // Do sends calls pipelined and returns one reply per call, in order. | 78 | // Do sends calls pipelined and returns one reply per call, in order. | ||
| 78 | - | // The batch ends at deadline; it also ends when no reply arrives for | 79 | + | // The batch ends at deadline. A call not answered idle after it was |
| 79 | - | // idle. Calls still unanswered then get ErrTimeout. | 80 | + | // sent gets ErrTimeout. When the connection fails or the deadline |
| 81 | + | // finds calls unanswered, those calls get an error wrapping | |||
| 82 | + | // ErrBroken: the connection must be dropped. | |||
| 80 | Do(deadline time.Time, idle time.Duration, calls []call) []reply | 83 | Do(deadline time.Time, idle time.Duration, calls []call) []reply | ||
| 81 | Close() error | 84 | Close() error | ||
| 82 | } | 85 | } | ||
| @@ -136,13 +139,46 @@ func (b *conn) readable(d time.Duration) (bool, error) { | |||||
| 136 | return n > 0, perr | 139 | return n > 0, perr | ||
| 137 | } | 140 | } | ||
| 138 | 141 | ||||
| 142 | + | // ErrBroken is the error of calls on a connection that failed or was closed | |||
| 143 | + | // at the deadline; the owner must drop it and connect again. | |||
| 144 | + | var ErrBroken = errors.New("atspi: connection broken") | |||
| 145 | + | ||||
| 146 | + | // errDeadline is the error of calls still unanswered at the batch deadline: | |||
| 147 | + | // late replies and half-read messages leave the connection unusable. | |||
| 148 | + | var errDeadline = fmt.Errorf("%w: %w", ErrBroken, ErrTimeout) | |||
| 149 | + | ||||
| 150 | + | // watchdogGrace is how long after the deadline the watchdog closes a socket | |||
| 151 | + | // that the batch loop could not leave by itself (a stuck read or write). | |||
| 152 | + | const watchdogGrace = 20 * time.Millisecond | |||
| 153 | + | ||||
| 154 | + | // Do sends the calls pipelined. A watchdog closes the socket shortly after | |||
| 155 | + | // deadline, so neither a blocked write nor a read stuck in the middle of a | |||
| 156 | + | // message can outlive the batch; the connection is then broken. Each call | |||
| 157 | + | // has its own idle timeout, counted from its send: traffic for other calls | |||
| 158 | + | // does not extend it. | |||
| 139 | func (b *conn) Do(deadline time.Time, idle time.Duration, calls []call) []reply { | 159 | func (b *conn) Do(deadline time.Time, idle time.Duration, calls []call) []reply { | ||
| 140 | out := make([]reply, len(calls)) | 160 | out := make([]reply, len(calls)) | ||
| 141 | answered := make([]bool, len(calls)) | 161 | answered := make([]bool, len(calls)) | ||
| 142 | - | pending := make(map[uint32]int, min(len(calls), maxInflight)) | 162 | + | type inflight struct { |
| 163 | + | idx int | |||
| 164 | + | sent time.Time | |||
| 165 | + | } | |||
| 166 | + | pending := make(map[uint32]inflight, min(len(calls), maxInflight)) | |||
| 143 | next, done := 0, 0 | 167 | next, done := 0, 0 | ||
| 168 | + | var fired atomic.Bool | |||
| 169 | + | watchdog := time.AfterFunc(time.Until(deadline)+watchdogGrace, func() { | |||
| 170 | + | fired.Store(true) | |||
| 171 | + | _ = b.c.Close() | |||
| 172 | + | }) | |||
| 173 | + | defer watchdog.Stop() | |||
| 144 | // finish gives err to every call without an answer. | 174 | // finish gives err to every call without an answer. | ||
| 145 | finish := func(err error) []reply { | 175 | finish := func(err error) []reply { | ||
| 176 | + | if !errors.Is(err, ErrBroken) && (fired.Load() || !errors.Is(err, ErrTimeout)) { | |||
| 177 | + | err = fmt.Errorf("%w: %w", ErrBroken, err) | |||
| 178 | + | } | |||
| 179 | + | if errors.Is(err, ErrBroken) { | |||
| 180 | + | _ = b.c.Close() // late replies would desynchronize the next batch | |||
| 181 | + | } | |||
| 146 | for i := range out { | 182 | for i := range out { | ||
| 147 | if !answered[i] { | 183 | if !answered[i] { | ||
| 148 | out[i].err = err | 184 | out[i].err = err | ||
| @@ -152,48 +188,64 @@ func (b *conn) Do(deadline time.Time, idle time.Duration, calls []call) []reply | |||||
| 152 | } | 188 | } | ||
| 153 | for done < len(calls) { | 189 | for done < len(calls) { | ||
| 154 | for next < len(calls) && len(pending) < maxInflight { | 190 | for next < len(calls) && len(pending) < maxInflight { | ||
| 191 | + | if !time.Now().Before(deadline) { | |||
| 192 | + | return finish(errDeadline) | |||
| 193 | + | } | |||
| 155 | serial, err := b.send(calls[next]) | 194 | serial, err := b.send(calls[next]) | ||
| 156 | if err != nil { | 195 | if err != nil { | ||
| 157 | - | if errors.Is(err, zerobus.ErrClosed) { | 196 | + | return finish(err) |
| 158 | - | return finish(err) | |||
| 159 | - | } | |||
| 160 | - | out[next].err = err | |||
| 161 | - | answered[next] = true | |||
| 162 | - | done++ | |||
| 163 | - | } else { | |||
| 164 | - | pending[serial] = next | |||
| 165 | } | 197 | } | ||
| 198 | + | pending[serial] = inflight{idx: next, sent: time.Now()} | |||
| 166 | next++ | 199 | next++ | ||
| 167 | } | 200 | } | ||
| 201 | + | // Calls that waited longer than idle get no reply. | |||
| 202 | + | now := time.Now() | |||
| 203 | + | var oldest time.Time | |||
| 204 | + | for serial, p := range pending { | |||
| 205 | + | if now.Sub(p.sent) >= idle { | |||
| 206 | + | out[p.idx].err = ErrTimeout | |||
| 207 | + | answered[p.idx] = true | |||
| 208 | + | done++ | |||
| 209 | + | delete(pending, serial) | |||
| 210 | + | } else if oldest.IsZero() || p.sent.Before(oldest) { | |||
| 211 | + | oldest = p.sent | |||
| 212 | + | } | |||
| 213 | + | } | |||
| 168 | if len(pending) == 0 { | 214 | if len(pending) == 0 { | ||
| 169 | continue | 215 | continue | ||
| 170 | } | 216 | } | ||
| 171 | - | wait := min(idle, time.Until(deadline)) | 217 | + | wait := min(time.Until(oldest.Add(idle)), time.Until(deadline)) |
| 172 | if wait <= 0 { | 218 | if wait <= 0 { | ||
| 173 | - | break | 219 | + | if !time.Now().Before(deadline) { |
| 220 | + | break | |||
| 221 | + | } | |||
| 222 | + | continue | |||
| 174 | } | 223 | } | ||
| 175 | ok, err := b.readable(wait) | 224 | ok, err := b.readable(wait) | ||
| 176 | if err != nil { | 225 | if err != nil { | ||
| 177 | return finish(err) | 226 | return finish(err) | ||
| 178 | } | 227 | } | ||
| 179 | if !ok { | 228 | if !ok { | ||
| 180 | - | break | 229 | + | continue |
| 181 | } | 230 | } | ||
| 182 | m, err := b.c.ReadMessage() | 231 | m, err := b.c.ReadMessage() | ||
| 183 | if err != nil { | 232 | if err != nil { | ||
| 184 | return finish(err) | 233 | return finish(err) | ||
| 185 | } | 234 | } | ||
| 186 | - | i, found := pending[m.ReplySerial] | 235 | + | p, found := pending[m.ReplySerial] |
| 187 | if !found || (m.Type != zerobus.TypeMethodReturn && m.Type != zerobus.TypeError) { | 236 | if !found || (m.Type != zerobus.TypeMethodReturn && m.Type != zerobus.TypeError) { | ||
| 188 | // A signal, or the late reply of an earlier batch. | 237 | // A signal, or the late reply of an earlier batch. | ||
| 189 | continue | 238 | continue | ||
| 190 | } | 239 | } | ||
| 191 | delete(pending, m.ReplySerial) | 240 | delete(pending, m.ReplySerial) | ||
| 192 | - | decode(m, calls[i], &out[i]) | 241 | + | decode(m, calls[p.idx], &out[p.idx]) |
| 193 | - | answered[i] = true | 242 | + | answered[p.idx] = true |
| 194 | done++ | 243 | done++ | ||
| 195 | } | 244 | } | ||
| 196 | - | return finish(ErrTimeout) | 245 | + | if done == len(calls) { |
| 246 | + | return out | |||
| 247 | + | } | |||
| 248 | + | return finish(errDeadline) | |||
| 197 | } | 249 | } | ||
| 198 | 250 | ||||
| 199 | // send builds and sends one call and returns its serial. | 251 | // send builds and sends one call and returns its serial. | ||
M internal/app/hintdebug.go +4 -8
| @@ -12,16 +12,12 @@ import ( | |||
| 12 | 12 | ||
| 13 | 13 | // hintOptions builds the hint worker options: the accessibility source, and, | |
| 14 | 14 | // when NEFERWL_HINT_DEBUG is "1", the dump directory of scanned frames, | |
| 15 | - | // $XDG_STATE_HOME/neferwl/hints. Without the accessibility bus the worker | |
| 16 | - | // reads uncovered windows with OCR only. | |
| 15 | + | // $XDG_STATE_HOME/neferwl/hints. The accessibility bus is only contacted | |
| 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 | - | src, err := atspi.New(ctx) | |
| 20 | - | if err != nil { | |
| 21 | - | log := logging.For(ctx, "atspi") | |
| 22 | - | log.Warn().Err(err).Msg("accessibility hint targets unavailable, using OCR only") | |
| 23 | - | return opts | |
| 24 | - | } | |
| 19 | + | // The source connects on its first scan, and again after a failure. | |
| 20 | + | src := atspi.New() | |
| 25 | 21 | context.AfterFunc(ctx, func() { _ = src.Close() }) | |
| 26 | 22 | opts.Source = src | |
| 27 | 23 | return opts | |
| @@ -12,16 +12,12 @@ import ( | |||||
| 12 | 12 | ||||
| 13 | // hintOptions builds the hint worker options: the accessibility source, and, | 13 | // hintOptions builds the hint worker options: the accessibility source, and, | ||
| 14 | // when NEFERWL_HINT_DEBUG is "1", the dump directory of scanned frames, | 14 | // when NEFERWL_HINT_DEBUG is "1", the dump directory of scanned frames, | ||
| 15 | - | // $XDG_STATE_HOME/neferwl/hints. Without the accessibility bus the worker | 15 | + | // $XDG_STATE_HOME/neferwl/hints. The accessibility bus is only contacted |
| 16 | - | // reads uncovered windows with OCR only. | 16 | + | // by the first scan; without it the worker reads windows with OCR only. |
| 17 | func hintOptions(ctx context.Context) hints.Options { | 17 | func hintOptions(ctx context.Context) hints.Options { | ||
| 18 | opts := hintDump(ctx) | 18 | opts := hintDump(ctx) | ||
| 19 | - | src, err := atspi.New(ctx) | 19 | + | // The source connects on its first scan, and again after a failure. |
| 20 | - | if err != nil { | 20 | + | src := atspi.New() |
| 21 | - | log := logging.For(ctx, "atspi") | |||
| 22 | - | log.Warn().Err(err).Msg("accessibility hint targets unavailable, using OCR only") | |||
| 23 | - | return opts | |||
| 24 | - | } | |||
| 25 | context.AfterFunc(ctx, func() { _ = src.Close() }) | 21 | context.AfterFunc(ctx, func() { _ = src.Close() }) | ||
| 26 | opts.Source = src | 22 | opts.Source = src | ||
| 27 | return opts | 23 | return opts | ||