detached 3711516

fix(client): hand off queued terminal input

commit
3711516eddcfdd42a1f3949896687e77b08e3145
author
brice · · 2026-07-18 07:25 UTC
parent
cdca174
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +162 -17
Unified
UnifiedSplit
M internal/usecase/client/client.go +113 -17
@@ -609,6 +609,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
609609 input.start()
610610 defer input.stop()
611611 }
612+ inputConsumer := input.claim()
612613
613614 sendCh := make(chan ports.Frame, sendQueueDepth)
614615 ackQueue := newCumulativeAckQueue()
@@ -723,12 +724,12 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
723724 stdinDone := make(chan struct{})
724725 go func() {
725726 defer close(stdinDone)
726- (&stdinPump{ctx: loopCtx, cancel: cancel, input: input, out: sendCh, clock: clk, clipboard: clip, logger: log, paletteEvents: paletteEvents, activeGeneration: &activeGeneration}).run()
727+ (&stdinPump{ctx: loopCtx, cancel: cancel, input: input, consumer: inputConsumer, out: sendCh, clock: clk, clipboard: clip, logger: log, paletteEvents: paletteEvents, activeGeneration: &activeGeneration}).run()
727728 }()
728- // A scanner must finish before the next attach starts. The lifecycle-owned
729- // reader remains available for reconnect, but a cancelled scanner cannot
730- // consume a later read result intended for its replacement.
729+ // Revoke this scanner's dequeue permission before waiting for it to exit.
730+ // The lifecycle-owned reader retains queued bytes for the next attempt.
731731 defer func() {
732+ input.revoke(inputConsumer)
732733 cancel()
733734 <-stdinDone
734735 }()
@@ -946,25 +947,112 @@ type terminalReadResult struct {
946947 // blocked after Run exits; stop drops its result and never closes caller-owned
947948 // stdin. Reconnects never add another reader.
948949 type terminalInputPump struct {
949- in io.Reader
950- reads chan terminalReadResult
951- done chan struct{}
952- startMu sync.Once
953- stopMu sync.Once
950+ in io.Reader
951+ done chan struct{}
952+
953+ mu sync.Mutex
954+ pending *terminalReadResult
955+ consumer uint64
956+ nextID uint64
957+ closed bool
958+ ready chan struct{}
959+ space chan struct{}
960+ startMu sync.Once
961+ stopMu sync.Once
954962 }
955963
956964 func newTerminalInputPump(in io.Reader) *terminalInputPump {
965+ space := make(chan struct{}, 1)
966+ space <- struct{}{}
957967 return &terminalInputPump{
958968 in: in,
959- reads: make(chan terminalReadResult, 1),
960969 done: make(chan struct{}),
970+ ready: make(chan struct{}, 1),
971+ space: space,
961972 }
962973 }
963974
975+// claim grants one attach scanner exclusive permission to dequeue terminal
976+// input. The reader itself remains lifecycle-owned across reconnects.
977+func (p *terminalInputPump) claim() uint64 {
978+ p.mu.Lock()
979+ defer p.mu.Unlock()
980+ if p.consumer != 0 {
981+ panic("terminal input consumer already claimed")
982+ }
983+ p.nextID++
984+ p.consumer = p.nextID
985+ return p.consumer
986+}
987+
988+// revoke invalidates an attempt before its replacement is allowed to claim
989+// input. Pending bytes are deliberately retained for that replacement.
990+func (p *terminalInputPump) revoke(consumer uint64) {
991+ p.mu.Lock()
992+ if p.consumer == consumer {
993+ p.consumer = 0
994+ }
995+ p.mu.Unlock()
996+ p.signalReady()
997+}
998+
999+func (p *terminalInputPump) signalReady() {
1000+ select {
1001+ case p.ready <- struct{}{}:
1002+ default:
1003+ }
1004+}
1005+
1006+// enqueue retains at most one completed terminal read. The reader blocks
1007+// before publishing another result, keeping handoff buffering bounded.
1008+func (p *terminalInputPump) enqueue(result terminalReadResult) bool {
1009+ select {
1010+ case <-p.done:
1011+ return false
1012+ case <-p.space:
1013+ }
1014+ p.mu.Lock()
1015+ if p.closed {
1016+ p.mu.Unlock()
1017+ p.space <- struct{}{}
1018+ return false
1019+ }
1020+ p.pending = &result
1021+ p.mu.Unlock()
1022+ p.signalReady()
1023+ return true
1024+}
1025+
1026+// take gives a ready result only to the current, non-cancelled consumer. It
1027+// checks cancellation after readiness is observed, so cancellation wins a
1028+// simultaneous cancellation/read race and leaves the result queued.
1029+func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
1030+ p.mu.Lock()
1031+ if ctx.Err() != nil || p.consumer != consumer || p.pending == nil {
1032+ pending := p.pending != nil
1033+ p.mu.Unlock()
1034+ if pending {
1035+ p.signalReady()
1036+ }
1037+ return terminalReadResult{}, false
1038+ }
1039+ result := *p.pending
1040+ p.pending = nil
1041+ p.mu.Unlock()
1042+ p.space <- struct{}{}
1043+ return result, true
1044+}
1045+
1046+func (p *terminalInputPump) finish() {
1047+ p.mu.Lock()
1048+ p.closed = true
1049+ p.mu.Unlock()
1050+ p.signalReady()
1051+}
1052+
9641053 func (p *terminalInputPump) start() {
9651054 p.startMu.Do(func() {
9661055 go func() {
967- defer close(p.reads)
9681056 buf := make([]byte, stdinBufSize)
9691057 for {
9701058 n, err := p.in.Read(buf)
@@ -972,12 +1060,11 @@ func (p *terminalInputPump) start() {
9721060 if n > 0 {
9731061 result.data = append([]byte(nil), buf[:n]...)
9741062 }
975- select {
976- case p.reads <- result:
977- case <-p.done:
1063+ if !p.enqueue(result) {
9781064 return
9791065 }
9801066 if err != nil {
1067+ p.finish()
9811068 return
9821069 }
9831070 }
@@ -998,6 +1085,7 @@ type stdinPump struct {
9981085 cancel context.CancelFunc
9991086 in io.Reader // compatibility for isolated scanner tests
10001087 input *terminalInputPump
1088+ consumer uint64
10011089 out chan<- ports.Frame
10021090 clock ports.Clock
10031091 clipboard ports.ClipboardReader
@@ -1060,7 +1148,11 @@ func (p *stdinPump) run() {
10601148 input.start()
10611149 defer input.stop()
10621150 }
1063- reads := input.reads
1151+ consumer := p.consumer
1152+ if consumer == 0 {
1153+ consumer = input.claim()
1154+ defer input.revoke(consumer)
1155+ }
10641156
10651157 var markerTimer ports.Timer
10661158 var markerTimerC <-chan time.Time
@@ -1105,9 +1197,13 @@ func (p *stdinPump) run() {
11051197 return
11061198 case <-markerTimerC:
11071199 flushMarkerPrefix()
1108- case result, ok := <-reads:
1200+ case <-input.ready:
1201+ result, ok := input.take(p.ctx, consumer)
11091202 if !ok {
1110- return
1203+ if p.ctx.Err() != nil {
1204+ return
1205+ }
1206+ continue
11111207 }
11121208 disarmMarkerDeadline()
11131209 // All scanner callbacks from this Read must retain this one generation.
@@ -609,6 +609,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
609 input.start()609 input.start()
610 defer input.stop()610 defer input.stop()
611 }611 }
612+ inputConsumer := input.claim()
612 613
613 sendCh := make(chan ports.Frame, sendQueueDepth)614 sendCh := make(chan ports.Frame, sendQueueDepth)
614 ackQueue := newCumulativeAckQueue()615 ackQueue := newCumulativeAckQueue()
@@ -723,12 +724,12 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
723 stdinDone := make(chan struct{})724 stdinDone := make(chan struct{})
724 go func() {725 go func() {
725 defer close(stdinDone)726 defer close(stdinDone)
726- (&stdinPump{ctx: loopCtx, cancel: cancel, input: input, out: sendCh, clock: clk, clipboard: clip, logger: log, paletteEvents: paletteEvents, activeGeneration: &activeGeneration}).run()727+ (&stdinPump{ctx: loopCtx, cancel: cancel, input: input, consumer: inputConsumer, out: sendCh, clock: clk, clipboard: clip, logger: log, paletteEvents: paletteEvents, activeGeneration: &activeGeneration}).run()
727 }()728 }()
728- // A scanner must finish before the next attach starts. The lifecycle-owned729+ // Revoke this scanner's dequeue permission before waiting for it to exit.
729- // reader remains available for reconnect, but a cancelled scanner cannot730+ // The lifecycle-owned reader retains queued bytes for the next attempt.
730- // consume a later read result intended for its replacement.
731 defer func() {731 defer func() {
732+ input.revoke(inputConsumer)
732 cancel()733 cancel()
733 <-stdinDone734 <-stdinDone
734 }()735 }()
@@ -946,25 +947,112 @@ type terminalReadResult struct {
946 // blocked after Run exits; stop drops its result and never closes caller-owned947 // blocked after Run exits; stop drops its result and never closes caller-owned
947 // stdin. Reconnects never add another reader.948 // stdin. Reconnects never add another reader.
948 type terminalInputPump struct {949 type terminalInputPump struct {
949- in io.Reader950+ in io.Reader
950- reads chan terminalReadResult951+ done chan struct{}
951- done chan struct{}952+
952- startMu sync.Once953+ mu sync.Mutex
953- stopMu sync.Once954+ pending *terminalReadResult
955+ consumer uint64
956+ nextID uint64
957+ closed bool
958+ ready chan struct{}
959+ space chan struct{}
960+ startMu sync.Once
961+ stopMu sync.Once
954 }962 }
955 963
956 func newTerminalInputPump(in io.Reader) *terminalInputPump {964 func newTerminalInputPump(in io.Reader) *terminalInputPump {
965+ space := make(chan struct{}, 1)
966+ space <- struct{}{}
957 return &terminalInputPump{967 return &terminalInputPump{
958 in: in,968 in: in,
959- reads: make(chan terminalReadResult, 1),
960 done: make(chan struct{}),969 done: make(chan struct{}),
970+ ready: make(chan struct{}, 1),
971+ space: space,
961 }972 }
962 }973 }
963 974
975+// claim grants one attach scanner exclusive permission to dequeue terminal
976+// input. The reader itself remains lifecycle-owned across reconnects.
977+func (p *terminalInputPump) claim() uint64 {
978+ p.mu.Lock()
979+ defer p.mu.Unlock()
980+ if p.consumer != 0 {
981+ panic("terminal input consumer already claimed")
982+ }
983+ p.nextID++
984+ p.consumer = p.nextID
985+ return p.consumer
986+}
987+
988+// revoke invalidates an attempt before its replacement is allowed to claim
989+// input. Pending bytes are deliberately retained for that replacement.
990+func (p *terminalInputPump) revoke(consumer uint64) {
991+ p.mu.Lock()
992+ if p.consumer == consumer {
993+ p.consumer = 0
994+ }
995+ p.mu.Unlock()
996+ p.signalReady()
997+}
998+
999+func (p *terminalInputPump) signalReady() {
1000+ select {
1001+ case p.ready <- struct{}{}:
1002+ default:
1003+ }
1004+}
1005+
1006+// enqueue retains at most one completed terminal read. The reader blocks
1007+// before publishing another result, keeping handoff buffering bounded.
1008+func (p *terminalInputPump) enqueue(result terminalReadResult) bool {
1009+ select {
1010+ case <-p.done:
1011+ return false
1012+ case <-p.space:
1013+ }
1014+ p.mu.Lock()
1015+ if p.closed {
1016+ p.mu.Unlock()
1017+ p.space <- struct{}{}
1018+ return false
1019+ }
1020+ p.pending = &result
1021+ p.mu.Unlock()
1022+ p.signalReady()
1023+ return true
1024+}
1025+
1026+// take gives a ready result only to the current, non-cancelled consumer. It
1027+// checks cancellation after readiness is observed, so cancellation wins a
1028+// simultaneous cancellation/read race and leaves the result queued.
1029+func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
1030+ p.mu.Lock()
1031+ if ctx.Err() != nil || p.consumer != consumer || p.pending == nil {
1032+ pending := p.pending != nil
1033+ p.mu.Unlock()
1034+ if pending {
1035+ p.signalReady()
1036+ }
1037+ return terminalReadResult{}, false
1038+ }
1039+ result := *p.pending
1040+ p.pending = nil
1041+ p.mu.Unlock()
1042+ p.space <- struct{}{}
1043+ return result, true
1044+}
1045+
1046+func (p *terminalInputPump) finish() {
1047+ p.mu.Lock()
1048+ p.closed = true
1049+ p.mu.Unlock()
1050+ p.signalReady()
1051+}
1052+
964 func (p *terminalInputPump) start() {1053 func (p *terminalInputPump) start() {
965 p.startMu.Do(func() {1054 p.startMu.Do(func() {
966 go func() {1055 go func() {
967- defer close(p.reads)
968 buf := make([]byte, stdinBufSize)1056 buf := make([]byte, stdinBufSize)
969 for {1057 for {
970 n, err := p.in.Read(buf)1058 n, err := p.in.Read(buf)
@@ -972,12 +1060,11 @@ func (p *terminalInputPump) start() {
972 if n > 0 {1060 if n > 0 {
973 result.data = append([]byte(nil), buf[:n]...)1061 result.data = append([]byte(nil), buf[:n]...)
974 }1062 }
975- select {1063+ if !p.enqueue(result) {
976- case p.reads <- result:
977- case <-p.done:
978 return1064 return
979 }1065 }
980 if err != nil {1066 if err != nil {
1067+ p.finish()
981 return1068 return
982 }1069 }
983 }1070 }
@@ -998,6 +1085,7 @@ type stdinPump struct {
998 cancel context.CancelFunc1085 cancel context.CancelFunc
999 in io.Reader // compatibility for isolated scanner tests1086 in io.Reader // compatibility for isolated scanner tests
1000 input *terminalInputPump1087 input *terminalInputPump
1088+ consumer uint64
1001 out chan<- ports.Frame1089 out chan<- ports.Frame
1002 clock ports.Clock1090 clock ports.Clock
1003 clipboard ports.ClipboardReader1091 clipboard ports.ClipboardReader
@@ -1060,7 +1148,11 @@ func (p *stdinPump) run() {
1060 input.start()1148 input.start()
1061 defer input.stop()1149 defer input.stop()
1062 }1150 }
1063- reads := input.reads1151+ consumer := p.consumer
1152+ if consumer == 0 {
1153+ consumer = input.claim()
1154+ defer input.revoke(consumer)
1155+ }
1064 1156
1065 var markerTimer ports.Timer1157 var markerTimer ports.Timer
1066 var markerTimerC <-chan time.Time1158 var markerTimerC <-chan time.Time
@@ -1105,9 +1197,13 @@ func (p *stdinPump) run() {
1105 return1197 return
1106 case <-markerTimerC:1198 case <-markerTimerC:
1107 flushMarkerPrefix()1199 flushMarkerPrefix()
1108- case result, ok := <-reads:1200+ case <-input.ready:
1201+ result, ok := input.take(p.ctx, consumer)
1109 if !ok {1202 if !ok {
1110- return1203+ if p.ctx.Err() != nil {
1204+ return
1205+ }
1206+ continue
1111 }1207 }
1112 disarmMarkerDeadline()1208 disarmMarkerDeadline()
1113 // All scanner callbacks from this Read must retain this one generation.1209 // All scanner callbacks from this Read must retain this one generation.
M internal/usecase/client/terminal_input_pump_test.go +49 -0
@@ -50,6 +50,55 @@ func (r *lifecycleReader) maxActive() int {
5050 return r.max
5151 }
5252
53+func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T) {
54+ input := newTerminalInputPump(nil)
55+ // Both cancellation and a terminal read are ready before the old scanner
56+ // tries to dequeue. It must surrender the read to its replacement.
57+ want := []byte("first\x00second\xff")
58+ input.enqueue(terminalReadResult{data: append([]byte(nil), want...)})
59+
60+ var activeGeneration atomic.Uint64
61+ cancelledCtx, cancelCancelled := context.WithCancel(context.Background())
62+ cancelCancelled()
63+ cancelledDone := make(chan struct{})
64+ go func() {
65+ (&stdinPump{
66+ ctx: cancelledCtx, cancel: cancelCancelled, input: input,
67+ out: make(chan ports.Frame, 1), clock: newAttachPaletteClock(),
68+ logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
69+ activeGeneration: &activeGeneration,
70+ }).run()
71+ close(cancelledDone)
72+ }()
73+ <-cancelledDone
74+
75+ replacementCtx, cancelReplacement := context.WithCancel(context.Background())
76+ defer cancelReplacement()
77+ out := make(chan ports.Frame, len(want))
78+ replacementDone := make(chan struct{})
79+ go func() {
80+ (&stdinPump{
81+ ctx: replacementCtx, cancel: cancelReplacement, input: input,
82+ out: out, clock: newAttachPaletteClock(),
83+ logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
84+ activeGeneration: &activeGeneration,
85+ }).run()
86+ close(replacementDone)
87+ }()
88+
89+ var got []byte
90+ for len(got) < len(want) {
91+ frame := <-out
92+ input, err := ports.UnmarshalInput(frame.Payload)
93+ require.NoError(t, err)
94+ got = append(got, input.Data...)
95+ }
96+ require.Equal(t, want, got, "handoff must preserve every queued byte in order")
97+
98+ cancelReplacement()
99+ <-replacementDone
100+}
101+
53102 func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {
54103 reader := newLifecycleReader()
55104 input := newTerminalInputPump(reader)
@@ -50,6 +50,55 @@ func (r *lifecycleReader) maxActive() int {
50 return r.max50 return r.max
51 }51 }
52 52
53+func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T) {
54+ input := newTerminalInputPump(nil)
55+ // Both cancellation and a terminal read are ready before the old scanner
56+ // tries to dequeue. It must surrender the read to its replacement.
57+ want := []byte("first\x00second\xff")
58+ input.enqueue(terminalReadResult{data: append([]byte(nil), want...)})
59+
60+ var activeGeneration atomic.Uint64
61+ cancelledCtx, cancelCancelled := context.WithCancel(context.Background())
62+ cancelCancelled()
63+ cancelledDone := make(chan struct{})
64+ go func() {
65+ (&stdinPump{
66+ ctx: cancelledCtx, cancel: cancelCancelled, input: input,
67+ out: make(chan ports.Frame, 1), clock: newAttachPaletteClock(),
68+ logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
69+ activeGeneration: &activeGeneration,
70+ }).run()
71+ close(cancelledDone)
72+ }()
73+ <-cancelledDone
74+
75+ replacementCtx, cancelReplacement := context.WithCancel(context.Background())
76+ defer cancelReplacement()
77+ out := make(chan ports.Frame, len(want))
78+ replacementDone := make(chan struct{})
79+ go func() {
80+ (&stdinPump{
81+ ctx: replacementCtx, cancel: cancelReplacement, input: input,
82+ out: out, clock: newAttachPaletteClock(),
83+ logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
84+ activeGeneration: &activeGeneration,
85+ }).run()
86+ close(replacementDone)
87+ }()
88+
89+ var got []byte
90+ for len(got) < len(want) {
91+ frame := <-out
92+ input, err := ports.UnmarshalInput(frame.Payload)
93+ require.NoError(t, err)
94+ got = append(got, input.Data...)
95+ }
96+ require.Equal(t, want, got, "handoff must preserve every queued byte in order")
97+
98+ cancelReplacement()
99+ <-replacementDone
100+}
101+
53 func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {102 func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {
54 reader := newLifecycleReader()103 reader := newLifecycleReader()
55 input := newTerminalInputPump(reader)104 input := newTerminalInputPump(reader)

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