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 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) (ports.SessionState, chan struct{}) {
23 done := make(chan struct{})
24 state := ports.SessionDown
25 if record.DegradedReason != "" {
26 state = ports.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, err := d.catalogue.Records()
42 if err != nil {
43 d.log.Error("loading catalogue for snapshot restoration failed", "err", err)
44 return
45 }
46 d.restoreCatalogue(ctx, records)
47}
48
49// restoreCatalogue gives every expected record an independent bounded job.
50// A failed record is degraded without cancelling any sibling restoration.
51func (d *Daemon) restoreCatalogue(ctx context.Context, records []domain.CatalogueRecord) {
52 jobs := make(chan domain.CatalogueRecord)
53 workers := min(catalogueRestoreConcurrency, len(records))
54 var wg sync.WaitGroup
55 for range workers {
56 wg.Go(func() {
57 for record := range jobs {
58 done := d.ensureCatalogueRegistryEntry(record)
59 err := d.restoreRecord(ctx, record)
60 d.finishRecordRestore(record, err, done)
61 }
62 })
63 }
64 for _, record := range records {
65 jobs <- record
66 }
67 close(jobs)
68 wg.Wait()
69}
70
71func stoppedSessionFromRecord(record domain.CatalogueRecord, state ports.SessionState, done chan struct{}) stoppedSession {
72 tabRecords := append([]domain.CatalogueTabRecord(nil), record.TabRecords...)
73 tabNames := append([]string(nil), record.TabNames...)
74 if len(tabRecords) != 0 {
75 tabNames = make([]string, len(tabRecords))
76 for i, tab := range tabRecords {
77 tabNames[i] = tab.Name
78 }
79 } else if len(tabNames) != 0 {
80 // Legacy catalogue records had no stable IDs. Preserve every encoded
81 // ordinal, including unnamed entries, and resolve them by exact name
82 // and expected count during stopped restoration.
83 tabRecords = make([]domain.CatalogueTabRecord, len(tabNames))
84 for i, name := range tabNames {
85 tabRecords[i] = domain.CatalogueTabRecord{Name: name}
86 }
87 }
88 return stoppedSession{
89 name: record.Name,
90 cwd: record.Cwd,
91 createdAt: record.CreatedAt,
92 incarnation: record.IncarnationID,
93 lastUsedSeq: record.LastUsedSeq,
94 tabNames: tabNames,
95 tabRecords: tabRecords,
96 record: record,
97 state: state,
98 restoreDone: done,
99 }
100}
101
102func (d *Daemon) ensureCatalogueRegistryEntry(record domain.CatalogueRecord) chan struct{} {
103 d.mu.Lock()
104 defer d.mu.Unlock()
105 if entry, ok := d.stopped[record.Name]; ok {
106 return entry.restoreDone
107 }
108 state, done := initialSessionState(record)
109 d.stopped[record.Name] = stoppedSessionFromRecord(record, state, done)
110 return done
111}
112
113// closeRuntimeRestoreDoneLocked closes a per-record restoration barrier.
114// Caller must hold d.mu so state transition and waiter release are atomic.
115func closeRuntimeRestoreDoneLocked(done chan struct{}) {
116 if done == nil {
117 return
118 }
119 select {
120 case <-done:
121 default:
122 close(done)
123 }
124}
125
126func (d *Daemon) setStoppedRecovery(record domain.CatalogueRecord, state ports.SessionState) {
127 d.mu.Lock()
128 defer d.mu.Unlock()
129 entry, ok := d.stopped[record.Name]
130 if !ok {
131 return
132 }
133 d.stopped[record.Name] = stoppedSessionFromRecord(record, state, entry.restoreDone)
134}
135
136func (d *Daemon) finishRecordRestore(record domain.CatalogueRecord, restoreErr error, done chan struct{}) {
137 retryable := errors.Is(restoreErr, errRetryableRestoreLoad)
138 degraded := false
139 if restoreErr != nil {
140 reason := "checkpoint validation failed"
141 switch {
142 case retryable:
143 reason = "checkpoint load failed"
144 case errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded):
145 reason = "restore interrupted"
146 }
147 if d.recovery == nil {
148 d.log.Error("marking session degraded failed", "session", record.Name, "err", errors.New("recovery coordinator unavailable"))
149 } else if current, marked, err := d.recovery.MarkBroken(context.Background(), record.Name, record.IncarnationID, reason); err != nil {
150 d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
151 } else if marked {
152 record = current
153 degraded = true
154 }
155 }
156
157 d.mu.Lock()
158 if entry, ok := d.stopped[record.Name]; ok && degraded {
159 d.stopped[record.Name] = stoppedSessionFromRecord(record, ports.SessionBroken, entry.restoreDone)
160 }
161 closeRuntimeRestoreDoneLocked(done)
162 d.mu.Unlock()
163 if degraded {
164 reasonCode := "checkpoint-invalid"
165 switch {
166 case retryable:
167 reasonCode = "checkpoint-load-failed"
168 case errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded):
169 reasonCode = "restore-interrupted"
170 }
171 d.logSessionDegraded(record, reasonCode)
172 }
173}
174
175func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecord) error {
176 if record.DegradedReason != "" {
177 d.setStoppedRecovery(record, ports.SessionBroken)
178 d.logSessionDegraded(record, "persisted-broken")
179 return nil
180 }
181 if record.Committed == nil {
182 d.setStoppedRecovery(record, ports.SessionDown)
183 d.logSessionRestoreComplete(record, 0, false)
184 return nil
185 }
186 if d.snapshotRepository == nil {
187 return errors.New("snapshot: repository unavailable")
188 }
189
190 if record.Committed == nil {
191 return errors.New("snapshot: healthy record has no checkpoint")
192 }
193 selected := *record.Committed
194 selectedGeneration, err := d.snapshotRepository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, selected)
195 if err != nil {
196 if ctxErr := ctx.Err(); ctxErr != nil {
197 return ctxErr
198 }
199 if isUnambiguousBadVersionError(err) {
200 return d.resetIncompatibleCheckpoint(ctx, record, selected)
201 }
202 if isRetryableRestoreLoadError(err) {
203 return fmt.Errorf("%w: %w", errRetryableRestoreLoad, err)
204 }
205 d.logRejectedRestoreCandidate(record, selected, false, "load-failed", err)
206 return errors.New("snapshot: committed checkpoint did not validate")
207 }
208 selectedSnapshot, err := sessionFromGeneration(selectedGeneration)
209 if err != nil {
210 d.logRejectedRestoreCandidate(record, selected, false, "decode-failed", err)
211 return errors.New("snapshot: committed checkpoint did not validate")
212 }
213 if selectedGeneration.IncarnationID != record.IncarnationID {
214 d.logRejectedRestoreCandidate(record, selected, false, "incarnation-mismatch", nil)
215 return errors.New("snapshot: committed checkpoint did not validate")
216 }
217 if selectedGeneration.Name != record.Name {
218 d.logRejectedRestoreCandidate(record, selected, false, "name-mismatch", nil)
219 return errors.New("snapshot: committed checkpoint did not validate")
220 }
221 if selectedGeneration.Generation != selected.Generation {
222 d.logRejectedRestoreCandidate(record, selected, false, "generation-mismatch", nil)
223 return errors.New("snapshot: committed checkpoint did not validate")
224 }
225 if err := d.snapshotRepository.ReconcileCheckpoint(ctx, record.IncarnationID, selected); err != nil {
226 d.log.Warn("snapshot_checkpoint_reconciliation_pending", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation, "reason_code", "reconcile-failed")
227 } else {
228 d.log.Info("snapshot_checkpoint_reconciliation_complete", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)
229 }
230
231 if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation, selected); err != nil {
232 return err
233 }
234 d.setStoppedRecovery(record, ports.SessionDown)
235 d.logSessionRestoreComplete(record, selected.Generation, false)
236 return nil
237}
238
239func (d *Daemon) resetIncompatibleCheckpoint(ctx context.Context, record domain.CatalogueRecord, selected domain.CheckpointRef) error {
240 fresh, committed, err := d.recovery.ResetIncompatible(ctx, record.Name, record.IncarnationID, selected)
241 if !committed {
242 if err != nil {
243 return fmt.Errorf("snapshot: reset incompatible checkpoint: %w", err)
244 }
245 d.log.Info("snapshot_incompatible_reset_superseded",
246 "session", record.Name,
247 "incarnation", record.IncarnationID.String(),
248 "generation", selected.Generation,
249 )
250 return nil
251 }
252
253 d.mu.Lock()
254 if entry, ok := d.stopped[record.Name]; ok {
255 d.stopped[record.Name] = stoppedSessionFromRecord(fresh, ports.SessionDown, entry.restoreDone)
256 }
257 d.mu.Unlock()
258
259 if err != nil {
260 d.log.Warn("snapshot_incompatible_reset_cleanup_pending",
261 "session", fresh.Name,
262 "incarnation", fresh.IncarnationID.String(),
263 "replaced_incarnation", record.IncarnationID.String(),
264 "generation", selected.Generation,
265 "err", err,
266 )
267 return nil
268 }
269 d.log.Info("snapshot_incompatible_reset_complete",
270 "session", fresh.Name,
271 "incarnation", fresh.IncarnationID.String(),
272 "replaced_incarnation", record.IncarnationID.String(),
273 "generation", selected.Generation,
274 )
275 return nil
276}
277
278func isUnambiguousBadVersionError(err error) bool {
279 seen := make(map[error]struct{})
280 for err != nil {
281 if !reflect.TypeOf(err).Comparable() {
282 return false
283 }
284 if _, duplicate := seen[err]; duplicate {
285 return false
286 }
287 seen[err] = struct{}{}
288 if err == snapcodec.ErrBadVersion {
289 return true
290 }
291 if _, ambiguous := err.(interface{ Unwrap() []error }); ambiguous {
292 return false
293 }
294 unwrapper, ok := err.(interface{ Unwrap() error })
295 if !ok {
296 return false
297 }
298 err = unwrapper.Unwrap()
299 }
300 return false
301}
302
303func isRetryableRestoreLoadError(err error) bool {
304 if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
305 return false
306 }
307 if errors.Is(err, syscall.ENOSPC) ||
308 errors.Is(err, syscall.ENOMEM) ||
309 errors.Is(err, syscall.EMFILE) ||
310 errors.Is(err, syscall.ENFILE) ||
311 errors.Is(err, syscall.EDQUOT) {
312 return true
313 }
314 if netErr, ok := errors.AsType[net.Error](err); ok && netErr.Timeout() {
315 return true
316 }
317 temporary, ok := errors.AsType[interface {
318 error
319 Temporary() bool
320 }](err)
321 return ok && temporary.Temporary()
322}
323
324func (d *Daemon) logRejectedRestoreCandidate(record domain.CatalogueRecord, candidate domain.CheckpointRef, fallback bool, reason string, err error) {
325 kind := "committed"
326 if fallback {
327 kind = "fallback"
328 }
329 d.log.Warn("snapshot_restore_candidate_rejected",
330 "session", record.Name,
331 "incarnation", record.IncarnationID.String(),
332 "generation", candidate.Generation,
333 "candidate", kind,
334 "reason_code", reason,
335 "err", err,
336 )
337}
338
339func sessionFromGeneration(generation ports.SnapshotGeneration) (snapcodec.Session, error) {
340 manifest, err := snapcodec.UnmarshalManifest(generation.Manifest)
341 if err != nil {
342 return snapcodec.Session{}, err
343 }
344 if manifest.IncarnationID != generation.IncarnationID || manifest.Name != generation.Name || manifest.Generation != generation.Generation {
345 return snapcodec.Session{}, fmt.Errorf("snapshot: generation identity mismatch")
346 }
347 result := snapcodec.Session{Name: manifest.Name, CreatedAt: manifest.CreatedAt, Active: manifest.Active, Tabs: make([]snapcodec.Tab, 0, len(manifest.Tabs))}
348 for _, tab := range manifest.Tabs {
349 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))}
350 for _, pane := range tab.Panes {
351 outPane := snapcodec.Pane{ID: pane.ID, StableID: pane.StableID, Cwd: pane.Cwd, Process: pane.Process}
352 for _, ref := range pane.Sealed {
353 data, err := generationObject(generation, ref, snapcodec.HistoryChunk)
354 if err != nil {
355 return snapcodec.Session{}, err
356 }
357 outPane.SealedChunks = append(outPane.SealedChunks, data)
358 }
359 if outPane.Tail, err = generationObject(generation, pane.Tail, snapcodec.HistoryTail); err != nil {
360 return snapcodec.Session{}, err
361 }
362 if outPane.Transcript, err = generationObject(generation, pane.Transcript, snapcodec.RecoveryTranscript); err != nil {
363 return snapcodec.Session{}, err
364 }
365 outTab.Panes = append(outTab.Panes, outPane)
366 }
367 result.Tabs = append(result.Tabs, outTab)
368 }
369 return result, nil
370}
371
372func generationObject(generation ports.SnapshotGeneration, ref snapcodec.ObjectRef, kind snapcodec.ObjectKind) ([]byte, error) {
373 if ref.Kind != kind {
374 return nil, fmt.Errorf("snapshot: object kind mismatch")
375 }
376 data, ok := generation.Objects[ref.Digest]
377 if !ok || uint32(len(data)) != ref.Size || sha256.Sum256(data) != ref.Digest {
378 return nil, fmt.Errorf("snapshot: missing object")
379 }
380 gotKind, payload, err := snapcodec.UnmarshalObject(data)
381 if err != nil || gotKind != kind {
382 return nil, fmt.Errorf("snapshot: invalid object")
383 }
384 return append([]byte(nil), payload...), nil
385}