snapshot_coordinator.go Raw History Folder

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
53// rollbackSnapshotQuarantine reverses quarantineSnapshotCoordinator when a
54// teardown aborts before it owns any destructive repository operation and the
55// session therefore stays live and registered. The quarantine cancelled the
56// publication context and flipped eligibility, so a surviving session would
57// otherwise never checkpoint again: this discards the cancelled context so the
58// next schedule allocates a fresh one, restores the captured pre-quarantine
59// eligibility, and marks the session dirty so the mutations its ineligible fast
60// path suppressed during the quarantine window are captured by a later
61// publication. A queued capture the quarantine discarded stays discarded:
62// startSnapshotPublication rejects it by identity and the forced dirty
63// generation guarantees a replacement capture.
64func rollbackSnapshotQuarantine(sess *session, wasEligible bool) {
65	if sess == nil {
66		return
67	}
68	sess.snapshotMu.Lock()
69	sess.snapshotQuarantined = false
70	sess.snapshotPublicationContext = nil
71	sess.snapshotPublicationCancel = nil
72	sess.snapshotMu.Unlock()
73	sess.snapEligible.Store(wasEligible)
74	if wasEligible {
75		markSnapshotDirty(sess)
76	}
77}
78
79func markSnapshotDirty(sess *session) {
80	if sess == nil || !sess.snapEligible.Load() {
81		return
82	}
83	sess.snapshotMu.Lock()
84	sess.snapshotGeneration++
85	sess.snapDirty.Store(true)
86	sess.signalSnapshotChangedLocked()
87	wake := sess.snapshotWake
88	sess.snapshotMu.Unlock()
89	if wake != nil {
90		select {
91		case wake <- struct{}{}:
92		default:
93		}
94	}
95}
96
97// snapshotChangeLocked returns a channel closed by the next snapshot state
98// transition. snapshotMu must be held.
99func (sess *session) snapshotChangeLocked() chan struct{} {
100	if sess.snapshotChanged == nil {
101		sess.snapshotChanged = make(chan struct{})
102	}
103	return sess.snapshotChanged
104}
105
106// signalSnapshotChangedLocked wakes waiters after a snapshot state transition.
107// snapshotMu must be held.
108func (sess *session) signalSnapshotChangedLocked() {
109	close(sess.snapshotChangeLocked())
110	sess.snapshotChanged = make(chan struct{})
111}
112
113// waitForSnapshotGeneration waits only for the bounded terminal checkpoint
114// deadline. It observes successful publication of the requested mutation
115// revision, rather than dirty state, so a concurrent later mutation cannot
116// turn a completed forced checkpoint into a false timeout.
117func (d *Daemon) waitForSnapshotGeneration(sess *session, generation uint64) bool {
118	if d == nil || sess == nil || generation == 0 {
119		return true
120	}
121	deadline := newSnapshotShutdownDeadline(d.clock)
122	defer deadline.stop()
123	return d.waitForSnapshotGenerationWithDeadline(sess, generation, deadline)
124}
125
126// waitForSnapshotGenerationWithDeadline waits for a forced checkpoint without
127// allocating another shutdown interval. Serve shares one deadline across every
128// session and the subsequent worker drain.
129func (d *Daemon) waitForSnapshotGenerationWithDeadline(sess *session, generation uint64, deadline *snapshotShutdownDeadline) bool {
130	if d == nil || sess == nil || generation == 0 {
131		return true
132	}
133	if deadline == nil {
134		return d.waitForSnapshotGeneration(sess, generation)
135	}
136	for {
137		// Once the shared deadline is observable, timeout wins even if publication
138		// completion becomes ready concurrently. This preserves the timed-out
139		// session identity and avoids nondeterministic success after the budget.
140		select {
141		case <-deadline.Done():
142			return false
143		default:
144		}
145		sess.snapshotMu.Lock()
146		published := sess.snapshotPublishedMutationRevision >= generation
147		changed := sess.snapshotChangeLocked()
148		sess.snapshotMu.Unlock()
149		if published {
150			return true
151		}
152		select {
153		case <-changed:
154		case <-deadline.Done():
155			return false
156		}
157	}
158}
159
160// startSnapshotPublication moves a retained capture from the producer queue to
161// the repository worker. Quarantine and this transition share snapshotMu, so a
162// repository call cannot begin after destructive teardown has quarantined the
163// session.
164func startSnapshotPublication(capture *snapshotCapture) bool {
165	if capture == nil || capture.session == nil {
166		return false
167	}
168	sess := capture.session
169	sess.snapshotMu.Lock()
170	defer sess.snapshotMu.Unlock()
171	if sess.snapshotQuarantined || sess.snapshotQueuedCapture != capture {
172		return false
173	}
174	sess.snapshotQueuedCapture = nil
175	sess.snapshotInFlightCapture = capture
176	sess.snapshotPublicationDone = make(chan struct{})
177	sess.snapshotPending = true
178	sess.signalSnapshotChangedLocked()
179	return true
180}
181
182func (d *Daemon) finishSnapshotCapture(capture *snapshotCapture, succeeded bool) {
183	if capture == nil {
184		return
185	}
186	capture.finishOnce.Do(func() {
187		var shouldScheduleForcedSuccessor bool
188		capture.session.snapshotMu.Lock()
189		// A capture the quarantine discarded already dropped its pending count when
190		// it was detached from the producer queue; the worker still calls finish on
191		// it after rejecting the stale identity. Decrementing again here would
192		// under-count a replacement capture scheduled after a rollback, so a
193		// discarded capture must not decrement twice.
194		if !capture.coordinatorDiscarded && capture.session.snapshotPendingCaptures > 0 {
195			capture.session.snapshotPendingCaptures--
196		}
197		if capture.session.snapshotQueuedCapture == capture {
198			capture.session.snapshotQueuedCapture = nil
199		}
200		if capture.session.snapshotInFlightCapture == capture {
201			capture.session.snapshotInFlightCapture = nil
202			if capture.session.snapshotPublicationDone != nil {
203				close(capture.session.snapshotPublicationDone)
204				capture.session.snapshotPublicationDone = nil
205			}
206		}
207		capture.session.snapshotPending = capture.session.snapshotPendingCaptures > 0 ||
208			capture.session.snapshotQueuedCapture != nil || capture.session.snapshotInFlightCapture != nil
209		if !capture.coordinatorDiscarded && capture.attemptKind == snapshotAttemptRoutine && d.snapshotRepository != nil {
210			// The interval is measured from completion, including a failed
211			// attempt. A forced teardown checkpoint deliberately leaves it alone.
212			capture.session.snapshotNextEligibleAt = d.clock.Now().Add(snapshotInterval)
213		}
214		mutationRevision := capture.mutationRevision
215		// Older unit fixtures created captures before mutation revisions were
216		// recorded explicitly. Production captures always set it at scheduling;
217		// retaining this fallback keeps those fixtures representative.
218		if mutationRevision == 0 {
219			mutationRevision = capture.generation
220		}
221		if succeeded {
222			if capture.generation >= capture.session.snapshotPublishedGeneration &&
223				capture.checkpoint.Generation == capture.generation && capture.checkpoint.ManifestDigest != ([32]byte{}) {
224				checkpoint := capture.checkpoint
225				capture.session.snapshotPublishedCheckpoint = &checkpoint
226			}
227			capture.session.snapshotPublishedGeneration = max(capture.session.snapshotPublishedGeneration, capture.generation)
228			capture.session.snapshotPublishedMutationRevision = max(capture.session.snapshotPublishedMutationRevision, mutationRevision)
229		}
230		if succeeded && capture.session.snapshotGeneration == mutationRevision {
231			capture.session.snapDirty.Store(false)
232		} else if !succeeded {
233			capture.session.snapDirty.Store(true)
234		}
235		if succeeded && capture.session.snapshotForcedGeneration != 0 &&
236			mutationRevision >= capture.session.snapshotForcedGeneration {
237			capture.session.snapshotForcedGeneration = 0
238		}
239		// A failed forced publication remains dirty for ordinary retry rather
240		// than recursively creating forced jobs. A later forced request records
241		// a newer intent and can create its one bounded successor.
242		shouldScheduleForcedSuccessor = capture.session.snapshotForcedGeneration != 0 &&
243			mutationRevision < capture.session.snapshotForcedGeneration &&
244			capture.session.snapshotQueuedCapture == nil && capture.session.snapshotInFlightCapture == nil &&
245			(succeeded || capture.attemptKind != snapshotAttemptForced)
246		capture.session.signalSnapshotChangedLocked()
247		wake := capture.session.snapshotWake
248		capture.session.snapshotMu.Unlock()
249		if capture.normalWorkerAdmitted {
250			d.snapshots.mu.Lock()
251			delete(d.snapshots.admitted, capture)
252			d.snapshots.mu.Unlock()
253		}
254		if wake != nil {
255			select {
256			case wake <- struct{}{}:
257			default:
258			}
259		}
260		if shouldScheduleForcedSuccessor {
261			// This runs after publication completion and outside snapshotMu. It
262			// bypasses routine eligibility while preserving the single queued and
263			// single in-flight capture bound.
264			_ = d.scheduleFinalSnapshot(capture.session)
265		}
266		if succeeded {
267			d.clearSnapshotFailure()
268		}
269	})
270}
snapshot_coordinator.go RawHistoryFolder

snapshot_coordinator.go Raw History Folder
