detached f5d748a

fix(client): prevent late terminal input after stop

commit
f5d748a777a867e99f9e3b677632d6d0896ffc25
author
brice · · 2026-07-18 11:08 UTC
committer
brice · · 2026-07-18 11:13 UTC
parent
a18957d
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +189 -13
M internal/usecase/client/client.go +124 -13
@@ -167,6 +167,9 @@ type Runner struct {
167167 clipboard ports.ClipboardReader
168168 logger *slog.Logger
169169 runtimeObserver ports.SerializedRuntimeObserver
170+
171+ inputMu sync.Mutex
172+ input *terminalInputPump
170173 }
171174
172175 // NewRunner constructs a client runner. A nil logger uses the process default.
@@ -185,6 +188,21 @@ func NewRunner(deps Dependencies) *Runner {
185188 }
186189 }
187190
191+// terminalInput returns the Runner-owned sole reader. It survives sequential
192+// Run calls because a caller-owned reader may remain blocked after an attach
193+// exits and therefore cannot safely be replaced by another reader.
194+func (r *Runner) terminalInput() *terminalInputPump {
195+ r.inputMu.Lock()
196+ defer r.inputMu.Unlock()
197+
198+ if r.input == nil || r.input.hasExited() || r.input.isClosed() {
199+ r.input = newTerminalInputPump(r.term.In())
200+ r.input.start()
201+ }
202+ r.input.resume()
203+ return r.input
204+}
205+
188206 // Run connects and runs the attach client. It owns the terminal lifecycle
189207 // above attach attempts so raw mode remains active while a live client process
190208 // redials a lost link.
@@ -233,7 +251,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
233251 var input *terminalInputPump
234252 defer func() {
235253 if input != nil {
236- input.stop()
254+ input.suspend()
237255 }
238256 }()
239257 originalEnterRaw := enterRaw
@@ -242,8 +260,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
242260 return err
243261 }
244262 if input == nil {
245- input = newTerminalInputPump(r.term.In())
246- input.start()
263+ input = r.terminalInput()
247264 }
248265 return nil
249266 }
@@ -956,9 +973,9 @@ type terminalReadResult struct {
956973
957974 // terminalInputPump is the sole consumer of the caller-owned terminal reader.
958975 // It is started once after raw mode is entered and is reused by each attach
959-// scanner. A bare io.Reader cannot be interrupted, so one final Read may stay
960-// blocked after Run exits; stop drops its result and never closes caller-owned
961-// stdin. Reconnects never add another reader.
976+// scanner. A bare io.Reader cannot be interrupted, so a Read may stay blocked
977+// after Run exits; suspend drops any result received while inactive and never
978+// closes caller-owned stdin. Reconnects and later runs never add another reader.
962979 type terminalInputPump struct {
963980 in io.Reader
964981 done chan struct{}
@@ -969,8 +986,11 @@ type terminalInputPump struct {
969986 delivering uint64
970987 nextID uint64
971988 closed bool
989+ active bool
972990 ready chan struct{}
973991 space chan struct{}
992+ state chan struct{}
993+ exited chan struct{}
974994 startMu sync.Once
975995 stopMu sync.Once
976996 }
@@ -979,10 +999,13 @@ func newTerminalInputPump(in io.Reader) *terminalInputPump {
979999 space := make(chan struct{}, 1)
9801000 space <- struct{}{}
9811001 return &terminalInputPump{
982- in: in,
983- done: make(chan struct{}),
984- ready: make(chan struct{}, 1),
985- space: space,
1002+ in: in,
1003+ done: make(chan struct{}),
1004+ active: true,
1005+ ready: make(chan struct{}, 1),
1006+ space: space,
1007+ state: make(chan struct{}, 1),
1008+ exited: make(chan struct{}),
9861009 }
9871010 }
9881011
@@ -1031,7 +1054,7 @@ func (p *terminalInputPump) enqueue(result terminalReadResult) bool {
10311054 case <-p.space:
10321055 }
10331056 p.mu.Lock()
1034- if p.closed {
1057+ if p.closed || !p.active {
10351058 p.mu.Unlock()
10361059 p.space <- struct{}{}
10371060 return false
@@ -1080,20 +1103,54 @@ func (p *terminalInputPump) finish() {
10801103 p.closed = true
10811104 p.mu.Unlock()
10821105 p.signalReady()
1106+ p.signalState()
1107+}
1108+
1109+func (p *terminalInputPump) signalState() {
1110+ select {
1111+ case p.state <- struct{}{}:
1112+ default:
1113+ }
1114+}
1115+
1116+func (p *terminalInputPump) waitForActive() bool {
1117+ for {
1118+ p.mu.Lock()
1119+ active, closed := p.active, p.closed
1120+ p.mu.Unlock()
1121+ if closed {
1122+ return false
1123+ }
1124+ if active {
1125+ return true
1126+ }
1127+ select {
1128+ case <-p.done:
1129+ return false
1130+ case <-p.state:
1131+ }
1132+ }
10831133 }
10841134
10851135 func (p *terminalInputPump) start() {
10861136 p.startMu.Do(func() {
10871137 go func() {
1138+ defer close(p.exited)
10881139 buf := make([]byte, stdinBufSize)
10891140 for {
1141+ if !p.waitForActive() {
1142+ return
1143+ }
10901144 n, err := p.in.Read(buf)
10911145 result := terminalReadResult{err: err}
10921146 if n > 0 {
10931147 result.data = append([]byte(nil), buf[:n]...)
10941148 }
10951149 if !p.enqueue(result) {
1096- return
1150+ if p.isClosed() {
1151+ return
1152+ }
1153+ continue
10971154 }
10981155 if err != nil {
10991156 p.finish()
@@ -1104,8 +1161,62 @@ func (p *terminalInputPump) start() {
11041161 })
11051162 }
11061163
1164+func (p *terminalInputPump) hasExited() bool {
1165+ select {
1166+ case <-p.exited:
1167+ return true
1168+ default:
1169+ return false
1170+ }
1171+}
1172+
1173+func (p *terminalInputPump) isClosed() bool {
1174+ p.mu.Lock()
1175+ defer p.mu.Unlock()
1176+ return p.closed
1177+}
1178+
1179+// suspend drops input completed after an attach ends and parks the lifecycle
1180+// reader before its next Read. A later Run resumes this same reader, avoiding
1181+// both input replay between runs and a competing read on caller-owned stdin.
1182+func (p *terminalInputPump) suspend() {
1183+ p.mu.Lock()
1184+ p.active = false
1185+ discarded := p.pending != nil
1186+ p.pending = nil
1187+ p.delivering = 0
1188+ p.mu.Unlock()
1189+ if discarded {
1190+ select {
1191+ case p.space <- struct{}{}:
1192+ default:
1193+ }
1194+ }
1195+ p.signalState()
1196+ p.signalReady()
1197+}
1198+
1199+func (p *terminalInputPump) resume() {
1200+ p.mu.Lock()
1201+ if !p.closed {
1202+ p.active = true
1203+ }
1204+ p.mu.Unlock()
1205+ p.signalState()
1206+}
1207+
11071208 func (p *terminalInputPump) stop() {
1108- p.stopMu.Do(func() { close(p.done) })
1209+ p.stopMu.Do(func() {
1210+ // Mark closure while holding the same mutex that publishes a completed
1211+ // Read. A Read which returns after stop therefore cannot publish even
1212+ // when both done and space are selectable in enqueue.
1213+ p.mu.Lock()
1214+ p.closed = true
1215+ p.mu.Unlock()
1216+ close(p.done)
1217+ p.signalReady()
1218+ p.signalState()
1219+ })
11091220 }
11101221
11111222 // stdinPump parses terminal color reports into generation-tagged events while
@@ -167,6 +167,9 @@ type Runner struct {
167 clipboard ports.ClipboardReader167 clipboard ports.ClipboardReader
168 logger *slog.Logger168 logger *slog.Logger
169 runtimeObserver ports.SerializedRuntimeObserver169 runtimeObserver ports.SerializedRuntimeObserver
170+
171+ inputMu sync.Mutex
172+ input *terminalInputPump
170 }173 }
171 174
172 // NewRunner constructs a client runner. A nil logger uses the process default.175 // NewRunner constructs a client runner. A nil logger uses the process default.
@@ -185,6 +188,21 @@ func NewRunner(deps Dependencies) *Runner {
185 }188 }
186 }189 }
187 190
191+// terminalInput returns the Runner-owned sole reader. It survives sequential
192+// Run calls because a caller-owned reader may remain blocked after an attach
193+// exits and therefore cannot safely be replaced by another reader.
194+func (r *Runner) terminalInput() *terminalInputPump {
195+ r.inputMu.Lock()
196+ defer r.inputMu.Unlock()
197+
198+ if r.input == nil || r.input.hasExited() || r.input.isClosed() {
199+ r.input = newTerminalInputPump(r.term.In())
200+ r.input.start()
201+ }
202+ r.input.resume()
203+ return r.input
204+}
205+
188 // Run connects and runs the attach client. It owns the terminal lifecycle206 // Run connects and runs the attach client. It owns the terminal lifecycle
189 // above attach attempts so raw mode remains active while a live client process207 // above attach attempts so raw mode remains active while a live client process
190 // redials a lost link.208 // redials a lost link.
@@ -233,7 +251,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
233 var input *terminalInputPump251 var input *terminalInputPump
234 defer func() {252 defer func() {
235 if input != nil {253 if input != nil {
236- input.stop()254+ input.suspend()
237 }255 }
238 }()256 }()
239 originalEnterRaw := enterRaw257 originalEnterRaw := enterRaw
@@ -242,8 +260,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
242 return err260 return err
243 }261 }
244 if input == nil {262 if input == nil {
245- input = newTerminalInputPump(r.term.In())263+ input = r.terminalInput()
246- input.start()
247 }264 }
248 return nil265 return nil
249 }266 }
@@ -956,9 +973,9 @@ type terminalReadResult struct {
956 973
957 // terminalInputPump is the sole consumer of the caller-owned terminal reader.974 // terminalInputPump is the sole consumer of the caller-owned terminal reader.
958 // It is started once after raw mode is entered and is reused by each attach975 // It is started once after raw mode is entered and is reused by each attach
959-// scanner. A bare io.Reader cannot be interrupted, so one final Read may stay976+// scanner. A bare io.Reader cannot be interrupted, so a Read may stay blocked
960-// blocked after Run exits; stop drops its result and never closes caller-owned977+// after Run exits; suspend drops any result received while inactive and never
961-// stdin. Reconnects never add another reader.978+// closes caller-owned stdin. Reconnects and later runs never add another reader.
962 type terminalInputPump struct {979 type terminalInputPump struct {
963 in io.Reader980 in io.Reader
964 done chan struct{}981 done chan struct{}
@@ -969,8 +986,11 @@ type terminalInputPump struct {
969 delivering uint64986 delivering uint64
970 nextID uint64987 nextID uint64
971 closed bool988 closed bool
989+ active bool
972 ready chan struct{}990 ready chan struct{}
973 space chan struct{}991 space chan struct{}
992+ state chan struct{}
993+ exited chan struct{}
974 startMu sync.Once994 startMu sync.Once
975 stopMu sync.Once995 stopMu sync.Once
976 }996 }
@@ -979,10 +999,13 @@ func newTerminalInputPump(in io.Reader) *terminalInputPump {
979 space := make(chan struct{}, 1)999 space := make(chan struct{}, 1)
980 space <- struct{}{}1000 space <- struct{}{}
981 return &terminalInputPump{1001 return &terminalInputPump{
982- in: in,1002+ in: in,
983- done: make(chan struct{}),1003+ done: make(chan struct{}),
984- ready: make(chan struct{}, 1),1004+ active: true,
985- space: space,1005+ ready: make(chan struct{}, 1),
1006+ space: space,
1007+ state: make(chan struct{}, 1),
1008+ exited: make(chan struct{}),
986 }1009 }
987 }1010 }
988 1011
@@ -1031,7 +1054,7 @@ func (p *terminalInputPump) enqueue(result terminalReadResult) bool {
1031 case <-p.space:1054 case <-p.space:
1032 }1055 }
1033 p.mu.Lock()1056 p.mu.Lock()
1034- if p.closed {1057+ if p.closed || !p.active {
1035 p.mu.Unlock()1058 p.mu.Unlock()
1036 p.space <- struct{}{}1059 p.space <- struct{}{}
1037 return false1060 return false
@@ -1080,20 +1103,54 @@ func (p *terminalInputPump) finish() {
1080 p.closed = true1103 p.closed = true
1081 p.mu.Unlock()1104 p.mu.Unlock()
1082 p.signalReady()1105 p.signalReady()
1106+ p.signalState()
1107+}
1108+
1109+func (p *terminalInputPump) signalState() {
1110+ select {
1111+ case p.state <- struct{}{}:
1112+ default:
1113+ }
1114+}
1115+
1116+func (p *terminalInputPump) waitForActive() bool {
1117+ for {
1118+ p.mu.Lock()
1119+ active, closed := p.active, p.closed
1120+ p.mu.Unlock()
1121+ if closed {
1122+ return false
1123+ }
1124+ if active {
1125+ return true
1126+ }
1127+ select {
1128+ case <-p.done:
1129+ return false
1130+ case <-p.state:
1131+ }
1132+ }
1083 }1133 }
1084 1134
1085 func (p *terminalInputPump) start() {1135 func (p *terminalInputPump) start() {
1086 p.startMu.Do(func() {1136 p.startMu.Do(func() {
1087 go func() {1137 go func() {
1138+ defer close(p.exited)
1088 buf := make([]byte, stdinBufSize)1139 buf := make([]byte, stdinBufSize)
1089 for {1140 for {
1141+ if !p.waitForActive() {
1142+ return
1143+ }
1090 n, err := p.in.Read(buf)1144 n, err := p.in.Read(buf)
1091 result := terminalReadResult{err: err}1145 result := terminalReadResult{err: err}
1092 if n > 0 {1146 if n > 0 {
1093 result.data = append([]byte(nil), buf[:n]...)1147 result.data = append([]byte(nil), buf[:n]...)
1094 }1148 }
1095 if !p.enqueue(result) {1149 if !p.enqueue(result) {
1096- return1150+ if p.isClosed() {
1151+ return
1152+ }
1153+ continue
1097 }1154 }
1098 if err != nil {1155 if err != nil {
1099 p.finish()1156 p.finish()
@@ -1104,8 +1161,62 @@ func (p *terminalInputPump) start() {
1104 })1161 })
1105 }1162 }
1106 1163
1164+func (p *terminalInputPump) hasExited() bool {
1165+ select {
1166+ case <-p.exited:
1167+ return true
1168+ default:
1169+ return false
1170+ }
1171+}
1172+
1173+func (p *terminalInputPump) isClosed() bool {
1174+ p.mu.Lock()
1175+ defer p.mu.Unlock()
1176+ return p.closed
1177+}
1178+
1179+// suspend drops input completed after an attach ends and parks the lifecycle
1180+// reader before its next Read. A later Run resumes this same reader, avoiding
1181+// both input replay between runs and a competing read on caller-owned stdin.
1182+func (p *terminalInputPump) suspend() {
1183+ p.mu.Lock()
1184+ p.active = false
1185+ discarded := p.pending != nil
1186+ p.pending = nil
1187+ p.delivering = 0
1188+ p.mu.Unlock()
1189+ if discarded {
1190+ select {
1191+ case p.space <- struct{}{}:
1192+ default:
1193+ }
1194+ }
1195+ p.signalState()
1196+ p.signalReady()
1197+}
1198+
1199+func (p *terminalInputPump) resume() {
1200+ p.mu.Lock()
1201+ if !p.closed {
1202+ p.active = true
1203+ }
1204+ p.mu.Unlock()
1205+ p.signalState()
1206+}
1207+
1107 func (p *terminalInputPump) stop() {1208 func (p *terminalInputPump) stop() {
1108- p.stopMu.Do(func() { close(p.done) })1209+ p.stopMu.Do(func() {
1210+ // Mark closure while holding the same mutex that publishes a completed
1211+ // Read. A Read which returns after stop therefore cannot publish even
1212+ // when both done and space are selectable in enqueue.
1213+ p.mu.Lock()
1214+ p.closed = true
1215+ p.mu.Unlock()
1216+ close(p.done)
1217+ p.signalReady()
1218+ p.signalState()
1219+ })
1109 }1220 }
1110 1221
1111 // stdinPump parses terminal color reports into generation-tagged events while1222 // stdinPump parses terminal color reports into generation-tagged events while
M internal/usecase/client/terminal_input_pump_test.go +65 -0
@@ -10,6 +10,7 @@ import (
1010
1111 "github.com/stretchr/testify/require"
1212
13+ "github.com/bnema/vev/internal/domain"
1314 "github.com/bnema/vev/internal/ports"
1415 )
1516
@@ -50,6 +51,70 @@ func (r *lifecycleReader) maxActive() int {
5051 return r.max
5152 }
5253
54+type lateStopReader struct {
55+ started chan struct{}
56+ release chan terminalReadResult
57+ reads atomic.Int32
58+}
59+
60+func newLateStopReader() *lateStopReader {
61+ return &lateStopReader{
62+ started: make(chan struct{}, 2),
63+ release: make(chan terminalReadResult, 1),
64+ }
65+}
66+
67+func (r *lateStopReader) Read(p []byte) (int, error) {
68+ r.reads.Add(1)
69+ r.started <- struct{}{}
70+ result := <-r.release
71+ return copy(p, result.data), result.err
72+}
73+
74+func TestTerminalInputPumpStopDropsLateReadWithoutStartingAnother(t *testing.T) {
75+ reader := newLateStopReader()
76+ input := newTerminalInputPump(reader)
77+ input.start()
78+ <-reader.started // The sole lifecycle read is blocked in the caller-owned reader.
79+
80+ input.stop()
81+ reader.release <- terminalReadResult{data: []byte("late")}
82+
83+ // The late read must be discarded rather than published. Both possible
84+ // outcomes are channel-synchronized: a correct pump exits; the historical
85+ // race published and immediately issued a second Read.
86+ select {
87+ case <-input.exited:
88+ case <-reader.started:
89+ require.Fail(t, "pump issued another Read after stop")
90+ }
91+ require.Equal(t, int32(1), reader.reads.Load())
92+
93+ consumer := input.claim()
94+ _, ok := input.take(context.Background(), consumer)
95+ require.False(t, ok, "late read must not be available after stop")
96+ input.revoke(consumer)
97+ input.stop() // stop is deliberately idempotent.
98+}
99+
100+func TestRunnerReusesBlockedTerminalInputPumpAcrossRuns(t *testing.T) {
101+ reader := newLateStopReader()
102+ runner := &Runner{term: &attachPaletteTerminalHarness{
103+ in: reader, out: newAttachPaletteWriter(), resize: make(chan domain.Size),
104+ }}
105+
106+ first := runner.terminalInput()
107+ <-reader.started
108+ first.suspend()
109+ second := runner.terminalInput()
110+ require.Same(t, first, second, "a later Run must reuse a blocked caller-owned reader")
111+ require.Equal(t, int32(1), reader.reads.Load())
112+
113+ first.stop()
114+ reader.release <- terminalReadResult{}
115+ <-first.exited
116+}
117+
53118 func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T) {
54119 input := newTerminalInputPump(nil)
55120 // Both cancellation and a terminal read are ready before the old scanner
@@ -10,6 +10,7 @@ import (
10 10
11 "github.com/stretchr/testify/require"11 "github.com/stretchr/testify/require"
12 12
13+ "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"14 "github.com/bnema/vev/internal/ports"
14 )15 )
15 16
@@ -50,6 +51,70 @@ func (r *lifecycleReader) maxActive() int {
50 return r.max51 return r.max
51 }52 }
52 53
54+type lateStopReader struct {
55+ started chan struct{}
56+ release chan terminalReadResult
57+ reads atomic.Int32
58+}
59+
60+func newLateStopReader() *lateStopReader {
61+ return &lateStopReader{
62+ started: make(chan struct{}, 2),
63+ release: make(chan terminalReadResult, 1),
64+ }
65+}
66+
67+func (r *lateStopReader) Read(p []byte) (int, error) {
68+ r.reads.Add(1)
69+ r.started <- struct{}{}
70+ result := <-r.release
71+ return copy(p, result.data), result.err
72+}
73+
74+func TestTerminalInputPumpStopDropsLateReadWithoutStartingAnother(t *testing.T) {
75+ reader := newLateStopReader()
76+ input := newTerminalInputPump(reader)
77+ input.start()
78+ <-reader.started // The sole lifecycle read is blocked in the caller-owned reader.
79+
80+ input.stop()
81+ reader.release <- terminalReadResult{data: []byte("late")}
82+
83+ // The late read must be discarded rather than published. Both possible
84+ // outcomes are channel-synchronized: a correct pump exits; the historical
85+ // race published and immediately issued a second Read.
86+ select {
87+ case <-input.exited:
88+ case <-reader.started:
89+ require.Fail(t, "pump issued another Read after stop")
90+ }
91+ require.Equal(t, int32(1), reader.reads.Load())
92+
93+ consumer := input.claim()
94+ _, ok := input.take(context.Background(), consumer)
95+ require.False(t, ok, "late read must not be available after stop")
96+ input.revoke(consumer)
97+ input.stop() // stop is deliberately idempotent.
98+}
99+
100+func TestRunnerReusesBlockedTerminalInputPumpAcrossRuns(t *testing.T) {
101+ reader := newLateStopReader()
102+ runner := &Runner{term: &attachPaletteTerminalHarness{
103+ in: reader, out: newAttachPaletteWriter(), resize: make(chan domain.Size),
104+ }}
105+
106+ first := runner.terminalInput()
107+ <-reader.started
108+ first.suspend()
109+ second := runner.terminalInput()
110+ require.Same(t, first, second, "a later Run must reuse a blocked caller-owned reader")
111+ require.Equal(t, int32(1), reader.reads.Load())
112+
113+ first.stop()
114+ reader.release <- terminalReadResult{}
115+ <-first.exited
116+}
117+
53 func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T) {118 func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T) {
54 input := newTerminalInputPump(nil)119 input := newTerminalInputPump(nil)
55 // Both cancellation and a terminal read are ready before the old scanner120 // Both cancellation and a terminal read are ready before the old scanner

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