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, nil)
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(), nil)
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, nil)
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 lifetime := &terminalInputLifetime{}
272 w := &resumeWatch{lifetime: lifetime}
273 w.hold(bytes.Repeat([]byte("a"), resumeHeldInputLimit-1))
274 w.hold([]byte("bb"))
275 w.hold([]byte("c"))
276 require.Equal(t, resumeHeldInputLimit-1, w.heldLen(), "keys after an overflow must be dropped too")
277
278 // The next resume phase of the same outage starts a new watch; it must
279 // keep dropping so the session never receives input with a gap.
280 next := &resumeWatch{lifetime: lifetime}
281 next.hold([]byte("d"))
282 require.Zero(t, next.heldLen(), "a later phase must not accept keys after an overflow")
283}
284
285// TestResumeWatchOverflowKeepsHeldPrefix overflows one phase and checks the
286// next phase still replays the gap-free prefix the first phase handed back.
287// TestResumeWatchHeldOverflowIsSticky covers dropping keys typed after it.
288func TestResumeWatchOverflowKeepsHeldPrefix(t *testing.T) {
289 reader := &inputTestReader{chunks: make(chan []byte, 2)}
290 pump := newTerminalInputPump(reader)
291 pump.start()
292 t.Cleanup(func() {
293 pump.stop()
294 close(reader.chunks)
295 })
296 lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
297 clock := newSupervisorTestClock()
298
299 first := lifetime.watchResume(clock, nil)
300 reader.chunks <- []byte("ls")
301 require.Eventually(t, func() bool { return first.heldLen() == 2 }, 5*time.Second, time.Millisecond)
302 first.hold(bytes.Repeat([]byte("x"), resumeHeldInputLimit))
303 require.True(t, lifetime.resumeOverflowed.Load())
304 first.release()
305
306 second := lifetime.watchResume(clock, nil)
307 require.Eventually(t, func() bool { return second.heldLen() == 2 }, 5*time.Second, time.Millisecond)
308 second.release()
309
310 pump.mu.Lock()
311 defer pump.mu.Unlock()
312 require.Equal(t, "ls", string(pump.residual))
313}
314
315// stallingSessionStream passes writes through until stall is set; stalled
316// writes block until release closes, then fail with failErr when set.
317type stallingSessionStream struct {
318 *sessionTestStream
319 stall atomic.Bool
320 entered chan struct{}
321 release chan struct{}
322 failErr error
323}
324
325func (s *stallingSessionStream) SendClient(message protocol.ClientMessage) error {
326 if s.stall.Load() {
327 select {
328 case s.entered <- struct{}{}:
329 default:
330 }
331 select {
332 case <-s.release:
333 case <-s.closed:
334 return errSessionTestClosed
335 }
336 if s.failErr != nil {
337 return s.failErr
338 }
339 }
340 return s.sessionTestStream.SendClient(message)
341}
342
343// TestAttachmentWorkerStalledLink drives the real worker over a link that
344// stops accepting writes after attach.
345func TestAttachmentWorkerStalledLink(t *testing.T) {
346 linkErr := errors.New("link lost")
347 tests := []struct {
348 name string
349 run func(t *testing.T, h *inputHarness, s *stallingSessionStream)
350 }{
351 {
352 name: "output keeps painting while a write is stalled",
353 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
354 h.send("a")
355 <-s.entered
356 later := sessionTestOutput(2, "\x1b[Hlater")
357 later.Epoch = 2
358 h.stream.deliver(later)
359 require.Eventually(t, func() bool { return strings.Contains(h.term.written(), "later") }, 5*time.Second, time.Millisecond, "a stalled write blocked painting")
360 close(s.release)
361 require.Equal(t, []string{"a"}, h.awaitInputs(t, 1))
362 h.end(t)
363 },
364 },
365 {
366 name: "a read that queues nothing is acked while an earlier write is stalled",
367 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
368 // A queued output Ack stalls behind the link.
369 next := sessionTestOutput(2, "\x1b[Hmore")
370 next.Epoch = 2
371 h.stream.deliver(next)
372 <-s.entered
373 // An open bracketed paste is held by the coalescer and queues
374 // nothing, so there is no write to wait for.
375 h.send("\x1b[200~abc")
376 h.send("def")
377 require.Eventually(t, func() bool {
378 h.pump.mu.Lock()
379 defer h.pump.mu.Unlock()
380 return h.pump.pending == nil && h.pump.delivering == 0 && len(h.reader.chunks) == 0
381 }, 5*time.Second, time.Millisecond, "an unqueued read held the pump behind a stalled write")
382 close(s.release)
383 h.end(t)
384 },
385 },
386 {
387 name: "keys behind a failed write are preserved for the next attempt",
388 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
389 s.failErr = linkErr
390 h.send("a")
391 <-s.entered
392 close(s.release)
393 select {
394 case event := <-h.events:
395 require.Equal(t, AttachmentEventFailed, event.Kind)
396 require.ErrorIs(t, event.Err, linkErr)
397 case <-time.After(5 * time.Second):
398 t.Fatal("the failed write never settled the run")
399 }
400 h.pump.mu.Lock()
401 defer h.pump.mu.Unlock()
402 require.Equal(t, "a", string(h.pump.residual), "the unsent key is kept for the resume")
403 },
404 },
405 {
406 name: "detach on a stalled link is bounded",
407 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
408 h.send("a")
409 <-s.entered
410 h.stream.deliver(protocol.Detached{Reason: protocol.ReasonDetach})
411 select {
412 case event := <-h.events:
413 require.Equal(t, AttachmentEventEnded, event.Kind)
414 case <-time.After(5 * time.Second):
415 t.Fatal("the daemon-confirmed detach never settled")
416 }
417 h.pump.mu.Lock()
418 defer h.pump.mu.Unlock()
419 require.Equal(t, "a", string(h.pump.residual), "a key the link never took is not lost")
420 },
421 },
422 }
423 for _, tt := range tests {
424 t.Run(tt.name, func(t *testing.T) {
425 base := newSessionTestStream()
426 s := &stallingSessionStream{sessionTestStream: base, entered: make(chan struct{}, 1), release: make(chan struct{})}
427 h := startInputHarnessOn(t, nil, sessionTestRequest(true), nil, base, s)
428 require.Eventually(t, func() bool { return strings.Contains(h.term.written(), "ready") }, 5*time.Second, time.Millisecond)
429 // Let the initial Ack and focus report through, then stall.
430 require.Eventually(t, func() bool { return len(base.messages()) >= 2 }, 5*time.Second, time.Millisecond)
431 s.stall.Store(true)
432 tt.run(t, h, s)
433 })
434 }
435}