1package client
2
3import (
4 "context"
5 "io"
6 "log/slog"
7 "sync"
8 "sync/atomic"
9 "testing"
10 "time"
11
12 "github.com/stretchr/testify/require"
13
14 "github.com/bnema/vev/internal/domain"
15 "github.com/bnema/vev/internal/ports"
16)
17
18type lifecycleReader struct {
19 chunks chan []byte
20 started chan struct{}
21
22 mu sync.Mutex
23 active int
24 max int
25}
26
27func newLifecycleReader() *lifecycleReader {
28 return &lifecycleReader{chunks: make(chan []byte), started: make(chan struct{}, 4)}
29}
30
31func (r *lifecycleReader) Read(p []byte) (int, error) {
32 r.mu.Lock()
33 r.active++
34 if r.active > r.max {
35 r.max = r.active
36 }
37 r.mu.Unlock()
38 r.started <- struct{}{}
39 data, ok := <-r.chunks
40 r.mu.Lock()
41 r.active--
42 r.mu.Unlock()
43 if !ok {
44 return 0, io.EOF
45 }
46 return copy(p, data), nil
47}
48
49func (r *lifecycleReader) maxActive() int {
50 r.mu.Lock()
51 defer r.mu.Unlock()
52 return r.max
53}
54
55type lateStopReader struct {
56 started chan struct{}
57 release chan terminalReadResult
58 reads atomic.Int32
59}
60
61func newLateStopReader() *lateStopReader {
62 return &lateStopReader{
63 started: make(chan struct{}, 2),
64 release: make(chan terminalReadResult, 1),
65 }
66}
67
68func (r *lateStopReader) Read(p []byte) (int, error) {
69 r.reads.Add(1)
70 r.started <- struct{}{}
71 result := <-r.release
72 return copy(p, result.data), result.err
73}
74
75func requireSignal(t *testing.T, ch <-chan struct{}, message string) {
76 t.Helper()
77 select {
78 case <-ch:
79 case <-time.After(time.Second):
80 t.Fatal(message)
81 }
82}
83
84func requireFrame(t *testing.T, ch <-chan ports.Frame, message string) ports.Frame {
85 t.Helper()
86 select {
87 case frame := <-ch:
88 return frame
89 case <-time.After(time.Second):
90 t.Fatal(message)
91 return ports.Frame{}
92 }
93}
94
95func requireTimer(t *testing.T, ch <-chan *attachPaletteTimer, message string) *attachPaletteTimer {
96 t.Helper()
97 select {
98 case timer := <-ch:
99 return timer
100 case <-time.After(time.Second):
101 t.Fatal(message)
102 return nil
103 }
104}
105
106func TestPaletteSlotRejectsOutOfRangeIntBeforeNarrowing(t *testing.T) {
107 for _, tc := range []struct {
108 slot int
109 want uint8
110 ok bool
111 }{
112 {slot: -1},
113 {slot: 0, want: 0, ok: true},
114 {slot: 15, want: 15, ok: true},
115 {slot: 16},
116 {slot: 256},
117 } {
118 got, ok := paletteSlot(tc.slot)
119 require.Equal(t, tc.ok, ok, "slot %d", tc.slot)
120 if ok {
121 require.Equal(t, tc.want, got, "slot %d", tc.slot)
122 }
123 }
124}
125
126func TestTerminalInputPumpStopDropsLateReadWithoutStartingAnother(t *testing.T) {
127 reader := newLateStopReader()
128 input := newTerminalInputPump(reader)
129 input.start()
130 <-reader.started // The sole lifecycle read is blocked in the caller-owned reader.
131
132 input.stop()
133 reader.release <- terminalReadResult{data: []byte("late")}
134
135 // The late read must be discarded rather than published. Both possible
136 // outcomes are channel-synchronized: a correct pump exits; the historical
137 // race published and immediately issued a second Read.
138 select {
139 case <-input.exited:
140 case <-reader.started:
141 require.Fail(t, "pump issued another Read after stop")
142 }
143 require.Equal(t, int32(1), reader.reads.Load())
144
145 consumer := input.claim()
146 _, ok := input.take(context.Background(), consumer)
147 require.False(t, ok, "late read must not be available after stop")
148 input.revoke(consumer)
149 input.stop() // stop is deliberately idempotent.
150}
151
152func TestRunnerReusesBlockedTerminalInputPumpAcrossRuns(t *testing.T) {
153 reader := newLateStopReader()
154 runner := &Runner{term: &attachPaletteTerminalHarness{
155 in: reader, out: newAttachPaletteWriter(), resize: make(chan domain.Size),
156 }}
157
158 first := runner.terminalInput()
159 <-reader.started
160 first.suspend()
161 second := runner.terminalInput()
162 require.Same(t, first, second, "a later Run must reuse a blocked caller-owned reader")
163 require.Equal(t, int32(1), reader.reads.Load())
164
165 first.stop()
166 reader.release <- terminalReadResult{}
167 <-first.exited
168}
169
170func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T) {
171 input := newTerminalInputPump(nil)
172 // Both cancellation and a terminal read are ready before the old scanner
173 // tries to dequeue. It must surrender the read to its replacement.
174 want := []byte("first\x00second\xff")
175 input.enqueue(terminalReadResult{data: append([]byte(nil), want...)}, 1)
176
177 var activeGeneration atomic.Uint64
178 cancelledCtx, cancelCancelled := context.WithCancel(context.Background())
179 cancelCancelled()
180 cancelledDone := make(chan struct{})
181 go func() {
182 (&stdinPump{
183 ctx: cancelledCtx, cancel: cancelCancelled, input: input,
184 out: make(chan ports.Frame, 1), clock: newAttachPaletteClock(),
185 logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
186 activeGeneration: &activeGeneration,
187 }).run()
188 close(cancelledDone)
189 }()
190 <-cancelledDone
191
192 replacementCtx, cancelReplacement := context.WithCancel(context.Background())
193 defer cancelReplacement()
194 out := make(chan ports.Frame, len(want))
195 replacementDone := make(chan struct{})
196 go func() {
197 (&stdinPump{
198 ctx: replacementCtx, cancel: cancelReplacement, input: input,
199 out: out, clock: newAttachPaletteClock(),
200 logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
201 activeGeneration: &activeGeneration,
202 }).run()
203 close(replacementDone)
204 }()
205
206 var got []byte
207 for len(got) < len(want) {
208 frame := <-out
209 input, err := ports.UnmarshalInput(frame.Payload)
210 require.NoError(t, err)
211 got = append(got, input.Data...)
212 }
213 require.Equal(t, want, got, "handoff must preserve every queued byte in order")
214
215 cancelReplacement()
216 <-replacementDone
217}
218
219func TestTerminalInputPumpCancellationAfterTakeRequeuesRawRead(t *testing.T) {
220 input := newTerminalInputPump(nil)
221 want := []byte("first\x00second\xff")
222 input.enqueue(terminalReadResult{data: append([]byte(nil), want...)}, 1)
223
224 ctx, cancel := context.WithCancel(context.Background())
225 consumer := input.claim()
226 taken := make(chan struct{})
227 release := make(chan struct{})
228 done := make(chan struct{})
229 go func() {
230 (&stdinPump{
231 ctx: ctx, cancel: cancel, input: input, consumer: consumer,
232 out: make(chan ports.Frame, 1), clock: newAttachPaletteClock(),
233 logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
234 afterInputTake: func() {
235 close(taken)
236 <-release
237 },
238 }).run()
239 close(done)
240 }()
241
242 <-taken
243 cancel()
244 close(release)
245 <-done
246 input.revoke(consumer)
247
248 replacement := input.claim()
249 got, ok := input.take(context.Background(), replacement)
250 require.True(t, ok, "cancelled delivery must leave the read for the replacement scanner")
251 require.Equal(t, want, got.data)
252 input.ack(replacement)
253 input.revoke(replacement)
254}
255
256func TestTerminalInputPumpCancellationPreservesStandaloneEscapeForReplacement(t *testing.T) {
257 input := newTerminalInputPump(nil)
258 input.enqueue(terminalReadResult{data: []byte("\x1b")}, 1)
259 var activeGeneration atomic.Uint64
260
261 oldCtx, cancelOld := context.WithCancel(context.Background())
262 oldClock := newAttachPaletteClock()
263 oldDone := make(chan struct{})
264 go func() {
265 (&stdinPump{
266 ctx: oldCtx, cancel: cancelOld, input: input, out: make(chan ports.Frame, 1),
267 clock: oldClock, logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
268 activeGeneration: &activeGeneration,
269 }).run()
270 close(oldDone)
271 }()
272 // The ambiguity timer proves the ESC was scanned and its raw read was
273 // committed; only the marker scanner's undecided suffix remains.
274 <-oldClock.timers
275 cancelOld()
276 <-oldDone
277
278 newCtx, cancelNew := context.WithCancel(context.Background())
279 defer cancelNew()
280 newClock := newAttachPaletteClock()
281 out := make(chan ports.Frame, 1)
282 newDone := make(chan struct{})
283 go func() {
284 (&stdinPump{
285 ctx: newCtx, cancel: cancelNew, input: input, out: out,
286 clock: newClock, logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
287 activeGeneration: &activeGeneration,
288 }).run()
289 close(newDone)
290 }()
291 (<-newClock.timers).fire() // marker ambiguity deadline
292 (<-newClock.timers).fire() // paste coalescer's lone-ESC deadline
293 frame := <-out
294 got, err := ports.UnmarshalInput(frame.Payload)
295 require.NoError(t, err)
296 require.Equal(t, []byte("\x1b"), got.Data)
297
298 cancelNew()
299 <-newDone
300}
301
302func TestTerminalInputPumpCancellationAfterInputBeforeOSCPreservesOnlyUndeliveredSuffix(t *testing.T) {
303 input := newTerminalInputPump(nil)
304 input.enqueue(terminalReadResult{data: []byte("x\x1b]10;rgb:ffff/0000/0000\x07y")}, 1)
305 var activeGeneration atomic.Uint64
306 ctx, cancel := context.WithCancel(context.Background())
307 delivered := make(chan struct{})
308 done := make(chan struct{})
309 out := make(chan ports.Frame, 1)
310 go func() {
311 (&stdinPump{
312 ctx: ctx, cancel: cancel, input: input, out: out, clock: newAttachPaletteClock(),
313 logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
314 activeGeneration: &activeGeneration,
315 afterInputDelivered: func() {
316 cancel()
317 close(delivered)
318 },
319 }).run()
320 close(done)
321 }()
322 frame := requireFrame(t, out, "timed out waiting for delivered input prefix")
323 got, err := ports.UnmarshalInput(frame.Payload)
324 require.NoError(t, err)
325 require.Equal(t, []byte("x"), got.Data)
326 requireSignal(t, delivered, "timed out waiting for cancellation after input delivery")
327 requireSignal(t, done, "timed out waiting for cancelled stdin pump")
328
329 replacementCtx, cancelReplacement := context.WithCancel(context.Background())
330 replacementOut := make(chan ports.Frame, 1)
331 replacementDone := make(chan struct{})
332 go func() {
333 (&stdinPump{
334 ctx: replacementCtx, cancel: cancelReplacement, input: input,
335 out: replacementOut, clock: newAttachPaletteClock(),
336 logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
337 activeGeneration: &activeGeneration,
338 }).run()
339 close(replacementDone)
340 }()
341 frame = requireFrame(t, replacementOut, "timed out waiting for undelivered input suffix")
342 got, err = ports.UnmarshalInput(frame.Payload)
343 require.NoError(t, err)
344 require.Equal(t, []byte("y"), got.Data, "replacement must receive only the undelivered ordinary suffix")
345
346 cancelReplacement()
347 requireSignal(t, replacementDone, "timed out waiting for replacement stdin pump shutdown")
348}
349
350func TestTerminalInputPumpCancellationPreservesCoalescerHeldSuffix(t *testing.T) {
351 input := newTerminalInputPump(nil)
352 held := ports.BracketedPasteOpenMarker[:3]
353 raw := append([]byte("x\x1b]10;rgb:ffff/0000/0000\x07"), held...)
354 input.enqueue(terminalReadResult{data: raw}, 1)
355 var activeGeneration atomic.Uint64
356 ctx, cancel := context.WithCancel(context.Background())
357 out := make(chan ports.Frame, 1)
358 done := make(chan struct{})
359 go func() {
360 (&stdinPump{
361 ctx: ctx, cancel: cancel, input: input, out: out, clock: newAttachPaletteClock(),
362 logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
363 activeGeneration: &activeGeneration,
364 afterInputDelivered: cancel,
365 }).run()
366 close(done)
367 }()
368 frame := requireFrame(t, out, "timed out waiting for delivered input prefix")
369 got, err := ports.UnmarshalInput(frame.Payload)
370 require.NoError(t, err)
371 require.Equal(t, []byte("x"), got.Data)
372 requireSignal(t, done, "timed out waiting for cancelled stdin pump")
373
374 replacementCtx, cancelReplacement := context.WithCancel(context.Background())
375 replacementClock := newAttachPaletteClock()
376 replacementOut := make(chan ports.Frame, 1)
377 replacementDone := make(chan struct{})
378 go func() {
379 (&stdinPump{
380 ctx: replacementCtx, cancel: cancelReplacement, input: input,
381 out: replacementOut, clock: replacementClock,
382 logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
383 activeGeneration: &activeGeneration,
384 }).run()
385 close(replacementDone)
386 }()
387 requireTimer(t, replacementClock.timers, "timed out waiting for paste-prefix ambiguity timer").fire()
388 frame = requireFrame(t, replacementOut, "timed out waiting for coalescer-held suffix")
389 got, err = ports.UnmarshalInput(frame.Payload)
390 require.NoError(t, err)
391 require.Equal(t, held, got.Data, "replacement must receive the coalescer-held suffix exactly once")
392
393 cancelReplacement()
394 requireSignal(t, replacementDone, "timed out waiting for replacement stdin pump shutdown")
395}
396
397func TestTerminalInputPumpRejectsReadFromPriorActivation(t *testing.T) {
398 reader := newLateStopReader()
399 input := newTerminalInputPump(reader)
400 input.start()
401 requireSignal(t, reader.started, "timed out waiting for initial activation Read")
402
403 input.suspend()
404 input.resume()
405 reader.release <- terminalReadResult{data: []byte("stale")}
406
407 // The old read must be discarded after resume, then the reader may begin a
408 // new Read for the current activation.
409 requireSignal(t, reader.started, "timed out waiting for current activation Read")
410 consumer := input.claim()
411 _, ok := input.take(context.Background(), consumer)
412 require.False(t, ok, "a Read authorized before suspend/resume must not reach the new activation")
413 input.revoke(consumer)
414
415 input.stop()
416 reader.release <- terminalReadResult{}
417 requireSignal(t, input.exited, "timed out waiting for terminal input pump shutdown")
418}
419
420func TestTerminalInputPumpCancellationAfterDeliveredCallbackDoesNotReplay(t *testing.T) {
421 input := newTerminalInputPump(nil)
422 input.enqueue(terminalReadResult{data: []byte("x")}, 1)
423 var activeGeneration atomic.Uint64
424 ctx, cancel := context.WithCancel(context.Background())
425 delivered := make(chan struct{})
426 done := make(chan struct{})
427 out := make(chan ports.Frame, 1)
428 go func() {
429 (&stdinPump{
430 ctx: ctx, cancel: cancel, input: input, out: out, clock: newAttachPaletteClock(),
431 logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
432 activeGeneration: &activeGeneration,
433 afterInputDelivered: func() {
434 cancel()
435 close(delivered)
436 },
437 }).run()
438 close(done)
439 }()
440 frame := <-out
441 got, err := ports.UnmarshalInput(frame.Payload)
442 require.NoError(t, err)
443 require.Equal(t, []byte("x"), got.Data)
444 <-delivered
445 <-done
446
447 consumer := input.claim()
448 _, ok := input.take(context.Background(), consumer)
449 require.False(t, ok, "a callback already delivered to the sender must not replay after reconnect")
450 input.revoke(consumer)
451}
452
453func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {
454 reader := newLifecycleReader()
455 input := newTerminalInputPump(reader)
456 input.start()
457 t.Cleanup(input.stop)
458 <-reader.started // The lifecycle pump owns the sole blocking Read.
459
460 var activeGeneration atomic.Uint64
461 firstCtx, cancelFirst := context.WithCancel(context.Background())
462 firstDone := make(chan struct{})
463 go func() {
464 (&stdinPump{
465 ctx: firstCtx, cancel: cancelFirst, input: input,
466 out: make(chan ports.Frame), clock: newAttachPaletteClock(),
467 logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
468 activeGeneration: &activeGeneration,
469 }).run()
470 close(firstDone)
471 }()
472 cancelFirst()
473 <-firstDone
474
475 secondCtx, cancelSecond := context.WithCancel(context.Background())
476 defer cancelSecond()
477 out := make(chan ports.Frame, 1)
478 secondDone := make(chan struct{})
479 go func() {
480 (&stdinPump{
481 ctx: secondCtx, cancel: cancelSecond, input: input,
482 out: out, clock: newAttachPaletteClock(),
483 logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
484 activeGeneration: &activeGeneration,
485 }).run()
486 close(secondDone)
487 }()
488
489 reader.chunks <- []byte("x")
490 frame := <-out
491 got, err := ports.UnmarshalInput(frame.Payload)
492 require.NoError(t, err)
493 require.Equal(t, []byte("x"), got.Data)
494 require.Equal(t, 1, reader.maxActive(), "cancelled attempts must not leave competing reads")
495
496 cancelSecond()
497 <-secondDone
498}