1package client
2
3import (
4 "bytes"
5 "errors"
6 "strings"
7 "sync/atomic"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/require"
12
13 "github.com/bnema/vev/internal/protocol"
14)
15
16// gatedSessionStream blocks every write until release is closed, like a
17// carriage stalled on a dead link, and can fail writes after release.
18type gatedSessionStream struct {
19 *sessionTestStream
20 release chan struct{}
21 fail error
22}
23
24func (s *gatedSessionStream) SendClient(message protocol.ClientMessage) error {
25 select {
26 case <-s.release:
27 case <-s.closed:
28 return errSessionTestClosed
29 }
30 if s.fail != nil {
31 return s.fail
32 }
33 return s.sessionTestStream.SendClient(message)
34}
35
36func TestAttachmentOutbox(t *testing.T) {
37 writeErr := errors.New("link lost")
38 tests := []struct {
39 name string
40 fail error
41 wantErr error
42 }{
43 {name: "stalled link does not block enqueue and preserves order"},
44 {name: "write failure is sticky and signalled", fail: writeErr, wantErr: writeErr},
45 }
46 for _, tt := range tests {
47 t.Run(tt.name, func(t *testing.T) {
48 stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{}), fail: tt.fail}
49 outbox := newAttachmentOutbox(stream)
50 defer outbox.stop()
51
52 messages := []protocol.ClientMessage{
53 protocol.Input{InputSeq: 1, Data: []byte("a")},
54 protocol.Input{InputSeq: 2, Data: []byte("b")},
55 protocol.Input{InputSeq: 3, Data: []byte("c")},
56 }
57 enqueued := make(chan struct{})
58 go func() {
59 for _, m := range messages {
60 if err := outbox.SendClient(m); err != nil {
61 t.Errorf("SendClient: %v", err)
62 }
63 }
64 close(enqueued)
65 }()
66 select {
67 case <-enqueued:
68 case <-time.After(5 * time.Second):
69 t.Fatal("SendClient blocked on a stalled link")
70 }
71
72 close(stream.release)
73 if tt.wantErr != nil {
74 <-outbox.Failed()
75 require.ErrorIs(t, outbox.failure(), tt.wantErr)
76 require.ErrorIs(t, outbox.SendClient(protocol.Detach{}), tt.wantErr)
77 return
78 }
79 <-outbox.Idle()
80 stream.mu.Lock()
81 defer stream.mu.Unlock()
82 require.Equal(t, messages, stream.sent)
83 })
84 }
85}
86
87func TestAttachmentOutboxCloseUnblocksStalledWrite(t *testing.T) {
88 stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
89 outbox := newAttachmentOutbox(stream)
90 require.NoError(t, outbox.SendClient(protocol.Detach{}))
91
92 require.NoError(t, outbox.Close())
93 select {
94 case <-outbox.done:
95 case <-time.After(5 * time.Second):
96 t.Fatal("sender did not exit after Close")
97 }
98}
99
100// TestResumeWatch proves the resume backoff reads the terminal: Ctrl-C or a
101// lone Esc cancels, anything else is replayed to the next owner in order.
102func TestResumeWatch(t *testing.T) {
103 tests := []struct {
104 name string
105 reads []string
106 wantCancel bool
107 fireEscape bool
108 wantResidual string
109 }{
110 {name: "ctrl-c cancels and drops held keys", reads: []string{"ls", "\x03"}, wantCancel: true},
111 {name: "lone esc cancels", reads: []string{"\x1b"}, wantCancel: true, fireEscape: true},
112 {name: "kitty esc cancels", reads: []string{"\x1b[27u"}, wantCancel: true},
113 {name: "kitty ctrl-c cancels", reads: []string{"\x1b[99;5u"}, wantCancel: true},
114 {name: "kitty ctrl-c with lock bits cancels", reads: []string{"\x1b[99;69u"}, wantCancel: true},
115 {name: "escape sequence is kept", reads: []string{"\x1b[A"}, wantResidual: "\x1b[A"},
116 {name: "typed keys are kept in order", reads: []string{"l", "s", "\r"}, wantResidual: "ls\r"},
117 }
118 for _, tt := range tests {
119 t.Run(tt.name, func(t *testing.T) {
120 reader := &inputTestReader{chunks: make(chan []byte, len(tt.reads))}
121 pump := newTerminalInputPump(reader)
122 pump.start()
123 t.Cleanup(func() {
124 pump.stop()
125 close(reader.chunks)
126 })
127 lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
128 clock := newSupervisorTestClock()
129 watch := lifetime.watchResume(clock)
130 for _, read := range tt.reads {
131 reader.chunks <- []byte(read)
132 }
133 if tt.wantCancel {
134 if tt.fireEscape {
135 clock.awaitTimer(t).fire()
136 }
137 select {
138 case <-watch.cancelled:
139 case <-time.After(5 * time.Second):
140 t.Fatal("cancel key ignored")
141 }
142 } else {
143 require.Eventually(t, func() bool { return watch.heldLen() == len(tt.wantResidual) }, 5*time.Second, time.Millisecond)
144 }
145 watch.release()
146 pump.mu.Lock()
147 defer pump.mu.Unlock()
148 require.Equal(t, tt.wantResidual, string(pump.residual))
149 require.Zero(t, pump.consumer, "release returns the claim")
150 })
151 }
152}
153
154// TestResumeWatchKeepsBacklogVerbatim starts a watch while session input the
155// lost attachment kept is still waiting: a Ctrl-C or trailing Esc in it was
156// typed for the remote shell, so it is replayed, not read as a cancel key.
157// Keys typed after the watch started still cancel.
158func TestResumeWatchKeepsBacklogVerbatim(t *testing.T) {
159 for _, tt := range []struct {
160 name, residual, pending, typed string
161 wantCancel bool
162 wantResidual string
163 }{
164 {name: "kept ctrl-c", residual: "\x03", wantResidual: "\x03"},
165 {name: "kept trailing esc", residual: "ihello\x1b", wantResidual: "ihello\x1b"},
166 {name: "pending ctrl-c", pending: "x\x03", wantResidual: "x\x03"},
167 {name: "residual then pending", residual: "a", pending: "\x03", wantResidual: "a\x03"},
168 {name: "new ctrl-c still cancels", residual: "\x03", typed: "\x03", wantCancel: true},
169 } {
170 t.Run(tt.name, func(t *testing.T) {
171 reader := &inputTestReader{chunks: make(chan []byte, 2)}
172 pump := newTerminalInputPump(reader)
173 pump.start()
174 t.Cleanup(func() {
175 pump.stop()
176 close(reader.chunks)
177 })
178 pump.mu.Lock()
179 pump.residual = []byte(tt.residual)
180 pump.mu.Unlock()
181 if tt.pending != "" {
182 reader.chunks <- []byte(tt.pending)
183 require.Eventually(t, func() bool {
184 pump.mu.Lock()
185 defer pump.mu.Unlock()
186 return pump.pending != nil
187 }, 5*time.Second, time.Millisecond)
188 }
189 lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
190 watch := lifetime.watchResume(newSupervisorTestClock())
191 require.Eventually(t, func() bool { return watch.heldLen() == len(tt.residual)+len(tt.pending) }, 5*time.Second, time.Millisecond)
192 if tt.typed != "" {
193 reader.chunks <- []byte(tt.typed)
194 }
195 if tt.wantCancel {
196 select {
197 case <-watch.cancelled:
198 case <-time.After(5 * time.Second):
199 t.Fatal("new cancel key ignored")
200 }
201 }
202 watch.release()
203 select {
204 case <-watch.cancelled:
205 require.True(t, tt.wantCancel, "backlog cancelled the resume")
206 default:
207 }
208 pump.mu.Lock()
209 defer pump.mu.Unlock()
210 require.Equal(t, tt.wantResidual, string(pump.residual))
211 })
212 }
213}
214
215// TestResumeWatchResolvesEscapeAtRelease presses Esc just before the phase
216// ends: release waits for the escape delay, so a lone Esc still cancels and an
217// escape sequence completed in time is kept for the session.
218func TestResumeWatchResolvesEscapeAtRelease(t *testing.T) {
219 for _, tt := range []struct {
220 name, after string
221 wantCancel bool
222 wantResidual string
223 }{
224 {name: "lone esc cancels", wantCancel: true},
225 {name: "arrow key is kept", after: "[A", wantResidual: "\x1b[A"},
226 } {
227 t.Run(tt.name, func(t *testing.T) {
228 reader := &inputTestReader{chunks: make(chan []byte, 2)}
229 pump := newTerminalInputPump(reader)
230 pump.start()
231 t.Cleanup(func() {
232 pump.stop()
233 close(reader.chunks)
234 })
235 lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
236 clock := newSupervisorTestClock()
237 watch := lifetime.watchResume(clock)
238 reader.chunks <- []byte("\x1b")
239 escape := clock.awaitTimer(t)
240 released := make(chan struct{})
241 go func() { watch.release(); close(released) }()
242 select {
243 case <-released:
244 t.Fatal("release returned before the escape was resolved")
245 case <-time.After(50 * time.Millisecond):
246 }
247 if tt.after != "" {
248 reader.chunks <- []byte(tt.after)
249 } else {
250 escape.fire()
251 }
252 select {
253 case <-released:
254 case <-time.After(5 * time.Second):
255 t.Fatal("release never returned")
256 }
257 select {
258 case <-watch.cancelled:
259 require.True(t, tt.wantCancel, "escape sequence cancelled the resume")
260 default:
261 require.False(t, tt.wantCancel, "lone esc at release did not cancel")
262 }
263 pump.mu.Lock()
264 defer pump.mu.Unlock()
265 require.Equal(t, tt.wantResidual, string(pump.residual))
266 })
267 }
268}
269
270func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {
271 w := &resumeWatch{}
272 w.hold(bytes.Repeat([]byte("a"), resumeHeldInputLimit-1))
273 w.hold([]byte("bb"))
274 w.hold([]byte("c"))
275 require.Equal(t, resumeHeldInputLimit-1, w.heldLen(), "keys after an overflow must be dropped too")
276}
277
278// stallingSessionStream passes writes through until stall is set; stalled
279// writes block until release closes, then fail with failErr when set.
280type stallingSessionStream struct {
281 *sessionTestStream
282 stall atomic.Bool
283 entered chan struct{}
284 release chan struct{}
285 failErr error
286}
287
288func (s *stallingSessionStream) SendClient(message protocol.ClientMessage) error {
289 if s.stall.Load() {
290 select {
291 case s.entered <- struct{}{}:
292 default:
293 }
294 select {
295 case <-s.release:
296 case <-s.closed:
297 return errSessionTestClosed
298 }
299 if s.failErr != nil {
300 return s.failErr
301 }
302 }
303 return s.sessionTestStream.SendClient(message)
304}
305
306// TestAttachmentWorkerStalledLink drives the real worker over a link that
307// stops accepting writes after attach.
308func TestAttachmentWorkerStalledLink(t *testing.T) {
309 linkErr := errors.New("link lost")
310 tests := []struct {
311 name string
312 run func(t *testing.T, h *inputHarness, s *stallingSessionStream)
313 }{
314 {
315 name: "output keeps painting while a write is stalled",
316 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
317 h.send("a")
318 <-s.entered
319 later := sessionTestOutput(2, "\x1b[Hlater")
320 later.Epoch = 2
321 h.stream.deliver(later)
322 require.Eventually(t, func() bool { return strings.Contains(h.term.written(), "later") }, 5*time.Second, time.Millisecond, "a stalled write blocked painting")
323 close(s.release)
324 require.Equal(t, []string{"a"}, h.awaitInputs(t, 1))
325 h.end(t)
326 },
327 },
328 {
329 name: "a read that queues nothing is acked while an earlier write is stalled",
330 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
331 // A queued output Ack stalls behind the link.
332 next := sessionTestOutput(2, "\x1b[Hmore")
333 next.Epoch = 2
334 h.stream.deliver(next)
335 <-s.entered
336 // An open bracketed paste is held by the coalescer and queues
337 // nothing, so there is no write to wait for.
338 h.send("\x1b[200~abc")
339 h.send("def")
340 require.Eventually(t, func() bool {
341 h.pump.mu.Lock()
342 defer h.pump.mu.Unlock()
343 return h.pump.pending == nil && h.pump.delivering == 0 && len(h.reader.chunks) == 0
344 }, 5*time.Second, time.Millisecond, "an unqueued read held the pump behind a stalled write")
345 close(s.release)
346 h.end(t)
347 },
348 },
349 {
350 name: "keys behind a failed write are preserved for the next attempt",
351 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
352 s.failErr = linkErr
353 h.send("a")
354 <-s.entered
355 close(s.release)
356 select {
357 case event := <-h.events:
358 require.Equal(t, AttachmentEventFailed, event.Kind)
359 require.ErrorIs(t, event.Err, linkErr)
360 case <-time.After(5 * time.Second):
361 t.Fatal("the failed write never settled the run")
362 }
363 h.pump.mu.Lock()
364 defer h.pump.mu.Unlock()
365 require.Equal(t, "a", string(h.pump.residual), "the unsent key is kept for the resume")
366 },
367 },
368 {
369 name: "detach on a stalled link is bounded",
370 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
371 h.send("a")
372 <-s.entered
373 h.stream.deliver(protocol.Detached{Reason: protocol.ReasonDetach})
374 select {
375 case event := <-h.events:
376 require.Equal(t, AttachmentEventEnded, event.Kind)
377 case <-time.After(5 * time.Second):
378 t.Fatal("the daemon-confirmed detach never settled")
379 }
380 h.pump.mu.Lock()
381 defer h.pump.mu.Unlock()
382 require.Equal(t, "a", string(h.pump.residual), "a key the link never took is not lost")
383 },
384 },
385 }
386 for _, tt := range tests {
387 t.Run(tt.name, func(t *testing.T) {
388 base := newSessionTestStream()
389 s := &stallingSessionStream{sessionTestStream: base, entered: make(chan struct{}, 1), release: make(chan struct{})}
390 h := startInputHarnessOn(t, nil, sessionTestRequest(true), nil, base, s)
391 require.Eventually(t, func() bool { return strings.Contains(h.term.written(), "ready") }, 5*time.Second, time.Millisecond)
392 // Let the initial Ack and focus report through, then stall.
393 require.Eventually(t, func() bool { return len(base.messages()) >= 2 }, 5*time.Second, time.Millisecond)
394 s.stall.Store(true)
395 tt.run(t, h, s)
396 })
397 }
398}