1package daemon
2
3import (
4 "context"
5 "crypto/rand"
6 "io"
7 "log/slog"
8 "sync"
9 "testing"
10 "time"
11
12 "github.com/stretchr/testify/mock"
13 "github.com/stretchr/testify/require"
14
15 vt "github.com/bnema/vev-vt"
16 renderer "github.com/bnema/vev-vt/ansi"
17 "github.com/bnema/vev/internal/domain"
18 "github.com/bnema/vev/internal/ports"
19 portsmocks "github.com/bnema/vev/internal/ports/mocks"
20 "github.com/bnema/vev/internal/protocol"
21 "github.com/bnema/vev/internal/protocol/wire"
22 "github.com/bnema/vev/internal/usecase/keys"
23)
24
25// --- test doubles -----------------------------------------------------------
26
27// stubClock returns timers whose channel never fires, so a scheduler under it
28// blocks in its debounce loop until the session context is cancelled. Used by
29// every test that is not specifically exercising the debounce.
30type stubClock struct{}
31
32func (stubClock) Now() time.Time { return time.Time{} }
33func (stubClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} }
34
35type stubTimer struct{}
36
37func (stubTimer) C() <-chan time.Time { return nil }
38func (stubTimer) Reset(time.Duration) bool { return false }
39func (stubTimer) Stop() bool { return true }
40
41type signalClock struct {
42 called chan struct{}
43 timers chan *signalTimer
44 once sync.Once
45}
46
47func (c *signalClock) Now() time.Time { return time.Time{} }
48func (c *signalClock) NewTimer(d time.Duration) ports.Timer {
49 if c.called != nil {
50 c.once.Do(func() { close(c.called) })
51 }
52 if c.timers == nil {
53 return stubTimer{}
54 }
55 t := &signalTimer{ch: make(chan time.Time, 1), duration: d}
56 c.timers <- t
57 return t
58}
59
60type signalTimer struct {
61 ch chan time.Time
62 duration time.Duration
63}
64
65func (t *signalTimer) C() <-chan time.Time { return t.ch }
66func (t *signalTimer) Reset(time.Duration) bool { return false }
67func (t *signalTimer) Stop() bool { return true }
68
69func newTestDaemon(t testing.TB, ptys ports.PTYFactory, clk ports.Clock) *Daemon {
70 t.Helper()
71 return newTestDaemonWithCleanup(t, ptys, clk, true)
72}
73
74type terminalEnv struct {
75 TrueColor bool
76}
77
78func createSessionForTest(d *Daemon, name string, ephemeral bool, cwd string, sz domain.Size, _ terminalEnv, env []string, restoredTabNames ...[]string) (*session, error) {
79 d.mu.Lock()
80 defer d.mu.Unlock()
81 return d.createSessionLocked(name, ephemeral, cwd, domain.Geometry{Size: sz}, env, restoredTabNames...)
82}
83
84func newTestDaemonWithCleanup(t testing.TB, ptys ports.PTYFactory, clk ports.Clock, registerCleanup bool) *Daemon {
85 t.Helper()
86 d := New(ptys, clk, slog.New(slog.NewTextHandler(io.Discard, nil)))
87 d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
88 d.hardCtx, d.hardCancel = context.WithCancel(context.Background())
89 if registerCleanup {
90 t.Cleanup(d.serveCancel)
91 t.Cleanup(d.hardCancel)
92 }
93 return d
94}
95
96// newBlockingPTY returns a MockPTY whose Read blocks until release is called,
97// then reports io.EOF (the "child exited" signal).
98func newBlockingPTY(t *testing.T) (*portsmocks.MockPTY, func()) {
99 t.Helper()
100 return newBlockingPTYWithWrites(t, nil)
101}
102
103func newBlockingPTYWithWrites(t *testing.T, writes chan<- []byte) (*portsmocks.MockPTY, func()) {
104 t.Helper()
105 p := portsmocks.NewMockPTY(t)
106 release := make(chan struct{})
107 var once sync.Once
108 unblock := func() { once.Do(func() { close(release) }) }
109 p.EXPECT().Read(mock.Anything).RunAndReturn(func(b []byte) (int, error) {
110 <-release
111 return 0, io.EOF
112 }).Maybe()
113 p.EXPECT().Write(mock.Anything).RunAndReturn(func(b []byte) (int, error) {
114 if writes != nil {
115 cp := append([]byte(nil), b...)
116 writes <- cp
117 }
118 return len(b), nil
119 }).Maybe()
120 p.EXPECT().Resize(mock.Anything).Return(nil).Maybe()
121 // Close unblocks a parked Read, exactly as the real PTY does (its Close
122 // closes the master fd, failing the in-flight read).
123 p.EXPECT().Close().RunAndReturn(func() error { unblock(); return nil }).Maybe()
124 p.EXPECT().Pid().Return(4242).Maybe()
125 return p, unblock
126}
127
128// newBlockingPTYWithFailingWrite returns a MockPTY whose Write always fails
129// with writeErr, simulating a wedged pty (or a dead child) that cannot
130// accept forwarded keystrokes.
131func newBlockingPTYWithFailingWrite(t *testing.T, writeErr error) (*portsmocks.MockPTY, func()) {
132 t.Helper()
133 p := portsmocks.NewMockPTY(t)
134 release := make(chan struct{})
135 var once sync.Once
136 unblock := func() { once.Do(func() { close(release) }) }
137 p.EXPECT().Read(mock.Anything).RunAndReturn(func(b []byte) (int, error) {
138 <-release
139 return 0, io.EOF
140 }).Maybe()
141 p.EXPECT().Write(mock.Anything).Return(0, writeErr).Maybe()
142 p.EXPECT().Resize(mock.Anything).Return(nil).Maybe()
143 p.EXPECT().Close().RunAndReturn(func() error { unblock(); return nil }).Maybe()
144 p.EXPECT().Pid().Return(4242).Maybe()
145 return p, unblock
146}
147
148// quietPTY is an independently closable PTY for background floating prewarms.
149// It intentionally has no testify expectations: tests using the factories below
150// are asserting their regular panes, not incidental asynchronous prewarms.
151type quietPTY struct{ done chan struct{} }
152
153func newQuietPTY() *quietPTY { return &quietPTY{done: make(chan struct{})} }
154func (p *quietPTY) Read([]byte) (int, error) { <-p.done; return 0, io.EOF }
155func (*quietPTY) Write(b []byte) (int, error) { return len(b), nil }
156func (p *quietPTY) Close() error {
157 select {
158 case <-p.done:
159 default:
160 close(p.done)
161 }
162 return nil
163}
164func (*quietPTY) Resize(domain.Geometry) error { return nil }
165func (*quietPTY) Pid() int { return 0 }
166func (*quietPTY) ForegroundPgid() (int, error) { return 0, nil }
167
168// blockingOpenFactory keeps activation tests deterministically in Warming
169// without starting pane reader/scheduler goroutines.
170type blockingOpenFactory struct {
171 release chan struct{}
172 once sync.Once
173}
174
175func newBlockingOpenFactory(t *testing.T, d *Daemon) *blockingOpenFactory {
176 t.Helper()
177 f := &blockingOpenFactory{release: make(chan struct{})}
178 t.Cleanup(func() {
179 f.once.Do(func() { close(f.release) })
180 d.sessWg.Wait()
181 })
182 return f
183}
184
185func (f *blockingOpenFactory) Open(_ context.Context, _ string, _ []string, _ []string, _ string, _ domain.Geometry) (ports.PTY, error) {
186 <-f.release
187 return nil, io.ErrClosedPipe
188}
189
190// newFactory keeps the supplied fixture for normal panes. Floating prewarms
191// have a distinct (smaller) geometry and receive independent quiet PTYs, so
192// they cannot consume test output or add post-test mock calls.
193func newFactory(t *testing.T, p ports.PTY) *portsmocks.MockPTYFactory {
194 t.Helper()
195 f := portsmocks.NewMockPTYFactory(t)
196 var normal domain.Size
197 f.EXPECT().Open(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
198 func(_ context.Context, _ string, _ []string, _ []string, _ string, geometry domain.Geometry) (ports.PTY, error) {
199 if !normal.Valid() {
200 normal = geometry.Size
201 }
202 if geometry.Size == normal {
203 return p, nil
204 }
205 return newQuietPTY(), nil
206 },
207 ).Maybe()
208 return f
209}
210
211func newFactorySeq(t *testing.T, ptys ...ports.PTY) *portsmocks.MockPTYFactory {
212 t.Helper()
213 f := portsmocks.NewMockPTYFactory(t)
214 var mu sync.Mutex
215 var normal domain.Size
216 next := 0
217 f.EXPECT().Open(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
218 func(_ context.Context, _ string, _ []string, _ []string, _ string, geometry domain.Geometry) (ports.PTY, error) {
219 mu.Lock()
220 defer mu.Unlock()
221 if !normal.Valid() {
222 normal = geometry.Size
223 }
224 if geometry.Size != normal {
225 return newQuietPTY(), nil
226 }
227 if next >= len(ptys) {
228 return newQuietPTY(), nil
229 }
230 p := ptys[next]
231 next++
232 return p, nil
233 },
234 ).Maybe()
235 return f
236}
237
238// newConn scripts a MockTransport: Recv yields first then more (in order),
239// then blocks until the connection is released or Closed (returning io.EOF).
240// Every Send is captured on the returned channel.
241func newConn(t *testing.T, first wire.Envelope, more ...wire.Envelope) (*mockServerConnection, chan wire.Envelope, func()) {
242 t.Helper()
243 tr := newMockServerConnection(t)
244 sends := make(chan wire.Envelope, 64)
245 recvCh := make(chan wire.Envelope, 1+len(more))
246 recvCh <- first
247 for _, f := range more {
248 recvCh <- f
249 }
250 done := make(chan struct{})
251 var once sync.Once
252 closeDone := func() { once.Do(func() { close(done) }) }
253
254 tr.EXPECT().Recv().RunAndReturn(func() (wire.Envelope, error) {
255 select {
256 case f := <-recvCh:
257 return f, nil
258 case <-done:
259 return wire.Envelope{}, io.EOF
260 }
261 }).Maybe()
262 tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f wire.Envelope) error {
263 sends <- f
264 return nil
265 }).Maybe()
266 tr.EXPECT().Close().RunAndReturn(func() error { closeDone(); return nil }).Maybe()
267 return tr, sends, closeDone
268}
269
270func mustHello(intent uint8, name string, sz domain.Size) wire.Envelope {
271 return mustClientEnvelope(protocol.Hello{
272 Version: protocol.Version, Intent: intent, Name: name, Size: sz, TermEnv: "xterm-256color",
273 })
274}
275
276func frameInput(data []byte) wire.Envelope {
277 return mustClientEnvelope(protocol.Input{Data: data})
278}
279
280func screenLineText(s *vt.Screen, y int) string {
281 out := make([]rune, s.Columns())
282 for x := range s.Columns() {
283 out[x] = s.Cell(x, y).Rune
284 }
285 return string(out)
286}
287
288// awaitFrame waits for the next envelope decoding to want on ch, failing
289// on timeout. The type name is the semantic message name ("Output",
290// "Welcome", ...).
291func awaitFrame(t *testing.T, ch chan wire.Envelope, want string) wire.Envelope {
292 t.Helper()
293 deadline := time.After(2 * time.Second)
294 for {
295 select {
296 case f := <-ch:
297 if envelopeMessageName(t, f.Payload) == want {
298 return f
299 }
300 case <-deadline:
301 t.Fatalf("timed out waiting for envelope %s", want)
302 }
303 }
304}
305
306func firstSession(d *Daemon) *session {
307 d.mu.Lock()
308 defer d.mu.Unlock()
309 for _, s := range d.sessions {
310 if s != nil {
311 return s
312 }
313 }
314 return nil
315}
316
317func sessionCount(d *Daemon) int {
318 d.mu.Lock()
319 defer d.mu.Unlock()
320 return len(d.sessions)
321}
322
323func listSessions(t *testing.T, d *Daemon) protocol.Sessions {
324 t.Helper()
325 tr, sends, _ := newConn(t, mustClientEnvelope(protocol.List{}))
326 d.handleList(tr)
327 f := awaitFrame(t, sends, "Sessions")
328 sessions, ok := decodeServerMessage(t, f).(protocol.Sessions)
329 require.True(t, ok)
330 return sessions
331}
332
333// welcomeBlockingTransport holds the route-level handshake precisely between
334// attachment publication and Welcome acceptance. It uses the generated
335// transport mock while channels make both sides of the wire deterministic.
336type welcomeBlockingTransport struct {
337 tr *mockServerConnection
338 sends chan wire.Envelope
339 welcomeEntered chan struct{}
340 releaseWelcome chan struct{}
341 recvDone chan struct{}
342 releaseOnce sync.Once
343 closeOnce sync.Once
344}
345
346func newWelcomeBlockingTransport(t *testing.T) *welcomeBlockingTransport {
347 t.Helper()
348 b := &welcomeBlockingTransport{
349 sends: make(chan wire.Envelope, 8),
350 welcomeEntered: make(chan struct{}),
351 releaseWelcome: make(chan struct{}),
352 recvDone: make(chan struct{}),
353 }
354 b.tr = newMockServerConnection(t)
355 b.tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f wire.Envelope) error {
356 b.sends <- f
357 if envelopeMessageName(nil, f.Payload) == "Welcome" {
358 close(b.welcomeEntered)
359 <-b.releaseWelcome
360 }
361 return nil
362 }).Maybe()
363 b.tr.EXPECT().Recv().RunAndReturn(func() (wire.Envelope, error) {
364 <-b.recvDone
365 return wire.Envelope{}, io.EOF
366 }).Maybe()
367 b.tr.EXPECT().Close().Return(nil).Maybe()
368 t.Cleanup(b.finish)
369 return b
370}
371
372func (b *welcomeBlockingTransport) release() {
373 b.releaseOnce.Do(func() { close(b.releaseWelcome) })
374}
375
376func (b *welcomeBlockingTransport) finish() {
377 b.release()
378 b.closeOnce.Do(func() { close(b.recvDone) })
379}
380
381func newCapturingTransport(t testing.TB) (*mockServerConnection, chan wire.Envelope) {
382 t.Helper()
383 tr := newMockServerConnection(t)
384 sends := make(chan wire.Envelope, 64)
385 tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f wire.Envelope) error {
386 sends <- f
387 return nil
388 }).Maybe()
389 tr.EXPECT().Close().Return(nil).Maybe()
390 return tr, sends
391}
392
393func newManualSessionWithPTYs(t testing.TB, ptys ...ports.PTY) (*Daemon, *session, *attachedClient, chan wire.Envelope) {
394 t.Helper()
395 return newManualSessionWithPTYsClock(t, stubClock{}, ptys...)
396}
397
398func newManualSessionWithPTYsClock(t testing.TB, clock ports.Clock, ptys ...ports.PTY) (*Daemon, *session, *attachedClient, chan wire.Envelope) {
399 t.Helper()
400 return newManualSessionWithPTYsClockCleanup(t, clock, true, ptys...)
401}
402
403func newManualSessionWithPTYsCleanup(t testing.TB, registerCleanup bool, ptys ...ports.PTY) (*Daemon, *session, *attachedClient, chan wire.Envelope) {
404 t.Helper()
405 return newManualSessionWithPTYsClockCleanup(t, stubClock{}, registerCleanup, ptys...)
406}
407
408func newTestLifecycle(t testing.TB) domain.SessionLifecycleID {
409 t.Helper()
410 id, err := domain.NewSessionLifecycleID(rand.Reader)
411 require.NoError(t, err)
412 return id
413}
414
415func newManualSessionWithPTYsClockCleanup(t testing.TB, clock ports.Clock, registerCleanup bool, ptys ...ports.PTY) (*Daemon, *session, *attachedClient, chan wire.Envelope) {
416 t.Helper()
417 d := newTestDaemonWithCleanup(t, nil, clock, registerCleanup)
418 tr, sends := newCapturingTransport(t)
419 ac := &attachedClient{tr: tr, output: newOutputStateStream(), size: domain.Size{Cols: 80, Rows: 24}}
420 ac.output.attachment = ac
421 ac.initOverlays()
422 sctx, cancel := context.WithCancel(d.serveCtx)
423 tabs := make([]*tab, 0, len(ptys))
424 for _, p := range ptys {
425 wctx, wcancel := context.WithCancel(sctx)
426 tb := newTab(p, domain.Size{Cols: 80, Rows: 23})
427 tb.ctx, tb.cancel = wctx, wcancel
428 for _, pane := range tb.panes {
429 pane.ctx, pane.cancel = wctx, wcancel
430 }
431 tabs = append(tabs, tb)
432 }
433 lifecycle := newTestLifecycle(t)
434 sess := &session{sessionCore: sessionCore{id: "manual", name: "work", incarnation: lifecycle, attachments: map[*attachedClient]struct{}{ac: {}}}, ctx: sctx, cancel: cancel, tabs: tabs}
435 for _, tb := range tabs {
436 publishTiledPaneOwners(sess, tb)
437 }
438 ac.setSession(sess)
439 // This fixture models an already attached client, with its initial route
440 // position known. Handshake tests exercise the first position publication.
441 if len(tabs) != 0 {
442 ac.output.lastRoutePosition = protocol.RoutePosition{
443 Target: protocol.ExactSessionTarget{LifecycleID: lifecycle, SessionName: sess.name},
444 ActiveTabID: domain.TabStableID(tabs[0].stableID),
445 }
446 }
447 ac.keys = keys.NewRouter(d.clock, daemonKeyHandler{d: d, ac: ac}, nil)
448 d.sessions[sess.id] = sess
449 if registerCleanup {
450 t.Cleanup(cancel)
451 }
452 return d, sess, ac, sends
453}
454
455func newManualTabSession(t *testing.T, n int) (*Daemon, *session, *attachedClient, chan wire.Envelope, []func()) {
456 t.Helper()
457 ptys := make([]ports.PTY, 0, n)
458 releases := make([]func(), 0, n)
459 for range n {
460 p, release := newBlockingPTY(t)
461 ptys = append(ptys, p)
462 releases = append(releases, release)
463 }
464 d, sess, ac, sends := newManualSessionWithPTYs(t, ptys...)
465 return d, sess, ac, sends, releases
466}
467
468func testAttachment(sess *session) *attachedClient {
469 if sess == nil {
470 return nil
471 }
472 attachments := sess.snapshotAttachments()
473 if len(attachments) == 0 {
474 return nil
475 }
476 return attachments[0]
477}
478
479func testAttachmentTabLocked(sess *session) *tab {
480 if sess == nil {
481 return nil
482 }
483 var ac *attachedClient
484 for candidate := range sess.attachments {
485 ac = candidate
486 break
487 }
488 if ac == nil {
489 if len(sess.tabs) == 0 {
490 return nil
491 }
492 return sess.tabs[0]
493 }
494 view := ac.viewSnapshot()
495 for _, tb := range sess.tabs {
496 if domain.TabStableID(tb.stableID) == view.tabID {
497 return tb
498 }
499 }
500 return nil
501}
502
503func testAttachmentTab(sess *session) *tab {
504 if ac := testAttachment(sess); ac != nil {
505 return sess.tabForAttachment(ac)
506 }
507 return sess.firstTab()
508}
509
510func selectTestAttachmentTab(sess *session, idx int) bool {
511 ac := testAttachment(sess)
512 if ac == nil {
513 return idx == 0 && testAttachmentTab(sess) != nil
514 }
515 return sess.switchAttachmentTab(ac, idx)
516}
517
518// selectTestAttachmentTabLocked is used only by fixtures that already hold
519// sess.mu; production navigation always goes through selectAttachmentTab.
520func selectTestAttachmentTabLocked(sess *session, idx int) bool {
521 if sess == nil {
522 return false
523 }
524 var ac *attachedClient
525 for candidate := range sess.attachments {
526 ac = candidate
527 break
528 }
529 if ac == nil {
530 return idx == 0 && len(sess.tabs) > 0
531 }
532 return sess.activateAttachmentViewLocked(ac, idx)
533}
534
535func testAttachmentTabIndexLocked(sess *session) int {
536 if sess == nil {
537 return 0
538 }
539 var ac *attachedClient
540 for candidate := range sess.attachments {
541 ac = candidate
542 break
543 }
544 if ac == nil {
545 return 0
546 }
547 view := ac.viewSnapshot()
548 for i, tb := range sess.tabs {
549 if domain.TabStableID(tb.stableID) == view.tabID {
550 return i
551 }
552 }
553 return 0
554}
555
556func testAttachmentTabIndex(sess *session) int {
557 if sess == nil {
558 return 0
559 }
560 sess.mu.Lock()
561 defer sess.mu.Unlock()
562 return testAttachmentTabIndexLocked(sess)
563}
564
565func testRow(text string) []renderer.Cell {
566 cells := make([]renderer.Cell, 0, len(text))
567 for _, r := range text {
568 cells = append(cells, renderer.Cell{Rune: r, Style: renderer.DefaultStyle()})
569 }
570 return cells
571}
572
573// appendHistoryRow records a row whose logical extent covers it entirely and
574// that ends its logical line — the shape of a line scrolled off a grid it never
575// wrapped out of. Tests that care about wrapped extents build the bound instead.
576func appendHistoryRow(tb testing.TB, history *vt.History, row []renderer.Cell) {
577 tb.Helper()
578 id := max(vt.RowID(1_000_000), history.NextRowID())
579 require.NoError(tb, history.AppendWithID(row, vt.LineBound{End: len(row)}, id))
580}
581
582func TestAppendHistoryRowAllocatesUniqueIDsAfterEviction(t *testing.T) {
583 history := vt.NewHistory(vt.HistoryConfig{MaxRows: 2, ChunkRows: 1})
584 ids := make(map[vt.RowID]struct{})
585 for range 4 {
586 appendHistoryRow(t, history, testRow("row"))
587 view := history.View()
588 id := view.RowID(view.Len() - 1)
589 require.NotContains(t, ids, id)
590 ids[id] = struct{}{}
591 }
592 require.Len(t, ids, 4)
593}
594
595func tabCount(sess *session) int {
596 sess.mu.Lock()
597 defer sess.mu.Unlock()
598 return len(sess.tabs)
599}
600
601// --- handshake --------------------------------------------------------------
602
603func TestRouteRejectsInvalidTerminalSizeBeforeResumeMutation(t *testing.T) {
604 tests := []struct {
605 name string
606 size domain.Size
607 }{
608 {name: "zero columns", size: domain.Size{Cols: 0, Rows: 24}},
609 {name: "zero rows", size: domain.Size{Cols: 80, Rows: 0}},
610 }
611 for _, tt := range tests {
612 t.Run(tt.name, func(t *testing.T) {
613 pty, release := newBlockingPTY(t)
614 defer release()
615 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
616
617 transport := &closeTrackingTransport{}
618 sess, ac, err := d.route(helloResumeCapable(protocol.IntentNew, "work", 0), transport)
619 require.NoError(t, err)
620 token := ac.resumeToken
621 d.clientGone(sess, ac, transport, false)
622
623 d.mu.Lock()
624 parked := d.resume.parked[token]
625 d.mu.Unlock()
626 require.NotNil(t, parked)
627
628 invalid := helloResumeCapable(protocol.IntentResume, "work", token)
629 invalid.Size = tt.size
630 _, _, err = d.route(invalid, &closeTrackingTransport{})
631 var protocolErr *protoErr
632 require.ErrorAs(t, err, &protocolErr)
633 require.Equal(t, protocol.ErrInternal, protocolErr.code)
634
635 d.mu.Lock()
636 require.Same(t, parked, d.resume.parked[token], "invalid resume must not consume the parked attachment")
637 d.mu.Unlock()
638 require.Same(t, transport, ac.transport(), "invalid resume must not bind a replacement transport")
639 })
640 }
641}
642
643func waitGroupDone(wg *sync.WaitGroup) <-chan struct{} {
644 done := make(chan struct{})
645 go func() {
646 wg.Wait()
647 close(done)
648 }()
649 return done
650}
651
652func rowText(row []renderer.Cell) string {
653 runes := make([]rune, len(row))
654 for i, c := range row {
655 if c.Continuation {
656 runes[i] = ' '
657 continue
658 }
659 runes[i] = c.Rune
660 }
661 return string(runes)
662}
663
664// notifyClock stubs debounceInterval so schedulers park, and shortens
665// detachNotifyTimeout to 5ms so a wedged client's Detached send is
666// force-closed quickly. Every other budget, including the handshake deadline,
667// runs at its real length. Timers are matched by duration value, so another
668// budget equal to one of these constants gets the same treatment.
669type notifyClock struct{}
670
671func (notifyClock) Now() time.Time { return time.Now() }
672func (notifyClock) NewTimer(d time.Duration) ports.Timer {
673 switch d {
674 case debounceInterval:
675 return stubTimer{}
676 case detachNotifyTimeout:
677 return realTimer{t: time.NewTimer(5 * time.Millisecond)}
678 default:
679 return realTimer{t: time.NewTimer(d)}
680 }
681}
682
683type realTimer struct{ t *time.Timer }
684
685func (r realTimer) C() <-chan time.Time { return r.t.C }
686func (r realTimer) Reset(d time.Duration) bool { return r.t.Reset(d) }
687func (r realTimer) Stop() bool { return r.t.Stop() }
688
689func TestHandleHelloDefersFreshOutputUntilWelcome(t *testing.T) {
690 pty, releasePTY := newBlockingPTY(t)
691 defer releasePTY()
692 clock := newCoordinatorMockClock(t, 4)
693 d := newTestDaemon(t, newFactory(t, pty), clock.clock)
694 tr := newWelcomeBlockingTransport(t)
695 done := make(chan struct{})
696 go func() {
697 d.handleHelloFrame(tr.tr, mustHello(protocol.IntentNew, "welcome-gate", domain.Size{Cols: 80, Rows: 24}))
698 close(done)
699 }()
700
701 awaitTestCompletion(t, tr.welcomeEntered, "timed out waiting for Welcome send")
702 handshakeTimer := awaitHandshakeTimer(t, clock)
703 welcome := awaitFrame(t, tr.sends, "Welcome")
704 require.Equal(t, "Welcome", envelopeMessageName(t, welcome.Payload))
705 sess := firstSession(d)
706 sess.mu.Lock()
707 ac := sess.snapshotAttachmentsLocked()[0]
708 sess.mu.Unlock()
709 require.NotNil(t, ac, "route must publish the attachment before Welcome returns")
710
711 // Exercise the real producer fan-in after publication. The deadline worker
712 // has completed before asserting the gate, so this is not a timing race.
713 d.invalidateRender(sess, ac, true, "welcome-gate-test")
714 // Attach-time theme setup may have armed an older bulk timer. The fresh
715 // urgent producer replaces it, so advance the newest published deadline.
716 timer := awaitLatestCoordinatorTimer(t, clock)
717 rc := sess.renderCoordinator()
718 rc.mu.Lock()
719 workerDone := rc.normalLane.token.done
720 rc.mu.Unlock()
721 require.NotNil(t, workerDone)
722 timer.ch <- time.Time{}
723 awaitTestCompletion(t, workerDone, "coordinator deadline worker did not complete")
724 requireNoCoordinatorOutputFrame(t, tr.sends)
725
726 tr.release()
727 output := awaitFrame(t, tr.sends, "Output")
728 first := unmarshalTestOutput(t, output.Payload)
729 require.Zero(t, first.Base, "the first post-Welcome frame is full")
730 tr.finish()
731 awaitTestCompletion(t, done, "timed out waiting for Welcome handler completion")
732 awaitTestCompletion(t, handshakeTimer.stopped, "handshake timer was not canceled after publication")
733 requireNoCoordinatorOutputFrame(t, tr.sends)
734}
735
736// TestServeReturnsDespiteWedgedClientOnShutdown: killSession's teardown must
737// not be gated behind the Detached notice. The client transport wedges every
738// Detached send until the transport is closed (mirroring a full kernel
739// send buffer: only Close fails the in-flight write). Serve must still tear
740
741func mustOutputData(t *testing.T, sends chan wire.Envelope) []byte {
742 t.Helper()
743 f := awaitFrame(t, sends, "Output")
744 out := unmarshalTestOutput(t, f.Payload)
745 return out.Data
746}