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