snapshot_queue.go RawHistoryFolder

1package daemon
2
3import (
4	"context"
5
6	"github.com/bnema/vev/internal/domain"
7)
8
9func (d *Daemon) scheduleSnapshot(sess *session) bool {
10	return d.scheduleSnapshotWithFinalFallback(sess, false)
11}
12
13// scheduleFinalSnapshot captures the terminal state even when an older capture
14// is pending. A removed session cannot be retried by the repository scheduler, so the
15// worker drains its retained terminal state before it stops.
16func (d *Daemon) scheduleFinalSnapshot(sess *session) bool {
17	return d.scheduleSnapshotWithFinalFallback(sess, true)
18}
19
20func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bool {
21	if d == nil || sess == nil || !d.snapsEnabled || d.snapshotRepository == nil || !sess.snapEligible.Load() {
22		return true
23	}
24	kind := snapshotAttemptRoutine
25	if final {
26		kind = snapshotAttemptForced
27	}
28	sess.snapshotMu.Lock()
29	if sess.snapshotQuarantined || !sess.snapDirty.Load() {
30		available := !sess.snapshotQuarantined
31		sess.snapshotMu.Unlock()
32		return available
33	}
34	if final && sess.snapshotGeneration > sess.snapshotForcedGeneration {
35		// Do not let an older queued or in-flight routine publication satisfy a
36		// terminal checkpoint. The intent is retained until a publication at
37		// this generation (or newer) succeeds.
38		sess.snapshotForcedGeneration = sess.snapshotGeneration
39	}
40	if sess.snapshotPending ||
41		(!final && !sess.snapshotNextEligibleAt.IsZero() && d.clock.Now().Before(sess.snapshotNextEligibleAt)) {
42		sess.snapshotMu.Unlock()
43		return true
44	}
45	if sess.snapshotPublicationContext == nil {
46		sess.snapshotPublicationContext, sess.snapshotPublicationCancel = context.WithCancel(context.Background())
47	}
48	// A capture's repository generation advances only from the last successful
49	// repository publication. Mutations are independently versioned, so several
50	// changes coalesced before this checkpoint still publish the immediate next
51	// repository generation and a failed attempt retries that same generation.
52	generation := sess.snapshotPublishedGeneration + 1
53	mutationRevision := sess.snapshotGeneration
54	// A restored or test-created dirty session can predate mutation revision
55	// tracking. Normalize that state into the first revision before capture.
56	if mutationRevision == 0 {
57		mutationRevision = 1
58		sess.snapshotGeneration = mutationRevision
59	}
60	publicationContext := sess.snapshotPublicationContext
61	var parentCheckpoint *domain.CheckpointRef
62	if sess.snapshotPublishedCheckpoint != nil {
63		parent := *sess.snapshotPublishedCheckpoint
64		parentCheckpoint = &parent
65	}
66	sess.snapshotPendingCaptures++
67	sess.snapshotPending = true
68	sess.signalSnapshotChangedLocked()
69	sess.snapshotMu.Unlock()
70
71	capture, ok := d.captureSnapshotState(sess, generation)
72	if capture != nil {
73		capture.mutationRevision = mutationRevision
74		capture.attemptKind = kind
75		capture.publicationContext = publicationContext
76		capture.parentCheckpoint = parentCheckpoint
77	}
78	if !ok {
79		d.finishSnapshotCapture(capture, false)
80		return false
81	}
82	sess.snapshotMu.Lock()
83	quarantined := sess.snapshotQuarantined
84	if !quarantined {
85		sess.snapshotQueuedCapture = capture
86	}
87	sess.snapshotMu.Unlock()
88	if quarantined {
89		d.finishSnapshotCapture(capture, false)
90		return false
91	}
92	if d.enqueueSnapshotCapture(capture) || (final && d.enqueueFinalSnapshotCapture(capture)) {
93		return true
94	}
95	// Coalesce under saturation or shutdown: the latest state stays dirty and
96	// will be captured on a later tick once an active worker has room.
97	d.finishSnapshotCapture(capture, false)
98	return false
99}
100
101// enqueueSnapshotCapture serializes worker shutdown with the non-blocking
102// queue send. The queue is deliberately never closed: producers can race
103// shutdown without risking a send-on-closed panic.
104func (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 {
108		return false
109	}
110	capture.normalWorkerAdmitted = true
111	select {
112	case d.snapshotJobs <- capture:
113		if d.snapshotAdmitted == nil {
114			d.snapshotAdmitted = make(map[*snapshotCapture]struct{})
115		}
116		d.snapshotAdmitted[capture] = struct{}{}
117		return true
118	default:
119		capture.normalWorkerAdmitted = false
120		return false
121	}
122}
123
124// enqueueFinalSnapshotCapture hands terminal captures to the existing worker
125// without making session teardown wait behind its normal bounded queue. A
126// blocked worker retains at most one terminal capture per session, replacing
127// stale state with the newest immutable capture.
128func (d *Daemon) enqueueFinalSnapshotCapture(capture *snapshotCapture) bool {
129	if capture == nil || capture.session == nil {
130		return false
131	}
132	d.snapshotWorkerMu.Lock()
133	if d.snapshotWorkerClosing || d.snapshotWorkerCancel == nil || d.snapshotWorkerCtx == nil || d.snapshotWorkerCtx.Err() != nil {
134		d.snapshotWorkerMu.Unlock()
135		return false
136	}
137	if d.snapshotFinalJobs == nil {
138		d.snapshotFinalJobs = make(map[*session]*snapshotCapture)
139	}
140	replaced, exists := d.snapshotFinalJobs[capture.session]
141	if !exists && len(d.snapshotFinalJobs) >= snapshotFinalQueueCapacity {
142		d.snapshotWorkerMu.Unlock()
143		d.log.Warn("terminal snapshot retention saturated; capture rejected", "session", capture.name, "capacity", snapshotFinalQueueCapacity)
144		return false
145	}
146	// Finish the stale capture before publishing its replacement. Otherwise a
147	// fast worker could persist the replacement and then have this stale failure
148	// mark the session dirty again.
149	d.finishSnapshotCapture(replaced, false)
150	if !exists {
151		d.snapshotFinalOrder = append(d.snapshotFinalOrder, capture.session)
152	}
153	d.snapshotFinalJobs[capture.session] = capture
154	select {
155	case d.snapshotWorkerFinalWake <- struct{}{}:
156	default:
157	}
158	d.snapshotWorkerMu.Unlock()
159	return true
160}