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