detached 4e35cb5

fix(client): cancel blocked reconnect opens and decode held resume input

Decode resume keys incrementally (grouped Ctrl-C, fragmented Kitty keys, paste, lone-Esc timer), cancel a blocked reconnect OpenStream from the resume watch, settle a failed outbox before rearming progress, and cover outbox bounds, drain outcomes and held-input replay.
commit
4e35cb59fc43e9edbbebeb5fbeb22aee1a0c249c
author
brice · · 2026-10-07 05:34 UTC
parent
f5e4a3e
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
11 files · +548 -26
Unified
UnifiedSplit
M internal/usecase/client/attachment_outbox_test.go +7 -2
@@ -103,10 +103,11 @@ func TestResumeWatch(t *testing.T) {
103103 name string
104104 reads []string
105105 wantCancel bool
106+ fireEscape bool
106107 wantResidual string
107108 }{
108109 {name: "ctrl-c cancels and drops held keys", reads: []string{"ls", "\x03"}, wantCancel: true},
109- {name: "lone esc cancels", reads: []string{"\x1b"}, wantCancel: true},
110+ {name: "lone esc cancels", reads: []string{"\x1b"}, wantCancel: true, fireEscape: true},
110111 {name: "kitty esc cancels", reads: []string{"\x1b[27u"}, wantCancel: true},
111112 {name: "kitty ctrl-c cancels", reads: []string{"\x1b[99;5u"}, wantCancel: true},
112113 {name: "kitty ctrl-c with lock bits cancels", reads: []string{"\x1b[99;69u"}, wantCancel: true},
@@ -123,11 +124,15 @@ func TestResumeWatch(t *testing.T) {
123124 close(reader.chunks)
124125 })
125126 lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
126- watch := lifetime.watchResume()
127+ clock := newSupervisorTestClock()
128+ watch := lifetime.watchResume(clock)
127129 for _, read := range tt.reads {
128130 reader.chunks <- []byte(read)
129131 }
130132 if tt.wantCancel {
133+ if tt.fireEscape {
134+ clock.awaitTimer(t).fire()
135+ }
131136 select {
132137 case <-watch.cancelled:
133138 case <-time.After(5 * time.Second):
@@ -103,10 +103,11 @@ func TestResumeWatch(t *testing.T) {
103 name string103 name string
104 reads []string104 reads []string
105 wantCancel bool105 wantCancel bool
106+ fireEscape bool
106 wantResidual string107 wantResidual string
107 }{108 }{
108 {name: "ctrl-c cancels and drops held keys", reads: []string{"ls", "\x03"}, wantCancel: true},109 {name: "ctrl-c cancels and drops held keys", reads: []string{"ls", "\x03"}, wantCancel: true},
109- {name: "lone esc cancels", reads: []string{"\x1b"}, wantCancel: true},110+ {name: "lone esc cancels", reads: []string{"\x1b"}, wantCancel: true, fireEscape: true},
110 {name: "kitty esc cancels", reads: []string{"\x1b[27u"}, wantCancel: true},111 {name: "kitty esc cancels", reads: []string{"\x1b[27u"}, wantCancel: true},
111 {name: "kitty ctrl-c cancels", reads: []string{"\x1b[99;5u"}, wantCancel: true},112 {name: "kitty ctrl-c cancels", reads: []string{"\x1b[99;5u"}, wantCancel: true},
112 {name: "kitty ctrl-c with lock bits cancels", reads: []string{"\x1b[99;69u"}, wantCancel: true},113 {name: "kitty ctrl-c with lock bits cancels", reads: []string{"\x1b[99;69u"}, wantCancel: true},
@@ -123,11 +124,15 @@ func TestResumeWatch(t *testing.T) {
123 close(reader.chunks)124 close(reader.chunks)
124 })125 })
125 lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}126 lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
126- watch := lifetime.watchResume()127+ clock := newSupervisorTestClock()
128+ watch := lifetime.watchResume(clock)
127 for _, read := range tt.reads {129 for _, read := range tt.reads {
128 reader.chunks <- []byte(read)130 reader.chunks <- []byte(read)
129 }131 }
130 if tt.wantCancel {132 if tt.wantCancel {
133+ if tt.fireEscape {
134+ clock.awaitTimer(t).fire()
135+ }
131 select {136 select {
132 case <-watch.cancelled:137 case <-watch.cancelled:
133 case <-time.After(5 * time.Second):138 case <-time.After(5 * time.Second):
M internal/usecase/client/attachment_session.go +12 -3
@@ -355,9 +355,10 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
355355 if pending == nil {
356356 return
357357 }
358- // The run ends before the delivery was written: join the sender so
359- // the outbox is final, then decide the delivery once, preserving
360- // exactly the session input that never reached the carriage.
358+ // Join the sender before deciding which input reached the carriage.
359+ // A written automation action remains OutcomeUnknown if teardown
360+ // prevents its UIFence receipt; acknowledging input is not proof
361+ // that the daemon processed the action.
361362 stream.join()
362363 if unsent := stream.unsentInput(pending.from, pending.mark); len(unsent) != 0 {
363364 fg.PreserveInput(unsent)
@@ -397,6 +398,14 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
397398 return ackWritten()
398399 }
399400 for {
401+ // Failure also closes progress. Settle it before rearming that
402+ // channel, rather than spinning on an acknowledgement that cannot
403+ // advance.
404+ select {
405+ case <-stream.Failed():
406+ return w.settle(ctx, fg, stream, token, stream.failure())
407+ default:
408+ }
400409 var progress <-chan struct{}
401410 if pending != nil {
402411 progress = stream.Progress()
@@ -355,9 +355,10 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
355 if pending == nil {355 if pending == nil {
356 return356 return
357 }357 }
358- // The run ends before the delivery was written: join the sender so358+ // Join the sender before deciding which input reached the carriage.
359- // the outbox is final, then decide the delivery once, preserving359+ // A written automation action remains OutcomeUnknown if teardown
360- // exactly the session input that never reached the carriage.360+ // prevents its UIFence receipt; acknowledging input is not proof
361+ // that the daemon processed the action.
361 stream.join()362 stream.join()
362 if unsent := stream.unsentInput(pending.from, pending.mark); len(unsent) != 0 {363 if unsent := stream.unsentInput(pending.from, pending.mark); len(unsent) != 0 {
363 fg.PreserveInput(unsent)364 fg.PreserveInput(unsent)
@@ -397,6 +398,14 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
397 return ackWritten()398 return ackWritten()
398 }399 }
399 for {400 for {
401+ // Failure also closes progress. Settle it before rearming that
402+ // channel, rather than spinning on an acknowledgement that cannot
403+ // advance.
404+ select {
405+ case <-stream.Failed():
406+ return w.settle(ctx, fg, stream, token, stream.failure())
407+ default:
408+ }
400 var progress <-chan struct{}409 var progress <-chan struct{}
401 if pending != nil {410 if pending != nil {
402 progress = stream.Progress()411 progress = stream.Progress()
A internal/usecase/client/outbox_bounds_test.go +80 -0
@@ -0,0 +1,80 @@
1+package client
2+
3+import (
4+ "context"
5+ "testing"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/protocol"
9+ "github.com/stretchr/testify/require"
10+)
11+
12+func TestOutboxFullCancellation(t *testing.T) {
13+ for _, foreground := range []bool{false, true} {
14+ t.Run(map[bool]string{false: "context", true: "foreground"}[foreground], func(t *testing.T) {
15+ stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
16+ box := newAttachmentOutbox(stream)
17+ defer box.join()
18+ for range outboxMaxPending {
19+ require.NoError(t, box.SendClient(protocol.Ack{}))
20+ }
21+ ctx, cancel := context.WithCancel(t.Context())
22+ defer cancel()
23+ revoked := make(chan struct{})
24+ result := make(chan error, 1)
25+ go func() { result <- box.enqueue(ctx, revoked, protocol.Ack{}) }()
26+ if foreground {
27+ close(revoked)
28+ } else {
29+ cancel()
30+ }
31+ select {
32+ case err := <-result:
33+ if foreground {
34+ require.ErrorIs(t, err, errAttachmentForegroundRevoked)
35+ } else {
36+ require.ErrorIs(t, err, context.Canceled)
37+ }
38+ case <-time.After(time.Second):
39+ t.Fatal("full outbox ignored cancellation")
40+ }
41+ })
42+ }
43+}
44+
45+// TestOutboxFullAdmitsAfterProgress checks that a caller blocked on a full
46+// outbox is admitted once the sender writes and frees space.
47+func TestOutboxFullAdmitsAfterProgress(t *testing.T) {
48+ stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
49+ box := newAttachmentOutbox(stream)
50+ defer box.join()
51+ for range outboxMaxPending {
52+ require.NoError(t, box.SendClient(protocol.Ack{}))
53+ }
54+ result := make(chan error, 1)
55+ go func() { result <- box.enqueue(t.Context(), nil, protocol.Detach{}) }()
56+ select {
57+ case err := <-result:
58+ t.Fatalf("full outbox admitted before progress: %v", err)
59+ case <-time.After(50 * time.Millisecond):
60+ }
61+ close(stream.release)
62+ select {
63+ case err := <-result:
64+ require.NoError(t, err)
65+ case <-time.After(time.Second):
66+ t.Fatal("blocked caller was not admitted after progress")
67+ }
68+ require.Eventually(t, func() bool { return len(stream.messages()) == outboxMaxPending+1 }, time.Second, time.Millisecond)
69+}
70+
71+func TestOutboxMixedUnsentInput(t *testing.T) {
72+ stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
73+ box := newAttachmentOutbox(stream)
74+ defer box.join()
75+ for _, message := range []protocol.ClientMessage{protocol.Ack{}, protocol.Input{Data: []byte("a")}, protocol.TerminalFocus{}, protocol.Input{Data: []byte("b")}} {
76+ require.NoError(t, box.SendClient(message))
77+ }
78+ require.Equal(t, []byte("ab"), box.unsentInput(0, box.accepted()))
79+ require.Equal(t, []byte("b"), box.unsentInput(2, box.accepted()))
80+}
@@ -0,0 +1,80 @@
1+package client
2+
3+import (
4+ "context"
5+ "testing"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/protocol"
9+ "github.com/stretchr/testify/require"
10+)
11+
12+func TestOutboxFullCancellation(t *testing.T) {
13+ for _, foreground := range []bool{false, true} {
14+ t.Run(map[bool]string{false: "context", true: "foreground"}[foreground], func(t *testing.T) {
15+ stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
16+ box := newAttachmentOutbox(stream)
17+ defer box.join()
18+ for range outboxMaxPending {
19+ require.NoError(t, box.SendClient(protocol.Ack{}))
20+ }
21+ ctx, cancel := context.WithCancel(t.Context())
22+ defer cancel()
23+ revoked := make(chan struct{})
24+ result := make(chan error, 1)
25+ go func() { result <- box.enqueue(ctx, revoked, protocol.Ack{}) }()
26+ if foreground {
27+ close(revoked)
28+ } else {
29+ cancel()
30+ }
31+ select {
32+ case err := <-result:
33+ if foreground {
34+ require.ErrorIs(t, err, errAttachmentForegroundRevoked)
35+ } else {
36+ require.ErrorIs(t, err, context.Canceled)
37+ }
38+ case <-time.After(time.Second):
39+ t.Fatal("full outbox ignored cancellation")
40+ }
41+ })
42+ }
43+}
44+
45+// TestOutboxFullAdmitsAfterProgress checks that a caller blocked on a full
46+// outbox is admitted once the sender writes and frees space.
47+func TestOutboxFullAdmitsAfterProgress(t *testing.T) {
48+ stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
49+ box := newAttachmentOutbox(stream)
50+ defer box.join()
51+ for range outboxMaxPending {
52+ require.NoError(t, box.SendClient(protocol.Ack{}))
53+ }
54+ result := make(chan error, 1)
55+ go func() { result <- box.enqueue(t.Context(), nil, protocol.Detach{}) }()
56+ select {
57+ case err := <-result:
58+ t.Fatalf("full outbox admitted before progress: %v", err)
59+ case <-time.After(50 * time.Millisecond):
60+ }
61+ close(stream.release)
62+ select {
63+ case err := <-result:
64+ require.NoError(t, err)
65+ case <-time.After(time.Second):
66+ t.Fatal("blocked caller was not admitted after progress")
67+ }
68+ require.Eventually(t, func() bool { return len(stream.messages()) == outboxMaxPending+1 }, time.Second, time.Millisecond)
69+}
70+
71+func TestOutboxMixedUnsentInput(t *testing.T) {
72+ stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
73+ box := newAttachmentOutbox(stream)
74+ defer box.join()
75+ for _, message := range []protocol.ClientMessage{protocol.Ack{}, protocol.Input{Data: []byte("a")}, protocol.TerminalFocus{}, protocol.Input{Data: []byte("b")}} {
76+ require.NoError(t, box.SendClient(message))
77+ }
78+ require.Equal(t, []byte("ab"), box.unsentInput(0, box.accepted()))
79+ require.Equal(t, []byte("b"), box.unsentInput(2, box.accepted()))
80+}
A internal/usecase/client/outbox_drain_test.go +64 -0
@@ -0,0 +1,64 @@
1+package client
2+
3+import (
4+ "context"
5+ "errors"
6+ "testing"
7+ "time"
8+
9+ "github.com/bnema/vev/internal/protocol"
10+ "github.com/stretchr/testify/require"
11+)
12+
13+func TestOutboxDrain(t *testing.T) {
14+ linkErr := errors.New("link lost")
15+ for _, mode := range []string{"written", "timeout", "cancelled", "failed", "revoked"} {
16+ t.Run(mode, func(t *testing.T) {
17+ stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
18+ box := newAttachmentOutbox(stream)
19+ defer box.join()
20+ require.NoError(t, box.SendClient(protocol.Detach{}))
21+ clock := newSupervisorTestClock()
22+ worker := &sessionAttachmentWorker{cfg: sessionAttachmentConfig{Clock: clock}}
23+ host := newAttachmentHost(attachmentHostConfig{})
24+ fg := host.newForeground(AttachmentToken{Generation: 1}, stream)
25+ ctx, cancel := context.WithCancel(t.Context())
26+ defer cancel()
27+ result := make(chan error, 1)
28+ go func() { result <- worker.drainOutbox(ctx, fg, box) }()
29+ timer := clock.awaitTimer(t)
30+ require.Equal(t, 2*time.Second, timer.delay)
31+ switch mode {
32+ case "written":
33+ close(stream.release)
34+ case "timeout":
35+ timer.fire()
36+ case "cancelled":
37+ cancel()
38+ case "failed":
39+ stream.fail = linkErr
40+ close(stream.release)
41+ case "revoked":
42+ fg.finalize()
43+ }
44+ select {
45+ case err := <-result:
46+ switch mode {
47+ case "cancelled":
48+ require.ErrorIs(t, err, context.Canceled)
49+ case "failed":
50+ require.ErrorIs(t, err, linkErr)
51+ case "revoked":
52+ require.ErrorIs(t, err, errAttachmentForegroundRevoked)
53+ default:
54+ require.NoError(t, err)
55+ }
56+ case <-time.After(time.Second):
57+ t.Fatal("drain did not settle")
58+ }
59+ if mode == "written" {
60+ require.Len(t, stream.messages(), 1)
61+ }
62+ })
63+ }
64+}
@@ -0,0 +1,64 @@
1+package client
2+
3+import (
4+ "context"
5+ "errors"
6+ "testing"
7+ "time"
8+
9+ "github.com/bnema/vev/internal/protocol"
10+ "github.com/stretchr/testify/require"
11+)
12+
13+func TestOutboxDrain(t *testing.T) {
14+ linkErr := errors.New("link lost")
15+ for _, mode := range []string{"written", "timeout", "cancelled", "failed", "revoked"} {
16+ t.Run(mode, func(t *testing.T) {
17+ stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
18+ box := newAttachmentOutbox(stream)
19+ defer box.join()
20+ require.NoError(t, box.SendClient(protocol.Detach{}))
21+ clock := newSupervisorTestClock()
22+ worker := &sessionAttachmentWorker{cfg: sessionAttachmentConfig{Clock: clock}}
23+ host := newAttachmentHost(attachmentHostConfig{})
24+ fg := host.newForeground(AttachmentToken{Generation: 1}, stream)
25+ ctx, cancel := context.WithCancel(t.Context())
26+ defer cancel()
27+ result := make(chan error, 1)
28+ go func() { result <- worker.drainOutbox(ctx, fg, box) }()
29+ timer := clock.awaitTimer(t)
30+ require.Equal(t, 2*time.Second, timer.delay)
31+ switch mode {
32+ case "written":
33+ close(stream.release)
34+ case "timeout":
35+ timer.fire()
36+ case "cancelled":
37+ cancel()
38+ case "failed":
39+ stream.fail = linkErr
40+ close(stream.release)
41+ case "revoked":
42+ fg.finalize()
43+ }
44+ select {
45+ case err := <-result:
46+ switch mode {
47+ case "cancelled":
48+ require.ErrorIs(t, err, context.Canceled)
49+ case "failed":
50+ require.ErrorIs(t, err, linkErr)
51+ case "revoked":
52+ require.ErrorIs(t, err, errAttachmentForegroundRevoked)
53+ default:
54+ require.NoError(t, err)
55+ }
56+ case <-time.After(time.Second):
57+ t.Fatal("drain did not settle")
58+ }
59+ if mode == "written" {
60+ require.Len(t, stream.messages(), 1)
61+ }
62+ })
63+ }
64+}
A internal/usecase/client/resume_held_input_test.go +115 -0
@@ -0,0 +1,115 @@
1+package client
2+
3+import (
4+ "context"
5+ "strings"
6+ "testing"
7+ "time"
8+
9+ "github.com/bnema/vev/internal/ports"
10+ "github.com/bnema/vev/internal/protocol"
11+ "github.com/stretchr/testify/require"
12+)
13+
14+// sentInput joins every Input message a stream carried, in order.
15+func sentInput(stream *sessionTestStream) string {
16+ var out strings.Builder
17+ for _, message := range stream.messages() {
18+ if input, ok := message.(protocol.Input); ok {
19+ out.Write(input.Data)
20+ }
21+ }
22+ return out.String()
23+}
24+
25+// pushKeys feeds the terminal one byte per read, like a keyboard, and waits
26+// until the shared reader consumed them all.
27+func pushKeys(t *testing.T, h *attachTestHarness, keys string) {
28+ t.Helper()
29+ for _, b := range []byte(keys) {
30+ h.reader.push([]byte{b})
31+ }
32+ require.Eventually(t, func() bool { return len(h.reader.chunks) == 0 }, 5*time.Second, time.Millisecond)
33+}
34+
35+// TestResumeReplaysKeysHeldWhileConnecting types while a lost attachment waits
36+// for its backoff and while the resume stream opens (including a failed open
37+// that is retried), then proves the resumed session receives every key, once,
38+// in order, and the lost attachment none.
39+func TestResumeReplaysKeysHeldWhileConnecting(t *testing.T) {
40+ for _, tt := range []struct {
41+ name string
42+ backoffKeys string
43+ openKeys string
44+ failFirst bool
45+ retryKeys string
46+ }{
47+ {name: "typed during backoff", backoffKeys: "HELD_ONE"},
48+ {name: "typed during stream open", openKeys: "HELD_TWO\r"},
49+ {name: "typed during backoff and open", backoffKeys: "HELD_ONE", openKeys: "HELD_TWO\r"},
50+ {name: "kept across a failed open", openKeys: "HELD_ONE", failFirst: true, retryKeys: "HELD_TWO\r"},
51+ } {
52+ t.Run(tt.name, func(t *testing.T) {
53+ picker := newAttachTestPicker()
54+ h := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { cfg.Jitter = resumeTestJitter })
55+ first := newSessionTestStream()
56+ second := newSessionTestStream()
57+ opened := make(chan int, 4)
58+ failed := make(chan struct{})
59+ release := make(chan struct{})
60+ count := 0
61+ h.service.setOpenStream(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
62+ count++
63+ switch {
64+ case count == 1:
65+ return first, nil
66+ case tt.failFirst && count == 2:
67+ opened <- count
68+ select {
69+ case <-failed:
70+ case <-ctx.Done():
71+ }
72+ return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "route down"}
73+ }
74+ opened <- count
75+ select {
76+ case <-release:
77+ case <-ctx.Done():
78+ return nil, ctx.Err()
79+ }
80+ return second, nil
81+ })
82+ awaitOpen := func() {
83+ select {
84+ case <-opened:
85+ case <-time.After(5 * time.Second):
86+ t.Fatal("resume did not open")
87+ }
88+ }
89+ picker.commit(sessionTestRequest(false))
90+ deliverReadyStream(t, first)
91+ awaitAttachedState(t, h.sup)
92+ first.fail(testStreamLoss())
93+
94+ require.Eventually(t, func() bool { return h.sup.State().Presentation == PresentConnecting }, 5*time.Second, time.Millisecond)
95+ pushKeys(t, h, tt.backoffKeys)
96+ fireResumeTimerAfter(t, h.clock, 1, func() {})
97+ awaitOpen()
98+ pushKeys(t, h, tt.openKeys)
99+ if tt.failFirst {
100+ close(failed)
101+ fireResumeTimerAfter(t, h.clock, 2, func() {})
102+ awaitOpen()
103+ pushKeys(t, h, tt.retryKeys)
104+ }
105+ close(release)
106+ deliverReadyStream(t, second)
107+ awaitAttachedState(t, h.sup)
108+
109+ want := tt.backoffKeys + tt.openKeys + tt.retryKeys
110+ require.Eventually(t, func() bool { return sentInput(second) == want }, 5*time.Second, time.Millisecond,
111+ "resumed session input = %q, want %q", sentInput(second), want)
112+ require.Empty(t, sentInput(first))
113+ })
114+ }
115+}
@@ -0,0 +1,115 @@
1+package client
2+
3+import (
4+ "context"
5+ "strings"
6+ "testing"
7+ "time"
8+
9+ "github.com/bnema/vev/internal/ports"
10+ "github.com/bnema/vev/internal/protocol"
11+ "github.com/stretchr/testify/require"
12+)
13+
14+// sentInput joins every Input message a stream carried, in order.
15+func sentInput(stream *sessionTestStream) string {
16+ var out strings.Builder
17+ for _, message := range stream.messages() {
18+ if input, ok := message.(protocol.Input); ok {
19+ out.Write(input.Data)
20+ }
21+ }
22+ return out.String()
23+}
24+
25+// pushKeys feeds the terminal one byte per read, like a keyboard, and waits
26+// until the shared reader consumed them all.
27+func pushKeys(t *testing.T, h *attachTestHarness, keys string) {
28+ t.Helper()
29+ for _, b := range []byte(keys) {
30+ h.reader.push([]byte{b})
31+ }
32+ require.Eventually(t, func() bool { return len(h.reader.chunks) == 0 }, 5*time.Second, time.Millisecond)
33+}
34+
35+// TestResumeReplaysKeysHeldWhileConnecting types while a lost attachment waits
36+// for its backoff and while the resume stream opens (including a failed open
37+// that is retried), then proves the resumed session receives every key, once,
38+// in order, and the lost attachment none.
39+func TestResumeReplaysKeysHeldWhileConnecting(t *testing.T) {
40+ for _, tt := range []struct {
41+ name string
42+ backoffKeys string
43+ openKeys string
44+ failFirst bool
45+ retryKeys string
46+ }{
47+ {name: "typed during backoff", backoffKeys: "HELD_ONE"},
48+ {name: "typed during stream open", openKeys: "HELD_TWO\r"},
49+ {name: "typed during backoff and open", backoffKeys: "HELD_ONE", openKeys: "HELD_TWO\r"},
50+ {name: "kept across a failed open", openKeys: "HELD_ONE", failFirst: true, retryKeys: "HELD_TWO\r"},
51+ } {
52+ t.Run(tt.name, func(t *testing.T) {
53+ picker := newAttachTestPicker()
54+ h := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { cfg.Jitter = resumeTestJitter })
55+ first := newSessionTestStream()
56+ second := newSessionTestStream()
57+ opened := make(chan int, 4)
58+ failed := make(chan struct{})
59+ release := make(chan struct{})
60+ count := 0
61+ h.service.setOpenStream(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
62+ count++
63+ switch {
64+ case count == 1:
65+ return first, nil
66+ case tt.failFirst && count == 2:
67+ opened <- count
68+ select {
69+ case <-failed:
70+ case <-ctx.Done():
71+ }
72+ return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "route down"}
73+ }
74+ opened <- count
75+ select {
76+ case <-release:
77+ case <-ctx.Done():
78+ return nil, ctx.Err()
79+ }
80+ return second, nil
81+ })
82+ awaitOpen := func() {
83+ select {
84+ case <-opened:
85+ case <-time.After(5 * time.Second):
86+ t.Fatal("resume did not open")
87+ }
88+ }
89+ picker.commit(sessionTestRequest(false))
90+ deliverReadyStream(t, first)
91+ awaitAttachedState(t, h.sup)
92+ first.fail(testStreamLoss())
93+
94+ require.Eventually(t, func() bool { return h.sup.State().Presentation == PresentConnecting }, 5*time.Second, time.Millisecond)
95+ pushKeys(t, h, tt.backoffKeys)
96+ fireResumeTimerAfter(t, h.clock, 1, func() {})
97+ awaitOpen()
98+ pushKeys(t, h, tt.openKeys)
99+ if tt.failFirst {
100+ close(failed)
101+ fireResumeTimerAfter(t, h.clock, 2, func() {})
102+ awaitOpen()
103+ pushKeys(t, h, tt.retryKeys)
104+ }
105+ close(release)
106+ deliverReadyStream(t, second)
107+ awaitAttachedState(t, h.sup)
108+
109+ want := tt.backoffKeys + tt.openKeys + tt.retryKeys
110+ require.Eventually(t, func() bool { return sentInput(second) == want }, 5*time.Second, time.Millisecond,
111+ "resumed session input = %q, want %q", sentInput(second), want)
112+ require.Empty(t, sentInput(first))
113+ })
114+ }
115+}
A internal/usecase/client/resume_input.go +86 -0
@@ -0,0 +1,86 @@
1+package client
2+
3+import (
4+ "bytes"
5+
6+ "github.com/bnema/vev/internal/usecase/keys/kittykey"
7+)
8+
9+const resumeEscapeLimit = 256
10+
11+// resumeInputDecoder retains only an undecided escape prefix. Held bytes stay
12+// in their original terminal encoding; paste content is never a cancel key.
13+type resumeInputDecoder struct {
14+ prefix []byte
15+ paste bool
16+}
17+
18+func (d *resumeInputDecoder) feed(data []byte) (held []byte, cancel bool) {
19+ for _, b := range data {
20+ if len(d.prefix) == 0 {
21+ if b == 0x1b {
22+ d.prefix = append(d.prefix, b)
23+ continue
24+ }
25+ if b == 3 && !d.paste {
26+ return held, true
27+ }
28+ held = append(held, b)
29+ continue
30+ }
31+ if b == 0x1b || (b < 0x20 && !d.paste) {
32+ held = append(held, d.prefix...)
33+ d.prefix = d.prefix[:0]
34+ if b == 3 {
35+ return held, true
36+ }
37+ if b == 0x1b {
38+ d.prefix = append(d.prefix, b)
39+ } else {
40+ held = append(held, b)
41+ }
42+ continue
43+ }
44+ d.prefix = append(d.prefix, b)
45+ if d.paste {
46+ end := []byte("\x1b[201~")
47+ if bytes.Equal(d.prefix, end) {
48+ d.paste = false
49+ } else if bytes.HasPrefix(end, d.prefix) {
50+ continue
51+ }
52+ } else {
53+ start := []byte("\x1b[200~")
54+ if bytes.Equal(d.prefix, start) {
55+ d.paste = true
56+ } else if bytes.HasPrefix(start, d.prefix) {
57+ continue
58+ } else {
59+ _, n, ok, partial := kittykey.Parse(d.prefix)
60+ if ok {
61+ translated := kittykey.Translate(d.prefix[:n], 0)
62+ if len(translated) == 1 && (translated[0] == 3 || translated[0] == 0x1b) {
63+ return held, true
64+ }
65+ } else if partial && len(d.prefix) < resumeEscapeLimit {
66+ continue
67+ } else if len(d.prefix) >= 2 && (d.prefix[1] == '[' || d.prefix[1] == 'O') && b >= 0x20 && b < 0x40 && len(d.prefix) < resumeEscapeLimit {
68+ continue
69+ }
70+ }
71+ }
72+ held = append(held, d.prefix...)
73+ d.prefix = d.prefix[:0]
74+ }
75+ return held, false
76+}
77+
78+func (d *resumeInputDecoder) loneEscape() bool {
79+ return !d.paste && bytes.Equal(d.prefix, []byte{0x1b})
80+}
81+
82+func (d *resumeInputDecoder) flush() []byte {
83+ data := d.prefix
84+ d.prefix = nil
85+ return data
86+}
@@ -0,0 +1,86 @@
1+package client
2+
3+import (
4+ "bytes"
5+
6+ "github.com/bnema/vev/internal/usecase/keys/kittykey"
7+)
8+
9+const resumeEscapeLimit = 256
10+
11+// resumeInputDecoder retains only an undecided escape prefix. Held bytes stay
12+// in their original terminal encoding; paste content is never a cancel key.
13+type resumeInputDecoder struct {
14+ prefix []byte
15+ paste bool
16+}
17+
18+func (d *resumeInputDecoder) feed(data []byte) (held []byte, cancel bool) {
19+ for _, b := range data {
20+ if len(d.prefix) == 0 {
21+ if b == 0x1b {
22+ d.prefix = append(d.prefix, b)
23+ continue
24+ }
25+ if b == 3 && !d.paste {
26+ return held, true
27+ }
28+ held = append(held, b)
29+ continue
30+ }
31+ if b == 0x1b || (b < 0x20 && !d.paste) {
32+ held = append(held, d.prefix...)
33+ d.prefix = d.prefix[:0]
34+ if b == 3 {
35+ return held, true
36+ }
37+ if b == 0x1b {
38+ d.prefix = append(d.prefix, b)
39+ } else {
40+ held = append(held, b)
41+ }
42+ continue
43+ }
44+ d.prefix = append(d.prefix, b)
45+ if d.paste {
46+ end := []byte("\x1b[201~")
47+ if bytes.Equal(d.prefix, end) {
48+ d.paste = false
49+ } else if bytes.HasPrefix(end, d.prefix) {
50+ continue
51+ }
52+ } else {
53+ start := []byte("\x1b[200~")
54+ if bytes.Equal(d.prefix, start) {
55+ d.paste = true
56+ } else if bytes.HasPrefix(start, d.prefix) {
57+ continue
58+ } else {
59+ _, n, ok, partial := kittykey.Parse(d.prefix)
60+ if ok {
61+ translated := kittykey.Translate(d.prefix[:n], 0)
62+ if len(translated) == 1 && (translated[0] == 3 || translated[0] == 0x1b) {
63+ return held, true
64+ }
65+ } else if partial && len(d.prefix) < resumeEscapeLimit {
66+ continue
67+ } else if len(d.prefix) >= 2 && (d.prefix[1] == '[' || d.prefix[1] == 'O') && b >= 0x20 && b < 0x40 && len(d.prefix) < resumeEscapeLimit {
68+ continue
69+ }
70+ }
71+ }
72+ held = append(held, d.prefix...)
73+ d.prefix = d.prefix[:0]
74+ }
75+ return held, false
76+}
77+
78+func (d *resumeInputDecoder) loneEscape() bool {
79+ return !d.paste && bytes.Equal(d.prefix, []byte{0x1b})
80+}
81+
82+func (d *resumeInputDecoder) flush() []byte {
83+ data := d.prefix
84+ d.prefix = nil
85+ return data
86+}
A internal/usecase/client/resume_input_test.go +54 -0
@@ -0,0 +1,54 @@
1+package client
2+
3+import (
4+ "testing"
5+
6+ "github.com/stretchr/testify/require"
7+)
8+
9+func TestResumeInputDecoder(t *testing.T) {
10+ for _, tt := range []struct {
11+ name, input string
12+ cancel bool
13+ }{
14+ {"grouped ctrl-c", "text\x03tail", true},
15+ {"paste escape followed by cancel", "\x1b[200~a\x1b\x1b[201~\x03", true},
16+ {"back to back escape", "\x1b\x1b[27u", true},
17+ {"ctrl-c in unfinished CSI", "\x1b[\x03", true},
18+ {"kitty ctrl-c", "\x1b[99;5u", true},
19+ {"kitty escape", "\x1b[27u", true},
20+ {"arrow", "\x1b[A", false},
21+ {"function", "\x1bOP", false},
22+ {"paste", "\x1b[200~a\x03\x1b[27u\x1b[201~", false},
23+ {"text", "hello", false},
24+ {"alt enter", "\x1b\r", false},
25+ {"alt tab", "\x1b\t", false},
26+ } {
27+ t.Run(tt.name, func(t *testing.T) {
28+ for split := 0; split <= len(tt.input); split++ {
29+ var d resumeInputDecoder
30+ first, cancelled := d.feed([]byte(tt.input[:split]))
31+ second := []byte(nil)
32+ if !cancelled {
33+ second, cancelled = d.feed([]byte(tt.input[split:]))
34+ }
35+ require.Equal(t, tt.cancel, cancelled, "split %d", split)
36+ if !tt.cancel {
37+ got := append(first, second...)
38+ got = append(got, d.flush()...)
39+ require.Equal(t, tt.input, string(got), "split %d", split)
40+ }
41+ }
42+ })
43+ }
44+}
45+
46+func TestResumeInputDecoderUndecidedEscape(t *testing.T) {
47+ var d resumeInputDecoder
48+ held, cancel := d.feed([]byte("\x1b"))
49+ require.Empty(t, held)
50+ require.False(t, cancel)
51+ require.True(t, d.loneEscape())
52+ require.Equal(t, []byte{0x1b}, d.flush())
53+ require.False(t, d.loneEscape())
54+}
@@ -0,0 +1,54 @@
1+package client
2+
3+import (
4+ "testing"
5+
6+ "github.com/stretchr/testify/require"
7+)
8+
9+func TestResumeInputDecoder(t *testing.T) {
10+ for _, tt := range []struct {
11+ name, input string
12+ cancel bool
13+ }{
14+ {"grouped ctrl-c", "text\x03tail", true},
15+ {"paste escape followed by cancel", "\x1b[200~a\x1b\x1b[201~\x03", true},
16+ {"back to back escape", "\x1b\x1b[27u", true},
17+ {"ctrl-c in unfinished CSI", "\x1b[\x03", true},
18+ {"kitty ctrl-c", "\x1b[99;5u", true},
19+ {"kitty escape", "\x1b[27u", true},
20+ {"arrow", "\x1b[A", false},
21+ {"function", "\x1bOP", false},
22+ {"paste", "\x1b[200~a\x03\x1b[27u\x1b[201~", false},
23+ {"text", "hello", false},
24+ {"alt enter", "\x1b\r", false},
25+ {"alt tab", "\x1b\t", false},
26+ } {
27+ t.Run(tt.name, func(t *testing.T) {
28+ for split := 0; split <= len(tt.input); split++ {
29+ var d resumeInputDecoder
30+ first, cancelled := d.feed([]byte(tt.input[:split]))
31+ second := []byte(nil)
32+ if !cancelled {
33+ second, cancelled = d.feed([]byte(tt.input[split:]))
34+ }
35+ require.Equal(t, tt.cancel, cancelled, "split %d", split)
36+ if !tt.cancel {
37+ got := append(first, second...)
38+ got = append(got, d.flush()...)
39+ require.Equal(t, tt.input, string(got), "split %d", split)
40+ }
41+ }
42+ })
43+ }
44+}
45+
46+func TestResumeInputDecoderUndecidedEscape(t *testing.T) {
47+ var d resumeInputDecoder
48+ held, cancel := d.feed([]byte("\x1b"))
49+ require.Empty(t, held)
50+ require.False(t, cancel)
51+ require.True(t, d.loneEscape())
52+ require.Equal(t, []byte{0x1b}, d.flush())
53+ require.False(t, d.loneEscape())
54+}
A internal/usecase/client/resume_open_test.go +67 -0
@@ -0,0 +1,67 @@
1+package client
2+
3+import (
4+ "context"
5+ "testing"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/ports"
9+ "github.com/stretchr/testify/require"
10+)
11+
12+func TestResumeCancelDuringStreamOpen(t *testing.T) {
13+ for _, tt := range []struct {
14+ name, input string
15+ escape bool
16+ }{
17+ {name: "ctrl-c", input: "\x03"},
18+ {name: "kitty escape", input: "\x1b[27u"},
19+ {name: "legacy escape", input: "\x1b", escape: true},
20+ } {
21+ t.Run(tt.name, func(t *testing.T) {
22+ picker := newAttachTestPicker()
23+ h := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { cfg.Jitter = resumeTestJitter })
24+ first := newSessionTestStream()
25+ opened := make(chan struct{})
26+ cancelled := make(chan struct{})
27+ count := 0
28+ h.service.setOpenStream(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
29+ count++
30+ if count == 1 {
31+ return first, nil
32+ }
33+ close(opened)
34+ <-ctx.Done()
35+ close(cancelled)
36+ return nil, ctx.Err()
37+ })
38+ picker.commit(sessionTestRequest(false))
39+ deliverReadyStream(t, first)
40+ awaitAttachedState(t, h.sup)
41+ first.fail(testStreamLoss())
42+ fireResumeTimerAfter(t, h.clock, 1, func() {})
43+ select {
44+ case <-opened:
45+ case <-time.After(5 * time.Second):
46+ t.Fatal("resume did not open")
47+ }
48+ h.reader.push([]byte(tt.input))
49+ if tt.escape {
50+ for {
51+ timer := h.clock.awaitTimer(t)
52+ if timer.delay == 50*time.Millisecond {
53+ timer.fire()
54+ break
55+ }
56+ }
57+ }
58+ select {
59+ case <-cancelled:
60+ case <-time.After(5 * time.Second):
61+ t.Fatal("cancel did not reach open context")
62+ }
63+ require.Eventually(t, picker.owns, 5*time.Second, time.Millisecond)
64+ require.Equal(t, 2, len(h.service.openedRequests()))
65+ })
66+ }
67+}
@@ -0,0 +1,67 @@
1+package client
2+
3+import (
4+ "context"
5+ "testing"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/ports"
9+ "github.com/stretchr/testify/require"
10+)
11+
12+func TestResumeCancelDuringStreamOpen(t *testing.T) {
13+ for _, tt := range []struct {
14+ name, input string
15+ escape bool
16+ }{
17+ {name: "ctrl-c", input: "\x03"},
18+ {name: "kitty escape", input: "\x1b[27u"},
19+ {name: "legacy escape", input: "\x1b", escape: true},
20+ } {
21+ t.Run(tt.name, func(t *testing.T) {
22+ picker := newAttachTestPicker()
23+ h := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) { cfg.Jitter = resumeTestJitter })
24+ first := newSessionTestStream()
25+ opened := make(chan struct{})
26+ cancelled := make(chan struct{})
27+ count := 0
28+ h.service.setOpenStream(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
29+ count++
30+ if count == 1 {
31+ return first, nil
32+ }
33+ close(opened)
34+ <-ctx.Done()
35+ close(cancelled)
36+ return nil, ctx.Err()
37+ })
38+ picker.commit(sessionTestRequest(false))
39+ deliverReadyStream(t, first)
40+ awaitAttachedState(t, h.sup)
41+ first.fail(testStreamLoss())
42+ fireResumeTimerAfter(t, h.clock, 1, func() {})
43+ select {
44+ case <-opened:
45+ case <-time.After(5 * time.Second):
46+ t.Fatal("resume did not open")
47+ }
48+ h.reader.push([]byte(tt.input))
49+ if tt.escape {
50+ for {
51+ timer := h.clock.awaitTimer(t)
52+ if timer.delay == 50*time.Millisecond {
53+ timer.fire()
54+ break
55+ }
56+ }
57+ }
58+ select {
59+ case <-cancelled:
60+ case <-time.After(5 * time.Second):
61+ t.Fatal("cancel did not reach open context")
62+ }
63+ require.Eventually(t, picker.owns, 5*time.Second, time.Millisecond)
64+ require.Equal(t, 2, len(h.service.openedRequests()))
65+ })
66+ }
67+}
M internal/usecase/client/supervisor.go +29 -12
@@ -1503,6 +1503,7 @@ type resumeWatch struct {
15031503 done chan struct{}
15041504 lifetime *terminalInputLifetime
15051505 id uint64
1506+ clock ports.Clock
15061507
15071508 mu sync.Mutex
15081509 held []byte
@@ -1524,18 +1525,10 @@ func (w *resumeWatch) hold(data []byte) {
15241525 }
15251526 }
15261527
1527-// isResumeCancelKey reports a read that is exactly Ctrl-C or a lone Esc,
1528-// after decoding the kitty keyboard protocol (Esc is \x1b[27u there). A longer
1529-// read starting with Esc is an escape sequence (arrows, focus), not a cancel.
1530-func isResumeCancelKey(data []byte) bool {
1531- data = kittykey.Translate(data, 0)
1532- return len(data) == 1 && (data[0] == 0x03 || data[0] == 0x1b)
1533-}
1534-
15351528 // watchResume claims terminal input for one resume backoff. When the claim
15361529 // is unavailable it returns a watch that never cancels.
1537-func (l *terminalInputLifetime) watchResume() *resumeWatch {
1538- w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l}
1530+func (l *terminalInputLifetime) watchResume(clock ports.Clock) *resumeWatch {
1531+ w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock}
15391532 if l == nil || l.pump == nil {
15401533 close(w.done)
15411534 return w
@@ -1553,11 +1546,26 @@ func (l *terminalInputLifetime) watchResume() *resumeWatch {
15531546 func (w *resumeWatch) run() {
15541547 defer close(w.done)
15551548 pump := w.lifetime.pump
1549+ var decoder resumeInputDecoder
1550+ var escape ports.Timer
1551+ defer func() {
1552+ if escape != nil {
1553+ escape.Stop()
1554+ }
1555+ w.hold(decoder.flush())
1556+ }()
15561557 for {
1558+ var escapeC <-chan time.Time
1559+ if escape != nil {
1560+ escapeC = escape.C()
1561+ }
15571562 result, ok := pump.take(context.Background(), w.id)
15581563 if !ok {
15591564 select {
15601565 case <-pump.readyFor(w.id):
1566+ case <-escapeC:
1567+ close(w.cancelled)
1568+ return
15611569 case <-w.stop:
15621570 return
15631571 case <-pump.done:
@@ -1568,16 +1576,25 @@ func (w *resumeWatch) run() {
15681576 if result.err != nil {
15691577 // Leave the final read for the terminal EOF watcher.
15701578 pump.ack(w.id)
1579+ w.hold(decoder.flush())
15711580 w.hold(result.data)
15721581 w.lifetime.finish(result.err)
15731582 return
15741583 }
15751584 pump.ack(w.id)
1576- if isResumeCancelKey(result.data) {
1585+ held, cancel := decoder.feed(result.data)
1586+ w.hold(held)
1587+ if cancel {
15771588 close(w.cancelled)
15781589 return
15791590 }
1580- w.hold(result.data)
1591+ if escape != nil {
1592+ escape.Stop()
1593+ escape = nil
1594+ }
1595+ if decoder.loneEscape() {
1596+ escape = w.clock.NewTimer(50 * time.Millisecond)
1597+ }
15811598 }
15821599 }
15831600
@@ -1503,6 +1503,7 @@ type resumeWatch struct {
1503 done chan struct{}1503 done chan struct{}
1504 lifetime *terminalInputLifetime1504 lifetime *terminalInputLifetime
1505 id uint641505 id uint64
1506+ clock ports.Clock
1506 1507
1507 mu sync.Mutex1508 mu sync.Mutex
1508 held []byte1509 held []byte
@@ -1524,18 +1525,10 @@ func (w *resumeWatch) hold(data []byte) {
1524 }1525 }
1525 }1526 }
1526 1527
1527-// isResumeCancelKey reports a read that is exactly Ctrl-C or a lone Esc,
1528-// after decoding the kitty keyboard protocol (Esc is \x1b[27u there). A longer
1529-// read starting with Esc is an escape sequence (arrows, focus), not a cancel.
1530-func isResumeCancelKey(data []byte) bool {
1531- data = kittykey.Translate(data, 0)
1532- return len(data) == 1 && (data[0] == 0x03 || data[0] == 0x1b)
1533-}
1534-
1535 // watchResume claims terminal input for one resume backoff. When the claim1528 // watchResume claims terminal input for one resume backoff. When the claim
1536 // is unavailable it returns a watch that never cancels.1529 // is unavailable it returns a watch that never cancels.
1537-func (l *terminalInputLifetime) watchResume() *resumeWatch {1530+func (l *terminalInputLifetime) watchResume(clock ports.Clock) *resumeWatch {
1538- w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l}1531+ w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock}
1539 if l == nil || l.pump == nil {1532 if l == nil || l.pump == nil {
1540 close(w.done)1533 close(w.done)
1541 return w1534 return w
@@ -1553,11 +1546,26 @@ func (l *terminalInputLifetime) watchResume() *resumeWatch {
1553 func (w *resumeWatch) run() {1546 func (w *resumeWatch) run() {
1554 defer close(w.done)1547 defer close(w.done)
1555 pump := w.lifetime.pump1548 pump := w.lifetime.pump
1549+ var decoder resumeInputDecoder
1550+ var escape ports.Timer
1551+ defer func() {
1552+ if escape != nil {
1553+ escape.Stop()
1554+ }
1555+ w.hold(decoder.flush())
1556+ }()
1556 for {1557 for {
1558+ var escapeC <-chan time.Time
1559+ if escape != nil {
1560+ escapeC = escape.C()
1561+ }
1557 result, ok := pump.take(context.Background(), w.id)1562 result, ok := pump.take(context.Background(), w.id)
1558 if !ok {1563 if !ok {
1559 select {1564 select {
1560 case <-pump.readyFor(w.id):1565 case <-pump.readyFor(w.id):
1566+ case <-escapeC:
1567+ close(w.cancelled)
1568+ return
1561 case <-w.stop:1569 case <-w.stop:
1562 return1570 return
1563 case <-pump.done:1571 case <-pump.done:
@@ -1568,16 +1576,25 @@ func (w *resumeWatch) run() {
1568 if result.err != nil {1576 if result.err != nil {
1569 // Leave the final read for the terminal EOF watcher.1577 // Leave the final read for the terminal EOF watcher.
1570 pump.ack(w.id)1578 pump.ack(w.id)
1579+ w.hold(decoder.flush())
1571 w.hold(result.data)1580 w.hold(result.data)
1572 w.lifetime.finish(result.err)1581 w.lifetime.finish(result.err)
1573 return1582 return
1574 }1583 }
1575 pump.ack(w.id)1584 pump.ack(w.id)
1576- if isResumeCancelKey(result.data) {1585+ held, cancel := decoder.feed(result.data)
1586+ w.hold(held)
1587+ if cancel {
1577 close(w.cancelled)1588 close(w.cancelled)
1578 return1589 return
1579 }1590 }
1580- w.hold(result.data)1591+ if escape != nil {
1592+ escape.Stop()
1593+ escape = nil
1594+ }
1595+ if decoder.loneEscape() {
1596+ escape = w.clock.NewTimer(50 * time.Millisecond)
1597+ }
1581 }1598 }
1582 }1599 }
1583 1600
M internal/usecase/client/supervisor_attach.go +34 -2
@@ -721,7 +721,39 @@ func (s *Supervisor) attachResolved(ctx context.Context, input *terminalInputLif
721721 request.Env = append([]string(nil), sessionEnv.Env...)
722722
723723 s.logger.Debug("client_stream_open_begin", "local", request.Local, "endpoint", request.Endpoint, "admission", request.Admission, "generation", s.State().Generation)
724- stream, err := service.OpenStream(deadline.Context(), request)
724+ openCtx := deadline.Context()
725+ var keys *resumeWatch
726+ var cancelOpen context.CancelFunc
727+ var watchDone chan struct{}
728+ if resuming {
729+ keys = input.watchResume(s.cfg.Clock)
730+ openCtx, cancelOpen = context.WithCancel(openCtx)
731+ watchDone = make(chan struct{})
732+ go func() {
733+ defer close(watchDone)
734+ select {
735+ case <-keys.cancelled:
736+ cancelOpen()
737+ case <-openCtx.Done():
738+ }
739+ }()
740+ }
741+ stream, err := service.OpenStream(openCtx, request)
742+ if keys != nil {
743+ keys.release()
744+ cancelOpen()
745+ <-watchDone
746+ select {
747+ case <-keys.cancelled:
748+ s.logger.Info("client_attachment_resume_cancelled", "phase", "stream_open")
749+ if !supervisorNil(stream) {
750+ _ = stream.Close()
751+ }
752+ s.transition(supervisorEvent{kind: supervisorAttachEnded})
753+ return attachmentEndedOutcome()
754+ default:
755+ }
756+ }
725757 if err != nil {
726758 s.logger.Warn("client_stream_open_failed", "local", request.Local, "endpoint", request.Endpoint, "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
727759 return s.attemptFailed(deadline.cause(err), resuming)
@@ -964,7 +996,7 @@ func (s *Supervisor) waitResume(ctx context.Context, input *terminalInputLifetim
964996 delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)
965997 timer := s.cfg.Clock.NewTimer(delay)
966998 defer stopSupervisorTimer(timer)
967- keys := input.watchResume()
999+ keys := input.watchResume(s.cfg.Clock)
9681000 defer keys.release()
9691001 for {
9701002 select {
@@ -721,7 +721,39 @@ func (s *Supervisor) attachResolved(ctx context.Context, input *terminalInputLif
721 request.Env = append([]string(nil), sessionEnv.Env...)721 request.Env = append([]string(nil), sessionEnv.Env...)
722 722
723 s.logger.Debug("client_stream_open_begin", "local", request.Local, "endpoint", request.Endpoint, "admission", request.Admission, "generation", s.State().Generation)723 s.logger.Debug("client_stream_open_begin", "local", request.Local, "endpoint", request.Endpoint, "admission", request.Admission, "generation", s.State().Generation)
724- stream, err := service.OpenStream(deadline.Context(), request)724+ openCtx := deadline.Context()
725+ var keys *resumeWatch
726+ var cancelOpen context.CancelFunc
727+ var watchDone chan struct{}
728+ if resuming {
729+ keys = input.watchResume(s.cfg.Clock)
730+ openCtx, cancelOpen = context.WithCancel(openCtx)
731+ watchDone = make(chan struct{})
732+ go func() {
733+ defer close(watchDone)
734+ select {
735+ case <-keys.cancelled:
736+ cancelOpen()
737+ case <-openCtx.Done():
738+ }
739+ }()
740+ }
741+ stream, err := service.OpenStream(openCtx, request)
742+ if keys != nil {
743+ keys.release()
744+ cancelOpen()
745+ <-watchDone
746+ select {
747+ case <-keys.cancelled:
748+ s.logger.Info("client_attachment_resume_cancelled", "phase", "stream_open")
749+ if !supervisorNil(stream) {
750+ _ = stream.Close()
751+ }
752+ s.transition(supervisorEvent{kind: supervisorAttachEnded})
753+ return attachmentEndedOutcome()
754+ default:
755+ }
756+ }
725 if err != nil {757 if err != nil {
726 s.logger.Warn("client_stream_open_failed", "local", request.Local, "endpoint", request.Endpoint, "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))758 s.logger.Warn("client_stream_open_failed", "local", request.Local, "endpoint", request.Endpoint, "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
727 return s.attemptFailed(deadline.cause(err), resuming)759 return s.attemptFailed(deadline.cause(err), resuming)
@@ -964,7 +996,7 @@ func (s *Supervisor) waitResume(ctx context.Context, input *terminalInputLifetim
964 delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)996 delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)
965 timer := s.cfg.Clock.NewTimer(delay)997 timer := s.cfg.Clock.NewTimer(delay)
966 defer stopSupervisorTimer(timer)998 defer stopSupervisorTimer(timer)
967- keys := input.watchResume()999+ keys := input.watchResume(s.cfg.Clock)
968 defer keys.release()1000 defer keys.release()
969 for {1001 for {
970 select {1002 select {
M internal/usecase/client/supervisor_attach_test.go +0 -7
@@ -1304,13 +1304,6 @@ func testStreamLoss() ports.BrokerStreamLost {
13041304 // 15s attachment deadline and the 200ms palette deadline).
13051305 func resumeTestJitter() float64 { return 0.5 }
13061306
1307-// fireResumeTimer fires the resume backoff timer for one attempt. Other timers
1308-// share the clock, so it matches the exact backoff delay.
1309-func fireResumeTimer(t *testing.T, clock *supervisorTestClock, attempt int) {
1310- t.Helper()
1311- fireResumeTimerAfter(t, clock, attempt, func() {})
1312-}
1313-
13141307 // fireResumeTimerAfter waits for the attempt's backoff timer, runs before, then
13151308 // fires it.
13161309 func fireResumeTimerAfter(t *testing.T, clock *supervisorTestClock, attempt int, before func()) {
@@ -1304,13 +1304,6 @@ func testStreamLoss() ports.BrokerStreamLost {
1304 // 15s attachment deadline and the 200ms palette deadline).1304 // 15s attachment deadline and the 200ms palette deadline).
1305 func resumeTestJitter() float64 { return 0.5 }1305 func resumeTestJitter() float64 { return 0.5 }
1306 1306
1307-// fireResumeTimer fires the resume backoff timer for one attempt. Other timers
1308-// share the clock, so it matches the exact backoff delay.
1309-func fireResumeTimer(t *testing.T, clock *supervisorTestClock, attempt int) {
1310- t.Helper()
1311- fireResumeTimerAfter(t, clock, attempt, func() {})
1312-}
1313-
1314 // fireResumeTimerAfter waits for the attempt's backoff timer, runs before, then1307 // fireResumeTimerAfter waits for the attempt's backoff timer, runs before, then
1315 // fires it.1308 // fires it.
1316 func fireResumeTimerAfter(t *testing.T, clock *supervisorTestClock, attempt int, before func()) {1309 func fireResumeTimerAfter(t *testing.T, clock *supervisorTestClock, attempt int, before 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