detached fd94b61

refactor(daemon): track attached commands per connection

commit
fd94b61c7d5ee8de33479f4528966d054281a029
author
brice · · 2026-08-03 12:19 UTC
committer
brice · · 2026-08-04 07:16 UTC
parent
c4be05e
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +416 -0
Unified
UnifiedSplit
A internal/usecase/daemon/attachment_commands.go +155 -0
@@ -0,0 +1,155 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "errors"
6+ "sync"
7+ "time"
8+
9+ "github.com/bnema/vev/internal/ports"
10+)
11+
12+const attachedCommandTimeout = 10 * time.Second
13+
14+var (
15+ errAttachedCommandTimeout = errors.New("attached command timed out")
16+ errAttachedCommandUnavailable = errors.New("attached command connection is unavailable")
17+)
18+
19+type attachedCommandOutcome struct {
20+ result ports.CommandResult
21+ err error
22+}
23+
24+type attachedCommandPending struct {
25+ generation uint64
26+ outcome chan attachedCommandOutcome
27+}
28+
29+// attachedCommandTracker owns command correlation for one attachment
30+// connection. Its lock protects only request publication state; command waits
31+// happen after the attachment sender lock is released.
32+type attachedCommandTracker struct {
33+ commandMu sync.Mutex
34+ mu sync.Mutex
35+ next uint64
36+ pending map[uint64]attachedCommandPending
37+}
38+
39+// sendCommand publishes one attached command and waits for its exact result.
40+// commandMu keeps interactive palette commands ordered for one attachment;
41+// other attachment traffic uses only sendMu during request publication.
42+func (ac *attachedClient) sendCommand(ctx context.Context, clock ports.Clock, slug string, args []string) (ports.CommandResult, error) {
43+ if ac == nil || clock == nil {
44+ return ports.CommandResult{}, errAttachedCommandUnavailable
45+ }
46+ if ctx == nil {
47+ ctx = context.Background()
48+ }
49+ if err := ctx.Err(); err != nil {
50+ return ports.CommandResult{}, err
51+ }
52+
53+ ac.commands.commandMu.Lock()
54+ defer ac.commands.commandMu.Unlock()
55+ if err := ctx.Err(); err != nil {
56+ return ports.CommandResult{}, err
57+ }
58+
59+ ac.sendMu.Lock()
60+ expected := ac.transportSnapshot()
61+ generation := ac.connectionGeneration.Load()
62+ if expected.transport == nil {
63+ ac.sendMu.Unlock()
64+ return ports.CommandResult{}, errAttachedCommandUnavailable
65+ }
66+ ac.commands.mu.Lock()
67+ if ac.commands.pending == nil {
68+ ac.commands.pending = make(map[uint64]attachedCommandPending)
69+ }
70+ ac.commands.next++
71+ if ac.commands.next == 0 {
72+ ac.commands.next++
73+ }
74+ requestID := ac.commands.next
75+ outcome := make(chan attachedCommandOutcome, 1)
76+ ac.commands.pending[requestID] = attachedCommandPending{generation: generation, outcome: outcome}
77+ ac.commands.mu.Unlock()
78+ if !ac.transportSnapshotCurrent(expected) || ac.connectionGeneration.Load() != generation {
79+ ac.removeAttachedCommand(requestID, generation)
80+ ac.sendMu.Unlock()
81+ return ports.CommandResult{}, errAttachedCommandUnavailable
82+ }
83+
84+ payload, err := ports.MarshalCommandRequest(ports.CommandRequest{
85+ Version: ports.ProtocolVersion, RequestID: requestID, Attached: true,
86+ Slug: slug, Args: append([]string(nil), args...),
87+ })
88+ if err != nil {
89+ ac.removeAttachedCommand(requestID, generation)
90+ ac.sendMu.Unlock()
91+ return ports.CommandResult{}, err
92+ }
93+ if err := expected.transport.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {
94+ ac.removeAttachedCommand(requestID, generation)
95+ ac.sendMu.Unlock()
96+ return ports.CommandResult{}, err
97+ }
98+ ac.sendMu.Unlock()
99+
100+ timer := clock.NewTimer(attachedCommandTimeout)
101+ if timer == nil {
102+ ac.removeAttachedCommand(requestID, generation)
103+ return ports.CommandResult{}, errAttachedCommandTimeout
104+ }
105+ defer timer.Stop()
106+ select {
107+ case completed := <-outcome:
108+ return completed.result, completed.err
109+ case <-ctx.Done():
110+ ac.removeAttachedCommand(requestID, generation)
111+ return ports.CommandResult{}, ctx.Err()
112+ case <-timer.C():
113+ ac.removeAttachedCommand(requestID, generation)
114+ return ports.CommandResult{}, errAttachedCommandTimeout
115+ }
116+}
117+
118+func (ac *attachedClient) removeAttachedCommand(requestID, generation uint64) {
119+ if ac == nil {
120+ return
121+ }
122+ ac.commands.mu.Lock()
123+ pending, ok := ac.commands.pending[requestID]
124+ if ok && pending.generation == generation {
125+ delete(ac.commands.pending, requestID)
126+ }
127+ ac.commands.mu.Unlock()
128+}
129+
130+// completeCommandResult accepts only a result for the exact request and
131+// connection generation. Unknown, late, and old-generation results are safe
132+// no-ops.
133+func (ac *attachedClient) completeCommandResult(generation uint64, result ports.CommandResult) {
134+ if ac == nil || result.RequestID == 0 || ac.connectionGeneration.Load() != generation {
135+ return
136+ }
137+ ac.commands.mu.Lock()
138+ pending, ok := ac.commands.pending[result.RequestID]
139+ if !ok || pending.generation != generation {
140+ ac.commands.mu.Unlock()
141+ return
142+ }
143+ delete(ac.commands.pending, result.RequestID)
144+ ac.commands.mu.Unlock()
145+ pending.outcome <- attachedCommandOutcome{result: result}
146+}
147+
148+func (ac *attachedClient) pendingAttachedCommandCount() int {
149+ if ac == nil {
150+ return 0
151+ }
152+ ac.commands.mu.Lock()
153+ defer ac.commands.mu.Unlock()
154+ return len(ac.commands.pending)
155+}
@@ -0,0 +1,155 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "errors"
6+ "sync"
7+ "time"
8+
9+ "github.com/bnema/vev/internal/ports"
10+)
11+
12+const attachedCommandTimeout = 10 * time.Second
13+
14+var (
15+ errAttachedCommandTimeout = errors.New("attached command timed out")
16+ errAttachedCommandUnavailable = errors.New("attached command connection is unavailable")
17+)
18+
19+type attachedCommandOutcome struct {
20+ result ports.CommandResult
21+ err error
22+}
23+
24+type attachedCommandPending struct {
25+ generation uint64
26+ outcome chan attachedCommandOutcome
27+}
28+
29+// attachedCommandTracker owns command correlation for one attachment
30+// connection. Its lock protects only request publication state; command waits
31+// happen after the attachment sender lock is released.
32+type attachedCommandTracker struct {
33+ commandMu sync.Mutex
34+ mu sync.Mutex
35+ next uint64
36+ pending map[uint64]attachedCommandPending
37+}
38+
39+// sendCommand publishes one attached command and waits for its exact result.
40+// commandMu keeps interactive palette commands ordered for one attachment;
41+// other attachment traffic uses only sendMu during request publication.
42+func (ac *attachedClient) sendCommand(ctx context.Context, clock ports.Clock, slug string, args []string) (ports.CommandResult, error) {
43+ if ac == nil || clock == nil {
44+ return ports.CommandResult{}, errAttachedCommandUnavailable
45+ }
46+ if ctx == nil {
47+ ctx = context.Background()
48+ }
49+ if err := ctx.Err(); err != nil {
50+ return ports.CommandResult{}, err
51+ }
52+
53+ ac.commands.commandMu.Lock()
54+ defer ac.commands.commandMu.Unlock()
55+ if err := ctx.Err(); err != nil {
56+ return ports.CommandResult{}, err
57+ }
58+
59+ ac.sendMu.Lock()
60+ expected := ac.transportSnapshot()
61+ generation := ac.connectionGeneration.Load()
62+ if expected.transport == nil {
63+ ac.sendMu.Unlock()
64+ return ports.CommandResult{}, errAttachedCommandUnavailable
65+ }
66+ ac.commands.mu.Lock()
67+ if ac.commands.pending == nil {
68+ ac.commands.pending = make(map[uint64]attachedCommandPending)
69+ }
70+ ac.commands.next++
71+ if ac.commands.next == 0 {
72+ ac.commands.next++
73+ }
74+ requestID := ac.commands.next
75+ outcome := make(chan attachedCommandOutcome, 1)
76+ ac.commands.pending[requestID] = attachedCommandPending{generation: generation, outcome: outcome}
77+ ac.commands.mu.Unlock()
78+ if !ac.transportSnapshotCurrent(expected) || ac.connectionGeneration.Load() != generation {
79+ ac.removeAttachedCommand(requestID, generation)
80+ ac.sendMu.Unlock()
81+ return ports.CommandResult{}, errAttachedCommandUnavailable
82+ }
83+
84+ payload, err := ports.MarshalCommandRequest(ports.CommandRequest{
85+ Version: ports.ProtocolVersion, RequestID: requestID, Attached: true,
86+ Slug: slug, Args: append([]string(nil), args...),
87+ })
88+ if err != nil {
89+ ac.removeAttachedCommand(requestID, generation)
90+ ac.sendMu.Unlock()
91+ return ports.CommandResult{}, err
92+ }
93+ if err := expected.transport.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {
94+ ac.removeAttachedCommand(requestID, generation)
95+ ac.sendMu.Unlock()
96+ return ports.CommandResult{}, err
97+ }
98+ ac.sendMu.Unlock()
99+
100+ timer := clock.NewTimer(attachedCommandTimeout)
101+ if timer == nil {
102+ ac.removeAttachedCommand(requestID, generation)
103+ return ports.CommandResult{}, errAttachedCommandTimeout
104+ }
105+ defer timer.Stop()
106+ select {
107+ case completed := <-outcome:
108+ return completed.result, completed.err
109+ case <-ctx.Done():
110+ ac.removeAttachedCommand(requestID, generation)
111+ return ports.CommandResult{}, ctx.Err()
112+ case <-timer.C():
113+ ac.removeAttachedCommand(requestID, generation)
114+ return ports.CommandResult{}, errAttachedCommandTimeout
115+ }
116+}
117+
118+func (ac *attachedClient) removeAttachedCommand(requestID, generation uint64) {
119+ if ac == nil {
120+ return
121+ }
122+ ac.commands.mu.Lock()
123+ pending, ok := ac.commands.pending[requestID]
124+ if ok && pending.generation == generation {
125+ delete(ac.commands.pending, requestID)
126+ }
127+ ac.commands.mu.Unlock()
128+}
129+
130+// completeCommandResult accepts only a result for the exact request and
131+// connection generation. Unknown, late, and old-generation results are safe
132+// no-ops.
133+func (ac *attachedClient) completeCommandResult(generation uint64, result ports.CommandResult) {
134+ if ac == nil || result.RequestID == 0 || ac.connectionGeneration.Load() != generation {
135+ return
136+ }
137+ ac.commands.mu.Lock()
138+ pending, ok := ac.commands.pending[result.RequestID]
139+ if !ok || pending.generation != generation {
140+ ac.commands.mu.Unlock()
141+ return
142+ }
143+ delete(ac.commands.pending, result.RequestID)
144+ ac.commands.mu.Unlock()
145+ pending.outcome <- attachedCommandOutcome{result: result}
146+}
147+
148+func (ac *attachedClient) pendingAttachedCommandCount() int {
149+ if ac == nil {
150+ return 0
151+ }
152+ ac.commands.mu.Lock()
153+ defer ac.commands.mu.Unlock()
154+ return len(ac.commands.pending)
155+}
A internal/usecase/daemon/attachment_commands_test.go +260 -0
@@ -0,0 +1,260 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "errors"
6+ "sync"
7+ "testing"
8+ "time"
9+
10+ "github.com/stretchr/testify/require"
11+
12+ "github.com/bnema/vev/internal/ports"
13+)
14+
15+type attachmentCommandTestTransport struct {
16+ sent chan ports.Frame
17+ entered chan struct{}
18+ gate <-chan struct{}
19+}
20+
21+func newAttachmentCommandTestTransport() *attachmentCommandTestTransport {
22+ return &attachmentCommandTestTransport{sent: make(chan ports.Frame, 8)}
23+}
24+
25+func (t *attachmentCommandTestTransport) Send(frame ports.Frame) error {
26+ if t.entered != nil {
27+ select {
28+ case t.entered <- struct{}{}:
29+ default:
30+ }
31+ }
32+ if t.gate != nil {
33+ <-t.gate
34+ }
35+ t.sent <- frame
36+ return nil
37+}
38+
39+func (t *attachmentCommandTestTransport) Recv() (ports.Frame, error) {
40+ return ports.Frame{}, nil
41+}
42+
43+func (t *attachmentCommandTestTransport) Close() error { return nil }
44+
45+type attachmentCommandTestTimer struct {
46+ ch chan time.Time
47+ once sync.Once
48+}
49+
50+func (t *attachmentCommandTestTimer) C() <-chan time.Time { return t.ch }
51+func (t *attachmentCommandTestTimer) Reset(time.Duration) bool { return false }
52+func (t *attachmentCommandTestTimer) Stop() bool { return true }
53+func (t *attachmentCommandTestTimer) fire() { t.once.Do(func() { t.ch <- time.Time{} }) }
54+
55+type attachmentCommandTestClock struct {
56+ timers chan *attachmentCommandTestTimer
57+}
58+
59+func newAttachmentCommandTestClock() *attachmentCommandTestClock {
60+ return &attachmentCommandTestClock{timers: make(chan *attachmentCommandTestTimer, 8)}
61+}
62+
63+func (c *attachmentCommandTestClock) Now() time.Time { return time.Time{} }
64+
65+func (c *attachmentCommandTestClock) NewTimer(delay time.Duration) ports.Timer {
66+ if delay != attachedCommandTimeout {
67+ return stubTimer{}
68+ }
69+ timer := &attachmentCommandTestTimer{ch: make(chan time.Time, 1)}
70+ c.timers <- timer
71+ return timer
72+}
73+
74+func awaitAttachedCommandFrame(t *testing.T, transport *attachmentCommandTestTransport) ports.CommandRequest {
75+ t.Helper()
76+ select {
77+ case frame := <-transport.sent:
78+ require.Equal(t, ports.MsgCommand, frame.Type)
79+ request, err := ports.UnmarshalCommandRequest(frame.Payload)
80+ require.NoError(t, err)
81+ return request
82+ case <-time.After(time.Second):
83+ t.Fatal("attached command was not published")
84+ return ports.CommandRequest{}
85+ }
86+}
87+
88+func awaitAttachedCommandTimer(t *testing.T, clock *attachmentCommandTestClock) *attachmentCommandTestTimer {
89+ t.Helper()
90+ select {
91+ case timer := <-clock.timers:
92+ return timer
93+ case <-time.After(time.Second):
94+ t.Fatal("attached command timeout was not armed")
95+ return nil
96+ }
97+}
98+
99+func awaitAttachedCommandResult(t *testing.T, done <-chan attachedCommandOutcome) attachedCommandOutcome {
100+ t.Helper()
101+ select {
102+ case result := <-done:
103+ return result
104+ case <-time.After(time.Second):
105+ t.Fatal("attached command did not finish")
106+ return attachedCommandOutcome{}
107+ }
108+}
109+
110+func startAttachedCommand(t *testing.T, ac *attachedClient, clock *attachmentCommandTestClock, slug string) (ports.CommandRequest, *attachmentCommandTestTimer, <-chan attachedCommandOutcome) {
111+ t.Helper()
112+ done := make(chan attachedCommandOutcome, 1)
113+ go func() {
114+ result, err := ac.sendCommand(context.Background(), clock, slug, nil)
115+ done <- attachedCommandOutcome{result: result, err: err}
116+ }()
117+ request := awaitAttachedCommandFrame(t, ac.transport().(*attachmentCommandTestTransport))
118+ return request, awaitAttachedCommandTimer(t, clock), done
119+}
120+
121+func TestAttachedCommandWaitReleasesSenderLockAfterPublication(t *testing.T) {
122+ transport := newAttachmentCommandTestTransport()
123+ gate := make(chan struct{})
124+ transport.gate = gate
125+ transport.entered = make(chan struct{}, 1)
126+ ac := &attachedClient{tr: transport}
127+ ac.connectionGeneration.Store(1)
128+ clock := newAttachmentCommandTestClock()
129+ done := make(chan attachedCommandOutcome, 1)
130+ go func() {
131+ result, err := ac.sendCommand(context.Background(), clock, "next-tab", nil)
132+ done <- attachedCommandOutcome{result: result, err: err}
133+ }()
134+
135+ select {
136+ case <-transport.entered:
137+ case <-time.After(time.Second):
138+ t.Fatal("command publication did not reach the transport")
139+ }
140+ close(gate)
141+ request := awaitAttachedCommandFrame(t, transport)
142+ _ = awaitAttachedCommandTimer(t, clock)
143+
144+ require.True(t, ac.sendMu.TryLock(), "command wait retained sender lock")
145+ ac.sendMu.Unlock()
146+ require.NoError(t, ac.sendExpectedTransport(ac.transportSnapshot(), ports.Frame{Type: ports.MsgPing}))
147+ select {
148+ case frame := <-transport.sent:
149+ require.Equal(t, ports.MsgPing, frame.Type)
150+ case <-time.After(time.Second):
151+ t.Fatal("ping was blocked by command wait")
152+ }
153+
154+ ac.completeCommandResult(1, ports.CommandResult{RequestID: request.RequestID, OK: true})
155+ result := awaitAttachedCommandResult(t, done)
156+ require.NoError(t, result.err)
157+ require.True(t, result.result.OK)
158+}
159+
160+func TestAttachedCommandTimeoutRemovesOnlyRequestAndIgnoresLateResult(t *testing.T) {
161+ transport := newAttachmentCommandTestTransport()
162+ ac := &attachedClient{tr: transport}
163+ ac.connectionGeneration.Store(4)
164+ clock := newAttachmentCommandTestClock()
165+
166+ first, timer, done := startAttachedCommand(t, ac, clock, "next-tab")
167+ timer.fire()
168+ result := awaitAttachedCommandResult(t, done)
169+ require.ErrorIs(t, result.err, errAttachedCommandTimeout)
170+ require.Zero(t, ac.pendingAttachedCommandCount())
171+
172+ ac.completeCommandResult(4, ports.CommandResult{RequestID: first.RequestID, OK: true})
173+ select {
174+ case <-done:
175+ t.Fatal("late result completed a timed-out command")
176+ default:
177+ }
178+
179+ second, secondTimer, secondDone := startAttachedCommand(t, ac, clock, "previous-tab")
180+ require.Equal(t, first.RequestID+1, second.RequestID)
181+ ac.completeCommandResult(4, ports.CommandResult{RequestID: first.RequestID, OK: true})
182+ select {
183+ case <-secondDone:
184+ t.Fatal("old request ID completed a newer command")
185+ default:
186+ }
187+ ac.completeCommandResult(4, ports.CommandResult{RequestID: second.RequestID, OK: true})
188+ result = awaitAttachedCommandResult(t, secondDone)
189+ require.NoError(t, result.err)
190+ require.True(t, result.result.OK)
191+ secondTimer.fire()
192+}
193+
194+func TestAttachedCommandIgnoresOldGenerationResults(t *testing.T) {
195+ transport := newAttachmentCommandTestTransport()
196+ ac := &attachedClient{tr: transport}
197+ ac.connectionGeneration.Store(7)
198+ clock := newAttachmentCommandTestClock()
199+
200+ request, timer, done := startAttachedCommand(t, ac, clock, "next-tab")
201+ ac.connectionGeneration.Store(8)
202+ ac.completeCommandResult(7, ports.CommandResult{RequestID: request.RequestID, OK: true})
203+ select {
204+ case <-done:
205+ t.Fatal("old-generation result completed a command")
206+ default:
207+ }
208+ timer.fire()
209+ result := awaitAttachedCommandResult(t, done)
210+ require.ErrorIs(t, result.err, errAttachedCommandTimeout)
211+}
212+
213+func TestAttachedCommandTrackingIsIndependentPerAttachment(t *testing.T) {
214+ firstTransport := newAttachmentCommandTestTransport()
215+ secondTransport := newAttachmentCommandTestTransport()
216+ first := &attachedClient{tr: firstTransport}
217+ second := &attachedClient{tr: secondTransport}
218+ first.connectionGeneration.Store(1)
219+ second.connectionGeneration.Store(1)
220+ clock := newAttachmentCommandTestClock()
221+
222+ firstDone := make(chan attachedCommandOutcome, 1)
223+ go func() {
224+ result, err := first.sendCommand(context.Background(), clock, "next-tab", nil)
225+ firstDone <- attachedCommandOutcome{result: result, err: err}
226+ }()
227+ secondDone := make(chan attachedCommandOutcome, 1)
228+ go func() {
229+ result, err := second.sendCommand(context.Background(), clock, "previous-tab", nil)
230+ secondDone <- attachedCommandOutcome{result: result, err: err}
231+ }()
232+
233+ firstRequest := awaitAttachedCommandFrame(t, firstTransport)
234+ secondRequest := awaitAttachedCommandFrame(t, secondTransport)
235+ require.Equal(t, uint64(1), firstRequest.RequestID)
236+ require.Equal(t, uint64(1), secondRequest.RequestID)
237+ _ = awaitAttachedCommandTimer(t, clock)
238+ _ = awaitAttachedCommandTimer(t, clock)
239+
240+ first.completeCommandResult(1, ports.CommandResult{RequestID: firstRequest.RequestID, OK: true})
241+ second.completeCommandResult(1, ports.CommandResult{RequestID: secondRequest.RequestID, OK: true})
242+ require.NoError(t, awaitAttachedCommandResult(t, firstDone).err)
243+ require.NoError(t, awaitAttachedCommandResult(t, secondDone).err)
244+ require.Equal(t, 0, first.pendingAttachedCommandCount())
245+ require.Equal(t, 0, second.pendingAttachedCommandCount())
246+}
247+
248+func TestAttachedCommandUnavailableAndContextCancellation(t *testing.T) {
249+ var nilAttachment *attachedClient
250+ clock := newAttachmentCommandTestClock()
251+ _, err := nilAttachment.sendCommand(context.Background(), clock, "next-tab", nil)
252+ require.True(t, errors.Is(err, errAttachedCommandUnavailable))
253+
254+ transport := newAttachmentCommandTestTransport()
255+ ac := &attachedClient{tr: transport}
256+ ctx, cancel := context.WithCancel(context.Background())
257+ cancel()
258+ _, err = ac.sendCommand(ctx, clock, "next-tab", nil)
259+ require.ErrorIs(t, err, context.Canceled)
260+}
@@ -0,0 +1,260 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "errors"
6+ "sync"
7+ "testing"
8+ "time"
9+
10+ "github.com/stretchr/testify/require"
11+
12+ "github.com/bnema/vev/internal/ports"
13+)
14+
15+type attachmentCommandTestTransport struct {
16+ sent chan ports.Frame
17+ entered chan struct{}
18+ gate <-chan struct{}
19+}
20+
21+func newAttachmentCommandTestTransport() *attachmentCommandTestTransport {
22+ return &attachmentCommandTestTransport{sent: make(chan ports.Frame, 8)}
23+}
24+
25+func (t *attachmentCommandTestTransport) Send(frame ports.Frame) error {
26+ if t.entered != nil {
27+ select {
28+ case t.entered <- struct{}{}:
29+ default:
30+ }
31+ }
32+ if t.gate != nil {
33+ <-t.gate
34+ }
35+ t.sent <- frame
36+ return nil
37+}
38+
39+func (t *attachmentCommandTestTransport) Recv() (ports.Frame, error) {
40+ return ports.Frame{}, nil
41+}
42+
43+func (t *attachmentCommandTestTransport) Close() error { return nil }
44+
45+type attachmentCommandTestTimer struct {
46+ ch chan time.Time
47+ once sync.Once
48+}
49+
50+func (t *attachmentCommandTestTimer) C() <-chan time.Time { return t.ch }
51+func (t *attachmentCommandTestTimer) Reset(time.Duration) bool { return false }
52+func (t *attachmentCommandTestTimer) Stop() bool { return true }
53+func (t *attachmentCommandTestTimer) fire() { t.once.Do(func() { t.ch <- time.Time{} }) }
54+
55+type attachmentCommandTestClock struct {
56+ timers chan *attachmentCommandTestTimer
57+}
58+
59+func newAttachmentCommandTestClock() *attachmentCommandTestClock {
60+ return &attachmentCommandTestClock{timers: make(chan *attachmentCommandTestTimer, 8)}
61+}
62+
63+func (c *attachmentCommandTestClock) Now() time.Time { return time.Time{} }
64+
65+func (c *attachmentCommandTestClock) NewTimer(delay time.Duration) ports.Timer {
66+ if delay != attachedCommandTimeout {
67+ return stubTimer{}
68+ }
69+ timer := &attachmentCommandTestTimer{ch: make(chan time.Time, 1)}
70+ c.timers <- timer
71+ return timer
72+}
73+
74+func awaitAttachedCommandFrame(t *testing.T, transport *attachmentCommandTestTransport) ports.CommandRequest {
75+ t.Helper()
76+ select {
77+ case frame := <-transport.sent:
78+ require.Equal(t, ports.MsgCommand, frame.Type)
79+ request, err := ports.UnmarshalCommandRequest(frame.Payload)
80+ require.NoError(t, err)
81+ return request
82+ case <-time.After(time.Second):
83+ t.Fatal("attached command was not published")
84+ return ports.CommandRequest{}
85+ }
86+}
87+
88+func awaitAttachedCommandTimer(t *testing.T, clock *attachmentCommandTestClock) *attachmentCommandTestTimer {
89+ t.Helper()
90+ select {
91+ case timer := <-clock.timers:
92+ return timer
93+ case <-time.After(time.Second):
94+ t.Fatal("attached command timeout was not armed")
95+ return nil
96+ }
97+}
98+
99+func awaitAttachedCommandResult(t *testing.T, done <-chan attachedCommandOutcome) attachedCommandOutcome {
100+ t.Helper()
101+ select {
102+ case result := <-done:
103+ return result
104+ case <-time.After(time.Second):
105+ t.Fatal("attached command did not finish")
106+ return attachedCommandOutcome{}
107+ }
108+}
109+
110+func startAttachedCommand(t *testing.T, ac *attachedClient, clock *attachmentCommandTestClock, slug string) (ports.CommandRequest, *attachmentCommandTestTimer, <-chan attachedCommandOutcome) {
111+ t.Helper()
112+ done := make(chan attachedCommandOutcome, 1)
113+ go func() {
114+ result, err := ac.sendCommand(context.Background(), clock, slug, nil)
115+ done <- attachedCommandOutcome{result: result, err: err}
116+ }()
117+ request := awaitAttachedCommandFrame(t, ac.transport().(*attachmentCommandTestTransport))
118+ return request, awaitAttachedCommandTimer(t, clock), done
119+}
120+
121+func TestAttachedCommandWaitReleasesSenderLockAfterPublication(t *testing.T) {
122+ transport := newAttachmentCommandTestTransport()
123+ gate := make(chan struct{})
124+ transport.gate = gate
125+ transport.entered = make(chan struct{}, 1)
126+ ac := &attachedClient{tr: transport}
127+ ac.connectionGeneration.Store(1)
128+ clock := newAttachmentCommandTestClock()
129+ done := make(chan attachedCommandOutcome, 1)
130+ go func() {
131+ result, err := ac.sendCommand(context.Background(), clock, "next-tab", nil)
132+ done <- attachedCommandOutcome{result: result, err: err}
133+ }()
134+
135+ select {
136+ case <-transport.entered:
137+ case <-time.After(time.Second):
138+ t.Fatal("command publication did not reach the transport")
139+ }
140+ close(gate)
141+ request := awaitAttachedCommandFrame(t, transport)
142+ _ = awaitAttachedCommandTimer(t, clock)
143+
144+ require.True(t, ac.sendMu.TryLock(), "command wait retained sender lock")
145+ ac.sendMu.Unlock()
146+ require.NoError(t, ac.sendExpectedTransport(ac.transportSnapshot(), ports.Frame{Type: ports.MsgPing}))
147+ select {
148+ case frame := <-transport.sent:
149+ require.Equal(t, ports.MsgPing, frame.Type)
150+ case <-time.After(time.Second):
151+ t.Fatal("ping was blocked by command wait")
152+ }
153+
154+ ac.completeCommandResult(1, ports.CommandResult{RequestID: request.RequestID, OK: true})
155+ result := awaitAttachedCommandResult(t, done)
156+ require.NoError(t, result.err)
157+ require.True(t, result.result.OK)
158+}
159+
160+func TestAttachedCommandTimeoutRemovesOnlyRequestAndIgnoresLateResult(t *testing.T) {
161+ transport := newAttachmentCommandTestTransport()
162+ ac := &attachedClient{tr: transport}
163+ ac.connectionGeneration.Store(4)
164+ clock := newAttachmentCommandTestClock()
165+
166+ first, timer, done := startAttachedCommand(t, ac, clock, "next-tab")
167+ timer.fire()
168+ result := awaitAttachedCommandResult(t, done)
169+ require.ErrorIs(t, result.err, errAttachedCommandTimeout)
170+ require.Zero(t, ac.pendingAttachedCommandCount())
171+
172+ ac.completeCommandResult(4, ports.CommandResult{RequestID: first.RequestID, OK: true})
173+ select {
174+ case <-done:
175+ t.Fatal("late result completed a timed-out command")
176+ default:
177+ }
178+
179+ second, secondTimer, secondDone := startAttachedCommand(t, ac, clock, "previous-tab")
180+ require.Equal(t, first.RequestID+1, second.RequestID)
181+ ac.completeCommandResult(4, ports.CommandResult{RequestID: first.RequestID, OK: true})
182+ select {
183+ case <-secondDone:
184+ t.Fatal("old request ID completed a newer command")
185+ default:
186+ }
187+ ac.completeCommandResult(4, ports.CommandResult{RequestID: second.RequestID, OK: true})
188+ result = awaitAttachedCommandResult(t, secondDone)
189+ require.NoError(t, result.err)
190+ require.True(t, result.result.OK)
191+ secondTimer.fire()
192+}
193+
194+func TestAttachedCommandIgnoresOldGenerationResults(t *testing.T) {
195+ transport := newAttachmentCommandTestTransport()
196+ ac := &attachedClient{tr: transport}
197+ ac.connectionGeneration.Store(7)
198+ clock := newAttachmentCommandTestClock()
199+
200+ request, timer, done := startAttachedCommand(t, ac, clock, "next-tab")
201+ ac.connectionGeneration.Store(8)
202+ ac.completeCommandResult(7, ports.CommandResult{RequestID: request.RequestID, OK: true})
203+ select {
204+ case <-done:
205+ t.Fatal("old-generation result completed a command")
206+ default:
207+ }
208+ timer.fire()
209+ result := awaitAttachedCommandResult(t, done)
210+ require.ErrorIs(t, result.err, errAttachedCommandTimeout)
211+}
212+
213+func TestAttachedCommandTrackingIsIndependentPerAttachment(t *testing.T) {
214+ firstTransport := newAttachmentCommandTestTransport()
215+ secondTransport := newAttachmentCommandTestTransport()
216+ first := &attachedClient{tr: firstTransport}
217+ second := &attachedClient{tr: secondTransport}
218+ first.connectionGeneration.Store(1)
219+ second.connectionGeneration.Store(1)
220+ clock := newAttachmentCommandTestClock()
221+
222+ firstDone := make(chan attachedCommandOutcome, 1)
223+ go func() {
224+ result, err := first.sendCommand(context.Background(), clock, "next-tab", nil)
225+ firstDone <- attachedCommandOutcome{result: result, err: err}
226+ }()
227+ secondDone := make(chan attachedCommandOutcome, 1)
228+ go func() {
229+ result, err := second.sendCommand(context.Background(), clock, "previous-tab", nil)
230+ secondDone <- attachedCommandOutcome{result: result, err: err}
231+ }()
232+
233+ firstRequest := awaitAttachedCommandFrame(t, firstTransport)
234+ secondRequest := awaitAttachedCommandFrame(t, secondTransport)
235+ require.Equal(t, uint64(1), firstRequest.RequestID)
236+ require.Equal(t, uint64(1), secondRequest.RequestID)
237+ _ = awaitAttachedCommandTimer(t, clock)
238+ _ = awaitAttachedCommandTimer(t, clock)
239+
240+ first.completeCommandResult(1, ports.CommandResult{RequestID: firstRequest.RequestID, OK: true})
241+ second.completeCommandResult(1, ports.CommandResult{RequestID: secondRequest.RequestID, OK: true})
242+ require.NoError(t, awaitAttachedCommandResult(t, firstDone).err)
243+ require.NoError(t, awaitAttachedCommandResult(t, secondDone).err)
244+ require.Equal(t, 0, first.pendingAttachedCommandCount())
245+ require.Equal(t, 0, second.pendingAttachedCommandCount())
246+}
247+
248+func TestAttachedCommandUnavailableAndContextCancellation(t *testing.T) {
249+ var nilAttachment *attachedClient
250+ clock := newAttachmentCommandTestClock()
251+ _, err := nilAttachment.sendCommand(context.Background(), clock, "next-tab", nil)
252+ require.True(t, errors.Is(err, errAttachedCommandUnavailable))
253+
254+ transport := newAttachmentCommandTestTransport()
255+ ac := &attachedClient{tr: transport}
256+ ctx, cancel := context.WithCancel(context.Background())
257+ cancel()
258+ _, err = ac.sendCommand(ctx, clock, "next-tab", nil)
259+ require.ErrorIs(t, err, context.Canceled)
260+}
M internal/usecase/daemon/client.go +1 -0
@@ -85,6 +85,7 @@ type attachedClient struct {
8585 previousSession Guarded[attachmentSession]
8686 linkMu sync.Mutex
8787 sendMu sync.Mutex
88+ commands attachedCommandTracker
8889 // beforeAttachmentTokenValidation is a deterministic lifecycle-race seam
8990 // used only by package tests.
9091 beforeAttachmentTokenValidation func()
@@ -85,6 +85,7 @@ type attachedClient struct {
85 previousSession Guarded[attachmentSession]85 previousSession Guarded[attachmentSession]
86 linkMu sync.Mutex86 linkMu sync.Mutex
87 sendMu sync.Mutex87 sendMu sync.Mutex
88+ commands attachedCommandTracker
88 // beforeAttachmentTokenValidation is a deterministic lifecycle-race seam89 // beforeAttachmentTokenValidation is a deterministic lifecycle-race seam
89 // used only by package tests.90 // used only by package tests.
90 beforeAttachmentTokenValidation func()91 beforeAttachmentTokenValidation func()

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close