detached a1296c0

refactor(daemon): extract snapshot worker

commit
a1296c07f73960588c41553e9e7ad1d4f7d34c18
author
brice · · 2026-10-08 06:06 UTC
committer
brice · · 2026-10-08 08:12 UTC
parent
4c53368
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
17 files · +256 -212
Unified
UnifiedSplit
M internal/usecase/daemon/client.go +11 -0
@@ -18,6 +18,17 @@
1818 // moveLifecycleReservation.Release must never be called with d.mu held: it takes
1919 // moveLifecycleMu first, releases both the move-gate and purge-gate reservations
2020 // in that order, and only then drops moveLifecycleMu.
21+//
22+// Snapshot worker locks (snapshotWorker in snapshot_worker_state.go):
23+// snapshotWorker.mu (d.snapshots.mu) is a leaf with respect to d.mu and the
24+// session (mu), tab, and pane locks: no path acquires any of them while holding
25+// it, and the capture, queue, and worker code never takes d.mu. The only nested
26+// acquisition is a session's snapshotMu inside it (enqueueFinalSnapshotCapture
27+// finishes a replaced capture), so snapshotMu holders must release snapshotMu
28+// before taking snapshotWorker.mu, as finishSnapshotCapture does.
29+// snapshotWorker.noticeMu guards the active persistence-failure signature; it
30+// is a leaf and is never held together with snapshotWorker.mu. Restoration
31+// takes d.mu only outside snapshotWorker.mu.
2132 package daemon
2233
2334 import (
@@ -18,6 +18,17 @@
18 // moveLifecycleReservation.Release must never be called with d.mu held: it takes18 // moveLifecycleReservation.Release must never be called with d.mu held: it takes
19 // moveLifecycleMu first, releases both the move-gate and purge-gate reservations19 // moveLifecycleMu first, releases both the move-gate and purge-gate reservations
20 // in that order, and only then drops moveLifecycleMu.20 // in that order, and only then drops moveLifecycleMu.
21+//
22+// Snapshot worker locks (snapshotWorker in snapshot_worker_state.go):
23+// snapshotWorker.mu (d.snapshots.mu) is a leaf with respect to d.mu and the
24+// session (mu), tab, and pane locks: no path acquires any of them while holding
25+// it, and the capture, queue, and worker code never takes d.mu. The only nested
26+// acquisition is a session's snapshotMu inside it (enqueueFinalSnapshotCapture
27+// finishes a replaced capture), so snapshotMu holders must release snapshotMu
28+// before taking snapshotWorker.mu, as finishSnapshotCapture does.
29+// snapshotWorker.noticeMu guards the active persistence-failure signature; it
30+// is a leaf and is never held together with snapshotWorker.mu. Restoration
31+// takes d.mu only outside snapshotWorker.mu.
21 package daemon32 package daemon
22 33
23 import (34 import (
M internal/usecase/daemon/daemon.go +31 -59
@@ -177,45 +177,15 @@ type Daemon struct {
177177 snapshotRepository ports.SnapshotRepository
178178 recovery *recoveryusecase.Coordinator
179179 snapshotGarbageCollection bool
180- maintenanceWorkerCancel context.CancelFunc
181- maintenanceWorkerDone chan struct{}
182- // restoreWorkerDone is the restoration goroutine's ownership signal. Startup
183- // restoration reconciles durable checkpoints, so it is a durable writer and
184- // is guarded by snapshotWorkerMu with the other two.
185- restoreWorkerDone chan struct{}
186- snapsEnabled bool
187- noticeStore ports.NoticeStore
188- snapshotJobs chan *snapshotCapture
189- // snapshotAdmitted contains every capture accepted by either worker queue,
190- // including captures buffered in snapshotJobs. Guarded by snapshotWorkerMu.
191- snapshotAdmitted map[*snapshotCapture]struct{}
192- // snapshotWake wakes the repository scheduler when a session becomes dirty
193- // or an attempt completes. It is never closed and producers only send
194- // non-blockingly.
195- snapshotWake chan struct{}
196- snapshotWorkerMu sync.Mutex
197- snapshotWorkerID uint64
198- snapshotWorkerCtx context.Context
199- snapshotWorkerCancel context.CancelFunc
200- snapshotWorkerDone chan struct{}
201- snapshotWorkerFlush chan struct{}
202- snapshotWorkerFinalWake chan struct{}
203- // snapshotFinalJobs coalesces terminal captures by session when the bounded
204- // regular queue is full. It retains at most snapshotFinalQueueCapacity named
205- // sessions, each with only its newest terminal state while the worker blocks.
206- snapshotFinalJobs map[*session]*snapshotCapture
207- snapshotFinalOrder []*session
208- snapshotWorkerClosing bool
209- snapshotWorkerInFlight *snapshotCapture
210- // snapshotNoticeMu guards the active global persistence failure signature.
211- // It is separate from snapshotWorkerMu so notice routing cannot block a
212- // producer or a repository worker.
213- snapshotNoticeMu sync.Mutex
214- snapshotActiveFailureSignature string
215- shutdownNoticeMu sync.Mutex
216- shutdownNoticedSessions map[string]struct{}
217- restoreDone chan struct{}
218- restoreOnce sync.Once
180+ snapsEnabled bool
181+ noticeStore ports.NoticeStore
182+ // snapshots owns the snapshot worker, the durable-writer lifecycle
183+ // signals, and the global persistence-failure notice state.
184+ snapshots snapshotWorker
185+ shutdownNoticeMu sync.Mutex
186+ shutdownNoticedSessions map[string]struct{}
187+ restoreDone chan struct{}
188+ restoreOnce sync.Once
219189 // proc inspects pane processes (cwd, foreground command, argv). Nil means
220190 // no inspection is available; see WithProcessInspector and WithCwdReader.
221191 proc ports.ProcessInspector
@@ -699,26 +669,28 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
699669 }
700670 paneProcessCtx, paneProcessCancel := context.WithCancel(context.Background())
701671 d := &Daemon{
702- sessions: make(map[domain.SessionID]*session),
703- inactive: make(map[string]inactiveSession),
704- creating: make(map[string]struct{}),
705- resume: newResumeCredentials(),
706- suspended: make(map[*attachedClient]*suspendedAttachmentRetention),
707- graphics: newGraphicsNamespacePool(),
708- paneProcessCtx: paneProcessCtx,
709- paneProcessCancel: paneProcessCancel,
710- ptys: ptys,
711- clock: clock,
712- log: log,
713- baseEnv: os.Environ(),
714- shell: defaultShellCommand,
715- dirOrHome: dirOrHome,
716- done: make(chan struct{}),
717- restoreDone: make(chan struct{}),
718- animWake: make(chan struct{}, 1),
719- snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),
720- snapshotAdmitted: make(map[*snapshotCapture]struct{}),
721- snapshotWake: make(chan struct{}, 1),
672+ sessions: make(map[domain.SessionID]*session),
673+ inactive: make(map[string]inactiveSession),
674+ creating: make(map[string]struct{}),
675+ resume: newResumeCredentials(),
676+ suspended: make(map[*attachedClient]*suspendedAttachmentRetention),
677+ graphics: newGraphicsNamespacePool(),
678+ paneProcessCtx: paneProcessCtx,
679+ paneProcessCancel: paneProcessCancel,
680+ ptys: ptys,
681+ clock: clock,
682+ log: log,
683+ baseEnv: os.Environ(),
684+ shell: defaultShellCommand,
685+ dirOrHome: dirOrHome,
686+ done: make(chan struct{}),
687+ restoreDone: make(chan struct{}),
688+ animWake: make(chan struct{}, 1),
689+ snapshots: snapshotWorker{
690+ jobs: make(chan *snapshotCapture, snapshotQueueCapacity),
691+ admitted: make(map[*snapshotCapture]struct{}),
692+ wake: make(chan struct{}, 1),
693+ },
722694 notices: newNoticeCenter(),
723695 resumeParkGrace: defaultResumeParkGrace,
724696 suspendedSafetyExpiry: defaultSuspendedSafetyExpiry,
@@ -177,45 +177,15 @@ type Daemon struct {
177 snapshotRepository ports.SnapshotRepository177 snapshotRepository ports.SnapshotRepository
178 recovery *recoveryusecase.Coordinator178 recovery *recoveryusecase.Coordinator
179 snapshotGarbageCollection bool179 snapshotGarbageCollection bool
180- maintenanceWorkerCancel context.CancelFunc180+ snapsEnabled bool
181- maintenanceWorkerDone chan struct{}181+ noticeStore ports.NoticeStore
182- // restoreWorkerDone is the restoration goroutine's ownership signal. Startup182+ // snapshots owns the snapshot worker, the durable-writer lifecycle
183- // restoration reconciles durable checkpoints, so it is a durable writer and183+ // signals, and the global persistence-failure notice state.
184- // is guarded by snapshotWorkerMu with the other two.184+ snapshots snapshotWorker
185- restoreWorkerDone chan struct{}185+ shutdownNoticeMu sync.Mutex
186- snapsEnabled bool186+ shutdownNoticedSessions map[string]struct{}
187- noticeStore ports.NoticeStore187+ restoreDone chan struct{}
188- snapshotJobs chan *snapshotCapture188+ restoreOnce sync.Once
189- // snapshotAdmitted contains every capture accepted by either worker queue,
190- // including captures buffered in snapshotJobs. Guarded by snapshotWorkerMu.
191- snapshotAdmitted map[*snapshotCapture]struct{}
192- // snapshotWake wakes the repository scheduler when a session becomes dirty
193- // or an attempt completes. It is never closed and producers only send
194- // non-blockingly.
195- snapshotWake chan struct{}
196- snapshotWorkerMu sync.Mutex
197- snapshotWorkerID uint64
198- snapshotWorkerCtx context.Context
199- snapshotWorkerCancel context.CancelFunc
200- snapshotWorkerDone chan struct{}
201- snapshotWorkerFlush chan struct{}
202- snapshotWorkerFinalWake chan struct{}
203- // snapshotFinalJobs coalesces terminal captures by session when the bounded
204- // regular queue is full. It retains at most snapshotFinalQueueCapacity named
205- // sessions, each with only its newest terminal state while the worker blocks.
206- snapshotFinalJobs map[*session]*snapshotCapture
207- snapshotFinalOrder []*session
208- snapshotWorkerClosing bool
209- snapshotWorkerInFlight *snapshotCapture
210- // snapshotNoticeMu guards the active global persistence failure signature.
211- // It is separate from snapshotWorkerMu so notice routing cannot block a
212- // producer or a repository worker.
213- snapshotNoticeMu sync.Mutex
214- snapshotActiveFailureSignature string
215- shutdownNoticeMu sync.Mutex
216- shutdownNoticedSessions map[string]struct{}
217- restoreDone chan struct{}
218- restoreOnce sync.Once
219 // proc inspects pane processes (cwd, foreground command, argv). Nil means189 // proc inspects pane processes (cwd, foreground command, argv). Nil means
220 // no inspection is available; see WithProcessInspector and WithCwdReader.190 // no inspection is available; see WithProcessInspector and WithCwdReader.
221 proc ports.ProcessInspector191 proc ports.ProcessInspector
@@ -699,26 +669,28 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
699 }669 }
700 paneProcessCtx, paneProcessCancel := context.WithCancel(context.Background())670 paneProcessCtx, paneProcessCancel := context.WithCancel(context.Background())
701 d := &Daemon{671 d := &Daemon{
702- sessions: make(map[domain.SessionID]*session),672+ sessions: make(map[domain.SessionID]*session),
703- inactive: make(map[string]inactiveSession),673+ inactive: make(map[string]inactiveSession),
704- creating: make(map[string]struct{}),674+ creating: make(map[string]struct{}),
705- resume: newResumeCredentials(),675+ resume: newResumeCredentials(),
706- suspended: make(map[*attachedClient]*suspendedAttachmentRetention),676+ suspended: make(map[*attachedClient]*suspendedAttachmentRetention),
707- graphics: newGraphicsNamespacePool(),677+ graphics: newGraphicsNamespacePool(),
708- paneProcessCtx: paneProcessCtx,678+ paneProcessCtx: paneProcessCtx,
709- paneProcessCancel: paneProcessCancel,679+ paneProcessCancel: paneProcessCancel,
710- ptys: ptys,680+ ptys: ptys,
711- clock: clock,681+ clock: clock,
712- log: log,682+ log: log,
713- baseEnv: os.Environ(),683+ baseEnv: os.Environ(),
714- shell: defaultShellCommand,684+ shell: defaultShellCommand,
715- dirOrHome: dirOrHome,685+ dirOrHome: dirOrHome,
716- done: make(chan struct{}),686+ done: make(chan struct{}),
717- restoreDone: make(chan struct{}),687+ restoreDone: make(chan struct{}),
718- animWake: make(chan struct{}, 1),688+ animWake: make(chan struct{}, 1),
719- snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),689+ snapshots: snapshotWorker{
720- snapshotAdmitted: make(map[*snapshotCapture]struct{}),690+ jobs: make(chan *snapshotCapture, snapshotQueueCapacity),
721- snapshotWake: make(chan struct{}, 1),691+ admitted: make(map[*snapshotCapture]struct{}),
692+ wake: make(chan struct{}, 1),
693+ },
722 notices: newNoticeCenter(),694 notices: newNoticeCenter(),
723 resumeParkGrace: defaultResumeParkGrace,695 resumeParkGrace: defaultResumeParkGrace,
724 suspendedSafetyExpiry: defaultSuspendedSafetyExpiry,696 suspendedSafetyExpiry: defaultSuspendedSafetyExpiry,
M internal/usecase/daemon/maintenance.go +11 -11
@@ -19,16 +19,16 @@ func (d *Daemon) startDurableMaintenance() {
1919 if d == nil || d.recovery == nil || !d.snapshotGarbageCollection {
2020 return
2121 }
22- d.snapshotWorkerMu.Lock()
23- if d.maintenanceWorkerCancel != nil {
24- d.snapshotWorkerMu.Unlock()
22+ d.snapshots.mu.Lock()
23+ if d.snapshots.maintenanceWorkerCancel != nil {
24+ d.snapshots.mu.Unlock()
2525 return
2626 }
2727 ctx, cancel := context.WithCancel(d.serveCtx)
28- d.maintenanceWorkerCancel = cancel
28+ d.snapshots.maintenanceWorkerCancel = cancel
2929 done := make(chan struct{})
30- d.maintenanceWorkerDone = done
31- d.snapshotWorkerMu.Unlock()
30+ d.snapshots.maintenanceWorkerDone = done
31+ d.snapshots.mu.Unlock()
3232
3333 go func() {
3434 defer close(done)
@@ -39,15 +39,15 @@ func (d *Daemon) startDurableMaintenance() {
3939 // cancelDurableMaintenance stops the GC worker without waiting for it;
4040 // WaitDurableWriters owns the join.
4141 func (d *Daemon) cancelDurableMaintenance() {
42- d.snapshotWorkerMu.Lock()
43- defer d.snapshotWorkerMu.Unlock()
42+ d.snapshots.mu.Lock()
43+ defer d.snapshots.mu.Unlock()
4444 d.cancelDurableMaintenanceLocked()
4545 }
4646
47-// cancelDurableMaintenanceLocked requires snapshotWorkerMu.
47+// cancelDurableMaintenanceLocked requires snapshotWorker.mu (d.snapshots.mu).
4848 func (d *Daemon) cancelDurableMaintenanceLocked() {
49- if d.maintenanceWorkerCancel != nil {
50- d.maintenanceWorkerCancel()
49+ if d.snapshots.maintenanceWorkerCancel != nil {
50+ d.snapshots.maintenanceWorkerCancel()
5151 }
5252 }
5353
@@ -19,16 +19,16 @@ func (d *Daemon) startDurableMaintenance() {
19 if d == nil || d.recovery == nil || !d.snapshotGarbageCollection {19 if d == nil || d.recovery == nil || !d.snapshotGarbageCollection {
20 return20 return
21 }21 }
22- d.snapshotWorkerMu.Lock()22+ d.snapshots.mu.Lock()
23- if d.maintenanceWorkerCancel != nil {23+ if d.snapshots.maintenanceWorkerCancel != nil {
24- d.snapshotWorkerMu.Unlock()24+ d.snapshots.mu.Unlock()
25 return25 return
26 }26 }
27 ctx, cancel := context.WithCancel(d.serveCtx)27 ctx, cancel := context.WithCancel(d.serveCtx)
28- d.maintenanceWorkerCancel = cancel28+ d.snapshots.maintenanceWorkerCancel = cancel
29 done := make(chan struct{})29 done := make(chan struct{})
30- d.maintenanceWorkerDone = done30+ d.snapshots.maintenanceWorkerDone = done
31- d.snapshotWorkerMu.Unlock()31+ d.snapshots.mu.Unlock()
32 32
33 go func() {33 go func() {
34 defer close(done)34 defer close(done)
@@ -39,15 +39,15 @@ func (d *Daemon) startDurableMaintenance() {
39 // cancelDurableMaintenance stops the GC worker without waiting for it;39 // cancelDurableMaintenance stops the GC worker without waiting for it;
40 // WaitDurableWriters owns the join.40 // WaitDurableWriters owns the join.
41 func (d *Daemon) cancelDurableMaintenance() {41 func (d *Daemon) cancelDurableMaintenance() {
42- d.snapshotWorkerMu.Lock()42+ d.snapshots.mu.Lock()
43- defer d.snapshotWorkerMu.Unlock()43+ defer d.snapshots.mu.Unlock()
44 d.cancelDurableMaintenanceLocked()44 d.cancelDurableMaintenanceLocked()
45 }45 }
46 46
47-// cancelDurableMaintenanceLocked requires snapshotWorkerMu.47+// cancelDurableMaintenanceLocked requires snapshotWorker.mu (d.snapshots.mu).
48 func (d *Daemon) cancelDurableMaintenanceLocked() {48 func (d *Daemon) cancelDurableMaintenanceLocked() {
49- if d.maintenanceWorkerCancel != nil {49+ if d.snapshots.maintenanceWorkerCancel != nil {
50- d.maintenanceWorkerCancel()50+ d.snapshots.maintenanceWorkerCancel()
51 }51 }
52 }52 }
53 53
M internal/usecase/daemon/maintenance_test.go +3 -3
@@ -78,7 +78,7 @@ func TestSnapshotGarbageCollectionRequiresExplicitOption(t *testing.T) {
7878 WithRecoveryCoordinator(coordinator)(d)
7979 d.startDurableMaintenance()
8080
81- d.snapshotWorkerMu.Lock()
82- defer d.snapshotWorkerMu.Unlock()
83- require.Nil(t, d.maintenanceWorkerDone)
81+ d.snapshots.mu.Lock()
82+ defer d.snapshots.mu.Unlock()
83+ require.Nil(t, d.snapshots.maintenanceWorkerDone)
8484 }
@@ -78,7 +78,7 @@ func TestSnapshotGarbageCollectionRequiresExplicitOption(t *testing.T) {
78 WithRecoveryCoordinator(coordinator)(d)78 WithRecoveryCoordinator(coordinator)(d)
79 d.startDurableMaintenance()79 d.startDurableMaintenance()
80 80
81- d.snapshotWorkerMu.Lock()81+ d.snapshots.mu.Lock()
82- defer d.snapshotWorkerMu.Unlock()82+ defer d.snapshots.mu.Unlock()
83- require.Nil(t, d.maintenanceWorkerDone)83+ require.Nil(t, d.snapshots.maintenanceWorkerDone)
84 }84 }
M internal/usecase/daemon/move_persistence_test.go +3 -3
@@ -183,14 +183,14 @@ func TestMoveTabPersistenceMatrix(t *testing.T) {
183183 source.incarnation = domain.IncarnationID{1}
184184 source.ephemeral = !tt.sourceNamed
185185 source.snapEligible.Store(tt.sourceNamed)
186- source.snapshotWake = d.snapshotWake
186+ source.snapshotWake = d.snapshots.wake
187187 moved := source.tabs[0]
188188 moved.stableID = "moved-tab"
189189
190190 destination := addMoveTabTestSession(d, "destination", "destination-tab")
191191 destination.ephemeral = !tt.destinationNamed
192192 destination.snapEligible.Store(tt.destinationNamed)
193- destination.snapshotWake = d.snapshotWake
193+ destination.snapshotWake = d.snapshots.wake
194194
195195 catalogue := &movePersistenceCatalogue{
196196 records: map[string]domain.CatalogueRecord{},
@@ -251,7 +251,7 @@ func TestMoveSnapshotPublicationFailureReportsWithoutRollback(t *testing.T) {
251251 destination := addMoveTabTestSession(d, "destination", "destination-tab")
252252 destination.ephemeral = false
253253 destination.snapEligible.Store(true)
254- destination.snapshotWake = d.snapshotWake
254+ destination.snapshotWake = d.snapshots.wake
255255
256256 repository := &movePersistenceRepository{
257257 publish: errors.New("injected publication failure"),
@@ -183,14 +183,14 @@ func TestMoveTabPersistenceMatrix(t *testing.T) {
183 source.incarnation = domain.IncarnationID{1}183 source.incarnation = domain.IncarnationID{1}
184 source.ephemeral = !tt.sourceNamed184 source.ephemeral = !tt.sourceNamed
185 source.snapEligible.Store(tt.sourceNamed)185 source.snapEligible.Store(tt.sourceNamed)
186- source.snapshotWake = d.snapshotWake186+ source.snapshotWake = d.snapshots.wake
187 moved := source.tabs[0]187 moved := source.tabs[0]
188 moved.stableID = "moved-tab"188 moved.stableID = "moved-tab"
189 189
190 destination := addMoveTabTestSession(d, "destination", "destination-tab")190 destination := addMoveTabTestSession(d, "destination", "destination-tab")
191 destination.ephemeral = !tt.destinationNamed191 destination.ephemeral = !tt.destinationNamed
192 destination.snapEligible.Store(tt.destinationNamed)192 destination.snapEligible.Store(tt.destinationNamed)
193- destination.snapshotWake = d.snapshotWake193+ destination.snapshotWake = d.snapshots.wake
194 194
195 catalogue := &movePersistenceCatalogue{195 catalogue := &movePersistenceCatalogue{
196 records: map[string]domain.CatalogueRecord{},196 records: map[string]domain.CatalogueRecord{},
@@ -251,7 +251,7 @@ func TestMoveSnapshotPublicationFailureReportsWithoutRollback(t *testing.T) {
251 destination := addMoveTabTestSession(d, "destination", "destination-tab")251 destination := addMoveTabTestSession(d, "destination", "destination-tab")
252 destination.ephemeral = false252 destination.ephemeral = false
253 destination.snapEligible.Store(true)253 destination.snapEligible.Store(true)
254- destination.snapshotWake = d.snapshotWake254+ destination.snapshotWake = d.snapshots.wake
255 255
256 repository := &movePersistenceRepository{256 repository := &movePersistenceRepository{
257 publish: errors.New("injected publication failure"),257 publish: errors.New("injected publication failure"),
M internal/usecase/daemon/performance_benchmark_test.go +3 -3
@@ -180,9 +180,9 @@ func TestPerformanceFixtureExplicitCloseReleasesIterationState(t *testing.T) {
180180 }
181181 d.mu.Unlock()
182182 require.NotContains(t, sessionIDs, sess.id)
183- d.snapshotWorkerMu.Lock()
184- workerCancel := d.snapshotWorkerCancel
185- d.snapshotWorkerMu.Unlock()
183+ d.snapshots.mu.Lock()
184+ workerCancel := d.snapshots.cancel
185+ d.snapshots.mu.Unlock()
186186 require.Nil(t, workerCancel)
187187 }
188188
@@ -180,9 +180,9 @@ func TestPerformanceFixtureExplicitCloseReleasesIterationState(t *testing.T) {
180 }180 }
181 d.mu.Unlock()181 d.mu.Unlock()
182 require.NotContains(t, sessionIDs, sess.id)182 require.NotContains(t, sessionIDs, sess.id)
183- d.snapshotWorkerMu.Lock()183+ d.snapshots.mu.Lock()
184- workerCancel := d.snapshotWorkerCancel184+ workerCancel := d.snapshots.cancel
185- d.snapshotWorkerMu.Unlock()185+ d.snapshots.mu.Unlock()
186 require.Nil(t, workerCancel)186 require.Nil(t, workerCancel)
187 }187 }
188 188
M internal/usecase/daemon/session.go +1 -1
@@ -380,7 +380,7 @@ func (d *Daemon) createSessionLockedWithModeAndInactiveFence(name string, epheme
380380 tabs: tabs,
381381 cwd: cwd,
382382 env: env,
383- snapshotWake: d.snapshotWake,
383+ snapshotWake: d.snapshots.wake,
384384 }
385385 if !ephemeral && name != "" {
386386 sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)
@@ -380,7 +380,7 @@ func (d *Daemon) createSessionLockedWithModeAndInactiveFence(name string, epheme
380 tabs: tabs,380 tabs: tabs,
381 cwd: cwd,381 cwd: cwd,
382 env: env,382 env: env,
383- snapshotWake: d.snapshotWake,383+ snapshotWake: d.snapshots.wake,
384 }384 }
385 if !ephemeral && name != "" {385 if !ephemeral && name != "" {
386 sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)386 sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)
M internal/usecase/daemon/snapshot_coordinator.go +3 -3
@@ -247,9 +247,9 @@ func (d *Daemon) finishSnapshotCapture(capture *snapshotCapture, succeeded bool)
247247 wake := capture.session.snapshotWake
248248 capture.session.snapshotMu.Unlock()
249249 if capture.normalWorkerAdmitted {
250- d.snapshotWorkerMu.Lock()
251- delete(d.snapshotAdmitted, capture)
252- d.snapshotWorkerMu.Unlock()
250+ d.snapshots.mu.Lock()
251+ delete(d.snapshots.admitted, capture)
252+ d.snapshots.mu.Unlock()
253253 }
254254 if wake != nil {
255255 select {
@@ -247,9 +247,9 @@ func (d *Daemon) finishSnapshotCapture(capture *snapshotCapture, succeeded bool)
247 wake := capture.session.snapshotWake247 wake := capture.session.snapshotWake
248 capture.session.snapshotMu.Unlock()248 capture.session.snapshotMu.Unlock()
249 if capture.normalWorkerAdmitted {249 if capture.normalWorkerAdmitted {
250- d.snapshotWorkerMu.Lock()250+ d.snapshots.mu.Lock()
251- delete(d.snapshotAdmitted, capture)251+ delete(d.snapshots.admitted, capture)
252- d.snapshotWorkerMu.Unlock()252+ d.snapshots.mu.Unlock()
253 }253 }
254 if wake != nil {254 if wake != nil {
255 select {255 select {
M internal/usecase/daemon/snapshot_queue.go +19 -19
@@ -102,18 +102,18 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo
102102 // queue send. The queue is deliberately never closed: producers can race
103103 // shutdown without risking a send-on-closed panic.
104104 func (d *Daemon) enqueueSnapshotCapture(capture *snapshotCapture) bool {
105- d.snapshotWorkerMu.Lock()
106- defer d.snapshotWorkerMu.Unlock()
107- if d.snapshotWorkerClosing || d.snapshotWorkerCancel == nil || d.snapshotWorkerCtx == nil || d.snapshotWorkerCtx.Err() != nil {
105+ d.snapshots.mu.Lock()
106+ defer d.snapshots.mu.Unlock()
107+ if d.snapshots.closing || d.snapshots.cancel == nil || d.snapshots.ctx == nil || d.snapshots.ctx.Err() != nil {
108108 return false
109109 }
110110 capture.normalWorkerAdmitted = true
111111 select {
112- case d.snapshotJobs <- capture:
113- if d.snapshotAdmitted == nil {
114- d.snapshotAdmitted = make(map[*snapshotCapture]struct{})
112+ case d.snapshots.jobs <- capture:
113+ if d.snapshots.admitted == nil {
114+ d.snapshots.admitted = make(map[*snapshotCapture]struct{})
115115 }
116- d.snapshotAdmitted[capture] = struct{}{}
116+ d.snapshots.admitted[capture] = struct{}{}
117117 return true
118118 default:
119119 capture.normalWorkerAdmitted = false
@@ -129,17 +129,17 @@ func (d *Daemon) enqueueFinalSnapshotCapture(capture *snapshotCapture) bool {
129129 if capture == nil || capture.session == nil {
130130 return false
131131 }
132- d.snapshotWorkerMu.Lock()
133- if d.snapshotWorkerClosing || d.snapshotWorkerCancel == nil || d.snapshotWorkerCtx == nil || d.snapshotWorkerCtx.Err() != nil {
134- d.snapshotWorkerMu.Unlock()
132+ d.snapshots.mu.Lock()
133+ if d.snapshots.closing || d.snapshots.cancel == nil || d.snapshots.ctx == nil || d.snapshots.ctx.Err() != nil {
134+ d.snapshots.mu.Unlock()
135135 return false
136136 }
137- if d.snapshotFinalJobs == nil {
138- d.snapshotFinalJobs = make(map[*session]*snapshotCapture)
137+ if d.snapshots.finalJobs == nil {
138+ d.snapshots.finalJobs = make(map[*session]*snapshotCapture)
139139 }
140- replaced, exists := d.snapshotFinalJobs[capture.session]
141- if !exists && len(d.snapshotFinalJobs) >= snapshotFinalQueueCapacity {
142- d.snapshotWorkerMu.Unlock()
140+ replaced, exists := d.snapshots.finalJobs[capture.session]
141+ if !exists && len(d.snapshots.finalJobs) >= snapshotFinalQueueCapacity {
142+ d.snapshots.mu.Unlock()
143143 d.log.Warn("terminal snapshot retention saturated; capture rejected", "session", capture.name, "capacity", snapshotFinalQueueCapacity)
144144 return false
145145 }
@@ -148,13 +148,13 @@ func (d *Daemon) enqueueFinalSnapshotCapture(capture *snapshotCapture) bool {
148148 // mark the session dirty again.
149149 d.finishSnapshotCapture(replaced, false)
150150 if !exists {
151- d.snapshotFinalOrder = append(d.snapshotFinalOrder, capture.session)
151+ d.snapshots.finalOrder = append(d.snapshots.finalOrder, capture.session)
152152 }
153- d.snapshotFinalJobs[capture.session] = capture
153+ d.snapshots.finalJobs[capture.session] = capture
154154 select {
155- case d.snapshotWorkerFinalWake <- struct{}{}:
155+ case d.snapshots.finalWake <- struct{}{}:
156156 default:
157157 }
158- d.snapshotWorkerMu.Unlock()
158+ d.snapshots.mu.Unlock()
159159 return true
160160 }
@@ -102,18 +102,18 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo
102 // queue send. The queue is deliberately never closed: producers can race102 // queue send. The queue is deliberately never closed: producers can race
103 // shutdown without risking a send-on-closed panic.103 // shutdown without risking a send-on-closed panic.
104 func (d *Daemon) enqueueSnapshotCapture(capture *snapshotCapture) bool {104 func (d *Daemon) enqueueSnapshotCapture(capture *snapshotCapture) bool {
105- d.snapshotWorkerMu.Lock()105+ d.snapshots.mu.Lock()
106- defer d.snapshotWorkerMu.Unlock()106+ defer d.snapshots.mu.Unlock()
107- if d.snapshotWorkerClosing || d.snapshotWorkerCancel == nil || d.snapshotWorkerCtx == nil || d.snapshotWorkerCtx.Err() != nil {107+ if d.snapshots.closing || d.snapshots.cancel == nil || d.snapshots.ctx == nil || d.snapshots.ctx.Err() != nil {
108 return false108 return false
109 }109 }
110 capture.normalWorkerAdmitted = true110 capture.normalWorkerAdmitted = true
111 select {111 select {
112- case d.snapshotJobs <- capture:112+ case d.snapshots.jobs <- capture:
113- if d.snapshotAdmitted == nil {113+ if d.snapshots.admitted == nil {
114- d.snapshotAdmitted = make(map[*snapshotCapture]struct{})114+ d.snapshots.admitted = make(map[*snapshotCapture]struct{})
115 }115 }
116- d.snapshotAdmitted[capture] = struct{}{}116+ d.snapshots.admitted[capture] = struct{}{}
117 return true117 return true
118 default:118 default:
119 capture.normalWorkerAdmitted = false119 capture.normalWorkerAdmitted = false
@@ -129,17 +129,17 @@ func (d *Daemon) enqueueFinalSnapshotCapture(capture *snapshotCapture) bool {
129 if capture == nil || capture.session == nil {129 if capture == nil || capture.session == nil {
130 return false130 return false
131 }131 }
132- d.snapshotWorkerMu.Lock()132+ d.snapshots.mu.Lock()
133- if d.snapshotWorkerClosing || d.snapshotWorkerCancel == nil || d.snapshotWorkerCtx == nil || d.snapshotWorkerCtx.Err() != nil {133+ if d.snapshots.closing || d.snapshots.cancel == nil || d.snapshots.ctx == nil || d.snapshots.ctx.Err() != nil {
134- d.snapshotWorkerMu.Unlock()134+ d.snapshots.mu.Unlock()
135 return false135 return false
136 }136 }
137- if d.snapshotFinalJobs == nil {137+ if d.snapshots.finalJobs == nil {
138- d.snapshotFinalJobs = make(map[*session]*snapshotCapture)138+ d.snapshots.finalJobs = make(map[*session]*snapshotCapture)
139 }139 }
140- replaced, exists := d.snapshotFinalJobs[capture.session]140+ replaced, exists := d.snapshots.finalJobs[capture.session]
141- if !exists && len(d.snapshotFinalJobs) >= snapshotFinalQueueCapacity {141+ if !exists && len(d.snapshots.finalJobs) >= snapshotFinalQueueCapacity {
142- d.snapshotWorkerMu.Unlock()142+ d.snapshots.mu.Unlock()
143 d.log.Warn("terminal snapshot retention saturated; capture rejected", "session", capture.name, "capacity", snapshotFinalQueueCapacity)143 d.log.Warn("terminal snapshot retention saturated; capture rejected", "session", capture.name, "capacity", snapshotFinalQueueCapacity)
144 return false144 return false
145 }145 }
@@ -148,13 +148,13 @@ func (d *Daemon) enqueueFinalSnapshotCapture(capture *snapshotCapture) bool {
148 // mark the session dirty again.148 // mark the session dirty again.
149 d.finishSnapshotCapture(replaced, false)149 d.finishSnapshotCapture(replaced, false)
150 if !exists {150 if !exists {
151- d.snapshotFinalOrder = append(d.snapshotFinalOrder, capture.session)151+ d.snapshots.finalOrder = append(d.snapshots.finalOrder, capture.session)
152 }152 }
153- d.snapshotFinalJobs[capture.session] = capture153+ d.snapshots.finalJobs[capture.session] = capture
154 select {154 select {
155- case d.snapshotWorkerFinalWake <- struct{}{}:155+ case d.snapshots.finalWake <- struct{}{}:
156 default:156 default:
157 }157 }
158- d.snapshotWorkerMu.Unlock()158+ d.snapshots.mu.Unlock()
159 return true159 return true
160 }160 }
M internal/usecase/daemon/snapshot_restore.go +6 -6
@@ -45,14 +45,14 @@ func (d *Daemon) startSnapshotRestoration() {
4545 if d == nil {
4646 return
4747 }
48- d.snapshotWorkerMu.Lock()
49- if d.restoreWorkerDone != nil {
50- d.snapshotWorkerMu.Unlock()
48+ d.snapshots.mu.Lock()
49+ if d.snapshots.restoreWorkerDone != nil {
50+ d.snapshots.mu.Unlock()
5151 return
5252 }
5353 done := make(chan struct{})
54- d.restoreWorkerDone = done
55- d.snapshotWorkerMu.Unlock()
54+ d.snapshots.restoreWorkerDone = done
55+ d.snapshots.mu.Unlock()
5656 d.sessWg.Go(func() {
5757 defer close(done)
5858 d.restoreSnapshots(d.serveCtx)
@@ -292,7 +292,7 @@ func (d *Daemon) restoreSnapshotPane(ctx context.Context, sessionName, tabStable
292292 func (d *Daemon) newRestoredSession(snap snapcodec.Session, sctx context.Context, cancel context.CancelFunc, tabs []*tab) *session {
293293 // Restored sessions keep ordered tabs; attachment repair deterministically
294294 // selects the first tab instead of restoring an interactive client view.
295- sess := &session{sessionCore: sessionCore{name: snap.Name, createdAt: int64(snap.CreatedAt)}, ctx: sctx, cancel: cancel, tabs: tabs, env: copyEnvironment(d.baseEnv), envProvisional: true, snapshotWake: d.snapshotWake, snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit)}
295+ sess := &session{sessionCore: sessionCore{name: snap.Name, createdAt: int64(snap.CreatedAt)}, ctx: sctx, cancel: cancel, tabs: tabs, env: copyEnvironment(d.baseEnv), envProvisional: true, snapshotWake: d.snapshots.wake, snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit)}
296296 // Restored tabs remain private until persistAndRegisterRestoredSession.
297297 // Initialize owners now so registration and reader startup cannot expose an
298298 // ownerless pane.
@@ -45,14 +45,14 @@ func (d *Daemon) startSnapshotRestoration() {
45 if d == nil {45 if d == nil {
46 return46 return
47 }47 }
48- d.snapshotWorkerMu.Lock()48+ d.snapshots.mu.Lock()
49- if d.restoreWorkerDone != nil {49+ if d.snapshots.restoreWorkerDone != nil {
50- d.snapshotWorkerMu.Unlock()50+ d.snapshots.mu.Unlock()
51 return51 return
52 }52 }
53 done := make(chan struct{})53 done := make(chan struct{})
54- d.restoreWorkerDone = done54+ d.snapshots.restoreWorkerDone = done
55- d.snapshotWorkerMu.Unlock()55+ d.snapshots.mu.Unlock()
56 d.sessWg.Go(func() {56 d.sessWg.Go(func() {
57 defer close(done)57 defer close(done)
58 d.restoreSnapshots(d.serveCtx)58 d.restoreSnapshots(d.serveCtx)
@@ -292,7 +292,7 @@ func (d *Daemon) restoreSnapshotPane(ctx context.Context, sessionName, tabStable
292 func (d *Daemon) newRestoredSession(snap snapcodec.Session, sctx context.Context, cancel context.CancelFunc, tabs []*tab) *session {292 func (d *Daemon) newRestoredSession(snap snapcodec.Session, sctx context.Context, cancel context.CancelFunc, tabs []*tab) *session {
293 // Restored sessions keep ordered tabs; attachment repair deterministically293 // Restored sessions keep ordered tabs; attachment repair deterministically
294 // selects the first tab instead of restoring an interactive client view.294 // selects the first tab instead of restoring an interactive client view.
295- sess := &session{sessionCore: sessionCore{name: snap.Name, createdAt: int64(snap.CreatedAt)}, ctx: sctx, cancel: cancel, tabs: tabs, env: copyEnvironment(d.baseEnv), envProvisional: true, snapshotWake: d.snapshotWake, snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit)}295+ sess := &session{sessionCore: sessionCore{name: snap.Name, createdAt: int64(snap.CreatedAt)}, ctx: sctx, cancel: cancel, tabs: tabs, env: copyEnvironment(d.baseEnv), envProvisional: true, snapshotWake: d.snapshots.wake, snapshotChunkCache: newSnapshotChunkCache(snapshotChunkCacheLimit)}
296 // Restored tabs remain private until persistAndRegisterRestoredSession.296 // Restored tabs remain private until persistAndRegisterRestoredSession.
297 // Initialize owners now so registration and reader startup cannot expose an297 // Initialize owners now so registration and reader startup cannot expose an
298 // ownerless pane.298 // ownerless pane.
M internal/usecase/daemon/snapshot_restore_test.go +3 -3
@@ -529,9 +529,9 @@ func TestRestoredSessionMetadataUpdatePreservesCheckpointLineage(t *testing.T) {
529529 markSnapshotDirty(restored)
530530 require.True(t, d.scheduleSnapshot(restored))
531531 awaitSnapshotIdle(t, restored)
532- d.snapshotNoticeMu.Lock()
533- failure := d.snapshotActiveFailureSignature
534- d.snapshotNoticeMu.Unlock()
532+ d.snapshots.noticeMu.Lock()
533+ failure := d.snapshots.activeFailureSignature
534+ d.snapshots.noticeMu.Unlock()
535535 require.Empty(t, failure)
536536 require.Len(t, repository.publishes, 1)
537537 require.Equal(t, record.Committed, repository.publishes[0].ParentCheckpoint)
@@ -529,9 +529,9 @@ func TestRestoredSessionMetadataUpdatePreservesCheckpointLineage(t *testing.T) {
529 markSnapshotDirty(restored)529 markSnapshotDirty(restored)
530 require.True(t, d.scheduleSnapshot(restored))530 require.True(t, d.scheduleSnapshot(restored))
531 awaitSnapshotIdle(t, restored)531 awaitSnapshotIdle(t, restored)
532- d.snapshotNoticeMu.Lock()532+ d.snapshots.noticeMu.Lock()
533- failure := d.snapshotActiveFailureSignature533+ failure := d.snapshots.activeFailureSignature
534- d.snapshotNoticeMu.Unlock()534+ d.snapshots.noticeMu.Unlock()
535 require.Empty(t, failure)535 require.Empty(t, failure)
536 require.Len(t, repository.publishes, 1)536 require.Len(t, repository.publishes, 1)
537 require.Equal(t, record.Committed, repository.publishes[0].ParentCheckpoint)537 require.Equal(t, record.Committed, repository.publishes[0].ParentCheckpoint)
M internal/usecase/daemon/snapshot_resume_publication_test.go +3 -3
@@ -204,9 +204,9 @@ func TestStoppedNamedSessionResumePublishesNextCheckpointInSameDaemon(t *testing
204204 require.Len(t, attempts, 3, "the resumed checkpoint must reach the repository after the initial and terminal checkpoints")
205205 require.NoError(t, publicationErrors[2])
206206 require.False(t, resumed.snapDirty.Load(), "the published resumed checkpoint must clear dirty state")
207- d.snapshotNoticeMu.Lock()
208- failure := d.snapshotActiveFailureSignature
209- d.snapshotNoticeMu.Unlock()
207+ d.snapshots.noticeMu.Lock()
208+ failure := d.snapshots.activeFailureSignature
209+ d.snapshots.noticeMu.Unlock()
210210 require.Empty(t, failure, "successful resumed publication must not retain a failure signature")
211211 require.Equal(t, closedRecord.Committed.Generation+1, resumedRecord.Committed.Generation,
212212 "same-daemon resume must publish exactly committed generation+1")
@@ -204,9 +204,9 @@ func TestStoppedNamedSessionResumePublishesNextCheckpointInSameDaemon(t *testing
204 require.Len(t, attempts, 3, "the resumed checkpoint must reach the repository after the initial and terminal checkpoints")204 require.Len(t, attempts, 3, "the resumed checkpoint must reach the repository after the initial and terminal checkpoints")
205 require.NoError(t, publicationErrors[2])205 require.NoError(t, publicationErrors[2])
206 require.False(t, resumed.snapDirty.Load(), "the published resumed checkpoint must clear dirty state")206 require.False(t, resumed.snapDirty.Load(), "the published resumed checkpoint must clear dirty state")
207- d.snapshotNoticeMu.Lock()207+ d.snapshots.noticeMu.Lock()
208- failure := d.snapshotActiveFailureSignature208+ failure := d.snapshots.activeFailureSignature
209- d.snapshotNoticeMu.Unlock()209+ d.snapshots.noticeMu.Unlock()
210 require.Empty(t, failure, "successful resumed publication must not retain a failure signature")210 require.Empty(t, failure, "successful resumed publication must not retain a failure signature")
211 require.Equal(t, closedRecord.Committed.Generation+1, resumedRecord.Committed.Generation,211 require.Equal(t, closedRecord.Committed.Generation+1, resumedRecord.Committed.Generation,
212 "same-daemon resume must publish exactly committed generation+1")212 "same-daemon resume must publish exactly committed generation+1")
M internal/usecase/daemon/snapshot_scheduler.go +1 -1
@@ -17,7 +17,7 @@ func (d *Daemon) snapshotRepositorySaver(ctx context.Context) {
1717 select {
1818 case <-ctx.Done():
1919 return
20- case <-d.snapshotWake:
20+ case <-d.snapshots.wake:
2121 case <-timer.C():
2222 }
2323 }
@@ -17,7 +17,7 @@ func (d *Daemon) snapshotRepositorySaver(ctx context.Context) {
17 select {17 select {
18 case <-ctx.Done():18 case <-ctx.Done():
19 return19 return
20- case <-d.snapshotWake:20+ case <-d.snapshots.wake:
21 case <-timer.C():21 case <-timer.C():
22 }22 }
23 }23 }
M internal/usecase/daemon/snapshot_scheduler_test.go +18 -18
@@ -135,7 +135,7 @@ func TestQuarantineDiscardDoesNotDoubleDecrementPendingCaptures(t *testing.T) {
135135 // Model the worker having already dequeued the stale capture: the global
136136 // bounded queue still holds it, so free the slot before the replacement.
137137 select {
138- case dequeued := <-d.snapshotJobs:
138+ case dequeued := <-d.snapshots.jobs:
139139 require.Same(t, first, dequeued)
140140 default:
141141 t.Fatal("the discarded capture must still be queued in the global worker queue")
@@ -356,9 +356,9 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
356356 require.NotNil(t, sess.snapshotQueuedCapture, "the one routine capture remains retryable")
357357 require.Nil(t, sess.snapshotInFlightCapture)
358358 }()
359- d.snapshotWorkerMu.Lock()
360- require.LessOrEqual(t, len(d.snapshotJobs), snapshotQueueCapacity)
361- d.snapshotWorkerMu.Unlock()
359+ d.snapshots.mu.Lock()
360+ require.LessOrEqual(t, len(d.snapshots.jobs), snapshotQueueCapacity)
361+ d.snapshots.mu.Unlock()
362362
363363 // The retained capture is not published after quarantine, but releasing the
364364 // unrelated blocked call lets the worker discard it and lets cleanup join.
@@ -572,7 +572,7 @@ func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {
572572
573573 awaitFrame(t, sends, "Welcome")
574574 markSnapshotDirty(firstSession(d))
575- d.snapshotWake <- struct{}{}
575+ d.snapshots.wake <- struct{}{}
576576 first := <-published
577577 require.Equal(t, uint64(1), first.Generation)
578578
@@ -629,7 +629,7 @@ func TestServeShutdownDeadlineStillJoinsUncooperativeSnapshotRepository(t *testi
629629
630630 awaitFrame(t, sends, "Welcome")
631631 markSnapshotDirty(firstSession(d))
632- d.snapshotWake <- struct{}{}
632+ d.snapshots.wake <- struct{}{}
633633 select {
634634 case <-firstPublicationCompleted:
635635 case <-time.After(testWaitTimeout):
@@ -642,9 +642,9 @@ func TestServeShutdownDeadlineStillJoinsUncooperativeSnapshotRepository(t *testi
642642 case <-time.After(testWaitTimeout):
643643 t.Fatal("shutdown terminal checkpoint did not reach the worker")
644644 }
645- d.snapshotWorkerMu.Lock()
646- workerDone := d.snapshotWorkerDone
647- d.snapshotWorkerMu.Unlock()
645+ d.snapshots.mu.Lock()
646+ workerDone := d.snapshots.done
647+ d.snapshots.mu.Unlock()
648648 require.NotNil(t, workerDone)
649649 clock.nextFinalTimer(t).fire()
650650 select {
@@ -926,7 +926,7 @@ func TestSnapshotSchedulerImmediateEligibilityAndStaleCapturesRemainDirty(t *tes
926926 }).Once()
927927
928928 sess := newSnapshotTestSession(t, "immediate", false, "/work")
929- sess.snapshotWake = d.snapshotWake
929+ sess.snapshotWake = d.snapshots.wake
930930 sess.snapDirty.Store(true)
931931 d.sessions[sess.id] = sess
932932
@@ -1000,10 +1000,10 @@ func TestSnapshotWorkerQueueIsBoundedAndForcedCapturesCoalesce(t *testing.T) {
10001000 markSnapshotDirty(queued)
10011001 require.True(t, d.scheduleSnapshot(queued))
10021002
1003- d.snapshotWorkerMu.Lock()
1004- require.NotNil(t, d.snapshotWorkerInFlight)
1005- require.Len(t, d.snapshotJobs, snapshotQueueCapacity)
1006- d.snapshotWorkerMu.Unlock()
1003+ d.snapshots.mu.Lock()
1004+ require.NotNil(t, d.snapshots.inFlight)
1005+ require.Len(t, d.snapshots.jobs, snapshotQueueCapacity)
1006+ d.snapshots.mu.Unlock()
10071007
10081008 saturated := newSnapshotTestSession(t, "saturated", false, "/work")
10091009 markSnapshotDirty(saturated)
@@ -1041,7 +1041,7 @@ func TestSnapshotRepositorySaverUsesEarliestDeadlineAndRecomputes(t *testing.T)
10411041 }).Once()
10421042
10431043 later := newSnapshotTestSession(t, "later", false, "/work")
1044- later.snapshotWake = d.snapshotWake
1044+ later.snapshotWake = d.snapshots.wake
10451045 later.snapDirty.Store(true)
10461046 later.snapshotNextEligibleAt = now.Add(10 * time.Minute)
10471047 d.sessions[later.id] = later
@@ -1056,11 +1056,11 @@ func TestSnapshotRepositorySaverUsesEarliestDeadlineAndRecomputes(t *testing.T)
10561056 require.Equal(t, 10*time.Minute, <-timer.resets)
10571057
10581058 earlier := newSnapshotTestSession(t, "earlier", false, "/work")
1059- earlier.snapshotWake = d.snapshotWake
1059+ earlier.snapshotWake = d.snapshots.wake
10601060 earlier.snapDirty.Store(true)
10611061 earlier.snapshotNextEligibleAt = now.Add(5 * time.Minute)
10621062 d.sessions[earlier.id] = earlier
1063- d.snapshotWake <- struct{}{}
1063+ d.snapshots.wake <- struct{}{}
10641064 require.Equal(t, 5*time.Minute, <-timer.resets)
10651065
10661066 clock.advance(now.Add(5 * time.Minute))
@@ -1083,7 +1083,7 @@ func TestSnapshotRepositorySaverUsesEarliestDeadlineAndRecomputes(t *testing.T)
10831083 require.Equal(t, 5*time.Minute, <-timer.resets)
10841084
10851085 delete(d.sessions, later.id)
1086- d.snapshotWake <- struct{}{}
1086+ d.snapshots.wake <- struct{}{}
10871087 require.Equal(t, 24*time.Hour, <-timer.resets)
10881088
10891089 cancel()
@@ -135,7 +135,7 @@ func TestQuarantineDiscardDoesNotDoubleDecrementPendingCaptures(t *testing.T) {
135 // Model the worker having already dequeued the stale capture: the global135 // Model the worker having already dequeued the stale capture: the global
136 // bounded queue still holds it, so free the slot before the replacement.136 // bounded queue still holds it, so free the slot before the replacement.
137 select {137 select {
138- case dequeued := <-d.snapshotJobs:138+ case dequeued := <-d.snapshots.jobs:
139 require.Same(t, first, dequeued)139 require.Same(t, first, dequeued)
140 default:140 default:
141 t.Fatal("the discarded capture must still be queued in the global worker queue")141 t.Fatal("the discarded capture must still be queued in the global worker queue")
@@ -356,9 +356,9 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
356 require.NotNil(t, sess.snapshotQueuedCapture, "the one routine capture remains retryable")356 require.NotNil(t, sess.snapshotQueuedCapture, "the one routine capture remains retryable")
357 require.Nil(t, sess.snapshotInFlightCapture)357 require.Nil(t, sess.snapshotInFlightCapture)
358 }()358 }()
359- d.snapshotWorkerMu.Lock()359+ d.snapshots.mu.Lock()
360- require.LessOrEqual(t, len(d.snapshotJobs), snapshotQueueCapacity)360+ require.LessOrEqual(t, len(d.snapshots.jobs), snapshotQueueCapacity)
361- d.snapshotWorkerMu.Unlock()361+ d.snapshots.mu.Unlock()
362 362
363 // The retained capture is not published after quarantine, but releasing the363 // The retained capture is not published after quarantine, but releasing the
364 // unrelated blocked call lets the worker discard it and lets cleanup join.364 // unrelated blocked call lets the worker discard it and lets cleanup join.
@@ -572,7 +572,7 @@ func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {
572 572
573 awaitFrame(t, sends, "Welcome")573 awaitFrame(t, sends, "Welcome")
574 markSnapshotDirty(firstSession(d))574 markSnapshotDirty(firstSession(d))
575- d.snapshotWake <- struct{}{}575+ d.snapshots.wake <- struct{}{}
576 first := <-published576 first := <-published
577 require.Equal(t, uint64(1), first.Generation)577 require.Equal(t, uint64(1), first.Generation)
578 578
@@ -629,7 +629,7 @@ func TestServeShutdownDeadlineStillJoinsUncooperativeSnapshotRepository(t *testi
629 629
630 awaitFrame(t, sends, "Welcome")630 awaitFrame(t, sends, "Welcome")
631 markSnapshotDirty(firstSession(d))631 markSnapshotDirty(firstSession(d))
632- d.snapshotWake <- struct{}{}632+ d.snapshots.wake <- struct{}{}
633 select {633 select {
634 case <-firstPublicationCompleted:634 case <-firstPublicationCompleted:
635 case <-time.After(testWaitTimeout):635 case <-time.After(testWaitTimeout):
@@ -642,9 +642,9 @@ func TestServeShutdownDeadlineStillJoinsUncooperativeSnapshotRepository(t *testi
642 case <-time.After(testWaitTimeout):642 case <-time.After(testWaitTimeout):
643 t.Fatal("shutdown terminal checkpoint did not reach the worker")643 t.Fatal("shutdown terminal checkpoint did not reach the worker")
644 }644 }
645- d.snapshotWorkerMu.Lock()645+ d.snapshots.mu.Lock()
646- workerDone := d.snapshotWorkerDone646+ workerDone := d.snapshots.done
647- d.snapshotWorkerMu.Unlock()647+ d.snapshots.mu.Unlock()
648 require.NotNil(t, workerDone)648 require.NotNil(t, workerDone)
649 clock.nextFinalTimer(t).fire()649 clock.nextFinalTimer(t).fire()
650 select {650 select {
@@ -926,7 +926,7 @@ func TestSnapshotSchedulerImmediateEligibilityAndStaleCapturesRemainDirty(t *tes
926 }).Once()926 }).Once()
927 927
928 sess := newSnapshotTestSession(t, "immediate", false, "/work")928 sess := newSnapshotTestSession(t, "immediate", false, "/work")
929- sess.snapshotWake = d.snapshotWake929+ sess.snapshotWake = d.snapshots.wake
930 sess.snapDirty.Store(true)930 sess.snapDirty.Store(true)
931 d.sessions[sess.id] = sess931 d.sessions[sess.id] = sess
932 932
@@ -1000,10 +1000,10 @@ func TestSnapshotWorkerQueueIsBoundedAndForcedCapturesCoalesce(t *testing.T) {
1000 markSnapshotDirty(queued)1000 markSnapshotDirty(queued)
1001 require.True(t, d.scheduleSnapshot(queued))1001 require.True(t, d.scheduleSnapshot(queued))
1002 1002
1003- d.snapshotWorkerMu.Lock()1003+ d.snapshots.mu.Lock()
1004- require.NotNil(t, d.snapshotWorkerInFlight)1004+ require.NotNil(t, d.snapshots.inFlight)
1005- require.Len(t, d.snapshotJobs, snapshotQueueCapacity)1005+ require.Len(t, d.snapshots.jobs, snapshotQueueCapacity)
1006- d.snapshotWorkerMu.Unlock()1006+ d.snapshots.mu.Unlock()
1007 1007
1008 saturated := newSnapshotTestSession(t, "saturated", false, "/work")1008 saturated := newSnapshotTestSession(t, "saturated", false, "/work")
1009 markSnapshotDirty(saturated)1009 markSnapshotDirty(saturated)
@@ -1041,7 +1041,7 @@ func TestSnapshotRepositorySaverUsesEarliestDeadlineAndRecomputes(t *testing.T)
1041 }).Once()1041 }).Once()
1042 1042
1043 later := newSnapshotTestSession(t, "later", false, "/work")1043 later := newSnapshotTestSession(t, "later", false, "/work")
1044- later.snapshotWake = d.snapshotWake1044+ later.snapshotWake = d.snapshots.wake
1045 later.snapDirty.Store(true)1045 later.snapDirty.Store(true)
1046 later.snapshotNextEligibleAt = now.Add(10 * time.Minute)1046 later.snapshotNextEligibleAt = now.Add(10 * time.Minute)
1047 d.sessions[later.id] = later1047 d.sessions[later.id] = later
@@ -1056,11 +1056,11 @@ func TestSnapshotRepositorySaverUsesEarliestDeadlineAndRecomputes(t *testing.T)
1056 require.Equal(t, 10*time.Minute, <-timer.resets)1056 require.Equal(t, 10*time.Minute, <-timer.resets)
1057 1057
1058 earlier := newSnapshotTestSession(t, "earlier", false, "/work")1058 earlier := newSnapshotTestSession(t, "earlier", false, "/work")
1059- earlier.snapshotWake = d.snapshotWake1059+ earlier.snapshotWake = d.snapshots.wake
1060 earlier.snapDirty.Store(true)1060 earlier.snapDirty.Store(true)
1061 earlier.snapshotNextEligibleAt = now.Add(5 * time.Minute)1061 earlier.snapshotNextEligibleAt = now.Add(5 * time.Minute)
1062 d.sessions[earlier.id] = earlier1062 d.sessions[earlier.id] = earlier
1063- d.snapshotWake <- struct{}{}1063+ d.snapshots.wake <- struct{}{}
1064 require.Equal(t, 5*time.Minute, <-timer.resets)1064 require.Equal(t, 5*time.Minute, <-timer.resets)
1065 1065
1066 clock.advance(now.Add(5 * time.Minute))1066 clock.advance(now.Add(5 * time.Minute))
@@ -1083,7 +1083,7 @@ func TestSnapshotRepositorySaverUsesEarliestDeadlineAndRecomputes(t *testing.T)
1083 require.Equal(t, 5*time.Minute, <-timer.resets)1083 require.Equal(t, 5*time.Minute, <-timer.resets)
1084 1084
1085 delete(d.sessions, later.id)1085 delete(d.sessions, later.id)
1086- d.snapshotWake <- struct{}{}1086+ d.snapshots.wake <- struct{}{}
1087 require.Equal(t, 24*time.Hour, <-timer.resets)1087 require.Equal(t, 24*time.Hour, <-timer.resets)
1088 1088
1089 cancel()1089 cancel()
M internal/usecase/daemon/snapshot_test.go +4 -4
@@ -30,10 +30,10 @@ func TestWithSnapshotRepositoryRejectsTypedNil(t *testing.T) {
3030 func TestDurableWriterFailureNamesIncludesBufferedCapture(t *testing.T) {
3131 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
3232 capture := &snapshotCapture{name: "work", session: newSnapshotTestSession(t, "work", false, "/work")}
33- d.snapshotWorkerMu.Lock()
34- d.snapshotAdmitted[capture] = struct{}{}
35- d.snapshotJobs <- capture
36- d.snapshotWorkerMu.Unlock()
33+ d.snapshots.mu.Lock()
34+ d.snapshots.admitted[capture] = struct{}{}
35+ d.snapshots.jobs <- capture
36+ d.snapshots.mu.Unlock()
3737
3838 require.Equal(t, []string{"work"}, d.durableWriterFailureNames())
3939 }
@@ -30,10 +30,10 @@ func TestWithSnapshotRepositoryRejectsTypedNil(t *testing.T) {
30 func TestDurableWriterFailureNamesIncludesBufferedCapture(t *testing.T) {30 func TestDurableWriterFailureNamesIncludesBufferedCapture(t *testing.T) {
31 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})31 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
32 capture := &snapshotCapture{name: "work", session: newSnapshotTestSession(t, "work", false, "/work")}32 capture := &snapshotCapture{name: "work", session: newSnapshotTestSession(t, "work", false, "/work")}
33- d.snapshotWorkerMu.Lock()33+ d.snapshots.mu.Lock()
34- d.snapshotAdmitted[capture] = struct{}{}34+ d.snapshots.admitted[capture] = struct{}{}
35- d.snapshotJobs <- capture35+ d.snapshots.jobs <- capture
36- d.snapshotWorkerMu.Unlock()36+ d.snapshots.mu.Unlock()
37 37
38 require.Equal(t, []string{"work"}, d.durableWriterFailureNames())38 require.Equal(t, []string{"work"}, d.durableWriterFailureNames())
39 }39 }
M internal/usecase/daemon/snapshot_worker.go +75 -75
@@ -13,10 +13,10 @@ func (d *Daemon) reportSnapshotFailure(capture *snapshotCapture, phase string, c
1313 return
1414 }
1515 signature := snapshotFailureSignature(phase, cause)
16- d.snapshotNoticeMu.Lock()
17- changed := d.snapshotActiveFailureSignature != signature
18- d.snapshotActiveFailureSignature = signature
19- d.snapshotNoticeMu.Unlock()
16+ d.snapshots.noticeMu.Lock()
17+ changed := d.snapshots.activeFailureSignature != signature
18+ d.snapshots.activeFailureSignature = signature
19+ d.snapshots.noticeMu.Unlock()
2020
2121 n := domain.Notification{
2222 Code: domain.NoticeSnapshotWrite,
@@ -38,38 +38,38 @@ func (d *Daemon) clearSnapshotFailure() {
3838 if d == nil {
3939 return
4040 }
41- d.snapshotNoticeMu.Lock()
42- d.snapshotActiveFailureSignature = ""
43- d.snapshotNoticeMu.Unlock()
41+ d.snapshots.noticeMu.Lock()
42+ d.snapshots.activeFailureSignature = ""
43+ d.snapshots.noticeMu.Unlock()
4444 }
4545
4646 func (d *Daemon) startSnapshotEncodeWorker() {
4747 if d == nil {
4848 return
4949 }
50- d.snapshotWorkerMu.Lock()
51- if d.snapshotWorkerCancel != nil {
52- d.snapshotWorkerMu.Unlock()
50+ d.snapshots.mu.Lock()
51+ if d.snapshots.cancel != nil {
52+ d.snapshots.mu.Unlock()
5353 return
5454 }
5555 // Shutdown owns the worker lifetime so it can flush captures after Serve
5656 // cancels its parent context. The worker is always stopped explicitly.
5757 workerCtx, cancel := context.WithCancel(context.Background())
58- d.snapshotWorkerID++
59- workerID := d.snapshotWorkerID
60- d.snapshotWorkerCtx = workerCtx
61- d.snapshotWorkerCancel = cancel
62- d.snapshotWorkerDone = make(chan struct{})
63- d.snapshotWorkerFlush = make(chan struct{})
64- d.snapshotWorkerFinalWake = make(chan struct{}, 1)
65- d.snapshotFinalJobs = make(map[*session]*snapshotCapture)
66- d.snapshotFinalOrder = nil
67- d.snapshotWorkerClosing = false
68- d.snapshotWorkerInFlight = nil
69- done := d.snapshotWorkerDone
70- flush := d.snapshotWorkerFlush
71- finalWake := d.snapshotWorkerFinalWake
72- d.snapshotWorkerMu.Unlock()
58+ d.snapshots.id++
59+ workerID := d.snapshots.id
60+ d.snapshots.ctx = workerCtx
61+ d.snapshots.cancel = cancel
62+ d.snapshots.done = make(chan struct{})
63+ d.snapshots.flush = make(chan struct{})
64+ d.snapshots.finalWake = make(chan struct{}, 1)
65+ d.snapshots.finalJobs = make(map[*session]*snapshotCapture)
66+ d.snapshots.finalOrder = nil
67+ d.snapshots.closing = false
68+ d.snapshots.inFlight = nil
69+ done := d.snapshots.done
70+ flush := d.snapshots.flush
71+ finalWake := d.snapshots.finalWake
72+ d.snapshots.mu.Unlock()
7373 go d.runSnapshotEncodeWorker(workerCtx, workerID, done, flush, finalWake)
7474 }
7575
@@ -89,7 +89,7 @@ func (d *Daemon) runSnapshotEncodeWorker(workerCtx context.Context, workerID uin
8989 if !d.drainFinalSnapshotCaptures(workerCtx, workerID) {
9090 return
9191 }
92- case capture := <-d.snapshotJobs:
92+ case capture := <-d.snapshots.jobs:
9393 if !d.publishSnapshotCapture(workerCtx, workerID, capture) {
9494 return
9595 }
@@ -100,7 +100,7 @@ func (d *Daemon) runSnapshotEncodeWorker(workerCtx context.Context, workerID uin
100100 func (d *Daemon) flushSnapshotCaptures(workerCtx context.Context, workerID uint64) {
101101 for {
102102 select {
103- case capture := <-d.snapshotJobs:
103+ case capture := <-d.snapshots.jobs:
104104 if !d.publishSnapshotCapture(workerCtx, workerID, capture) {
105105 return
106106 }
@@ -162,14 +162,14 @@ func (d *Daemon) publishSnapshotCapture(workerCtx context.Context, workerID uint
162162 }
163163
164164 func (d *Daemon) takeFinalSnapshotCapture() *snapshotCapture {
165- d.snapshotWorkerMu.Lock()
166- defer d.snapshotWorkerMu.Unlock()
167- for len(d.snapshotFinalOrder) > 0 {
168- sess := d.snapshotFinalOrder[0]
169- d.snapshotFinalOrder[0] = nil
170- d.snapshotFinalOrder = d.snapshotFinalOrder[1:]
171- capture := d.snapshotFinalJobs[sess]
172- delete(d.snapshotFinalJobs, sess)
165+ d.snapshots.mu.Lock()
166+ defer d.snapshots.mu.Unlock()
167+ for len(d.snapshots.finalOrder) > 0 {
168+ sess := d.snapshots.finalOrder[0]
169+ d.snapshots.finalOrder[0] = nil
170+ d.snapshots.finalOrder = d.snapshots.finalOrder[1:]
171+ capture := d.snapshots.finalJobs[sess]
172+ delete(d.snapshots.finalJobs, sess)
173173 if capture != nil {
174174 return capture
175175 }
@@ -178,20 +178,20 @@ func (d *Daemon) takeFinalSnapshotCapture() *snapshotCapture {
178178 }
179179
180180 func (d *Daemon) setSnapshotWorkerInFlight(workerID uint64, capture *snapshotCapture) bool {
181- d.snapshotWorkerMu.Lock()
182- defer d.snapshotWorkerMu.Unlock()
183- if d.snapshotWorkerID != workerID || d.snapshotWorkerCancel == nil || d.snapshotWorkerCtx == nil || d.snapshotWorkerCtx.Err() != nil {
181+ d.snapshots.mu.Lock()
182+ defer d.snapshots.mu.Unlock()
183+ if d.snapshots.id != workerID || d.snapshots.cancel == nil || d.snapshots.ctx == nil || d.snapshots.ctx.Err() != nil {
184184 return false
185185 }
186- d.snapshotWorkerInFlight = capture
186+ d.snapshots.inFlight = capture
187187 return true
188188 }
189189
190190 func (d *Daemon) clearSnapshotWorkerInFlight(workerID uint64, capture *snapshotCapture) {
191- d.snapshotWorkerMu.Lock()
192- defer d.snapshotWorkerMu.Unlock()
193- if d.snapshotWorkerID == workerID && d.snapshotWorkerInFlight == capture {
194- d.snapshotWorkerInFlight = nil
191+ d.snapshots.mu.Lock()
192+ defer d.snapshots.mu.Unlock()
193+ if d.snapshots.id == workerID && d.snapshots.inFlight == capture {
194+ d.snapshots.inFlight = nil
195195 }
196196 }
197197
@@ -240,11 +240,11 @@ func (d *Daemon) WaitDurableWriters() {
240240 if d == nil {
241241 return
242242 }
243- d.snapshotWorkerMu.Lock()
244- snapshotDone := d.snapshotWorkerDone
245- maintenanceDone := d.maintenanceWorkerDone
246- restoreDone := d.restoreWorkerDone
247- d.snapshotWorkerMu.Unlock()
243+ d.snapshots.mu.Lock()
244+ snapshotDone := d.snapshots.done
245+ maintenanceDone := d.snapshots.maintenanceWorkerDone
246+ restoreDone := d.snapshots.restoreWorkerDone
247+ d.snapshots.mu.Unlock()
248248 if snapshotDone != nil {
249249 <-snapshotDone
250250 d.finishStoppedSnapshotWorker(false)
@@ -260,37 +260,37 @@ func (d *Daemon) WaitDurableWriters() {
260260 }
261261
262262 func (d *Daemon) requestDurableWriterStop() (context.CancelFunc, <-chan struct{}) {
263- d.snapshotWorkerMu.Lock()
264- defer d.snapshotWorkerMu.Unlock()
263+ d.snapshots.mu.Lock()
264+ defer d.snapshots.mu.Unlock()
265265 d.cancelDurableMaintenanceLocked()
266- cancel := d.snapshotWorkerCancel
266+ cancel := d.snapshots.cancel
267267 if cancel == nil {
268268 return nil, nil
269269 }
270- if !d.snapshotWorkerClosing {
271- d.snapshotWorkerClosing = true
272- close(d.snapshotWorkerFlush)
270+ if !d.snapshots.closing {
271+ d.snapshots.closing = true
272+ close(d.snapshots.flush)
273273 }
274- return cancel, d.snapshotWorkerDone
274+ return cancel, d.snapshots.done
275275 }
276276
277277 // durableWriterFailureNames includes admitted buffered captures as well as the
278278 // active and final queues. snapshotAdmitted tracks normal captures from queue
279279 // admission through completion, so worker dequeue cannot make one disappear.
280280 func (d *Daemon) durableWriterFailureNames() []string {
281- d.snapshotWorkerMu.Lock()
282- defer d.snapshotWorkerMu.Unlock()
281+ d.snapshots.mu.Lock()
282+ defer d.snapshots.mu.Unlock()
283283 seen := make(map[string]struct{})
284284 add := func(capture *snapshotCapture) {
285285 if capture != nil && capture.name != "" {
286286 seen[capture.name] = struct{}{}
287287 }
288288 }
289- add(d.snapshotWorkerInFlight)
290- for _, capture := range d.snapshotFinalJobs {
289+ add(d.snapshots.inFlight)
290+ for _, capture := range d.snapshots.finalJobs {
291291 add(capture)
292292 }
293- for capture := range d.snapshotAdmitted {
293+ for capture := range d.snapshots.admitted {
294294 add(capture)
295295 }
296296 names := make([]string, 0, len(seen))
@@ -302,27 +302,27 @@ func (d *Daemon) durableWriterFailureNames() []string {
302302 }
303303
304304 func (d *Daemon) finishStoppedSnapshotWorker(abandoned bool) {
305- d.snapshotWorkerMu.Lock()
306- inFlight := d.snapshotWorkerInFlight
307- d.snapshotWorkerCtx = nil
308- d.snapshotWorkerCancel = nil
309- d.snapshotWorkerDone = nil
310- d.snapshotWorkerFlush = nil
311- d.snapshotWorkerFinalWake = nil
312- d.snapshotWorkerClosing = false
313- d.snapshotWorkerInFlight = nil
314- queued := make([]*snapshotCapture, 0, len(d.snapshotJobs)+len(d.snapshotFinalJobs))
315- for _, capture := range d.snapshotFinalJobs {
305+ d.snapshots.mu.Lock()
306+ inFlight := d.snapshots.inFlight
307+ d.snapshots.ctx = nil
308+ d.snapshots.cancel = nil
309+ d.snapshots.done = nil
310+ d.snapshots.flush = nil
311+ d.snapshots.finalWake = nil
312+ d.snapshots.closing = false
313+ d.snapshots.inFlight = nil
314+ queued := make([]*snapshotCapture, 0, len(d.snapshots.jobs)+len(d.snapshots.finalJobs))
315+ for _, capture := range d.snapshots.finalJobs {
316316 queued = append(queued, capture)
317317 }
318- d.snapshotFinalJobs = nil
319- d.snapshotFinalOrder = nil
318+ d.snapshots.finalJobs = nil
319+ d.snapshots.finalOrder = nil
320320 for {
321321 select {
322- case capture := <-d.snapshotJobs:
322+ case capture := <-d.snapshots.jobs:
323323 queued = append(queued, capture)
324324 default:
325- d.snapshotWorkerMu.Unlock()
325+ d.snapshots.mu.Unlock()
326326 if abandoned {
327327 d.finishSnapshotCapture(inFlight, false)
328328 }
@@ -13,10 +13,10 @@ func (d *Daemon) reportSnapshotFailure(capture *snapshotCapture, phase string, c
13 return13 return
14 }14 }
15 signature := snapshotFailureSignature(phase, cause)15 signature := snapshotFailureSignature(phase, cause)
16- d.snapshotNoticeMu.Lock()16+ d.snapshots.noticeMu.Lock()
17- changed := d.snapshotActiveFailureSignature != signature17+ changed := d.snapshots.activeFailureSignature != signature
18- d.snapshotActiveFailureSignature = signature18+ d.snapshots.activeFailureSignature = signature
19- d.snapshotNoticeMu.Unlock()19+ d.snapshots.noticeMu.Unlock()
20 20
21 n := domain.Notification{21 n := domain.Notification{
22 Code: domain.NoticeSnapshotWrite,22 Code: domain.NoticeSnapshotWrite,
@@ -38,38 +38,38 @@ func (d *Daemon) clearSnapshotFailure() {
38 if d == nil {38 if d == nil {
39 return39 return
40 }40 }
41- d.snapshotNoticeMu.Lock()41+ d.snapshots.noticeMu.Lock()
42- d.snapshotActiveFailureSignature = ""42+ d.snapshots.activeFailureSignature = ""
43- d.snapshotNoticeMu.Unlock()43+ d.snapshots.noticeMu.Unlock()
44 }44 }
45 45
46 func (d *Daemon) startSnapshotEncodeWorker() {46 func (d *Daemon) startSnapshotEncodeWorker() {
47 if d == nil {47 if d == nil {
48 return48 return
49 }49 }
50- d.snapshotWorkerMu.Lock()50+ d.snapshots.mu.Lock()
51- if d.snapshotWorkerCancel != nil {51+ if d.snapshots.cancel != nil {
52- d.snapshotWorkerMu.Unlock()52+ d.snapshots.mu.Unlock()
53 return53 return
54 }54 }
55 // Shutdown owns the worker lifetime so it can flush captures after Serve55 // Shutdown owns the worker lifetime so it can flush captures after Serve
56 // cancels its parent context. The worker is always stopped explicitly.56 // cancels its parent context. The worker is always stopped explicitly.
57 workerCtx, cancel := context.WithCancel(context.Background())57 workerCtx, cancel := context.WithCancel(context.Background())
58- d.snapshotWorkerID++58+ d.snapshots.id++
59- workerID := d.snapshotWorkerID59+ workerID := d.snapshots.id
60- d.snapshotWorkerCtx = workerCtx60+ d.snapshots.ctx = workerCtx
61- d.snapshotWorkerCancel = cancel61+ d.snapshots.cancel = cancel
62- d.snapshotWorkerDone = make(chan struct{})62+ d.snapshots.done = make(chan struct{})
63- d.snapshotWorkerFlush = make(chan struct{})63+ d.snapshots.flush = make(chan struct{})
64- d.snapshotWorkerFinalWake = make(chan struct{}, 1)64+ d.snapshots.finalWake = make(chan struct{}, 1)
65- d.snapshotFinalJobs = make(map[*session]*snapshotCapture)65+ d.snapshots.finalJobs = make(map[*session]*snapshotCapture)
66- d.snapshotFinalOrder = nil66+ d.snapshots.finalOrder = nil
67- d.snapshotWorkerClosing = false67+ d.snapshots.closing = false
68- d.snapshotWorkerInFlight = nil68+ d.snapshots.inFlight = nil
69- done := d.snapshotWorkerDone69+ done := d.snapshots.done
70- flush := d.snapshotWorkerFlush70+ flush := d.snapshots.flush
71- finalWake := d.snapshotWorkerFinalWake71+ finalWake := d.snapshots.finalWake
72- d.snapshotWorkerMu.Unlock()72+ d.snapshots.mu.Unlock()
73 go d.runSnapshotEncodeWorker(workerCtx, workerID, done, flush, finalWake)73 go d.runSnapshotEncodeWorker(workerCtx, workerID, done, flush, finalWake)
74 }74 }
75 75
@@ -89,7 +89,7 @@ func (d *Daemon) runSnapshotEncodeWorker(workerCtx context.Context, workerID uin
89 if !d.drainFinalSnapshotCaptures(workerCtx, workerID) {89 if !d.drainFinalSnapshotCaptures(workerCtx, workerID) {
90 return90 return
91 }91 }
92- case capture := <-d.snapshotJobs:92+ case capture := <-d.snapshots.jobs:
93 if !d.publishSnapshotCapture(workerCtx, workerID, capture) {93 if !d.publishSnapshotCapture(workerCtx, workerID, capture) {
94 return94 return
95 }95 }
@@ -100,7 +100,7 @@ func (d *Daemon) runSnapshotEncodeWorker(workerCtx context.Context, workerID uin
100 func (d *Daemon) flushSnapshotCaptures(workerCtx context.Context, workerID uint64) {100 func (d *Daemon) flushSnapshotCaptures(workerCtx context.Context, workerID uint64) {
101 for {101 for {
102 select {102 select {
103- case capture := <-d.snapshotJobs:103+ case capture := <-d.snapshots.jobs:
104 if !d.publishSnapshotCapture(workerCtx, workerID, capture) {104 if !d.publishSnapshotCapture(workerCtx, workerID, capture) {
105 return105 return
106 }106 }
@@ -162,14 +162,14 @@ func (d *Daemon) publishSnapshotCapture(workerCtx context.Context, workerID uint
162 }162 }
163 163
164 func (d *Daemon) takeFinalSnapshotCapture() *snapshotCapture {164 func (d *Daemon) takeFinalSnapshotCapture() *snapshotCapture {
165- d.snapshotWorkerMu.Lock()165+ d.snapshots.mu.Lock()
166- defer d.snapshotWorkerMu.Unlock()166+ defer d.snapshots.mu.Unlock()
167- for len(d.snapshotFinalOrder) > 0 {167+ for len(d.snapshots.finalOrder) > 0 {
168- sess := d.snapshotFinalOrder[0]168+ sess := d.snapshots.finalOrder[0]
169- d.snapshotFinalOrder[0] = nil169+ d.snapshots.finalOrder[0] = nil
170- d.snapshotFinalOrder = d.snapshotFinalOrder[1:]170+ d.snapshots.finalOrder = d.snapshots.finalOrder[1:]
171- capture := d.snapshotFinalJobs[sess]171+ capture := d.snapshots.finalJobs[sess]
172- delete(d.snapshotFinalJobs, sess)172+ delete(d.snapshots.finalJobs, sess)
173 if capture != nil {173 if capture != nil {
174 return capture174 return capture
175 }175 }
@@ -178,20 +178,20 @@ func (d *Daemon) takeFinalSnapshotCapture() *snapshotCapture {
178 }178 }
179 179
180 func (d *Daemon) setSnapshotWorkerInFlight(workerID uint64, capture *snapshotCapture) bool {180 func (d *Daemon) setSnapshotWorkerInFlight(workerID uint64, capture *snapshotCapture) bool {
181- d.snapshotWorkerMu.Lock()181+ d.snapshots.mu.Lock()
182- defer d.snapshotWorkerMu.Unlock()182+ defer d.snapshots.mu.Unlock()
183- if d.snapshotWorkerID != workerID || d.snapshotWorkerCancel == nil || d.snapshotWorkerCtx == nil || d.snapshotWorkerCtx.Err() != nil {183+ if d.snapshots.id != workerID || d.snapshots.cancel == nil || d.snapshots.ctx == nil || d.snapshots.ctx.Err() != nil {
184 return false184 return false
185 }185 }
186- d.snapshotWorkerInFlight = capture186+ d.snapshots.inFlight = capture
187 return true187 return true
188 }188 }
189 189
190 func (d *Daemon) clearSnapshotWorkerInFlight(workerID uint64, capture *snapshotCapture) {190 func (d *Daemon) clearSnapshotWorkerInFlight(workerID uint64, capture *snapshotCapture) {
191- d.snapshotWorkerMu.Lock()191+ d.snapshots.mu.Lock()
192- defer d.snapshotWorkerMu.Unlock()192+ defer d.snapshots.mu.Unlock()
193- if d.snapshotWorkerID == workerID && d.snapshotWorkerInFlight == capture {193+ if d.snapshots.id == workerID && d.snapshots.inFlight == capture {
194- d.snapshotWorkerInFlight = nil194+ d.snapshots.inFlight = nil
195 }195 }
196 }196 }
197 197
@@ -240,11 +240,11 @@ func (d *Daemon) WaitDurableWriters() {
240 if d == nil {240 if d == nil {
241 return241 return
242 }242 }
243- d.snapshotWorkerMu.Lock()243+ d.snapshots.mu.Lock()
244- snapshotDone := d.snapshotWorkerDone244+ snapshotDone := d.snapshots.done
245- maintenanceDone := d.maintenanceWorkerDone245+ maintenanceDone := d.snapshots.maintenanceWorkerDone
246- restoreDone := d.restoreWorkerDone246+ restoreDone := d.snapshots.restoreWorkerDone
247- d.snapshotWorkerMu.Unlock()247+ d.snapshots.mu.Unlock()
248 if snapshotDone != nil {248 if snapshotDone != nil {
249 <-snapshotDone249 <-snapshotDone
250 d.finishStoppedSnapshotWorker(false)250 d.finishStoppedSnapshotWorker(false)
@@ -260,37 +260,37 @@ func (d *Daemon) WaitDurableWriters() {
260 }260 }
261 261
262 func (d *Daemon) requestDurableWriterStop() (context.CancelFunc, <-chan struct{}) {262 func (d *Daemon) requestDurableWriterStop() (context.CancelFunc, <-chan struct{}) {
263- d.snapshotWorkerMu.Lock()263+ d.snapshots.mu.Lock()
264- defer d.snapshotWorkerMu.Unlock()264+ defer d.snapshots.mu.Unlock()
265 d.cancelDurableMaintenanceLocked()265 d.cancelDurableMaintenanceLocked()
266- cancel := d.snapshotWorkerCancel266+ cancel := d.snapshots.cancel
267 if cancel == nil {267 if cancel == nil {
268 return nil, nil268 return nil, nil
269 }269 }
270- if !d.snapshotWorkerClosing {270+ if !d.snapshots.closing {
271- d.snapshotWorkerClosing = true271+ d.snapshots.closing = true
272- close(d.snapshotWorkerFlush)272+ close(d.snapshots.flush)
273 }273 }
274- return cancel, d.snapshotWorkerDone274+ return cancel, d.snapshots.done
275 }275 }
276 276
277 // durableWriterFailureNames includes admitted buffered captures as well as the277 // durableWriterFailureNames includes admitted buffered captures as well as the
278 // active and final queues. snapshotAdmitted tracks normal captures from queue278 // active and final queues. snapshotAdmitted tracks normal captures from queue
279 // admission through completion, so worker dequeue cannot make one disappear.279 // admission through completion, so worker dequeue cannot make one disappear.
280 func (d *Daemon) durableWriterFailureNames() []string {280 func (d *Daemon) durableWriterFailureNames() []string {
281- d.snapshotWorkerMu.Lock()281+ d.snapshots.mu.Lock()
282- defer d.snapshotWorkerMu.Unlock()282+ defer d.snapshots.mu.Unlock()
283 seen := make(map[string]struct{})283 seen := make(map[string]struct{})
284 add := func(capture *snapshotCapture) {284 add := func(capture *snapshotCapture) {
285 if capture != nil && capture.name != "" {285 if capture != nil && capture.name != "" {
286 seen[capture.name] = struct{}{}286 seen[capture.name] = struct{}{}
287 }287 }
288 }288 }
289- add(d.snapshotWorkerInFlight)289+ add(d.snapshots.inFlight)
290- for _, capture := range d.snapshotFinalJobs {290+ for _, capture := range d.snapshots.finalJobs {
291 add(capture)291 add(capture)
292 }292 }
293- for capture := range d.snapshotAdmitted {293+ for capture := range d.snapshots.admitted {
294 add(capture)294 add(capture)
295 }295 }
296 names := make([]string, 0, len(seen))296 names := make([]string, 0, len(seen))
@@ -302,27 +302,27 @@ func (d *Daemon) durableWriterFailureNames() []string {
302 }302 }
303 303
304 func (d *Daemon) finishStoppedSnapshotWorker(abandoned bool) {304 func (d *Daemon) finishStoppedSnapshotWorker(abandoned bool) {
305- d.snapshotWorkerMu.Lock()305+ d.snapshots.mu.Lock()
306- inFlight := d.snapshotWorkerInFlight306+ inFlight := d.snapshots.inFlight
307- d.snapshotWorkerCtx = nil307+ d.snapshots.ctx = nil
308- d.snapshotWorkerCancel = nil308+ d.snapshots.cancel = nil
309- d.snapshotWorkerDone = nil309+ d.snapshots.done = nil
310- d.snapshotWorkerFlush = nil310+ d.snapshots.flush = nil
311- d.snapshotWorkerFinalWake = nil311+ d.snapshots.finalWake = nil
312- d.snapshotWorkerClosing = false312+ d.snapshots.closing = false
313- d.snapshotWorkerInFlight = nil313+ d.snapshots.inFlight = nil
314- queued := make([]*snapshotCapture, 0, len(d.snapshotJobs)+len(d.snapshotFinalJobs))314+ queued := make([]*snapshotCapture, 0, len(d.snapshots.jobs)+len(d.snapshots.finalJobs))
315- for _, capture := range d.snapshotFinalJobs {315+ for _, capture := range d.snapshots.finalJobs {
316 queued = append(queued, capture)316 queued = append(queued, capture)
317 }317 }
318- d.snapshotFinalJobs = nil318+ d.snapshots.finalJobs = nil
319- d.snapshotFinalOrder = nil319+ d.snapshots.finalOrder = nil
320 for {320 for {
321 select {321 select {
322- case capture := <-d.snapshotJobs:322+ case capture := <-d.snapshots.jobs:
323 queued = append(queued, capture)323 queued = append(queued, capture)
324 default:324 default:
325- d.snapshotWorkerMu.Unlock()325+ d.snapshots.mu.Unlock()
326 if abandoned {326 if abandoned {
327 d.finishSnapshotCapture(inFlight, false)327 d.finishSnapshotCapture(inFlight, false)
328 }328 }
A internal/usecase/daemon/snapshot_worker_state.go +61 -0
@@ -0,0 +1,61 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "sync"
6+)
7+
8+// snapshotWorker owns the state of the repository snapshot worker and of the
9+// other durable writers joined with it (maintenance and restoration), so
10+// WaitDurableWriters can read all three completion signals under one lock.
11+//
12+// Lock position: mu is a leaf with respect to Daemon.mu. No code path acquires
13+// Daemon.mu, a session lock, or a pane lock while holding it, and capture,
14+// queue, and worker code never takes Daemon.mu. noticeMu is likewise a leaf and
15+// is never held together with mu.
16+//
17+// The zero value is usable for tests that build &Daemon{}: nil maps are
18+// allocated lazily by the enqueue paths, and a nil jobs channel never becomes
19+// ready in a select.
20+type snapshotWorker struct {
21+ // jobs is the bounded regular capture queue. It is never closed: producers
22+ // can race shutdown without risking a send-on-closed panic.
23+ jobs chan *snapshotCapture
24+ // wake wakes the repository scheduler when a session becomes dirty
25+ // or an attempt completes. It is never closed and producers only send
26+ // non-blockingly.
27+ wake chan struct{}
28+
29+ // mu guards every field below it.
30+ mu sync.Mutex
31+ // admitted contains every capture accepted by either worker queue,
32+ // including captures buffered in jobs.
33+ admitted map[*snapshotCapture]struct{}
34+ id uint64
35+ ctx context.Context
36+ cancel context.CancelFunc
37+ done chan struct{}
38+ flush chan struct{}
39+ finalWake chan struct{}
40+ // finalJobs coalesces terminal captures by session when the bounded
41+ // regular queue is full. It retains at most snapshotFinalQueueCapacity named
42+ // sessions, each with only its newest terminal state while the worker blocks.
43+ finalJobs map[*session]*snapshotCapture
44+ finalOrder []*session
45+ closing bool
46+ inFlight *snapshotCapture
47+
48+ // maintenanceWorkerCancel/Done and restoreWorkerDone are the other durable
49+ // writers' ownership signals. They share mu with the snapshot worker and are
50+ // joined together in WaitDurableWriters; never split them. Startup
51+ // restoration reconciles durable checkpoints, so it is a durable writer.
52+ maintenanceWorkerCancel context.CancelFunc
53+ maintenanceWorkerDone chan struct{}
54+ restoreWorkerDone chan struct{}
55+
56+ // noticeMu guards the active global persistence failure signature. It is
57+ // separate from mu so notice routing cannot block a producer or a
58+ // repository worker.
59+ noticeMu sync.Mutex
60+ activeFailureSignature string
61+}
@@ -0,0 +1,61 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "sync"
6+)
7+
8+// snapshotWorker owns the state of the repository snapshot worker and of the
9+// other durable writers joined with it (maintenance and restoration), so
10+// WaitDurableWriters can read all three completion signals under one lock.
11+//
12+// Lock position: mu is a leaf with respect to Daemon.mu. No code path acquires
13+// Daemon.mu, a session lock, or a pane lock while holding it, and capture,
14+// queue, and worker code never takes Daemon.mu. noticeMu is likewise a leaf and
15+// is never held together with mu.
16+//
17+// The zero value is usable for tests that build &Daemon{}: nil maps are
18+// allocated lazily by the enqueue paths, and a nil jobs channel never becomes
19+// ready in a select.
20+type snapshotWorker struct {
21+ // jobs is the bounded regular capture queue. It is never closed: producers
22+ // can race shutdown without risking a send-on-closed panic.
23+ jobs chan *snapshotCapture
24+ // wake wakes the repository scheduler when a session becomes dirty
25+ // or an attempt completes. It is never closed and producers only send
26+ // non-blockingly.
27+ wake chan struct{}
28+
29+ // mu guards every field below it.
30+ mu sync.Mutex
31+ // admitted contains every capture accepted by either worker queue,
32+ // including captures buffered in jobs.
33+ admitted map[*snapshotCapture]struct{}
34+ id uint64
35+ ctx context.Context
36+ cancel context.CancelFunc
37+ done chan struct{}
38+ flush chan struct{}
39+ finalWake chan struct{}
40+ // finalJobs coalesces terminal captures by session when the bounded
41+ // regular queue is full. It retains at most snapshotFinalQueueCapacity named
42+ // sessions, each with only its newest terminal state while the worker blocks.
43+ finalJobs map[*session]*snapshotCapture
44+ finalOrder []*session
45+ closing bool
46+ inFlight *snapshotCapture
47+
48+ // maintenanceWorkerCancel/Done and restoreWorkerDone are the other durable
49+ // writers' ownership signals. They share mu with the snapshot worker and are
50+ // joined together in WaitDurableWriters; never split them. Startup
51+ // restoration reconciles durable checkpoints, so it is a durable writer.
52+ maintenanceWorkerCancel context.CancelFunc
53+ maintenanceWorkerDone chan struct{}
54+ restoreWorkerDone chan struct{}
55+
56+ // noticeMu guards the active global persistence failure signature. It is
57+ // separate from mu so notice routing cannot block a producer or a
58+ // repository worker.
59+ noticeMu sync.Mutex
60+ activeFailureSignature string
61+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close