fix(daemon): bind coordinator wakes to attachments
- commit
- de2ad79f93cf86ab9eb7720910f6296077331496
- author
- brice · · 2026-07-11 19:47 UTC
- parent
- 9a215e4
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +45 -11
M internal/usecase/daemon/client.go +6 -5
| @@ -385,11 +385,12 @@ func (d *Daemon) attachCoordinator(sess *session, old, current *attachedClient) | |||
| 385 | 385 | rc = newRenderCoordinator(renderCoordinatorOptions{ | |
| 386 | 386 | clock: d.clock, | |
| 387 | 387 | wake: func(w renderWake) { | |
| 388 | - | sess.mu.Lock() | |
| 389 | - | attached := sess.client | |
| 390 | - | sess.mu.Unlock() | |
| 391 | - | if attached != nil { | |
| 392 | - | d.paint(sess, attached, w.reset) | |
| 388 | + | // w.attachment was validated and captured under the coordinator | |
| 389 | + | // lock. Never reread sess.client here: attach publishes that field | |
| 390 | + | // before it replaces the coordinator, so doing so can paint the | |
| 391 | + | // new connection before Welcome. | |
| 392 | + | if w.attachment != nil { | |
| 393 | + | d.paint(sess, w.attachment, w.reset) | |
| 393 | 394 | } | |
| 394 | 395 | }, | |
| 395 | 396 | ackReady: func() bool { | |
| @@ -385,11 +385,12 @@ func (d *Daemon) attachCoordinator(sess *session, old, current *attachedClient) | |||||
| 385 | rc = newRenderCoordinator(renderCoordinatorOptions{ | 385 | rc = newRenderCoordinator(renderCoordinatorOptions{ | ||
| 386 | clock: d.clock, | 386 | clock: d.clock, | ||
| 387 | wake: func(w renderWake) { | 387 | wake: func(w renderWake) { | ||
| 388 | - | sess.mu.Lock() | 388 | + | // w.attachment was validated and captured under the coordinator |
| 389 | - | attached := sess.client | 389 | + | // lock. Never reread sess.client here: attach publishes that field |
| 390 | - | sess.mu.Unlock() | 390 | + | // before it replaces the coordinator, so doing so can paint the |
| 391 | - | if attached != nil { | 391 | + | // new connection before Welcome. |
| 392 | - | d.paint(sess, attached, w.reset) | 392 | + | if w.attachment != nil { |
| 393 | + | d.paint(sess, w.attachment, w.reset) | |||
| 393 | } | 394 | } | ||
| 394 | }, | 395 | }, | ||
| 395 | ackReady: func() bool { | 396 | ackReady: func() bool { | ||
M internal/usecase/daemon/picker.go +6 -0
| @@ -407,6 +407,12 @@ func (d *Daemon) stealClientForTarget(from *session, ac *attachedClient, targetS | |||
| 407 | 407 | ac.recordPreviousSession(from) | |
| 408 | 408 | d.mu.Unlock() | |
| 409 | 409 | d.handoffCoordinator(from, targetSess, old, ac) | |
| 410 | + | if old != nil && old != ac { | |
| 411 | + | // Displacement is a detach lifecycle transition too. Cancel the | |
| 412 | + | // attachment-owned PR #71 timer only after releasing daemon/session | |
| 413 | + | // locks, preserving sendMu > daemon/session lock ordering. | |
| 414 | + | old.cancelResizePaint() | |
| 415 | + | } | |
| 410 | 416 | return old | |
| 411 | 417 | } | |
| 412 | 418 | ||
| @@ -407,6 +407,12 @@ func (d *Daemon) stealClientForTarget(from *session, ac *attachedClient, targetS | |||||
| 407 | ac.recordPreviousSession(from) | 407 | ac.recordPreviousSession(from) | ||
| 408 | d.mu.Unlock() | 408 | d.mu.Unlock() | ||
| 409 | d.handoffCoordinator(from, targetSess, old, ac) | 409 | d.handoffCoordinator(from, targetSess, old, ac) | ||
| 410 | + | if old != nil && old != ac { | |||
| 411 | + | // Displacement is a detach lifecycle transition too. Cancel the | |||
| 412 | + | // attachment-owned PR #71 timer only after releasing daemon/session | |||
| 413 | + | // locks, preserving sendMu > daemon/session lock ordering. | |||
| 414 | + | old.cancelResizePaint() | |||
| 415 | + | } | |||
| 410 | return old | 416 | return old | ||
| 411 | } | 417 | } | ||
| 412 | 418 | ||||
M internal/usecase/daemon/render.go +7 -1
| @@ -288,7 +288,13 @@ func (d *Daemon) paintForResizeGeneration(sess *session, ac *attachedClient, res | |||
| 288 | 288 | } | |
| 289 | 289 | ||
| 290 | 290 | ac.sendMu.Lock() | |
| 291 | - | if ac.currentSession() != sess { | |
| 291 | + | // sendMu is the attachment ownership boundary. Check the session's | |
| 292 | + | // published identity while holding it so a deadline captured before an | |
| 293 | + | // attach/replace cannot emit on either the old or new output chain. | |
| 294 | + | sess.mu.Lock() | |
| 295 | + | owned := sess.client == ac | |
| 296 | + | sess.mu.Unlock() | |
| 297 | + | if !owned || ac.currentSession() != sess { | |
| 292 | 298 | ac.sendMu.Unlock() | |
| 293 | 299 | return | |
| 294 | 300 | } | |
| @@ -288,7 +288,13 @@ func (d *Daemon) paintForResizeGeneration(sess *session, ac *attachedClient, res | |||||
| 288 | } | 288 | } | ||
| 289 | 289 | ||||
| 290 | ac.sendMu.Lock() | 290 | ac.sendMu.Lock() | ||
| 291 | - | if ac.currentSession() != sess { | 291 | + | // sendMu is the attachment ownership boundary. Check the session's |
| 292 | + | // published identity while holding it so a deadline captured before an | |||
| 293 | + | // attach/replace cannot emit on either the old or new output chain. | |||
| 294 | + | sess.mu.Lock() | |||
| 295 | + | owned := sess.client == ac | |||
| 296 | + | sess.mu.Unlock() | |||
| 297 | + | if !owned || ac.currentSession() != sess { | |||
| 292 | ac.sendMu.Unlock() | 298 | ac.sendMu.Unlock() | ||
| 293 | return | 299 | return | ||
| 294 | } | 300 | } | ||
M internal/usecase/daemon/render_coordinator.go +10 -2
| @@ -54,6 +54,10 @@ type renderWake struct { | |||
| 54 | 54 | reset bool | |
| 55 | 55 | urgent bool | |
| 56 | 56 | coalesced int | |
| 57 | + | // attachment is snapshotted under coordinator ownership. The wake | |
| 58 | + | // consumer must compose only for this exact client, never by rereading | |
| 59 | + | // session.client after an attach publication. | |
| 60 | + | attachment *attachedClient | |
| 57 | 61 | // watchdog marks a flush forced by the synchronized-output watchdog. | |
| 58 | 62 | watchdog bool | |
| 59 | 63 | } | |
| @@ -510,7 +514,11 @@ func (c *renderCoordinator) noteDetach(ac *attachedClient) { | |||
| 510 | 514 | // captured by old become stale. | |
| 511 | 515 | func (c *renderCoordinator) noteReplace(old, replacement *attachedClient) { | |
| 512 | 516 | c.mu.Lock() | |
| 513 | - | if c.attachment == old { | |
| 517 | + | // A coordinator may be installed while replacing a legacy attachment | |
| 518 | + | // which predated coordinator ownership. In that case nil has no pending | |
| 519 | + | // identity to invalidate, and the replacement becomes the first bound | |
| 520 | + | // attachment atomically with this lifecycle transition. | |
| 521 | + | if c.attachment == old || c.attachment == nil { | |
| 514 | 522 | c.attachment = replacement | |
| 515 | 523 | c.pending = false | |
| 516 | 524 | c.ackDeferred = false | |
| @@ -647,7 +655,7 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | |||
| 647 | 655 | c.mu.Unlock() | |
| 648 | 656 | return | |
| 649 | 657 | } | |
| 650 | - | w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, watchdog: watchdog} | |
| 658 | + | w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, watchdog: watchdog, attachment: c.attachment} | |
| 651 | 659 | if !ready { | |
| 652 | 660 | if deadline { | |
| 653 | 661 | c.ackDeferred = true | |
| @@ -54,6 +54,10 @@ type renderWake struct { | |||||
| 54 | reset bool | 54 | reset bool | ||
| 55 | urgent bool | 55 | urgent bool | ||
| 56 | coalesced int | 56 | coalesced int | ||
| 57 | + | // attachment is snapshotted under coordinator ownership. The wake | |||
| 58 | + | // consumer must compose only for this exact client, never by rereading | |||
| 59 | + | // session.client after an attach publication. | |||
| 60 | + | attachment *attachedClient | |||
| 57 | // watchdog marks a flush forced by the synchronized-output watchdog. | 61 | // watchdog marks a flush forced by the synchronized-output watchdog. | ||
| 58 | watchdog bool | 62 | watchdog bool | ||
| 59 | } | 63 | } | ||
| @@ -510,7 +514,11 @@ func (c *renderCoordinator) noteDetach(ac *attachedClient) { | |||||
| 510 | // captured by old become stale. | 514 | // captured by old become stale. | ||
| 511 | func (c *renderCoordinator) noteReplace(old, replacement *attachedClient) { | 515 | func (c *renderCoordinator) noteReplace(old, replacement *attachedClient) { | ||
| 512 | c.mu.Lock() | 516 | c.mu.Lock() | ||
| 513 | - | if c.attachment == old { | 517 | + | // A coordinator may be installed while replacing a legacy attachment |
| 518 | + | // which predated coordinator ownership. In that case nil has no pending | |||
| 519 | + | // identity to invalidate, and the replacement becomes the first bound | |||
| 520 | + | // attachment atomically with this lifecycle transition. | |||
| 521 | + | if c.attachment == old || c.attachment == nil { | |||
| 514 | c.attachment = replacement | 522 | c.attachment = replacement | ||
| 515 | c.pending = false | 523 | c.pending = false | ||
| 516 | c.ackDeferred = false | 524 | c.ackDeferred = false | ||
| @@ -647,7 +655,7 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | |||||
| 647 | c.mu.Unlock() | 655 | c.mu.Unlock() | ||
| 648 | return | 656 | return | ||
| 649 | } | 657 | } | ||
| 650 | - | w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, watchdog: watchdog} | 658 | + | w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, watchdog: watchdog, attachment: c.attachment} |
| 651 | if !ready { | 659 | if !ready { | ||
| 652 | if deadline { | 660 | if deadline { | ||
| 653 | c.ackDeferred = true | 661 | c.ackDeferred = true | ||
M internal/usecase/daemon/render_coordinator_test.go +16 -3
| @@ -354,7 +354,10 @@ resize: | |||
| 354 | 354 | select { | |
| 355 | 355 | case epoch := <-resizeDone: | |
| 356 | 356 | require.Equal(t, uint64(1), epoch) | |
| 357 | - | require.Equal(t, renderWake{coalesced: 1}, awaitWake(t, wakes)) | |
| 357 | + | wake := awaitWake(t, wakes) | |
| 358 | + | require.Same(t, owner, wake.attachment) | |
| 359 | + | wake.attachment = nil | |
| 360 | + | require.Equal(t, renderWake{coalesced: 1}, wake) | |
| 358 | 361 | for range 4096 { | |
| 359 | 362 | select { | |
| 360 | 363 | case <-fireDone: | |
| @@ -737,14 +740,24 @@ func TestRenderCoordinatorPreviewWakesDoNotWaitForTargetAck(t *testing.T) { | |||
| 737 | 740 | require.Len(t, timers, 1) | |
| 738 | 741 | timers[0].ch <- time.Time{} | |
| 739 | 742 | ||
| 740 | - | require.Equal(t, renderWake{coalesced: 1}, awaitWake(t, h.previews), | |
| 743 | + | preview := awaitWake(t, h.previews) | |
| 744 | + | if tc.attach { | |
| 745 | + | require.NotNil(t, preview.attachment) | |
| 746 | + | } | |
| 747 | + | preview.attachment = nil | |
| 748 | + | require.Equal(t, renderWake{coalesced: 1}, preview, | |
| 741 | 749 | "the viewer preview must receive target output without target ACK capacity") | |
| 742 | 750 | requireNoWake(t, h.wakes) | |
| 743 | 751 | requireNoWake(t, h.previews) | |
| 744 | 752 | ||
| 745 | 753 | h.ackReady.Store(true) | |
| 746 | 754 | h.rc.notifyAck() | |
| 747 | - | require.Equal(t, renderWake{coalesced: 1}, awaitWake(t, h.wakes), | |
| 755 | + | wake := awaitWake(t, h.wakes) | |
| 756 | + | if tc.attach { | |
| 757 | + | require.NotNil(t, wake.attachment) | |
| 758 | + | } | |
| 759 | + | wake.attachment = nil | |
| 760 | + | require.Equal(t, renderWake{coalesced: 1}, wake, | |
| 748 | 761 | "the target primary frame remains pending for its own ACK") | |
| 749 | 762 | requireNoWake(t, h.previews) | |
| 750 | 763 | }) | |
| @@ -354,7 +354,10 @@ resize: | |||||
| 354 | select { | 354 | select { | ||
| 355 | case epoch := <-resizeDone: | 355 | case epoch := <-resizeDone: | ||
| 356 | require.Equal(t, uint64(1), epoch) | 356 | require.Equal(t, uint64(1), epoch) | ||
| 357 | - | require.Equal(t, renderWake{coalesced: 1}, awaitWake(t, wakes)) | 357 | + | wake := awaitWake(t, wakes) |
| 358 | + | require.Same(t, owner, wake.attachment) | |||
| 359 | + | wake.attachment = nil | |||
| 360 | + | require.Equal(t, renderWake{coalesced: 1}, wake) | |||
| 358 | for range 4096 { | 361 | for range 4096 { | ||
| 359 | select { | 362 | select { | ||
| 360 | case <-fireDone: | 363 | case <-fireDone: | ||
| @@ -737,14 +740,24 @@ func TestRenderCoordinatorPreviewWakesDoNotWaitForTargetAck(t *testing.T) { | |||||
| 737 | require.Len(t, timers, 1) | 740 | require.Len(t, timers, 1) | ||
| 738 | timers[0].ch <- time.Time{} | 741 | timers[0].ch <- time.Time{} | ||
| 739 | 742 | ||||
| 740 | - | require.Equal(t, renderWake{coalesced: 1}, awaitWake(t, h.previews), | 743 | + | preview := awaitWake(t, h.previews) |
| 744 | + | if tc.attach { | |||
| 745 | + | require.NotNil(t, preview.attachment) | |||
| 746 | + | } | |||
| 747 | + | preview.attachment = nil | |||
| 748 | + | require.Equal(t, renderWake{coalesced: 1}, preview, | |||
| 741 | "the viewer preview must receive target output without target ACK capacity") | 749 | "the viewer preview must receive target output without target ACK capacity") | ||
| 742 | requireNoWake(t, h.wakes) | 750 | requireNoWake(t, h.wakes) | ||
| 743 | requireNoWake(t, h.previews) | 751 | requireNoWake(t, h.previews) | ||
| 744 | 752 | ||||
| 745 | h.ackReady.Store(true) | 753 | h.ackReady.Store(true) | ||
| 746 | h.rc.notifyAck() | 754 | h.rc.notifyAck() | ||
| 747 | - | require.Equal(t, renderWake{coalesced: 1}, awaitWake(t, h.wakes), | 755 | + | wake := awaitWake(t, h.wakes) |
| 756 | + | if tc.attach { | |||
| 757 | + | require.NotNil(t, wake.attachment) | |||
| 758 | + | } | |||
| 759 | + | wake.attachment = nil | |||
| 760 | + | require.Equal(t, renderWake{coalesced: 1}, wake, | |||
| 748 | "the target primary frame remains pending for its own ACK") | 761 | "the target primary frame remains pending for its own ACK") | ||
| 749 | requireNoWake(t, h.previews) | 762 | requireNoWake(t, h.previews) | ||
| 750 | }) | 763 | }) | ||