snapshot_coordinator.go RawHistoryFolder

1package daemon
2
3// quarantineSnapshotCoordinator prevents a deleted session from starting any
4// further publication. The returned join channel is closed after an already
5// started repository call returns; callers must wait without daemon or session
6// locks before deleting repository state.
7func quarantineSnapshotCoordinator(sess *session) <-chan struct{} {
8	return quarantineSnapshotCoordinatorWithOptions(sess, false)
9}
10
11// quarantineSnapshotCoordinatorRetainingQueuedCapture is used after a forced
12// shutdown checkpoint times out. Its queued capture is logical retry state, not
13// a repository operation: startSnapshotPublication rejects it while
14// quarantined, and the worker eventually discards it after its blocked call
15// returns.
16func quarantineSnapshotCoordinatorRetainingQueuedCapture(sess *session) <-chan struct{} {
17	return quarantineSnapshotCoordinatorWithOptions(sess, true)
18}
19
20func quarantineSnapshotCoordinatorWithOptions(sess *session, retainQueuedCapture bool) <-chan struct{} {
21	if sess == nil {
22		done := make(chan struct{})
23		close(done)
24		return done
25	}
26	sess.snapshotMu.Lock()
27	sess.snapshotQuarantined = true
28	sess.snapEligible.Store(false)
29	if sess.snapshotPublicationCancel != nil {
30		sess.snapshotPublicationCancel()
31	}
32	if sess.snapshotQueuedCapture != nil && !retainQueuedCapture {
33		// A rename or purge must discard a capture under the old identity. The
34		// global queue may still contain it, but startSnapshotPublication
35		// compares identity and will discard it even if this session is renamed.
36		sess.snapshotQueuedCapture.coordinatorDiscarded = true
37		sess.snapshotQueuedCapture = nil
38		if sess.snapshotPendingCaptures > 0 {
39			sess.snapshotPendingCaptures--
40		}
41		sess.snapshotPending = sess.snapshotPendingCaptures > 0 || sess.snapshotInFlightCapture != nil
42	}
43	done := sess.snapshotPublicationDone
44	if done == nil {
45		done = make(chan struct{})
46		close(done)
47	}
48	sess.signalSnapshotChangedLocked()
49	sess.snapshotMu.Unlock()
50	return done
51}
52
53func markSnapshotDirty(sess *session) {
54	if sess == nil || !sess.snapEligible.Load() {
55		return
56	}
57	sess.snapshotMu.Lock()
58	sess.snapshotGeneration++
59	sess.snapDirty.Store(true)
60	sess.signalSnapshotChangedLocked()
61	wake := sess.snapshotWake
62	sess.snapshotMu.Unlock()
63	if wake != nil {
64		select {
65		case wake <- struct{}{}:
66		default:
67		}
68	}
69}
70
71// snapshotChangeLocked returns a channel closed by the next snapshot state
72// transition. snapshotMu must be held.
73func (sess *session) snapshotChangeLocked() chan struct{} {
74	if sess.snapshotChanged == nil {
75		sess.snapshotChanged = make(chan struct{})
76	}
77	return sess.snapshotChanged
78}
79
80// signalSnapshotChangedLocked wakes waiters after a snapshot state transition.
81// snapshotMu must be held.
82func (sess *session) signalSnapshotChangedLocked() {
83	close(sess.snapshotChangeLocked())
84	sess.snapshotChanged = make(chan struct{})
85}
86
87// waitForSnapshotGeneration waits only for the bounded terminal checkpoint
88// deadline. It observes successful publication of the requested mutation
89// revision, rather than dirty state, so a concurrent later mutation cannot
90// turn a completed forced checkpoint into a false timeout.
91func (d *Daemon) waitForSnapshotGeneration(sess *session, generation uint64) bool {
92	if d == nil || sess == nil || generation == 0 {
93		return true
94	}
95	deadline := newSnapshotShutdownDeadline(d.clock)
96	defer deadline.stop()
97	return d.waitForSnapshotGenerationWithDeadline(sess, generation, deadline)
98}
99
100// waitForSnapshotGenerationWithDeadline waits for a forced checkpoint without
101// allocating another shutdown interval. Serve shares one deadline across every
102// session and the subsequent worker drain.
103func (d *Daemon) waitForSnapshotGenerationWithDeadline(sess *session, generation uint64, deadline *snapshotShutdownDeadline) bool {
104	if d == nil || sess == nil || generation == 0 {
105		return true
106	}
107	if deadline == nil {
108		return d.waitForSnapshotGeneration(sess, generation)
109	}
110	for {
111		// Once the shared deadline is observable, timeout wins even if publication
112		// completion becomes ready concurrently. This preserves the timed-out
113		// session identity and avoids nondeterministic success after the budget.
114		select {
115		case <-deadline.Done():
116			return false
117		default:
118		}
119		sess.snapshotMu.Lock()
120		published := sess.snapshotPublishedMutationRevision >= generation
121		changed := sess.snapshotChangeLocked()
122		sess.snapshotMu.Unlock()
123		if published {
124			return true
125		}
126		select {
127		case <-changed:
128		case <-deadline.Done():
129			return false
130		}
131	}
132}
133
134// startSnapshotPublication moves a retained capture from the producer queue to
135// the repository worker. Quarantine and this transition share snapshotMu, so a
136// repository call cannot begin after destructive teardown has quarantined the
137// session.
138func startSnapshotPublication(capture *snapshotCapture) bool {
139	if capture == nil || capture.session == nil {
140		return false
141	}
142	sess := capture.session
143	sess.snapshotMu.Lock()
144	defer sess.snapshotMu.Unlock()
145	if sess.snapshotQuarantined || sess.snapshotQueuedCapture != capture {
146		return false
147	}
148	sess.snapshotQueuedCapture = nil
149	sess.snapshotInFlightCapture = capture
150	sess.snapshotPublicationDone = make(chan struct{})
151	sess.snapshotPending = true
152	sess.signalSnapshotChangedLocked()
153	return true
154}
155
156func (d *Daemon) finishSnapshotCapture(capture *snapshotCapture, succeeded bool) {
157	if capture == nil {
158		return
159	}
160	capture.finishOnce.Do(func() {
161		var shouldScheduleForcedSuccessor bool
162		capture.session.snapshotMu.Lock()
163		if capture.session.snapshotPendingCaptures > 0 {
164			capture.session.snapshotPendingCaptures--
165		}
166		if capture.session.snapshotQueuedCapture == capture {
167			capture.session.snapshotQueuedCapture = nil
168		}
169		if capture.session.snapshotInFlightCapture == capture {
170			capture.session.snapshotInFlightCapture = nil
171			if capture.session.snapshotPublicationDone != nil {
172				close(capture.session.snapshotPublicationDone)
173				capture.session.snapshotPublicationDone = nil
174			}
175		}
176		capture.session.snapshotPending = capture.session.snapshotPendingCaptures > 0 ||
177			capture.session.snapshotQueuedCapture != nil || capture.session.snapshotInFlightCapture != nil
178		if !capture.coordinatorDiscarded && capture.attemptKind == snapshotAttemptRoutine && d.snapshotRepository != nil {
179			// The interval is measured from completion, including a failed
180			// attempt. A forced teardown checkpoint deliberately leaves it alone.
181			capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval)
182		}
183		mutationRevision := capture.mutationRevision
184		// Older unit fixtures created captures before mutation revisions were
185		// recorded explicitly. Production captures always set it at scheduling;
186		// retaining this fallback keeps those fixtures representative.
187		if mutationRevision == 0 {
188			mutationRevision = capture.generation
189		}
190		if succeeded {
191			if capture.generation >= capture.session.snapshotPublishedGeneration &&
192				capture.checkpoint.Generation == capture.generation && capture.checkpoint.ManifestDigest != ([32]byte{}) {
193				checkpoint := capture.checkpoint
194				capture.session.snapshotPublishedCheckpoint = &checkpoint
195			}
196			capture.session.snapshotPublishedGeneration = max(capture.session.snapshotPublishedGeneration, capture.generation)
197			capture.session.snapshotPublishedMutationRevision = max(capture.session.snapshotPublishedMutationRevision, mutationRevision)
198		}
199		if succeeded && capture.session.snapshotGeneration == mutationRevision {
200			capture.session.snapDirty.Store(false)
201		} else if !succeeded {
202			capture.session.snapDirty.Store(true)
203		}
204		if succeeded && capture.session.snapshotForcedGeneration != 0 &&
205			mutationRevision >= capture.session.snapshotForcedGeneration {
206			capture.session.snapshotForcedGeneration = 0
207		}
208		// A failed forced publication remains dirty for ordinary retry rather
209		// than recursively creating forced jobs. A later forced request records
210		// a newer intent and can create its one bounded successor.
211		shouldScheduleForcedSuccessor = capture.session.snapshotForcedGeneration != 0 &&
212			mutationRevision < capture.session.snapshotForcedGeneration &&
213			capture.session.snapshotQueuedCapture == nil && capture.session.snapshotInFlightCapture == nil &&
214			(succeeded || capture.attemptKind != snapshotAttemptForced)
215		capture.session.signalSnapshotChangedLocked()
216		wake := capture.session.snapshotWake
217		capture.session.snapshotMu.Unlock()
218		if capture.normalWorkerAdmitted {
219			d.snapshotWorkerMu.Lock()
220			delete(d.snapshotAdmitted, capture)
221			d.snapshotWorkerMu.Unlock()
222		}
223		if wake != nil {
224			select {
225			case wake <- struct{}{}:
226			default:
227			}
228		}
229		if shouldScheduleForcedSuccessor {
230			// This runs after publication completion and outside snapshotMu. It
231			// bypasses routine eligibility while preserving the single queued and
232			// single in-flight capture bound.
233			_ = d.scheduleFinalSnapshot(capture.session)
234		}
235		if succeeded {
236			d.clearSnapshotFailure()
237		}
238	})
239}