1package daemon
2
3import (
4 "context"
5 "crypto/sha256"
6 "errors"
7 "fmt"
8 "net"
9 "reflect"
10 "sync"
11 "syscall"
12
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/ports"
15 "github.com/bnema/vev/internal/protocol"
16 snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
17)
18
19const catalogueRestoreConcurrency = 8
20
21var errRetryableRestoreLoad = errors.New("snapshot: retryable checkpoint load")
22
23func initialSessionState(record domain.CatalogueRecord) (protocol.SessionState, chan struct{}) {
24 done := make(chan struct{})
25 state := protocol.SessionDown
26 if record.DegradedReason != "" {
27 state = protocol.SessionBroken
28 close(done)
29 } else if record.Committed == nil {
30 close(done)
31 }
32 return state, done
33}
34
35// restoreIncrementalSnapshots restores only records named by the authoritative
36// catalogue. Snapshot repository discovery is deliberately not part of this
37// path, so stale or oversized namespaces cannot affect unrelated sessions.
38func (d *Daemon) restoreIncrementalSnapshots(ctx context.Context) {
39 if d.catalogue == nil {
40 return
41 }
42 records, epoch, ok := d.captureCatalogueSnapshot(ctx)
43 if !ok {
44 return
45 }
46 d.restoreCatalogueEpoch(ctx, records, epoch)
47}
48
49// captureCatalogueSnapshot reads the purge boundary and the authoritative
50// catalogue records together under purge admission. Holding the gate for both
51// reads keeps them atomic with any KillAll: a purge can never increment the
52// epoch between the two reads and pair a record set it already removed with a
53// boundary that looks current. The caller's context and the move/purge gates
54// bound the wait.
55func (d *Daemon) captureCatalogueSnapshot(ctx context.Context) ([]domain.CatalogueRecord, uint64, bool) {
56 if err := d.acquirePurgeAdmission(ctx); err != nil {
57 return nil, 0, false
58 }
59 defer d.releasePurgeAdmission()
60 epoch := d.purgeEpochSnapshot()
61 records, err := d.catalogue.Records()
62 if err != nil {
63 d.log.Error("loading catalogue for snapshot restoration failed", "err", err)
64 return nil, 0, false
65 }
66 return records, epoch, true
67}
68
69// restoreCatalogueEpoch restores records captured at the given purge boundary.
70// Each worker holds purge admission for exactly one record's transition — its
71// registry publication, checkpoint load, and publication or demotion — so a
72// KillAll drains the in-flight record before it snapshots, and a worker that
73// waited on admission abandons records superseded by a completed purge. Holding
74// admission per record rather than per worker keeps the purge's drain latency
75// proportional to the slowest single record, not the whole catalogue.
76func (d *Daemon) restoreCatalogueEpoch(ctx context.Context, records []domain.CatalogueRecord, epoch uint64) {
77 jobs := make(chan domain.CatalogueRecord)
78 workers := min(catalogueRestoreConcurrency, len(records))
79 var wg sync.WaitGroup
80 for range workers {
81 wg.Go(func() {
82 for record := range jobs {
83 if err := d.acquirePurgeAdmission(ctx); err != nil {
84 // Admission is closed for good (shutdown or cancellation): drain
85 // the queue so the producer can finish, and touch no registry state.
86 continue
87 }
88 if ctx.Err() != nil {
89 d.releasePurgeAdmission()
90 continue
91 }
92 done, ok := d.ensureCatalogueRegistryEntry(record, epoch)
93 if !ok {
94 d.releasePurgeAdmission()
95 continue
96 }
97 err := d.restoreRecord(ctx, record, epoch)
98 d.finishRecordRestore(record, err, done)
99 d.releasePurgeAdmission()
100 }
101 })
102 }
103 for _, record := range records {
104 jobs <- record
105 }
106 close(jobs)
107 wg.Wait()
108}
109
110func inactiveSessionFromRecord(record domain.CatalogueRecord, state protocol.SessionState, done chan struct{}) inactiveSession {
111 tabRecords := append([]domain.CatalogueTabRecord(nil), record.TabRecords...)
112 tabNames := append([]string(nil), record.TabNames...)
113 if len(tabRecords) != 0 {
114 tabNames = make([]string, len(tabRecords))
115 for i, tab := range tabRecords {
116 tabNames[i] = tab.Name
117 }
118 } else if len(tabNames) != 0 {
119 // Legacy catalogue records had no stable IDs. Preserve every encoded
120 // ordinal, including unnamed entries, and resolve them by exact name
121 // and expected count during stopped restoration.
122 tabRecords = make([]domain.CatalogueTabRecord, len(tabNames))
123 for i, name := range tabNames {
124 tabRecords[i] = domain.CatalogueTabRecord{Name: name}
125 }
126 }
127 return inactiveSession{
128 name: record.Name,
129 cwd: record.Cwd,
130 createdAt: record.CreatedAt,
131 incarnation: record.IncarnationID,
132 lastUsedSeq: record.LastUsedSeq,
133 tabNames: tabNames,
134 tabRecords: tabRecords,
135 record: record,
136 state: state,
137 restoreDone: done,
138 }
139}
140
141// ensureCatalogueRegistryEntry publishes a stopped registry entry for a record
142// captured at epoch, unless a purge or explicit shutdown has superseded it. The
143// false result means the record must be skipped: it no longer belongs to the
144// authoritative set, so creating an entry here would resurrect it.
145func (d *Daemon) ensureCatalogueRegistryEntry(record domain.CatalogueRecord, epoch uint64) (chan struct{}, bool) {
146 d.mu.Lock()
147 defer d.mu.Unlock()
148 if d.purgeSupersededLocked(epoch) {
149 return nil, false
150 }
151 if entry, ok := d.inactive[record.Name]; ok {
152 return entry.restoreDone, true
153 }
154 state, done := initialSessionState(record)
155 d.inactive[record.Name] = inactiveSessionFromRecord(record, state, done)
156 return done, true
157}
158
159// closeRuntimeRestoreDoneLocked closes a per-record restoration barrier.
160// Caller must hold d.mu so state transition and waiter release are atomic.
161func closeRuntimeRestoreDoneLocked(done chan struct{}) {
162 if done == nil {
163 return
164 }
165 select {
166 case <-done:
167 default:
168 close(done)
169 }
170}
171
172func (d *Daemon) setStoppedRecovery(record domain.CatalogueRecord, state protocol.SessionState) {
173 d.mu.Lock()
174 defer d.mu.Unlock()
175 entry, ok := d.inactive[record.Name]
176 if !ok {
177 return
178 }
179 d.inactive[record.Name] = inactiveSessionFromRecord(record, state, entry.restoreDone)
180}
181
182func (d *Daemon) finishRecordRestore(record domain.CatalogueRecord, restoreErr error, done chan struct{}) {
183 retryable := errors.Is(restoreErr, errRetryableRestoreLoad)
184 degraded := false
185 if restoreErr != nil {
186 reason := "checkpoint validation failed"
187 switch {
188 case retryable:
189 reason = "checkpoint load failed"
190 case errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded):
191 reason = "restore interrupted"
192 }
193 if d.recovery == nil {
194 d.log.Error("marking session degraded failed", "session", record.Name, "err", errors.New("recovery coordinator unavailable"))
195 } else if current, marked, err := d.recovery.MarkBroken(context.Background(), record.Name, record.IncarnationID, reason); err != nil {
196 d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
197 } else if marked {
198 record = current
199 degraded = true
200 }
201 }
202
203 d.mu.Lock()
204 if entry, ok := d.inactive[record.Name]; ok && degraded {
205 d.inactive[record.Name] = inactiveSessionFromRecord(record, protocol.SessionBroken, entry.restoreDone)
206 }
207 closeRuntimeRestoreDoneLocked(done)
208 d.mu.Unlock()
209 if degraded {
210 reasonCode := "checkpoint-invalid"
211 switch {
212 case retryable:
213 reasonCode = "checkpoint-load-failed"
214 case errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded):
215 reasonCode = "restore-interrupted"
216 }
217 d.logSessionDegraded(record, reasonCode)
218 }
219}
220
221func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecord, epoch uint64) error {
222 if d.purgeSuperseded(epoch) {
223 // A purge or shutdown owns this record now; skip the load entirely so no
224 // PTY is opened for a lifecycle that will not survive.
225 return nil
226 }
227 if record.DegradedReason != "" {
228 d.setStoppedRecovery(record, protocol.SessionBroken)
229 d.logSessionDegraded(record, "persisted-broken")
230 return nil
231 }
232 if record.Committed == nil {
233 d.setStoppedRecovery(record, protocol.SessionDown)
234 d.logSessionRestoreComplete(record, 0, false)
235 return nil
236 }
237 if d.snapshotRepository == nil {
238 return errors.New("snapshot: repository unavailable")
239 }
240
241 if record.Committed == nil {
242 return errors.New("snapshot: healthy record has no checkpoint")
243 }
244 selected := *record.Committed
245 selectedGeneration, err := d.snapshotRepository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, selected)
246 if err != nil {
247 if ctxErr := ctx.Err(); ctxErr != nil {
248 return ctxErr
249 }
250 if isUnambiguousBadVersionError(err) {
251 return d.resetIncompatibleCheckpoint(ctx, record, selected)
252 }
253 if isRetryableRestoreLoadError(err) {
254 return fmt.Errorf("%w: %w", errRetryableRestoreLoad, err)
255 }
256 d.logRejectedRestoreCandidate(record, selected, false, "load-failed", err)
257 return errors.New("snapshot: committed checkpoint did not validate")
258 }
259 selectedSnapshot, err := sessionFromGeneration(selectedGeneration)
260 if err != nil {
261 d.logRejectedRestoreCandidate(record, selected, false, "decode-failed", err)
262 return errors.New("snapshot: committed checkpoint did not validate")
263 }
264 if selectedGeneration.IncarnationID != record.IncarnationID {
265 d.logRejectedRestoreCandidate(record, selected, false, "incarnation-mismatch", nil)
266 return errors.New("snapshot: committed checkpoint did not validate")
267 }
268 if selectedGeneration.Name != record.Name {
269 d.logRejectedRestoreCandidate(record, selected, false, "name-mismatch", nil)
270 return errors.New("snapshot: committed checkpoint did not validate")
271 }
272 if selectedGeneration.Generation != selected.Generation {
273 d.logRejectedRestoreCandidate(record, selected, false, "generation-mismatch", nil)
274 return errors.New("snapshot: committed checkpoint did not validate")
275 }
276 if err := d.snapshotRepository.ReconcileCheckpoint(ctx, record.IncarnationID, selected); err != nil {
277 d.log.Warn("snapshot_checkpoint_reconciliation_pending", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation, "reason_code", "reconcile-failed")
278 } else {
279 d.log.Info("snapshot_checkpoint_reconciliation_complete", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)
280 }
281
282 if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation, selected, epoch); err != nil {
283 return err
284 }
285 d.setStoppedRecovery(record, protocol.SessionDown)
286 d.logSessionRestoreComplete(record, selected.Generation, false)
287 return nil
288}
289
290func (d *Daemon) resetIncompatibleCheckpoint(ctx context.Context, record domain.CatalogueRecord, selected domain.CheckpointRef) error {
291 fresh, committed, err := d.recovery.ResetIncompatible(ctx, record.Name, record.IncarnationID, selected)
292 if !committed {
293 if err != nil {
294 return fmt.Errorf("snapshot: reset incompatible checkpoint: %w", err)
295 }
296 d.log.Info("snapshot_incompatible_reset_superseded",
297 "session", record.Name,
298 "incarnation", record.IncarnationID.String(),
299 "generation", selected.Generation,
300 )
301 return nil
302 }
303
304 d.mu.Lock()
305 if entry, ok := d.inactive[record.Name]; ok {
306 d.inactive[record.Name] = inactiveSessionFromRecord(fresh, protocol.SessionDown, entry.restoreDone)
307 }
308 d.mu.Unlock()
309
310 if err != nil {
311 d.log.Warn("snapshot_incompatible_reset_cleanup_pending",
312 "session", fresh.Name,
313 "incarnation", fresh.IncarnationID.String(),
314 "replaced_incarnation", record.IncarnationID.String(),
315 "generation", selected.Generation,
316 "err", err,
317 )
318 return nil
319 }
320 d.log.Info("snapshot_incompatible_reset_complete",
321 "session", fresh.Name,
322 "incarnation", fresh.IncarnationID.String(),
323 "replaced_incarnation", record.IncarnationID.String(),
324 "generation", selected.Generation,
325 )
326 return nil
327}
328
329func isUnambiguousBadVersionError(err error) bool {
330 seen := make(map[error]struct{})
331 for err != nil {
332 if !reflect.TypeOf(err).Comparable() {
333 return false
334 }
335 if _, duplicate := seen[err]; duplicate {
336 return false
337 }
338 seen[err] = struct{}{}
339 if err == snapcodec.ErrBadVersion {
340 return true
341 }
342 if _, ambiguous := err.(interface{ Unwrap() []error }); ambiguous {
343 return false
344 }
345 unwrapper, ok := err.(interface{ Unwrap() error })
346 if !ok {
347 return false
348 }
349 err = unwrapper.Unwrap()
350 }
351 return false
352}
353
354func isRetryableRestoreLoadError(err error) bool {
355 if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
356 return false
357 }
358 if errors.Is(err, syscall.ENOSPC) ||
359 errors.Is(err, syscall.ENOMEM) ||
360 errors.Is(err, syscall.EMFILE) ||
361 errors.Is(err, syscall.ENFILE) ||
362 errors.Is(err, syscall.EDQUOT) {
363 return true
364 }
365 if netErr, ok := errors.AsType[net.Error](err); ok && netErr.Timeout() {
366 return true
367 }
368 temporary, ok := errors.AsType[interface {
369 error
370 Temporary() bool
371 }](err)
372 return ok && temporary.Temporary()
373}
374
375func (d *Daemon) logRejectedRestoreCandidate(record domain.CatalogueRecord, candidate domain.CheckpointRef, fallback bool, reason string, err error) {
376 kind := "committed"
377 if fallback {
378 kind = "fallback"
379 }
380 d.log.Warn("snapshot_restore_candidate_rejected",
381 "session", record.Name,
382 "incarnation", record.IncarnationID.String(),
383 "generation", candidate.Generation,
384 "candidate", kind,
385 "reason_code", reason,
386 "err", err,
387 )
388}
389
390func sessionFromGeneration(generation ports.SnapshotGeneration) (snapcodec.Session, error) {
391 manifest, err := snapcodec.UnmarshalManifest(generation.Manifest)
392 if err != nil {
393 return snapcodec.Session{}, err
394 }
395 if manifest.IncarnationID != generation.IncarnationID || manifest.Name != generation.Name || manifest.Generation != generation.Generation {
396 return snapcodec.Session{}, fmt.Errorf("snapshot: generation identity mismatch")
397 }
398 result := snapcodec.Session{Name: manifest.Name, CreatedAt: manifest.CreatedAt, Active: manifest.Active, Tabs: make([]snapcodec.Tab, 0, len(manifest.Tabs))}
399 for _, tab := range manifest.Tabs {
400 outTab := snapcodec.Tab{StableID: tab.StableID, Cols: tab.Cols, Rows: tab.Rows, NextPaneID: tab.NextPaneID, Focus: tab.Focus, Tree: tab.Tree, Panes: make([]snapcodec.Pane, 0, len(tab.Panes))}
401 for _, pane := range tab.Panes {
402 outPane := snapcodec.Pane{ID: pane.ID, StableID: pane.StableID, Cwd: pane.Cwd, Process: pane.Process}
403 for _, ref := range pane.Sealed {
404 data, err := generationObject(generation, ref, snapcodec.HistoryChunk)
405 if err != nil {
406 return snapcodec.Session{}, err
407 }
408 outPane.SealedChunks = append(outPane.SealedChunks, data)
409 }
410 if outPane.Tail, err = generationObject(generation, pane.Tail, snapcodec.HistoryTail); err != nil {
411 return snapcodec.Session{}, err
412 }
413 if outPane.Transcript, err = generationObject(generation, pane.Transcript, snapcodec.RecoveryTranscript); err != nil {
414 return snapcodec.Session{}, err
415 }
416 outTab.Panes = append(outTab.Panes, outPane)
417 }
418 result.Tabs = append(result.Tabs, outTab)
419 }
420 return result, nil
421}
422
423func generationObject(generation ports.SnapshotGeneration, ref snapcodec.ObjectRef, kind snapcodec.ObjectKind) ([]byte, error) {
424 if ref.Kind != kind {
425 return nil, fmt.Errorf("snapshot: object kind mismatch")
426 }
427 data, ok := generation.Objects[ref.Digest]
428 if !ok || uint32(len(data)) != ref.Size || sha256.Sum256(data) != ref.Digest {
429 return nil, fmt.Errorf("snapshot: missing object")
430 }
431 // The payload aliases the loaded object. A generation's objects are private
432 // to this restore and are only decoded, never mutated, so copying here would
433 // double the transient footprint of every restored history chunk.
434 gotKind, payload, err := snapcodec.PreflightObject(data)
435 if err != nil || gotKind != kind {
436 return nil, fmt.Errorf("snapshot: invalid object")
437 }
438 return payload, nil
439}