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// stallingSessionStream passes writes through until stall is set; stalled
286// writes block until release closes, then fail with failErr when set.
287type stallingSessionStream struct {
288 *sessionTestStream
289 stall atomic.Bool
290 entered chan struct{}
291 release chan struct{}
292 failErr error
293}
294
295func (s *stallingSessionStream) SendClient(message protocol.ClientMessage) error {
296 if s.stall.Load() {
297 select {
298 case s.entered <- struct{}{}:
299 default:
300 }
301 select {
302 case <-s.release:
303 case <-s.closed:
304 return errSessionTestClosed
305 }
306 if s.failErr != nil {
307 return s.failErr
308 }
309 }
310 return s.sessionTestStream.SendClient(message)
311}
312
313// TestAttachmentWorkerStalledLink drives the real worker over a link that
314// stops accepting writes after attach.
315func TestAttachmentWorkerStalledLink(t *testing.T) {
316 linkErr := errors.New("link lost")
317 tests := []struct {
318 name string
319 run func(t *testing.T, h *inputHarness, s *stallingSessionStream)
320 }{
321 {
322 name: "output keeps painting while a write is stalled",
323 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
324 h.send("a")
325 <-s.entered
326 later := sessionTestOutput(2, "\x1b[Hlater")
327 later.Epoch = 2
328 h.stream.deliver(later)
329 require.Eventually(t, func() bool { return strings.Contains(h.term.written(), "later") }, 5*time.Second, time.Millisecond, "a stalled write blocked painting")
330 close(s.release)
331 require.Equal(t, []string{"a"}, h.awaitInputs(t, 1))
332 h.end(t)
333 },
334 },
335 {
336 name: "a read that queues nothing is acked while an earlier write is stalled",
337 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
338 // A queued output Ack stalls behind the link.
339 next := sessionTestOutput(2, "\x1b[Hmore")
340 next.Epoch = 2
341 h.stream.deliver(next)
342 <-s.entered
343 // An open bracketed paste is held by the coalescer and queues
344 // nothing, so there is no write to wait for.
345 h.send("\x1b[200~abc")
346 h.send("def")
347 require.Eventually(t, func() bool {
348 h.pump.mu.Lock()
349 defer h.pump.mu.Unlock()
350 return h.pump.pending == nil && h.pump.delivering == 0 && len(h.reader.chunks) == 0
351 }, 5*time.Second, time.Millisecond, "an unqueued read held the pump behind a stalled write")
352 close(s.release)
353 h.end(t)
354 },
355 },
356 {
357 name: "keys behind a failed write are preserved for the next attempt",
358 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
359 s.failErr = linkErr
360 h.send("a")
361 <-s.entered
362 close(s.release)
363 select {
364 case event := <-h.events:
365 require.Equal(t, AttachmentEventFailed, event.Kind)
366 require.ErrorIs(t, event.Err, linkErr)
367 case <-time.After(5 * time.Second):
368 t.Fatal("the failed write never settled the run")
369 }
370 h.pump.mu.Lock()
371 defer h.pump.mu.Unlock()
372 require.Equal(t, "a", string(h.pump.residual), "the unsent key is kept for the resume")
373 },
374 },
375 {
376 name: "detach on a stalled link is bounded",
377 run: func(t *testing.T, h *inputHarness, s *stallingSessionStream) {
378 h.send("a")
379 <-s.entered
380 h.stream.deliver(protocol.Detached{Reason: protocol.ReasonDetach})
381 select {
382 case event := <-h.events:
383 require.Equal(t, AttachmentEventEnded, event.Kind)
384 case <-time.After(5 * time.Second):
385 t.Fatal("the daemon-confirmed detach never settled")
386 }
387 h.pump.mu.Lock()
388 defer h.pump.mu.Unlock()
389 require.Equal(t, "a", string(h.pump.residual), "a key the link never took is not lost")
390 },
391 },
392 }
393 for _, tt := range tests {
394 t.Run(tt.name, func(t *testing.T) {
395 base := newSessionTestStream()
396 s := &stallingSessionStream{sessionTestStream: base, entered: make(chan struct{}, 1), release: make(chan struct{})}
397 h := startInputHarnessOn(t, nil, sessionTestRequest(true), nil, base, s)
398 require.Eventually(t, func() bool { return strings.Contains(h.term.written(), "ready") }, 5*time.Second, time.Millisecond)
399 // Let the initial Ack and focus report through, then stall.
400 require.Eventually(t, func() bool { return len(base.messages()) >= 2 }, 5*time.Second, time.Millisecond)
401 s.stall.Store(true)
402 tt.run(t, h, s)
403 })
404 }
405}