1package daemon
2
3import (
4 "context"
5 "errors"
6 "sort"
7
8 "github.com/bnema/vev/internal/domain"
9)
10
11func (d *Daemon) reportSnapshotFailure(capture *snapshotCapture, phase string, cause error) {
12 if d == nil || capture == nil || capture.session == nil || cause == nil {
13 return
14 }
15 signature := snapshotFailureSignature(phase, cause)
16 d.snapshots.noticeMu.Lock()
17 changed := d.snapshots.activeFailureSignature != signature
18 d.snapshots.activeFailureSignature = signature
19 d.snapshots.noticeMu.Unlock()
20
21 n := domain.Notification{
22 Code: domain.NoticeSnapshotWrite,
23 Severity: domain.NoticeError,
24 Message: "couldn't save session state; recent state may be lost on restart",
25 // The history is intentionally diagnostic only at the error-class level.
26 // Raw causes can include user paths and terminal content.
27 Details: signature,
28 Time: d.clock.Now(),
29 }
30 n, _ = d.notices.recordSnapshotFailure(n)
31 d.log.Warn("writing session snapshot failed", "phase", phase, "class", signature, "session", capture.name)
32 if changed {
33 d.deliverGlobal(n)
34 }
35}
36
37func (d *Daemon) clearSnapshotFailure() {
38 if d == nil {
39 return
40 }
41 d.snapshots.noticeMu.Lock()
42 d.snapshots.activeFailureSignature = ""
43 d.snapshots.noticeMu.Unlock()
44}
45
46func (d *Daemon) startSnapshotEncodeWorker() {
47 if d == nil {
48 return
49 }
50 d.snapshots.mu.Lock()
51 if d.snapshots.cancel != nil {
52 d.snapshots.mu.Unlock()
53 return
54 }
55 // Shutdown owns the worker lifetime so it can flush captures after Serve
56 // cancels its parent context. The worker is always stopped explicitly.
57 workerCtx, cancel := context.WithCancel(context.Background())
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()
73 go d.runSnapshotEncodeWorker(workerCtx, workerID, done, flush, finalWake)
74}
75
76// runSnapshotEncodeWorker owns the worker state machine. Each event either
77// continues processing, drains the terminal queues for shutdown, or stops;
78// publication details are kept out of the select loop.
79func (d *Daemon) runSnapshotEncodeWorker(workerCtx context.Context, workerID uint64, done chan<- struct{}, flush, finalWake <-chan struct{}) {
80 defer close(done)
81 for {
82 select {
83 case <-workerCtx.Done():
84 return
85 case <-flush:
86 d.flushSnapshotCaptures(workerCtx, workerID)
87 return
88 case <-finalWake:
89 if !d.drainFinalSnapshotCaptures(workerCtx, workerID) {
90 return
91 }
92 case capture := <-d.snapshots.jobs:
93 if !d.publishSnapshotCapture(workerCtx, workerID, capture) {
94 return
95 }
96 }
97 }
98}
99
100func (d *Daemon) flushSnapshotCaptures(workerCtx context.Context, workerID uint64) {
101 for {
102 select {
103 case capture := <-d.snapshots.jobs:
104 if !d.publishSnapshotCapture(workerCtx, workerID, capture) {
105 return
106 }
107 default:
108 _ = d.drainFinalSnapshotCaptures(workerCtx, workerID)
109 return
110 }
111 }
112}
113
114func (d *Daemon) drainFinalSnapshotCaptures(workerCtx context.Context, workerID uint64) bool {
115 for capture := d.takeFinalSnapshotCapture(); capture != nil; capture = d.takeFinalSnapshotCapture() {
116 if !d.publishSnapshotCapture(workerCtx, workerID, capture) {
117 return false
118 }
119 }
120 return true
121}
122
123func (d *Daemon) publishSnapshotCapture(workerCtx context.Context, workerID uint64, capture *snapshotCapture) bool {
124 if workerCtx.Err() != nil || !startSnapshotPublication(capture) || !d.setSnapshotWorkerInFlight(workerID, capture) {
125 d.finishSnapshotCapture(capture, false)
126 return workerCtx.Err() == nil
127 }
128 publication, err := d.incrementalPublication(capture)
129 publicationContext := capture.publicationContext
130 if publicationContext == nil {
131 publicationContext = workerCtx
132 }
133 if err == nil && workerCtx.Err() == nil {
134 if err = publicationContext.Err(); err == nil {
135 if d.recovery != nil {
136 var record domain.CatalogueRecord
137 record, err = d.recovery.PublishCheckpoint(publicationContext, capture.name, publication)
138 if err == nil {
139 if record.Committed == nil {
140 err = errors.New("snapshot: checkpoint publication returned no committed checkpoint")
141 } else {
142 capture.checkpoint = *record.Committed
143 }
144 }
145 } else {
146 err = d.snapshotRepository.Publish(publicationContext, publication)
147 }
148 }
149 }
150 d.clearSnapshotWorkerInFlight(workerID, capture)
151 if err == nil && workerCtx.Err() == nil {
152 markSnapshotCaptureObjectsPublished(capture)
153 }
154 if err != nil && workerCtx.Err() == nil && publicationContext.Err() == nil {
155 // Global, not session-scoped: by now the session may already be torn
156 // down, so a session notice would be dead-on-arrival. No lock is held
157 // here (clearSnapshotWorkerInFlight released its own).
158 d.reportSnapshotFailure(capture, "publish", err)
159 }
160 d.finishSnapshotCapture(capture, err == nil && workerCtx.Err() == nil)
161 return workerCtx.Err() == nil
162}
163
164func (d *Daemon) takeFinalSnapshotCapture() *snapshotCapture {
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)
173 if capture != nil {
174 return capture
175 }
176 }
177 return nil
178}
179
180func (d *Daemon) setSnapshotWorkerInFlight(workerID uint64, capture *snapshotCapture) bool {
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 {
184 return false
185 }
186 d.snapshots.inFlight = capture
187 return true
188}
189
190func (d *Daemon) clearSnapshotWorkerInFlight(workerID uint64, capture *snapshotCapture) {
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
195 }
196}
197
198func (d *Daemon) stopSnapshotEncodeWorker() {
199 if d == nil {
200 return
201 }
202 _ = d.StopDurableWriters(context.Background())
203 d.WaitDurableWriters()
204}
205
206// StopDurableWriters stops maintenance scheduling, cancels cooperative
207// maintenance calls, stops snapshot admission, and requests the snapshot
208// worker's final drain. The context bounds checkpoint success only:
209// cancellation never detaches a writer from its owner.
210func (d *Daemon) StopDurableWriters(ctx context.Context) []string {
211 if d == nil {
212 return nil
213 }
214 cancel, done := d.requestDurableWriterStop()
215 if done == nil {
216 return nil
217 }
218 if ctx == nil {
219 ctx = context.Background()
220 }
221 select {
222 case <-done:
223 return nil
224 case <-ctx.Done():
225 // Snapshot identity before cancellation can let a cooperative worker clear
226 // its in-flight capture. The caller persists timeout notices before the
227 // unconditional ownership join.
228 names := d.durableWriterFailureNames()
229 cancel()
230 return names
231 }
232}
233
234// WaitDurableWriters is the unconditional ownership barrier. It has no timeout
235// or status branch: every writer goroutine must exit before Serve may return.
236// Restoration is joined with the others because it mutates durable state too;
237// it observes serveCtx cancellation, so waiting for it cannot outlive an
238// uncooperative repository call any longer than the snapshot worker does.
239func (d *Daemon) WaitDurableWriters() {
240 if d == nil {
241 return
242 }
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()
248 if snapshotDone != nil {
249 <-snapshotDone
250 d.finishStoppedSnapshotWorker(false)
251 }
252 if maintenanceDone != nil {
253 <-maintenanceDone
254 }
255 // Restoration never blocks on the workers above (snapshot admission is
256 // non-blocking), so joining it last cannot deadlock.
257 if restoreDone != nil {
258 <-restoreDone
259 }
260}
261
262func (d *Daemon) requestDurableWriterStop() (context.CancelFunc, <-chan struct{}) {
263 d.snapshots.mu.Lock()
264 defer d.snapshots.mu.Unlock()
265 d.cancelDurableMaintenanceLocked()
266 cancel := d.snapshots.cancel
267 if cancel == nil {
268 return nil, nil
269 }
270 if !d.snapshots.closing {
271 d.snapshots.closing = true
272 close(d.snapshots.flush)
273 }
274 return cancel, d.snapshots.done
275}
276
277// durableWriterFailureNames includes admitted buffered captures as well as the
278// active and final queues. snapshots.admitted tracks normal captures from queue
279// admission through completion, so worker dequeue cannot make one disappear.
280func (d *Daemon) durableWriterFailureNames() []string {
281 d.snapshots.mu.Lock()
282 defer d.snapshots.mu.Unlock()
283 seen := make(map[string]struct{})
284 add := func(capture *snapshotCapture) {
285 if capture != nil && capture.name != "" {
286 seen[capture.name] = struct{}{}
287 }
288 }
289 add(d.snapshots.inFlight)
290 for _, capture := range d.snapshots.finalJobs {
291 add(capture)
292 }
293 for capture := range d.snapshots.admitted {
294 add(capture)
295 }
296 names := make([]string, 0, len(seen))
297 for name := range seen {
298 names = append(names, name)
299 }
300 sort.Strings(names)
301 return names
302}
303
304func (d *Daemon) finishStoppedSnapshotWorker(abandoned bool) {
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 {
316 queued = append(queued, capture)
317 }
318 d.snapshots.finalJobs = nil
319 d.snapshots.finalOrder = nil
320 for {
321 select {
322 case capture := <-d.snapshots.jobs:
323 queued = append(queued, capture)
324 default:
325 d.snapshots.mu.Unlock()
326 if abandoned {
327 d.finishSnapshotCapture(inFlight, false)
328 }
329 for _, capture := range queued {
330 d.finishSnapshotCapture(capture, false)
331 }
332 return
333 }
334 }
335}