test(daemon): exercise welcome output gate
- commit
- 2653724c1af8f566c84ee35b5d19bb398d67376d
- author
- brice · · 2026-07-11 21:37 UTC
- parent
- 35dbd6b
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +193 -10
M internal/usecase/daemon/daemon_test.go +96 -0
| @@ -286,6 +286,47 @@ func listSessions(t *testing.T, d *Daemon) ports.Sessions { | |||
| 286 | 286 | return sessions | |
| 287 | 287 | } | |
| 288 | 288 | ||
| 289 | + | // welcomeBlockingTransport holds the route-level handshake precisely between | |
| 290 | + | // attachment publication and Welcome acceptance. It uses the generated | |
| 291 | + | // transport mock while channels make both sides of the wire deterministic. | |
| 292 | + | type welcomeBlockingTransport struct { | |
| 293 | + | tr *portsmocks.MockTransport | |
| 294 | + | sends chan ports.Frame | |
| 295 | + | welcomeEntered chan struct{} | |
| 296 | + | releaseWelcome chan struct{} | |
| 297 | + | recvDone chan struct{} | |
| 298 | + | closeOnce sync.Once | |
| 299 | + | } | |
| 300 | + | ||
| 301 | + | func newWelcomeBlockingTransport(t *testing.T) *welcomeBlockingTransport { | |
| 302 | + | t.Helper() | |
| 303 | + | b := &welcomeBlockingTransport{ | |
| 304 | + | sends: make(chan ports.Frame, 8), | |
| 305 | + | welcomeEntered: make(chan struct{}), | |
| 306 | + | releaseWelcome: make(chan struct{}), | |
| 307 | + | recvDone: make(chan struct{}), | |
| 308 | + | } | |
| 309 | + | b.tr = portsmocks.NewMockTransport(t) | |
| 310 | + | b.tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error { | |
| 311 | + | b.sends <- f | |
| 312 | + | if f.Type == ports.MsgWelcome { | |
| 313 | + | close(b.welcomeEntered) | |
| 314 | + | <-b.releaseWelcome | |
| 315 | + | } | |
| 316 | + | return nil | |
| 317 | + | }).Maybe() | |
| 318 | + | b.tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) { | |
| 319 | + | <-b.recvDone | |
| 320 | + | return ports.Frame{}, io.EOF | |
| 321 | + | }).Maybe() | |
| 322 | + | b.tr.EXPECT().Close().Return(nil).Maybe() | |
| 323 | + | return b | |
| 324 | + | } | |
| 325 | + | ||
| 326 | + | func (b *welcomeBlockingTransport) finish() { | |
| 327 | + | b.closeOnce.Do(func() { close(b.recvDone) }) | |
| 328 | + | } | |
| 329 | + | ||
| 289 | 330 | func newCapturingTransport(t testing.TB) (*portsmocks.MockTransport, chan ports.Frame) { | |
| 290 | 331 | t.Helper() | |
| 291 | 332 | tr := portsmocks.NewMockTransport(t) | |
| @@ -395,6 +436,61 @@ func (r realTimer) C() <-chan time.Time { return r.t.C } | |||
| 395 | 436 | func (r realTimer) Reset(d time.Duration) bool { return r.t.Reset(d) } | |
| 396 | 437 | func (r realTimer) Stop() bool { return r.t.Stop() } | |
| 397 | 438 | ||
| 439 | + | func TestHandleHelloDefersFreshOutputUntilWelcome(t *testing.T) { | |
| 440 | + | pty, releasePTY := newBlockingPTY(t) | |
| 441 | + | defer releasePTY() | |
| 442 | + | clock := newCoordinatorMockClock(t, 4) | |
| 443 | + | d := newTestDaemon(t, newFactory(t, pty), clock.clock) | |
| 444 | + | tr := newWelcomeBlockingTransport(t) | |
| 445 | + | done := make(chan struct{}) | |
| 446 | + | go func() { | |
| 447 | + | d.handleHello(tr.tr, mustHello(ports.IntentNew, "welcome-gate", domain.Size{Cols: 80, Rows: 24})) | |
| 448 | + | close(done) | |
| 449 | + | }() | |
| 450 | + | ||
| 451 | + | <-tr.welcomeEntered | |
| 452 | + | welcome := <-tr.sends | |
| 453 | + | require.Equal(t, ports.MsgWelcome, welcome.Type) | |
| 454 | + | sess := firstSession(d) | |
| 455 | + | sess.mu.Lock() | |
| 456 | + | ac := sess.client | |
| 457 | + | sess.mu.Unlock() | |
| 458 | + | require.NotNil(t, ac, "route must publish the attachment before Welcome returns") | |
| 459 | + | ||
| 460 | + | // Exercise the real producer fan-in after publication. The deadline worker | |
| 461 | + | // has completed before asserting the gate, so this is not a timing race. | |
| 462 | + | d.invalidateRender(sess, ac, true, "welcome-gate-test") | |
| 463 | + | timer := awaitCoordinatorScheduledTimer(t, clock) | |
| 464 | + | // attach-time theme setup may have armed an older bulk timer. The fresh | |
| 465 | + | // urgent producer replaces it, so advance the newest published deadline. | |
| 466 | + | for { | |
| 467 | + | select { | |
| 468 | + | case timer = <-clock.timers: | |
| 469 | + | default: | |
| 470 | + | goto currentDeadline | |
| 471 | + | } | |
| 472 | + | } | |
| 473 | + | ||
| 474 | + | currentDeadline: | |
| 475 | + | rc := sess.renderCoordinator() | |
| 476 | + | rc.mu.Lock() | |
| 477 | + | workerDone := rc.normalWorkerDone | |
| 478 | + | rc.mu.Unlock() | |
| 479 | + | require.NotNil(t, workerDone) | |
| 480 | + | timer.ch <- time.Time{} | |
| 481 | + | awaitHandshakeWorker(t, workerDone) | |
| 482 | + | requireNoCoordinatorOutputFrame(t, tr.sends) | |
| 483 | + | ||
| 484 | + | close(tr.releaseWelcome) | |
| 485 | + | output := awaitFrame(t, tr.sends, ports.MsgOutput) | |
| 486 | + | first, err := ports.UnmarshalOutput(output.Payload) | |
| 487 | + | require.NoError(t, err) | |
| 488 | + | require.Zero(t, first.BaseStateNum, "the first post-Welcome frame is full") | |
| 489 | + | tr.finish() | |
| 490 | + | <-done | |
| 491 | + | requireNoCoordinatorOutputFrame(t, tr.sends) | |
| 492 | + | } | |
| 493 | + | ||
| 398 | 494 | // TestServeReturnsDespiteWedgedClientOnShutdown: killSession's teardown must | |
| 399 | 495 | // not be gated behind the Detached notice. The client transport wedges every | |
| 400 | 496 | // MsgDetached send until the transport is closed (mirroring a full kernel | |
| @@ -286,6 +286,47 @@ func listSessions(t *testing.T, d *Daemon) ports.Sessions { | |||||
| 286 | return sessions | 286 | return sessions | ||
| 287 | } | 287 | } | ||
| 288 | 288 | ||||
| 289 | + | // welcomeBlockingTransport holds the route-level handshake precisely between | |||
| 290 | + | // attachment publication and Welcome acceptance. It uses the generated | |||
| 291 | + | // transport mock while channels make both sides of the wire deterministic. | |||
| 292 | + | type welcomeBlockingTransport struct { | |||
| 293 | + | tr *portsmocks.MockTransport | |||
| 294 | + | sends chan ports.Frame | |||
| 295 | + | welcomeEntered chan struct{} | |||
| 296 | + | releaseWelcome chan struct{} | |||
| 297 | + | recvDone chan struct{} | |||
| 298 | + | closeOnce sync.Once | |||
| 299 | + | } | |||
| 300 | + | ||||
| 301 | + | func newWelcomeBlockingTransport(t *testing.T) *welcomeBlockingTransport { | |||
| 302 | + | t.Helper() | |||
| 303 | + | b := &welcomeBlockingTransport{ | |||
| 304 | + | sends: make(chan ports.Frame, 8), | |||
| 305 | + | welcomeEntered: make(chan struct{}), | |||
| 306 | + | releaseWelcome: make(chan struct{}), | |||
| 307 | + | recvDone: make(chan struct{}), | |||
| 308 | + | } | |||
| 309 | + | b.tr = portsmocks.NewMockTransport(t) | |||
| 310 | + | b.tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error { | |||
| 311 | + | b.sends <- f | |||
| 312 | + | if f.Type == ports.MsgWelcome { | |||
| 313 | + | close(b.welcomeEntered) | |||
| 314 | + | <-b.releaseWelcome | |||
| 315 | + | } | |||
| 316 | + | return nil | |||
| 317 | + | }).Maybe() | |||
| 318 | + | b.tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) { | |||
| 319 | + | <-b.recvDone | |||
| 320 | + | return ports.Frame{}, io.EOF | |||
| 321 | + | }).Maybe() | |||
| 322 | + | b.tr.EXPECT().Close().Return(nil).Maybe() | |||
| 323 | + | return b | |||
| 324 | + | } | |||
| 325 | + | ||||
| 326 | + | func (b *welcomeBlockingTransport) finish() { | |||
| 327 | + | b.closeOnce.Do(func() { close(b.recvDone) }) | |||
| 328 | + | } | |||
| 329 | + | ||||
| 289 | func newCapturingTransport(t testing.TB) (*portsmocks.MockTransport, chan ports.Frame) { | 330 | func newCapturingTransport(t testing.TB) (*portsmocks.MockTransport, chan ports.Frame) { | ||
| 290 | t.Helper() | 331 | t.Helper() | ||
| 291 | tr := portsmocks.NewMockTransport(t) | 332 | tr := portsmocks.NewMockTransport(t) | ||
| @@ -395,6 +436,61 @@ func (r realTimer) C() <-chan time.Time { return r.t.C } | |||||
| 395 | func (r realTimer) Reset(d time.Duration) bool { return r.t.Reset(d) } | 436 | func (r realTimer) Reset(d time.Duration) bool { return r.t.Reset(d) } | ||
| 396 | func (r realTimer) Stop() bool { return r.t.Stop() } | 437 | func (r realTimer) Stop() bool { return r.t.Stop() } | ||
| 397 | 438 | ||||
| 439 | + | func TestHandleHelloDefersFreshOutputUntilWelcome(t *testing.T) { | |||
| 440 | + | pty, releasePTY := newBlockingPTY(t) | |||
| 441 | + | defer releasePTY() | |||
| 442 | + | clock := newCoordinatorMockClock(t, 4) | |||
| 443 | + | d := newTestDaemon(t, newFactory(t, pty), clock.clock) | |||
| 444 | + | tr := newWelcomeBlockingTransport(t) | |||
| 445 | + | done := make(chan struct{}) | |||
| 446 | + | go func() { | |||
| 447 | + | d.handleHello(tr.tr, mustHello(ports.IntentNew, "welcome-gate", domain.Size{Cols: 80, Rows: 24})) | |||
| 448 | + | close(done) | |||
| 449 | + | }() | |||
| 450 | + | ||||
| 451 | + | <-tr.welcomeEntered | |||
| 452 | + | welcome := <-tr.sends | |||
| 453 | + | require.Equal(t, ports.MsgWelcome, welcome.Type) | |||
| 454 | + | sess := firstSession(d) | |||
| 455 | + | sess.mu.Lock() | |||
| 456 | + | ac := sess.client | |||
| 457 | + | sess.mu.Unlock() | |||
| 458 | + | require.NotNil(t, ac, "route must publish the attachment before Welcome returns") | |||
| 459 | + | ||||
| 460 | + | // Exercise the real producer fan-in after publication. The deadline worker | |||
| 461 | + | // has completed before asserting the gate, so this is not a timing race. | |||
| 462 | + | d.invalidateRender(sess, ac, true, "welcome-gate-test") | |||
| 463 | + | timer := awaitCoordinatorScheduledTimer(t, clock) | |||
| 464 | + | // attach-time theme setup may have armed an older bulk timer. The fresh | |||
| 465 | + | // urgent producer replaces it, so advance the newest published deadline. | |||
| 466 | + | for { | |||
| 467 | + | select { | |||
| 468 | + | case timer = <-clock.timers: | |||
| 469 | + | default: | |||
| 470 | + | goto currentDeadline | |||
| 471 | + | } | |||
| 472 | + | } | |||
| 473 | + | ||||
| 474 | + | currentDeadline: | |||
| 475 | + | rc := sess.renderCoordinator() | |||
| 476 | + | rc.mu.Lock() | |||
| 477 | + | workerDone := rc.normalWorkerDone | |||
| 478 | + | rc.mu.Unlock() | |||
| 479 | + | require.NotNil(t, workerDone) | |||
| 480 | + | timer.ch <- time.Time{} | |||
| 481 | + | awaitHandshakeWorker(t, workerDone) | |||
| 482 | + | requireNoCoordinatorOutputFrame(t, tr.sends) | |||
| 483 | + | ||||
| 484 | + | close(tr.releaseWelcome) | |||
| 485 | + | output := awaitFrame(t, tr.sends, ports.MsgOutput) | |||
| 486 | + | first, err := ports.UnmarshalOutput(output.Payload) | |||
| 487 | + | require.NoError(t, err) | |||
| 488 | + | require.Zero(t, first.BaseStateNum, "the first post-Welcome frame is full") | |||
| 489 | + | tr.finish() | |||
| 490 | + | <-done | |||
| 491 | + | requireNoCoordinatorOutputFrame(t, tr.sends) | |||
| 492 | + | } | |||
| 493 | + | ||||
| 398 | // TestServeReturnsDespiteWedgedClientOnShutdown: killSession's teardown must | 494 | // TestServeReturnsDespiteWedgedClientOnShutdown: killSession's teardown must | ||
| 399 | // not be gated behind the Detached notice. The client transport wedges every | 495 | // not be gated behind the Detached notice. The client transport wedges every | ||
| 400 | // MsgDetached send until the transport is closed (mirroring a full kernel | 496 | // MsgDetached send until the transport is closed (mirroring a full kernel | ||
M internal/usecase/daemon/render_coordinator.go +6 -1
| @@ -767,9 +767,14 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | |||
| 767 | 767 | c.mu.Unlock() | |
| 768 | 768 | return | |
| 769 | 769 | } | |
| 770 | + | // Welcome holds attachment.sendMu while its transport Send is in flight. | |
| 771 | + | // Do not probe ACK capacity through that lock before this incarnation is | |
| 772 | + | // ready: a deadline that expires during the handshake must complete and | |
| 773 | + | // leave work pending, rather than wedging behind Welcome. | |
| 774 | + | handshakePending := c.attachment != nil && !c.attachmentReady | |
| 770 | 775 | ackReady := c.opts.ackReady | |
| 771 | 776 | c.mu.Unlock() | |
| 772 | - | ready := ackReady == nil || ackReady() | |
| 777 | + | ready := !handshakePending && (ackReady == nil || ackReady()) | |
| 773 | 778 | ||
| 774 | 779 | // The readiness callback can detach, replace, or publish a sync batch. | |
| 775 | 780 | // syncGateOpenLocked retries predicate evaluation on every registry change | |
| @@ -767,9 +767,14 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | |||||
| 767 | c.mu.Unlock() | 767 | c.mu.Unlock() | ||
| 768 | return | 768 | return | ||
| 769 | } | 769 | } | ||
| 770 | + | // Welcome holds attachment.sendMu while its transport Send is in flight. | |||
| 771 | + | // Do not probe ACK capacity through that lock before this incarnation is | |||
| 772 | + | // ready: a deadline that expires during the handshake must complete and | |||
| 773 | + | // leave work pending, rather than wedging behind Welcome. | |||
| 774 | + | handshakePending := c.attachment != nil && !c.attachmentReady | |||
| 770 | ackReady := c.opts.ackReady | 775 | ackReady := c.opts.ackReady | ||
| 771 | c.mu.Unlock() | 776 | c.mu.Unlock() | ||
| 772 | - | ready := ackReady == nil || ackReady() | 777 | + | ready := !handshakePending && (ackReady == nil || ackReady()) |
| 773 | 778 | ||||
| 774 | // The readiness callback can detach, replace, or publish a sync batch. | 779 | // The readiness callback can detach, replace, or publish a sync batch. | ||
| 775 | // syncGateOpenLocked retries predicate evaluation on every registry change | 780 | // syncGateOpenLocked retries predicate evaluation on every registry change | ||
M internal/usecase/daemon/render_coordinator_handshake_test.go +34 -9
| @@ -1,24 +1,45 @@ | |||
| 1 | 1 | package daemon | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | + | "runtime" | |
| 4 | 5 | "testing" | |
| 5 | 6 | "time" | |
| 6 | 7 | ||
| 7 | 8 | "github.com/stretchr/testify/require" | |
| 8 | 9 | ) | |
| 9 | 10 | ||
| 11 | + | // awaitHandshakeWorker observes the worker completion channel rather than a | |
| 12 | + | // scheduler-dependent no-wake window. The generated mock timer's channel is | |
| 13 | + | // the only event that advances it. | |
| 14 | + | func awaitHandshakeWorker(t *testing.T, done <-chan struct{}) { | |
| 15 | + | t.Helper() | |
| 16 | + | for range 1_000_000 { | |
| 17 | + | select { | |
| 18 | + | case <-done: | |
| 19 | + | return | |
| 20 | + | default: | |
| 21 | + | runtime.Gosched() | |
| 22 | + | } | |
| 23 | + | } | |
| 24 | + | t.Fatal("coordinator deadline worker did not complete") | |
| 25 | + | } | |
| 26 | + | ||
| 10 | 27 | // Fresh producer state may arrive after route has published an attachment but | |
| 11 | 28 | // before the transport has accepted Welcome. It must remain pending until the | |
| 12 | 29 | // exact attachment incarnation completes that handshake. | |
| 13 | 30 | func TestRenderCoordinatorFreshWakeWaitsForWelcome(t *testing.T) { | |
| 14 | 31 | h := newCoordinatorHarness(t) | |
| 15 | 32 | ac := &attachedClient{} | |
| 16 | - | h.rc.attach(ac) | |
| 33 | + | h.rc.attachWithReadiness(ac, false) | |
| 17 | 34 | ||
| 18 | 35 | h.rc.invalidate(renderInvalidation{class: invalidateOutput, reset: true, producer: "pty"}) | |
| 19 | - | timers := h.armedTimers(t) | |
| 20 | - | require.Len(t, timers, 1) | |
| 21 | - | timers[0].ch <- time.Time{} | |
| 36 | + | timer := awaitCoordinatorScheduledTimer(t, h.clk) | |
| 37 | + | h.rc.mu.Lock() | |
| 38 | + | workerDone := h.rc.normalWorkerDone | |
| 39 | + | h.rc.mu.Unlock() | |
| 40 | + | require.NotNil(t, workerDone) | |
| 41 | + | timer.ch <- time.Time{} | |
| 42 | + | requireWorkerExit(t, workerDone) | |
| 22 | 43 | requireNoWake(t, h.wakes) | |
| 23 | 44 | ||
| 24 | 45 | require.True(t, h.rc.markAttachmentReady(ac)) | |
| @@ -36,15 +57,19 @@ func TestRenderCoordinatorFreshWakeWaitsForWelcome(t *testing.T) { | |||
| 36 | 57 | func TestRenderCoordinatorParkedResumeRequiresNewWelcome(t *testing.T) { | |
| 37 | 58 | h := newCoordinatorHarness(t) | |
| 38 | 59 | ac := &attachedClient{} | |
| 39 | - | h.rc.attach(ac) | |
| 60 | + | h.rc.attachWithReadiness(ac, false) | |
| 40 | 61 | require.True(t, h.rc.markAttachmentReady(ac)) | |
| 41 | 62 | h.rc.notePark(ac) | |
| 42 | - | h.rc.attach(ac) | |
| 63 | + | h.rc.attachWithReadiness(ac, false) | |
| 43 | 64 | ||
| 44 | 65 | h.rc.invalidate(renderInvalidation{class: invalidateUrgent, reset: true, producer: "session"}) | |
| 45 | - | timers := h.armedTimers(t) | |
| 46 | - | require.Len(t, timers, 1) | |
| 47 | - | timers[0].ch <- time.Time{} | |
| 66 | + | timer := awaitCoordinatorScheduledTimer(t, h.clk) | |
| 67 | + | h.rc.mu.Lock() | |
| 68 | + | workerDone := h.rc.normalWorkerDone | |
| 69 | + | h.rc.mu.Unlock() | |
| 70 | + | require.NotNil(t, workerDone) | |
| 71 | + | timer.ch <- time.Time{} | |
| 72 | + | requireWorkerExit(t, workerDone) | |
| 48 | 73 | requireNoWake(t, h.wakes) | |
| 49 | 74 | ||
| 50 | 75 | require.True(t, h.rc.markAttachmentReady(ac)) | |
| @@ -1,24 +1,45 @@ | |||||
| 1 | package daemon | 1 | package daemon | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | + | "runtime" | |||
| 4 | "testing" | 5 | "testing" | ||
| 5 | "time" | 6 | "time" | ||
| 6 | 7 | ||||
| 7 | "github.com/stretchr/testify/require" | 8 | "github.com/stretchr/testify/require" | ||
| 8 | ) | 9 | ) | ||
| 9 | 10 | ||||
| 11 | + | // awaitHandshakeWorker observes the worker completion channel rather than a | |||
| 12 | + | // scheduler-dependent no-wake window. The generated mock timer's channel is | |||
| 13 | + | // the only event that advances it. | |||
| 14 | + | func awaitHandshakeWorker(t *testing.T, done <-chan struct{}) { | |||
| 15 | + | t.Helper() | |||
| 16 | + | for range 1_000_000 { | |||
| 17 | + | select { | |||
| 18 | + | case <-done: | |||
| 19 | + | return | |||
| 20 | + | default: | |||
| 21 | + | runtime.Gosched() | |||
| 22 | + | } | |||
| 23 | + | } | |||
| 24 | + | t.Fatal("coordinator deadline worker did not complete") | |||
| 25 | + | } | |||
| 26 | + | ||||
| 10 | // Fresh producer state may arrive after route has published an attachment but | 27 | // Fresh producer state may arrive after route has published an attachment but | ||
| 11 | // before the transport has accepted Welcome. It must remain pending until the | 28 | // before the transport has accepted Welcome. It must remain pending until the | ||
| 12 | // exact attachment incarnation completes that handshake. | 29 | // exact attachment incarnation completes that handshake. | ||
| 13 | func TestRenderCoordinatorFreshWakeWaitsForWelcome(t *testing.T) { | 30 | func TestRenderCoordinatorFreshWakeWaitsForWelcome(t *testing.T) { | ||
| 14 | h := newCoordinatorHarness(t) | 31 | h := newCoordinatorHarness(t) | ||
| 15 | ac := &attachedClient{} | 32 | ac := &attachedClient{} | ||
| 16 | - | h.rc.attach(ac) | 33 | + | h.rc.attachWithReadiness(ac, false) |
| 17 | 34 | ||||
| 18 | h.rc.invalidate(renderInvalidation{class: invalidateOutput, reset: true, producer: "pty"}) | 35 | h.rc.invalidate(renderInvalidation{class: invalidateOutput, reset: true, producer: "pty"}) | ||
| 19 | - | timers := h.armedTimers(t) | 36 | + | timer := awaitCoordinatorScheduledTimer(t, h.clk) |
| 20 | - | require.Len(t, timers, 1) | 37 | + | h.rc.mu.Lock() |
| 21 | - | timers[0].ch <- time.Time{} | 38 | + | workerDone := h.rc.normalWorkerDone |
| 39 | + | h.rc.mu.Unlock() | |||
| 40 | + | require.NotNil(t, workerDone) | |||
| 41 | + | timer.ch <- time.Time{} | |||
| 42 | + | requireWorkerExit(t, workerDone) | |||
| 22 | requireNoWake(t, h.wakes) | 43 | requireNoWake(t, h.wakes) | ||
| 23 | 44 | ||||
| 24 | require.True(t, h.rc.markAttachmentReady(ac)) | 45 | require.True(t, h.rc.markAttachmentReady(ac)) | ||
| @@ -36,15 +57,19 @@ func TestRenderCoordinatorFreshWakeWaitsForWelcome(t *testing.T) { | |||||
| 36 | func TestRenderCoordinatorParkedResumeRequiresNewWelcome(t *testing.T) { | 57 | func TestRenderCoordinatorParkedResumeRequiresNewWelcome(t *testing.T) { | ||
| 37 | h := newCoordinatorHarness(t) | 58 | h := newCoordinatorHarness(t) | ||
| 38 | ac := &attachedClient{} | 59 | ac := &attachedClient{} | ||
| 39 | - | h.rc.attach(ac) | 60 | + | h.rc.attachWithReadiness(ac, false) |
| 40 | require.True(t, h.rc.markAttachmentReady(ac)) | 61 | require.True(t, h.rc.markAttachmentReady(ac)) | ||
| 41 | h.rc.notePark(ac) | 62 | h.rc.notePark(ac) | ||
| 42 | - | h.rc.attach(ac) | 63 | + | h.rc.attachWithReadiness(ac, false) |
| 43 | 64 | ||||
| 44 | h.rc.invalidate(renderInvalidation{class: invalidateUrgent, reset: true, producer: "session"}) | 65 | h.rc.invalidate(renderInvalidation{class: invalidateUrgent, reset: true, producer: "session"}) | ||
| 45 | - | timers := h.armedTimers(t) | 66 | + | timer := awaitCoordinatorScheduledTimer(t, h.clk) |
| 46 | - | require.Len(t, timers, 1) | 67 | + | h.rc.mu.Lock() |
| 47 | - | timers[0].ch <- time.Time{} | 68 | + | workerDone := h.rc.normalWorkerDone |
| 69 | + | h.rc.mu.Unlock() | |||
| 70 | + | require.NotNil(t, workerDone) | |||
| 71 | + | timer.ch <- time.Time{} | |||
| 72 | + | requireWorkerExit(t, workerDone) | |||
| 48 | requireNoWake(t, h.wakes) | 73 | requireNoWake(t, h.wakes) | ||
| 49 | 74 | ||||
| 50 | require.True(t, h.rc.markAttachmentReady(ac)) | 75 | require.True(t, h.rc.markAttachmentReady(ac)) | ||
M internal/usecase/daemon/resume_test.go +57 -0
| @@ -77,6 +77,63 @@ func TestNamedLinkLossParks(t *testing.T) { | |||
| 77 | 77 | require.True(t, parked, "named resume-capable link loss is parked") | |
| 78 | 78 | } | |
| 79 | 79 | ||
| 80 | + | func TestHandleHelloResumeDefersFreshOutputUntilWelcome(t *testing.T) { | |
| 81 | + | pty, releasePTY := newBlockingPTY(t) | |
| 82 | + | defer releasePTY() | |
| 83 | + | clock := newCoordinatorMockClock(t, 4) | |
| 84 | + | d := newTestDaemon(t, newFactorySeq(t, pty), clock.clock) | |
| 85 | + | oldTransport := &closeTrackingTransport{} | |
| 86 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "resume-welcome-gate", 0), oldTransport) | |
| 87 | + | require.NoError(t, err) | |
| 88 | + | token := ac.resumeToken | |
| 89 | + | d.clientGone(sess, ac, oldTransport, false) | |
| 90 | + | ||
| 91 | + | tr := newWelcomeBlockingTransport(t) | |
| 92 | + | done := make(chan struct{}) | |
| 93 | + | resumeHello := helloResumeCapable(ports.IntentResume, sess.name, token) | |
| 94 | + | go func() { | |
| 95 | + | d.handleHello(tr.tr, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(resumeHello)}) | |
| 96 | + | close(done) | |
| 97 | + | }() | |
| 98 | + | ||
| 99 | + | <-tr.welcomeEntered | |
| 100 | + | welcome := <-tr.sends | |
| 101 | + | require.Equal(t, ports.MsgWelcome, welcome.Type) | |
| 102 | + | sess.mu.Lock() | |
| 103 | + | resumed := sess.client | |
| 104 | + | sess.mu.Unlock() | |
| 105 | + | require.Same(t, ac, resumed) | |
| 106 | + | ||
| 107 | + | d.invalidateRender(sess, resumed, true, "resume-welcome-gate-test") | |
| 108 | + | timer := awaitCoordinatorScheduledTimer(t, clock) | |
| 109 | + | for { | |
| 110 | + | select { | |
| 111 | + | case timer = <-clock.timers: | |
| 112 | + | default: | |
| 113 | + | goto currentDeadline | |
| 114 | + | } | |
| 115 | + | } | |
| 116 | + | ||
| 117 | + | currentDeadline: | |
| 118 | + | rc := sess.renderCoordinator() | |
| 119 | + | rc.mu.Lock() | |
| 120 | + | workerDone := rc.normalWorkerDone | |
| 121 | + | rc.mu.Unlock() | |
| 122 | + | require.NotNil(t, workerDone) | |
| 123 | + | timer.ch <- time.Time{} | |
| 124 | + | awaitHandshakeWorker(t, workerDone) | |
| 125 | + | requireNoCoordinatorOutputFrame(t, tr.sends) | |
| 126 | + | ||
| 127 | + | close(tr.releaseWelcome) | |
| 128 | + | output := awaitFrame(t, tr.sends, ports.MsgOutput) | |
| 129 | + | first, err := ports.UnmarshalOutput(output.Payload) | |
| 130 | + | require.NoError(t, err) | |
| 131 | + | require.Zero(t, first.BaseStateNum) | |
| 132 | + | tr.finish() | |
| 133 | + | <-done | |
| 134 | + | requireNoCoordinatorOutputFrame(t, tr.sends) | |
| 135 | + | } | |
| 136 | + | ||
| 80 | 137 | func TestEphemeralLinkLossParksAndResumes(t *testing.T) { | |
| 81 | 138 | pty, release := newBlockingPTY(t) | |
| 82 | 139 | defer release() | |
| @@ -77,6 +77,63 @@ func TestNamedLinkLossParks(t *testing.T) { | |||||
| 77 | require.True(t, parked, "named resume-capable link loss is parked") | 77 | require.True(t, parked, "named resume-capable link loss is parked") | ||
| 78 | } | 78 | } | ||
| 79 | 79 | ||||
| 80 | + | func TestHandleHelloResumeDefersFreshOutputUntilWelcome(t *testing.T) { | |||
| 81 | + | pty, releasePTY := newBlockingPTY(t) | |||
| 82 | + | defer releasePTY() | |||
| 83 | + | clock := newCoordinatorMockClock(t, 4) | |||
| 84 | + | d := newTestDaemon(t, newFactorySeq(t, pty), clock.clock) | |||
| 85 | + | oldTransport := &closeTrackingTransport{} | |||
| 86 | + | sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "resume-welcome-gate", 0), oldTransport) | |||
| 87 | + | require.NoError(t, err) | |||
| 88 | + | token := ac.resumeToken | |||
| 89 | + | d.clientGone(sess, ac, oldTransport, false) | |||
| 90 | + | ||||
| 91 | + | tr := newWelcomeBlockingTransport(t) | |||
| 92 | + | done := make(chan struct{}) | |||
| 93 | + | resumeHello := helloResumeCapable(ports.IntentResume, sess.name, token) | |||
| 94 | + | go func() { | |||
| 95 | + | d.handleHello(tr.tr, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(resumeHello)}) | |||
| 96 | + | close(done) | |||
| 97 | + | }() | |||
| 98 | + | ||||
| 99 | + | <-tr.welcomeEntered | |||
| 100 | + | welcome := <-tr.sends | |||
| 101 | + | require.Equal(t, ports.MsgWelcome, welcome.Type) | |||
| 102 | + | sess.mu.Lock() | |||
| 103 | + | resumed := sess.client | |||
| 104 | + | sess.mu.Unlock() | |||
| 105 | + | require.Same(t, ac, resumed) | |||
| 106 | + | ||||
| 107 | + | d.invalidateRender(sess, resumed, true, "resume-welcome-gate-test") | |||
| 108 | + | timer := awaitCoordinatorScheduledTimer(t, clock) | |||
| 109 | + | for { | |||
| 110 | + | select { | |||
| 111 | + | case timer = <-clock.timers: | |||
| 112 | + | default: | |||
| 113 | + | goto currentDeadline | |||
| 114 | + | } | |||
| 115 | + | } | |||
| 116 | + | ||||
| 117 | + | currentDeadline: | |||
| 118 | + | rc := sess.renderCoordinator() | |||
| 119 | + | rc.mu.Lock() | |||
| 120 | + | workerDone := rc.normalWorkerDone | |||
| 121 | + | rc.mu.Unlock() | |||
| 122 | + | require.NotNil(t, workerDone) | |||
| 123 | + | timer.ch <- time.Time{} | |||
| 124 | + | awaitHandshakeWorker(t, workerDone) | |||
| 125 | + | requireNoCoordinatorOutputFrame(t, tr.sends) | |||
| 126 | + | ||||
| 127 | + | close(tr.releaseWelcome) | |||
| 128 | + | output := awaitFrame(t, tr.sends, ports.MsgOutput) | |||
| 129 | + | first, err := ports.UnmarshalOutput(output.Payload) | |||
| 130 | + | require.NoError(t, err) | |||
| 131 | + | require.Zero(t, first.BaseStateNum) | |||
| 132 | + | tr.finish() | |||
| 133 | + | <-done | |||
| 134 | + | requireNoCoordinatorOutputFrame(t, tr.sends) | |||
| 135 | + | } | |||
| 136 | + | ||||
| 80 | func TestEphemeralLinkLossParksAndResumes(t *testing.T) { | 137 | func TestEphemeralLinkLossParksAndResumes(t *testing.T) { | ||
| 81 | pty, release := newBlockingPTY(t) | 138 | pty, release := newBlockingPTY(t) | ||
| 82 | defer release() | 139 | defer release() | ||