fix(daemon): join only started remote link workers
- commit
- 8d7a35460c108f4c7afc9914de44875317c7955d
- author
- brice · · 2026-08-06 20:05 UTC
- parent
- b902d96
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +21 -6
M internal/usecase/daemon/daemon.go +1 -1
| @@ -849,7 +849,7 @@ func (d *Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline *snapsho | |||
| 849 | 849 | warm.stop() | |
| 850 | 850 | } | |
| 851 | 851 | for _, link := range remoteLinks { | |
| 852 | - | <-link.done | |
| 852 | + | joinRemoteLink(link) | |
| 853 | 853 | } | |
| 854 | 854 | d.finishParkedAttachmentRetirements(parkedRetirements) | |
| 855 | 855 | d.log.Info("graceful shutdown begin", "reason", reason, "sessions", len(snapshot)) | |
| @@ -849,7 +849,7 @@ func (d *Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline *snapsho | |||||
| 849 | warm.stop() | 849 | warm.stop() | ||
| 850 | } | 850 | } | ||
| 851 | for _, link := range remoteLinks { | 851 | for _, link := range remoteLinks { | ||
| 852 | - | <-link.done | 852 | + | joinRemoteLink(link) |
| 853 | } | 853 | } | ||
| 854 | d.finishParkedAttachmentRetirements(parkedRetirements) | 854 | d.finishParkedAttachmentRetirements(parkedRetirements) | ||
| 855 | d.log.Info("graceful shutdown begin", "reason", reason, "sessions", len(snapshot)) | 855 | d.log.Info("graceful shutdown begin", "reason", reason, "sessions", len(snapshot)) | ||
M internal/usecase/daemon/remote_link.go +16 -4
| @@ -6,6 +6,7 @@ import ( | |||
| 6 | 6 | "errors" | |
| 7 | 7 | "fmt" | |
| 8 | 8 | "sync" | |
| 9 | + | "sync/atomic" | |
| 9 | 10 | ||
| 10 | 11 | "github.com/bnema/vev/internal/domain" | |
| 11 | 12 | "github.com/bnema/vev/internal/ports" | |
| @@ -37,10 +38,11 @@ type remoteLink struct { | |||
| 37 | 38 | transport ports.Transport | |
| 38 | 39 | cancel context.CancelFunc | |
| 39 | 40 | ||
| 40 | - | sendMu sync.Mutex | |
| 41 | - | startOnce sync.Once | |
| 42 | - | done chan struct{} | |
| 43 | - | active bool // guarded by view.mu | |
| 41 | + | sendMu sync.Mutex | |
| 42 | + | startOnce sync.Once | |
| 43 | + | workerStarted atomic.Bool | |
| 44 | + | done chan struct{} | |
| 45 | + | active bool // guarded by view.mu | |
| 44 | 46 | } | |
| 45 | 47 | ||
| 46 | 48 | // remoteOutputState is the remote-link equivalent of the thin client's | |
| @@ -458,10 +460,20 @@ func (d *Daemon) startRemoteLink(link *remoteLink) { | |||
| 458 | 460 | return | |
| 459 | 461 | } | |
| 460 | 462 | link.startOnce.Do(func() { | |
| 463 | + | link.workerStarted.Store(true) | |
| 461 | 464 | go d.runRemoteLink(link) | |
| 462 | 465 | }) | |
| 463 | 466 | } | |
| 464 | 467 | ||
| 468 | + | // joinRemoteLink waits only for a receiver that was actually started. Candidate | |
| 469 | + | // and test-only links may have no transport worker; treating their open done | |
| 470 | + | // channel as joinable would deadlock terminal cleanup. | |
| 471 | + | func joinRemoteLink(link *remoteLink) { | |
| 472 | + | if link != nil && link.workerStarted.Load() { | |
| 473 | + | <-link.done | |
| 474 | + | } | |
| 475 | + | } | |
| 476 | + | ||
| 465 | 477 | func (d *Daemon) runRemoteLink(link *remoteLink) { | |
| 466 | 478 | defer close(link.done) | |
| 467 | 479 | for { | |
| @@ -6,6 +6,7 @@ import ( | |||||
| 6 | "errors" | 6 | "errors" | ||
| 7 | "fmt" | 7 | "fmt" | ||
| 8 | "sync" | 8 | "sync" | ||
| 9 | + | "sync/atomic" | |||
| 9 | 10 | ||||
| 10 | "github.com/bnema/vev/internal/domain" | 11 | "github.com/bnema/vev/internal/domain" | ||
| 11 | "github.com/bnema/vev/internal/ports" | 12 | "github.com/bnema/vev/internal/ports" | ||
| @@ -37,10 +38,11 @@ type remoteLink struct { | |||||
| 37 | transport ports.Transport | 38 | transport ports.Transport | ||
| 38 | cancel context.CancelFunc | 39 | cancel context.CancelFunc | ||
| 39 | 40 | ||||
| 40 | - | sendMu sync.Mutex | 41 | + | sendMu sync.Mutex |
| 41 | - | startOnce sync.Once | 42 | + | startOnce sync.Once |
| 42 | - | done chan struct{} | 43 | + | workerStarted atomic.Bool |
| 43 | - | active bool // guarded by view.mu | 44 | + | done chan struct{} |
| 45 | + | active bool // guarded by view.mu | |||
| 44 | } | 46 | } | ||
| 45 | 47 | ||||
| 46 | // remoteOutputState is the remote-link equivalent of the thin client's | 48 | // remoteOutputState is the remote-link equivalent of the thin client's | ||
| @@ -458,10 +460,20 @@ func (d *Daemon) startRemoteLink(link *remoteLink) { | |||||
| 458 | return | 460 | return | ||
| 459 | } | 461 | } | ||
| 460 | link.startOnce.Do(func() { | 462 | link.startOnce.Do(func() { | ||
| 463 | + | link.workerStarted.Store(true) | |||
| 461 | go d.runRemoteLink(link) | 464 | go d.runRemoteLink(link) | ||
| 462 | }) | 465 | }) | ||
| 463 | } | 466 | } | ||
| 464 | 467 | ||||
| 468 | + | // joinRemoteLink waits only for a receiver that was actually started. Candidate | |||
| 469 | + | // and test-only links may have no transport worker; treating their open done | |||
| 470 | + | // channel as joinable would deadlock terminal cleanup. | |||
| 471 | + | func joinRemoteLink(link *remoteLink) { | |||
| 472 | + | if link != nil && link.workerStarted.Load() { | |||
| 473 | + | <-link.done | |||
| 474 | + | } | |||
| 475 | + | } | |||
| 476 | + | ||||
| 465 | func (d *Daemon) runRemoteLink(link *remoteLink) { | 477 | func (d *Daemon) runRemoteLink(link *remoteLink) { | ||
| 466 | defer close(link.done) | 478 | defer close(link.done) | ||
| 467 | for { | 479 | for { | ||
M internal/usecase/daemon/remote_warm.go +1 -1
| @@ -145,7 +145,7 @@ func (d *Daemon) expireRemoteViewWarm(view *remoteView, warm *remoteViewWarm) { | |||
| 145 | 145 | if link.transport != nil { | |
| 146 | 146 | _ = link.transport.Close() | |
| 147 | 147 | } | |
| 148 | - | <-link.done | |
| 148 | + | joinRemoteLink(link) | |
| 149 | 149 | } | |
| 150 | 150 | ||
| 151 | 151 | // stopRemoteViewWarm clears and returns only the current timer. Its caller is | |
| @@ -145,7 +145,7 @@ func (d *Daemon) expireRemoteViewWarm(view *remoteView, warm *remoteViewWarm) { | |||||
| 145 | if link.transport != nil { | 145 | if link.transport != nil { | ||
| 146 | _ = link.transport.Close() | 146 | _ = link.transport.Close() | ||
| 147 | } | 147 | } | ||
| 148 | - | <-link.done | 148 | + | joinRemoteLink(link) |
| 149 | } | 149 | } | ||
| 150 | 150 | ||||
| 151 | // stopRemoteViewWarm clears and returns only the current timer. Its caller is | 151 | // stopRemoteViewWarm clears and returns only the current timer. Its caller is | ||
M internal/usecase/daemon/remote_warm_test.go +3 -0
| @@ -13,6 +13,9 @@ func TestRemoteViewWarmExpiryRemovesOnlyDetachedExactView(t *testing.T) { | |||
| 13 | 13 | clock := &signalClock{timers: make(chan *signalTimer, 1)} | |
| 14 | 14 | d := newTestDaemon(t, nil, clock) | |
| 15 | 15 | view := registerRemoteWarmTestView(t, d) | |
| 16 | + | view.mu.Lock() | |
| 17 | + | view.link = &remoteLink{view: view, generation: view.linkGeneration, cancel: func() {}, done: make(chan struct{}), active: true} | |
| 18 | + | view.mu.Unlock() | |
| 16 | 19 | ||
| 17 | 20 | d.parkRemoteViewWarm(view) | |
| 18 | 21 | warm, timer := remoteWarmTimer(t, clock, view) | |
| @@ -13,6 +13,9 @@ func TestRemoteViewWarmExpiryRemovesOnlyDetachedExactView(t *testing.T) { | |||||
| 13 | clock := &signalClock{timers: make(chan *signalTimer, 1)} | 13 | clock := &signalClock{timers: make(chan *signalTimer, 1)} | ||
| 14 | d := newTestDaemon(t, nil, clock) | 14 | d := newTestDaemon(t, nil, clock) | ||
| 15 | view := registerRemoteWarmTestView(t, d) | 15 | view := registerRemoteWarmTestView(t, d) | ||
| 16 | + | view.mu.Lock() | |||
| 17 | + | view.link = &remoteLink{view: view, generation: view.linkGeneration, cancel: func() {}, done: make(chan struct{}), active: true} | |||
| 18 | + | view.mu.Unlock() | |||
| 16 | 19 | ||||
| 17 | d.parkRemoteViewWarm(view) | 20 | d.parkRemoteViewWarm(view) | ||
| 18 | warm, timer := remoteWarmTimer(t, clock, view) | 21 | warm, timer := remoteWarmTimer(t, clock, view) | ||