fix(render): preserve watchdog flush semantics
- commit
- c842af759411bf98cd0fbc0d1576e8da58286012
- author
- brice · · 2026-07-14 04:33 UTC
- parent
- a7dcf8b
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +30 -9
M internal/usecase/daemon/output_state_sync_test.go +4 -2
| @@ -370,6 +370,7 @@ func TestResizeBurstFlushesOnlyLatestGeometry(t *testing.T) { | |||
| 370 | 370 | case <-time.After(time.Second): | |
| 371 | 371 | t.Fatal("first resize did not schedule a bounded paint") | |
| 372 | 372 | } | |
| 373 | + | firstDone := captureResizeCallbackDone(t, sess.renderCoordinator()) | |
| 373 | 374 | d.resize(sess, ac, domain.Size{Cols: 120, Rows: 24}) | |
| 374 | 375 | var latest *signalTimer | |
| 375 | 376 | select { | |
| @@ -378,11 +379,12 @@ func TestResizeBurstFlushesOnlyLatestGeometry(t *testing.T) { | |||
| 378 | 379 | t.Fatal("latest resize did not replace the bounded paint") | |
| 379 | 380 | } | |
| 380 | 381 | ||
| 381 | - | done := captureResizeCallbackDone(t, sess.renderCoordinator()) | |
| 382 | + | latestDone := captureResizeCallbackDone(t, sess.renderCoordinator()) | |
| 382 | 383 | first.ch <- time.Now() | |
| 384 | + | awaitTestCompletion(t, firstDone, "obsolete resize callback did not complete") | |
| 383 | 385 | requireNoOutputFrame(t, sends) | |
| 384 | 386 | latest.ch <- time.Now() | |
| 385 | - | awaitTestCompletion(t, done, "latest resize callback did not complete") | |
| 387 | + | awaitTestCompletion(t, latestDone, "latest resize callback did not complete") | |
| 386 | 388 | awaitFrame(t, sends, ports.MsgOutput) | |
| 387 | 389 | require.Equal(t, domain.Size{Cols: 120, Rows: 24}, ac.size) | |
| 388 | 390 | require.Equal(t, 120, sess.activeTab().focusedPane().screen.Frame.Width) | |
| @@ -370,6 +370,7 @@ func TestResizeBurstFlushesOnlyLatestGeometry(t *testing.T) { | |||||
| 370 | case <-time.After(time.Second): | 370 | case <-time.After(time.Second): | ||
| 371 | t.Fatal("first resize did not schedule a bounded paint") | 371 | t.Fatal("first resize did not schedule a bounded paint") | ||
| 372 | } | 372 | } | ||
| 373 | + | firstDone := captureResizeCallbackDone(t, sess.renderCoordinator()) | |||
| 373 | d.resize(sess, ac, domain.Size{Cols: 120, Rows: 24}) | 374 | d.resize(sess, ac, domain.Size{Cols: 120, Rows: 24}) | ||
| 374 | var latest *signalTimer | 375 | var latest *signalTimer | ||
| 375 | select { | 376 | select { | ||
| @@ -378,11 +379,12 @@ func TestResizeBurstFlushesOnlyLatestGeometry(t *testing.T) { | |||||
| 378 | t.Fatal("latest resize did not replace the bounded paint") | 379 | t.Fatal("latest resize did not replace the bounded paint") | ||
| 379 | } | 380 | } | ||
| 380 | 381 | ||||
| 381 | - | done := captureResizeCallbackDone(t, sess.renderCoordinator()) | 382 | + | latestDone := captureResizeCallbackDone(t, sess.renderCoordinator()) |
| 382 | first.ch <- time.Now() | 383 | first.ch <- time.Now() | ||
| 384 | + | awaitTestCompletion(t, firstDone, "obsolete resize callback did not complete") | |||
| 383 | requireNoOutputFrame(t, sends) | 385 | requireNoOutputFrame(t, sends) | ||
| 384 | latest.ch <- time.Now() | 386 | latest.ch <- time.Now() | ||
| 385 | - | awaitTestCompletion(t, done, "latest resize callback did not complete") | 387 | + | awaitTestCompletion(t, latestDone, "latest resize callback did not complete") |
| 386 | awaitFrame(t, sends, ports.MsgOutput) | 388 | awaitFrame(t, sends, ports.MsgOutput) | ||
| 387 | require.Equal(t, domain.Size{Cols: 120, Rows: 24}, ac.size) | 389 | require.Equal(t, domain.Size{Cols: 120, Rows: 24}, ac.size) | ||
| 388 | require.Equal(t, 120, sess.activeTab().focusedPane().screen.Frame.Width) | 390 | require.Equal(t, 120, sess.activeTab().focusedPane().screen.Frame.Width) | ||
M internal/usecase/daemon/render_coordinator.go +3 -3
| @@ -215,14 +215,14 @@ func (c *renderCoordinator) invalidateForLeaseAtResizeEpoch(source *attachedClie | |||
| 215 | 215 | } | |
| 216 | 216 | c.mu.Unlock() | |
| 217 | 217 | if !valid { | |
| 218 | - | stopDetachedTimer(&timerToken{timer: timer}) | |
| 218 | + | timer.Stop() | |
| 219 | 219 | if onInvalidate != nil { | |
| 220 | 220 | onInvalidate(inv) | |
| 221 | 221 | } | |
| 222 | 222 | return true | |
| 223 | 223 | } | |
| 224 | 224 | if timerC == nil { | |
| 225 | - | stopDetachedTimer(&timerToken{timer: timer}) | |
| 225 | + | timer.Stop() | |
| 226 | 226 | c.fire(gen, false, true) | |
| 227 | 227 | if onInvalidate != nil { | |
| 228 | 228 | onInvalidate(inv) | |
| @@ -707,7 +707,7 @@ func (d *Daemon) invalidateRenderNow(sess *session, ac *attachedClient, reset bo | |||
| 707 | 707 | ||
| 708 | 708 | func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | |
| 709 | 709 | c.mu.Lock() | |
| 710 | - | if c.normalLane.generation != gen { | |
| 710 | + | if !watchdog && c.normalLane.generation != gen { | |
| 711 | 711 | c.mu.Unlock() | |
| 712 | 712 | return | |
| 713 | 713 | } | |
| @@ -215,14 +215,14 @@ func (c *renderCoordinator) invalidateForLeaseAtResizeEpoch(source *attachedClie | |||||
| 215 | } | 215 | } | ||
| 216 | c.mu.Unlock() | 216 | c.mu.Unlock() | ||
| 217 | if !valid { | 217 | if !valid { | ||
| 218 | - | stopDetachedTimer(&timerToken{timer: timer}) | 218 | + | timer.Stop() |
| 219 | if onInvalidate != nil { | 219 | if onInvalidate != nil { | ||
| 220 | onInvalidate(inv) | 220 | onInvalidate(inv) | ||
| 221 | } | 221 | } | ||
| 222 | return true | 222 | return true | ||
| 223 | } | 223 | } | ||
| 224 | if timerC == nil { | 224 | if timerC == nil { | ||
| 225 | - | stopDetachedTimer(&timerToken{timer: timer}) | 225 | + | timer.Stop() |
| 226 | c.fire(gen, false, true) | 226 | c.fire(gen, false, true) | ||
| 227 | if onInvalidate != nil { | 227 | if onInvalidate != nil { | ||
| 228 | onInvalidate(inv) | 228 | onInvalidate(inv) | ||
| @@ -707,7 +707,7 @@ func (d *Daemon) invalidateRenderNow(sess *session, ac *attachedClient, reset bo | |||||
| 707 | 707 | ||||
| 708 | func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | 708 | func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | ||
| 709 | c.mu.Lock() | 709 | c.mu.Lock() | ||
| 710 | - | if c.normalLane.generation != gen { | 710 | + | if !watchdog && c.normalLane.generation != gen { |
| 711 | c.mu.Unlock() | 711 | c.mu.Unlock() | ||
| 712 | return | 712 | return | ||
| 713 | } | 713 | } | ||
M internal/usecase/daemon/render_coordinator_test.go +19 -0
| @@ -712,6 +712,25 @@ func TestRenderCoordinatorSynchronizedOutput(t *testing.T) { | |||
| 712 | 712 | requireNoWake(t, h.wakes) | |
| 713 | 713 | }) | |
| 714 | 714 | ||
| 715 | + | t.Run("watchdog flushes after the snapshotted generation is replaced", func(t *testing.T) { | |
| 716 | + | h := newCoordinatorHarness(t) | |
| 717 | + | h.rc.invalidate(renderInvalidation{class: invalidateOutput}) | |
| 718 | + | h.rc.mu.Lock() | |
| 719 | + | staleGeneration := h.rc.normalLane.generation | |
| 720 | + | h.rc.mu.Unlock() | |
| 721 | + | ||
| 722 | + | // Simulate a new urgent publication after fireCurrent(true) snapshots its | |
| 723 | + | // generation but before fire validates it. | |
| 724 | + | h.rc.invalidate(renderInvalidation{class: invalidateUrgent}) | |
| 725 | + | h.rc.fire(staleGeneration, true, true) | |
| 726 | + | ||
| 727 | + | w := awaitWake(t, h.wakes) | |
| 728 | + | require.True(t, w.watchdog) | |
| 729 | + | require.True(t, w.urgent) | |
| 730 | + | require.Equal(t, 2, w.coalesced) | |
| 731 | + | requireNoWake(t, h.wakes) | |
| 732 | + | }) | |
| 733 | + | ||
| 715 | 734 | t.Run("stale watchdog generation cannot wake a completed batch", func(t *testing.T) { | |
| 716 | 735 | h := newCoordinatorHarness(t) | |
| 717 | 736 | h.syncActive.Store(true) | |
| @@ -712,6 +712,25 @@ func TestRenderCoordinatorSynchronizedOutput(t *testing.T) { | |||||
| 712 | requireNoWake(t, h.wakes) | 712 | requireNoWake(t, h.wakes) | ||
| 713 | }) | 713 | }) | ||
| 714 | 714 | ||||
| 715 | + | t.Run("watchdog flushes after the snapshotted generation is replaced", func(t *testing.T) { | |||
| 716 | + | h := newCoordinatorHarness(t) | |||
| 717 | + | h.rc.invalidate(renderInvalidation{class: invalidateOutput}) | |||
| 718 | + | h.rc.mu.Lock() | |||
| 719 | + | staleGeneration := h.rc.normalLane.generation | |||
| 720 | + | h.rc.mu.Unlock() | |||
| 721 | + | ||||
| 722 | + | // Simulate a new urgent publication after fireCurrent(true) snapshots its | |||
| 723 | + | // generation but before fire validates it. | |||
| 724 | + | h.rc.invalidate(renderInvalidation{class: invalidateUrgent}) | |||
| 725 | + | h.rc.fire(staleGeneration, true, true) | |||
| 726 | + | ||||
| 727 | + | w := awaitWake(t, h.wakes) | |||
| 728 | + | require.True(t, w.watchdog) | |||
| 729 | + | require.True(t, w.urgent) | |||
| 730 | + | require.Equal(t, 2, w.coalesced) | |||
| 731 | + | requireNoWake(t, h.wakes) | |||
| 732 | + | }) | |||
| 733 | + | ||||
| 715 | t.Run("stale watchdog generation cannot wake a completed batch", func(t *testing.T) { | 734 | t.Run("stale watchdog generation cannot wake a completed batch", func(t *testing.T) { | ||
| 716 | h := newCoordinatorHarness(t) | 735 | h := newCoordinatorHarness(t) | ||
| 717 | h.syncActive.Store(true) | 736 | h.syncActive.Store(true) | ||
M internal/usecase/daemon/render_resize_deadline.go +4 -4
| @@ -42,7 +42,7 @@ func (c *renderCoordinator) scheduleResizeForLease(size domain.Size, source *att | |||
| 42 | 42 | timer := clock.NewTimer(minOutputRenderDeadline) | |
| 43 | 43 | timerC := timer.C() | |
| 44 | 44 | if timerC == nil { | |
| 45 | - | stopDetachedTimer(&timerToken{timer: timer}) | |
| 45 | + | timer.Stop() | |
| 46 | 46 | if c.resizeCurrentForLease(epoch, source, lease, false) { | |
| 47 | 47 | run(epoch) | |
| 48 | 48 | } | |
| @@ -70,7 +70,7 @@ func (c *renderCoordinator) scheduleResizeForLease(size domain.Size, source *att | |||
| 70 | 70 | } | |
| 71 | 71 | c.mu.Unlock() | |
| 72 | 72 | if !valid { | |
| 73 | - | stopDetachedTimer(&timerToken{timer: timer}) | |
| 73 | + | timer.Stop() | |
| 74 | 74 | } | |
| 75 | 75 | return epoch | |
| 76 | 76 | } | |
| @@ -106,7 +106,7 @@ func (c *renderCoordinator) scheduleResizeRetryForLease(epoch uint64, source *at | |||
| 106 | 106 | timer := clock.NewTimer(minOutputRenderDeadline) | |
| 107 | 107 | timerC := timer.C() | |
| 108 | 108 | if timerC == nil { | |
| 109 | - | stopDetachedTimer(&timerToken{timer: timer}) | |
| 109 | + | timer.Stop() | |
| 110 | 110 | return | |
| 111 | 111 | } | |
| 112 | 112 | c.mu.Lock() | |
| @@ -130,7 +130,7 @@ func (c *renderCoordinator) scheduleResizeRetryForLease(epoch uint64, source *at | |||
| 130 | 130 | } | |
| 131 | 131 | c.mu.Unlock() | |
| 132 | 132 | if !valid { | |
| 133 | - | stopDetachedTimer(&timerToken{timer: timer}) | |
| 133 | + | timer.Stop() | |
| 134 | 134 | } | |
| 135 | 135 | } | |
| 136 | 136 | ||
| @@ -42,7 +42,7 @@ func (c *renderCoordinator) scheduleResizeForLease(size domain.Size, source *att | |||||
| 42 | timer := clock.NewTimer(minOutputRenderDeadline) | 42 | timer := clock.NewTimer(minOutputRenderDeadline) | ||
| 43 | timerC := timer.C() | 43 | timerC := timer.C() | ||
| 44 | if timerC == nil { | 44 | if timerC == nil { | ||
| 45 | - | stopDetachedTimer(&timerToken{timer: timer}) | 45 | + | timer.Stop() |
| 46 | if c.resizeCurrentForLease(epoch, source, lease, false) { | 46 | if c.resizeCurrentForLease(epoch, source, lease, false) { | ||
| 47 | run(epoch) | 47 | run(epoch) | ||
| 48 | } | 48 | } | ||
| @@ -70,7 +70,7 @@ func (c *renderCoordinator) scheduleResizeForLease(size domain.Size, source *att | |||||
| 70 | } | 70 | } | ||
| 71 | c.mu.Unlock() | 71 | c.mu.Unlock() | ||
| 72 | if !valid { | 72 | if !valid { | ||
| 73 | - | stopDetachedTimer(&timerToken{timer: timer}) | 73 | + | timer.Stop() |
| 74 | } | 74 | } | ||
| 75 | return epoch | 75 | return epoch | ||
| 76 | } | 76 | } | ||
| @@ -106,7 +106,7 @@ func (c *renderCoordinator) scheduleResizeRetryForLease(epoch uint64, source *at | |||||
| 106 | timer := clock.NewTimer(minOutputRenderDeadline) | 106 | timer := clock.NewTimer(minOutputRenderDeadline) | ||
| 107 | timerC := timer.C() | 107 | timerC := timer.C() | ||
| 108 | if timerC == nil { | 108 | if timerC == nil { | ||
| 109 | - | stopDetachedTimer(&timerToken{timer: timer}) | 109 | + | timer.Stop() |
| 110 | return | 110 | return | ||
| 111 | } | 111 | } | ||
| 112 | c.mu.Lock() | 112 | c.mu.Lock() | ||
| @@ -130,7 +130,7 @@ func (c *renderCoordinator) scheduleResizeRetryForLease(epoch uint64, source *at | |||||
| 130 | } | 130 | } | ||
| 131 | c.mu.Unlock() | 131 | c.mu.Unlock() | ||
| 132 | if !valid { | 132 | if !valid { | ||
| 133 | - | stopDetachedTimer(&timerToken{timer: timer}) | 133 | + | timer.Stop() |
| 134 | } | 134 | } | ||
| 135 | } | 135 | } | ||
| 136 | 136 | ||||