detached 3c189af

feat(snapshot): reconcile and retain catalogue checkpoints

commit
3c189af35d72a80bdc09569920b1975de90380a6
author
brice · · 2026-07-26 01:43 UTC
parent
c0e00f5
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
7 files · +846 -8
Unified
UnifiedSplit
M internal/adapters/snapshot/maintenance_delete.go +0 -8
@@ -675,11 +675,3 @@ func (r *Repository) QuarantineIncarnation(ctx context.Context, id domain.Incarn
675675 func (r *Repository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
676676 return errors.New("snapshot: direct incarnation deletion not implemented")
677677 }
678-
679-func (r *Repository) MaintainSession(context.Context, ports.RetentionPlan, ports.MaintenanceBudget) (bool, error) {
680- return false, errors.New("snapshot: incarnation retention not implemented")
681-}
682-
683-func (r *Repository) Reconcile(_ context.Context, _ []domain.CatalogueRecord, cursor ports.ReconcileCursor, _ ports.MaintenanceBudget) (ports.ReconcileCursor, []ports.ReconcileFinding, error) {
684- return cursor, nil, errors.New("snapshot: incarnation reconciliation not implemented")
685-}
@@ -675,11 +675,3 @@ func (r *Repository) QuarantineIncarnation(ctx context.Context, id domain.Incarn
675 func (r *Repository) DeleteIncarnation(context.Context, domain.IncarnationID) error {675 func (r *Repository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
676 return errors.New("snapshot: direct incarnation deletion not implemented")676 return errors.New("snapshot: direct incarnation deletion not implemented")
677 }677 }
678-
679-func (r *Repository) MaintainSession(context.Context, ports.RetentionPlan, ports.MaintenanceBudget) (bool, error) {
680- return false, errors.New("snapshot: incarnation retention not implemented")
681-}
682-
683-func (r *Repository) Reconcile(_ context.Context, _ []domain.CatalogueRecord, cursor ports.ReconcileCursor, _ ports.MaintenanceBudget) (ports.ReconcileCursor, []ports.ReconcileFinding, error) {
684- return cursor, nil, errors.New("snapshot: incarnation reconciliation not implemented")
685-}
M internal/adapters/snapshot/maintenance_gc.go +369 -0
@@ -10,10 +10,22 @@ import (
1010 "sort"
1111 "strings"
1212
13+ "github.com/bnema/vev/internal/domain"
1314 "github.com/bnema/vev/internal/ports"
1415 codec "github.com/bnema/vev/internal/usecase/snapshot"
1516 )
1617
18+type retentionMaintenance struct {
19+ token string
20+ references map[ports.SnapshotDigest]struct{}
21+ validateQueue []ports.CheckpointRef
22+ validateIndex int
23+ validated bool
24+ manifestsDone bool
25+ objectShard string
26+ objectRootDone bool
27+}
28+
1729 type manifestMaintenance struct {
1830 refs map[ports.SnapshotDigest]codec.ObjectRef
1931 complete bool
@@ -346,6 +358,363 @@ func (r *Repository) sweepSession(ctx context.Context, key string, state *sessio
346358 return nil
347359 }
348360
361+// keepSet is the complete catalogue-owned retention index. Directory order and
362+// HEAD never grant retention authority.
363+func keepSet(plan ports.RetentionPlan) map[uint64]ports.SnapshotDigest {
364+ keep := make(map[uint64]ports.SnapshotDigest, len(plan.Keep))
365+ for _, ref := range plan.Keep {
366+ keep[ref.Generation] = ref.ManifestDigest
367+ }
368+ return keep
369+}
370+
371+// MaintainSession incrementally validates and retains only catalogue-indexed
372+// checkpoints. Every incarnation owns its budget and continuation state.
373+func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPlan, budget ports.MaintenanceBudget) (bool, error) {
374+ if err := ctx.Err(); err != nil {
375+ return false, err
376+ }
377+ key, err := incarnationKey(plan.IncarnationID)
378+ if err != nil {
379+ return false, err
380+ }
381+ if plan.PinAll {
382+ r.maintenanceMu.Lock()
383+ r.clearRetentionMaintenance(key)
384+ r.maintenanceMu.Unlock()
385+ return true, nil
386+ }
387+ if budget.Entries == 0 || budget.Bytes == 0 {
388+ return false, nil
389+ }
390+
391+ r.maintenanceMu.Lock()
392+ defer r.maintenanceMu.Unlock()
393+ lock := r.lockSession(key)
394+ defer r.unlockSession(lock)
395+ if r.maintenanceCursors == nil {
396+ r.maintenanceCursors = make(map[string]*maintenanceCursor)
397+ }
398+ if r.retentionSessions == nil {
399+ r.retentionSessions = make(map[string]*retentionMaintenance)
400+ }
401+
402+ token := retentionToken(plan)
403+ state := r.retentionSessions[key]
404+ if state == nil || state.token != token {
405+ r.clearRetentionMaintenance(key)
406+ state = &retentionMaintenance{
407+ token: token,
408+ references: make(map[ports.SnapshotDigest]struct{}),
409+ validateQueue: append([]ports.CheckpointRef(nil), plan.Keep...),
410+ }
411+ r.retentionSessions[key] = state
412+ }
413+ keep := keepSet(plan)
414+ if !state.validated {
415+ for state.validateIndex < len(state.validateQueue) {
416+ ref := state.validateQueue[state.validateIndex]
417+ generation, digest := ref.Generation, ref.ManifestDigest
418+ data, err := r.readBounded(r.manifestPath(plan.IncarnationID, generation))
419+ if err != nil {
420+ return false, fmt.Errorf("validate retained manifest %d: %w", generation, err)
421+ }
422+ manifest, err := codec.UnmarshalManifest(data)
423+ if err != nil || manifest.Generation != generation || manifest.IncarnationID != plan.IncarnationID || sha256.Sum256(data) != digest {
424+ return false, fmt.Errorf("validate retained manifest %d: catalogue reference mismatch", generation)
425+ }
426+ loaded, err := r.loadGeneration(ctx, manifest.Name, key, generation)
427+ if err != nil {
428+ return false, fmt.Errorf("validate retained manifest %d: %w", generation, err)
429+ }
430+ consumedBytes := uint64(len(data))
431+ for _, object := range loaded.Objects {
432+ consumedBytes += uint64(len(object))
433+ }
434+ if budget.Entries == 0 || consumedBytes > budget.Bytes {
435+ return false, nil
436+ }
437+ budget.Entries--
438+ budget.Bytes -= consumedBytes
439+ for digest := range loaded.Objects {
440+ state.references[digest] = struct{}{}
441+ }
442+ state.validateIndex++
443+ }
444+ state.validated = true
445+ }
446+
447+ purpose := "retention-manifests:" + key + ":" + token
448+ for !state.manifestsDone && budget.Entries > 0 && budget.Bytes > 0 {
449+ limit := int(min(budget.Entries, uint64(maintenanceBatch)))
450+ entries, done, err := r.readMaintenanceDir(filepath.Join(r.sessionPath(plan.IncarnationID), repositoryGenerations), limit, purpose)
451+ if errors.Is(err, os.ErrNotExist) {
452+ state.manifestsDone = true
453+ break
454+ }
455+ if err != nil {
456+ return false, err
457+ }
458+ for i, entry := range entries {
459+ path := r.manifestPath(plan.IncarnationID, 0)
460+ path = filepath.Join(filepath.Dir(path), entry.name)
461+ info, statErr := r.stat(path)
462+ if statErr != nil && !errors.Is(statErr, os.ErrNotExist) {
463+ return false, statErr
464+ }
465+ size := uint64(0)
466+ if statErr == nil && info.Size() > 0 {
467+ size = uint64(info.Size())
468+ }
469+ if budget.Entries == 0 || size > budget.Bytes {
470+ r.requeueMaintenanceEntries(filepath.Dir(path), purpose, entries[i:])
471+ return false, nil
472+ }
473+ budget.Entries--
474+ budget.Bytes -= size
475+ generation, canonical := parseGenerationFilename(entry.name)
476+ if !entry.isDir && canonical {
477+ if _, retained := keep[generation]; !retained {
478+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
479+ return false, err
480+ }
481+ if err := r.syncDirectory(filepath.Dir(path)); err != nil {
482+ return false, err
483+ }
484+ }
485+ }
486+ }
487+ state.manifestsDone = done
488+ if len(entries) == 0 && !done {
489+ return false, nil
490+ }
491+ }
492+ if !state.manifestsDone {
493+ return false, nil
494+ }
495+
496+ objectsRoot := filepath.Join(r.sessionPath(plan.IncarnationID), repositoryObjectsDir)
497+ for budget.Entries > 0 && budget.Bytes > 0 {
498+ if state.objectShard != "" {
499+ shard := state.objectShard
500+ dir := filepath.Join(objectsRoot, shard)
501+ objectPurpose := "retention-objects:" + key + ":" + shard + ":" + token
502+ entries, done, err := r.readMaintenanceDir(dir, int(min(budget.Entries, uint64(maintenanceBatch))), objectPurpose)
503+ if errors.Is(err, os.ErrNotExist) {
504+ state.objectShard = ""
505+ continue
506+ }
507+ if err != nil {
508+ return false, err
509+ }
510+ for i, entry := range entries {
511+ path := filepath.Join(dir, entry.name)
512+ info, statErr := r.stat(path)
513+ if statErr != nil && !errors.Is(statErr, os.ErrNotExist) {
514+ return false, statErr
515+ }
516+ size := uint64(0)
517+ if statErr == nil && info.Size() > 0 {
518+ size = uint64(info.Size())
519+ }
520+ if budget.Entries == 0 || size > budget.Bytes {
521+ r.requeueMaintenanceEntries(dir, objectPurpose, entries[i:])
522+ return false, nil
523+ }
524+ budget.Entries--
525+ budget.Bytes -= size
526+ digest, canonical := parseObjectDigest(entry.name)
527+ if !entry.isDir && canonical {
528+ if _, retained := state.references[digest]; !retained {
529+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
530+ return false, err
531+ }
532+ if err := r.syncDirectory(dir); err != nil {
533+ return false, err
534+ }
535+ }
536+ }
537+ }
538+ if done {
539+ state.objectShard = ""
540+ } else if len(entries) == 0 {
541+ return false, nil
542+ }
543+ continue
544+ }
545+ if state.objectRootDone {
546+ delete(r.retentionSessions, key)
547+ return true, nil
548+ }
549+ entries, done, err := r.readMaintenanceDir(objectsRoot, 1, "retention-shards:"+key+":"+token)
550+ if errors.Is(err, os.ErrNotExist) {
551+ delete(r.retentionSessions, key)
552+ return true, nil
553+ }
554+ if err != nil {
555+ return false, err
556+ }
557+ state.objectRootDone = done
558+ if len(entries) != 0 {
559+ budget.Entries--
560+ if entries[0].isDir {
561+ state.objectShard = entries[0].name
562+ }
563+ }
564+ }
565+ return false, nil
566+}
567+
568+func (r *Repository) clearRetentionMaintenance(key string) {
569+ delete(r.retentionSessions, key)
570+ for id := range r.maintenanceCursors {
571+ if strings.Contains(id, "retention-") && strings.Contains(id, key) {
572+ delete(r.maintenanceCursors, id)
573+ }
574+ }
575+}
576+
577+func retentionToken(plan ports.RetentionPlan) string {
578+ var b strings.Builder
579+ b.WriteString(plan.IncarnationID.String())
580+ for _, ref := range plan.Keep {
581+ fmt.Fprintf(&b, ":%d:%x", ref.Generation, ref.ManifestDigest)
582+ }
583+ return b.String()
584+}
585+
586+// Reconcile scans top-level incarnation namespaces from an external seek
587+// cookie. Checkpoint-chain work receives a fresh budget for every directory.
588+func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRecord, cursor ports.ReconcileCursor, budget ports.MaintenanceBudget) (ports.ReconcileCursor, []ports.ReconcileFinding, error) {
589+ if err := ctx.Err(); err != nil {
590+ return cursor, nil, err
591+ }
592+ if budget.Entries == 0 || budget.Bytes == 0 {
593+ return cursor, nil, nil
594+ }
595+ offset := int64(cursor.DirectoryCookie)
596+ if offset < 0 {
597+ return cursor, nil, errors.New("snapshot: invalid reconciliation cursor")
598+ }
599+ byIncarnation := make(map[domain.IncarnationID]domain.CatalogueRecord, len(records))
600+ for _, record := range records {
601+ byIncarnation[record.IncarnationID] = record
602+ }
603+
604+ r.maintenanceMu.Lock()
605+ defer r.maintenanceMu.Unlock()
606+ root := filepath.Join(r.dir, repositorySessionsDir)
607+ state := &maintenanceCursor{offset: offset}
608+ findingCapacity := maintenanceBatch
609+ if budget.Entries < maintenanceBatch {
610+ findingCapacity = int(budget.Entries)
611+ }
612+ findings := make([]ports.ReconcileFinding, 0, findingCapacity)
613+ for scanned := uint64(0); scanned < budget.Entries; scanned++ {
614+ entries, done, err := r.readMaintenanceDirent(root, 1, state)
615+ if errors.Is(err, os.ErrNotExist) {
616+ return ports.ReconcileCursor{}, findings, nil
617+ }
618+ if err != nil {
619+ return cursor, nil, err
620+ }
621+ if len(entries) == 0 {
622+ if done {
623+ return ports.ReconcileCursor{}, findings, nil
624+ }
625+ continue
626+ }
627+ entry := entries[0]
628+ next := ports.ReconcileCursor{DirectoryCookie: uint64(state.offset)}
629+ if !entry.isDir || isQuarantine(entry.name) {
630+ if done {
631+ return ports.ReconcileCursor{}, findings, nil
632+ }
633+ continue
634+ }
635+ var id domain.IncarnationID
636+ if err := id.UnmarshalText([]byte(entry.name)); err != nil {
637+ findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name}, Cursor: next, Consumed: ports.MaintenanceBudget{Entries: 1}})
638+ } else if record, ok := byIncarnation[id]; !ok {
639+ findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name, IncarnationID: id}, Cursor: next, Consumed: ports.MaintenanceBudget{Entries: 1}})
640+ } else {
641+ finding := r.reconcileIncarnation(ctx, record, budget)
642+ finding.Cursor = next
643+ findings = append(findings, finding)
644+ }
645+ if done {
646+ return ports.ReconcileCursor{}, findings, nil
647+ }
648+ }
649+ return ports.ReconcileCursor{DirectoryCookie: uint64(state.offset)}, findings, nil
650+}
651+
652+func (r *Repository) reconcileIncarnation(ctx context.Context, record domain.CatalogueRecord, budget ports.MaintenanceBudget) ports.ReconcileFinding {
653+ finding := ports.ReconcileFinding{Kind: ports.ReconcileInvalidCandidate, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: record.Name, IncarnationID: record.IncarnationID}, Consumed: ports.MaintenanceBudget{Entries: 1}}
654+ if record.Committed == nil {
655+ return finding
656+ }
657+ generation, digest, err := r.readHead(record.IncarnationID)
658+ if err != nil {
659+ return finding
660+ }
661+ candidateRef := domain.CheckpointRef{Generation: generation, ManifestDigest: digest}
662+ finding.Candidate.Ref = candidateRef
663+ if generation <= record.Committed.Generation {
664+ finding.Status = ports.ReconcileValidated
665+ return finding
666+ }
667+ finding.Kind = ports.ReconcileForwardOrphan
668+
669+ candidate, exhausted, err := r.reconcileLoad(ctx, record, candidateRef, &finding.Consumed, budget)
670+ if exhausted {
671+ finding.Status = ports.ReconcileBudgetExhausted
672+ return finding
673+ }
674+ if err != nil {
675+ return finding
676+ }
677+ finding.Candidate.Parent = candidate.ParentCheckpoint
678+ parent := candidate.ParentCheckpoint
679+ for parent != nil && *parent != *record.Committed {
680+ ancestor, exhausted, err := r.reconcileLoad(ctx, record, *parent, &finding.Consumed, budget)
681+ if exhausted {
682+ finding.Status = ports.ReconcileBudgetExhausted
683+ return finding
684+ }
685+ if err != nil {
686+ return finding
687+ }
688+ finding.AncestorChain = append(finding.AncestorChain, ports.ValidatedCheckpoint{Ref: *parent, Parent: ancestor.ParentCheckpoint})
689+ parent = ancestor.ParentCheckpoint
690+ }
691+ if parent == nil {
692+ return finding
693+ }
694+ finding.Status = ports.ReconcileValidated
695+ return finding
696+}
697+
698+func (r *Repository) reconcileLoad(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef, consumed *ports.MaintenanceBudget, budget ports.MaintenanceBudget) (ports.SnapshotGeneration, bool, error) {
699+ if consumed.Entries >= budget.Entries {
700+ return ports.SnapshotGeneration{}, true, nil
701+ }
702+ generation, err := r.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)
703+ if err != nil {
704+ return ports.SnapshotGeneration{}, false, err
705+ }
706+ bytes := uint64(len(generation.Manifest))
707+ for _, object := range generation.Objects {
708+ bytes += uint64(len(object))
709+ }
710+ if bytes > budget.Bytes-consumed.Bytes {
711+ return ports.SnapshotGeneration{}, true, nil
712+ }
713+ consumed.Entries++
714+ consumed.Bytes += bytes
715+ return generation, false, nil
716+}
717+
349718 func retainedReferences(marked map[uint64]manifestMaintenance, conservative bool) map[ports.SnapshotDigest]struct{} {
350719 if conservative {
351720 references := make(map[ports.SnapshotDigest]struct{})
@@ -10,10 +10,22 @@ import (
10 "sort"10 "sort"
11 "strings"11 "strings"
12 12
13+ "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"14 "github.com/bnema/vev/internal/ports"
14 codec "github.com/bnema/vev/internal/usecase/snapshot"15 codec "github.com/bnema/vev/internal/usecase/snapshot"
15 )16 )
16 17
18+type retentionMaintenance struct {
19+ token string
20+ references map[ports.SnapshotDigest]struct{}
21+ validateQueue []ports.CheckpointRef
22+ validateIndex int
23+ validated bool
24+ manifestsDone bool
25+ objectShard string
26+ objectRootDone bool
27+}
28+
17 type manifestMaintenance struct {29 type manifestMaintenance struct {
18 refs map[ports.SnapshotDigest]codec.ObjectRef30 refs map[ports.SnapshotDigest]codec.ObjectRef
19 complete bool31 complete bool
@@ -346,6 +358,363 @@ func (r *Repository) sweepSession(ctx context.Context, key string, state *sessio
346 return nil358 return nil
347 }359 }
348 360
361+// keepSet is the complete catalogue-owned retention index. Directory order and
362+// HEAD never grant retention authority.
363+func keepSet(plan ports.RetentionPlan) map[uint64]ports.SnapshotDigest {
364+ keep := make(map[uint64]ports.SnapshotDigest, len(plan.Keep))
365+ for _, ref := range plan.Keep {
366+ keep[ref.Generation] = ref.ManifestDigest
367+ }
368+ return keep
369+}
370+
371+// MaintainSession incrementally validates and retains only catalogue-indexed
372+// checkpoints. Every incarnation owns its budget and continuation state.
373+func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPlan, budget ports.MaintenanceBudget) (bool, error) {
374+ if err := ctx.Err(); err != nil {
375+ return false, err
376+ }
377+ key, err := incarnationKey(plan.IncarnationID)
378+ if err != nil {
379+ return false, err
380+ }
381+ if plan.PinAll {
382+ r.maintenanceMu.Lock()
383+ r.clearRetentionMaintenance(key)
384+ r.maintenanceMu.Unlock()
385+ return true, nil
386+ }
387+ if budget.Entries == 0 || budget.Bytes == 0 {
388+ return false, nil
389+ }
390+
391+ r.maintenanceMu.Lock()
392+ defer r.maintenanceMu.Unlock()
393+ lock := r.lockSession(key)
394+ defer r.unlockSession(lock)
395+ if r.maintenanceCursors == nil {
396+ r.maintenanceCursors = make(map[string]*maintenanceCursor)
397+ }
398+ if r.retentionSessions == nil {
399+ r.retentionSessions = make(map[string]*retentionMaintenance)
400+ }
401+
402+ token := retentionToken(plan)
403+ state := r.retentionSessions[key]
404+ if state == nil || state.token != token {
405+ r.clearRetentionMaintenance(key)
406+ state = &retentionMaintenance{
407+ token: token,
408+ references: make(map[ports.SnapshotDigest]struct{}),
409+ validateQueue: append([]ports.CheckpointRef(nil), plan.Keep...),
410+ }
411+ r.retentionSessions[key] = state
412+ }
413+ keep := keepSet(plan)
414+ if !state.validated {
415+ for state.validateIndex < len(state.validateQueue) {
416+ ref := state.validateQueue[state.validateIndex]
417+ generation, digest := ref.Generation, ref.ManifestDigest
418+ data, err := r.readBounded(r.manifestPath(plan.IncarnationID, generation))
419+ if err != nil {
420+ return false, fmt.Errorf("validate retained manifest %d: %w", generation, err)
421+ }
422+ manifest, err := codec.UnmarshalManifest(data)
423+ if err != nil || manifest.Generation != generation || manifest.IncarnationID != plan.IncarnationID || sha256.Sum256(data) != digest {
424+ return false, fmt.Errorf("validate retained manifest %d: catalogue reference mismatch", generation)
425+ }
426+ loaded, err := r.loadGeneration(ctx, manifest.Name, key, generation)
427+ if err != nil {
428+ return false, fmt.Errorf("validate retained manifest %d: %w", generation, err)
429+ }
430+ consumedBytes := uint64(len(data))
431+ for _, object := range loaded.Objects {
432+ consumedBytes += uint64(len(object))
433+ }
434+ if budget.Entries == 0 || consumedBytes > budget.Bytes {
435+ return false, nil
436+ }
437+ budget.Entries--
438+ budget.Bytes -= consumedBytes
439+ for digest := range loaded.Objects {
440+ state.references[digest] = struct{}{}
441+ }
442+ state.validateIndex++
443+ }
444+ state.validated = true
445+ }
446+
447+ purpose := "retention-manifests:" + key + ":" + token
448+ for !state.manifestsDone && budget.Entries > 0 && budget.Bytes > 0 {
449+ limit := int(min(budget.Entries, uint64(maintenanceBatch)))
450+ entries, done, err := r.readMaintenanceDir(filepath.Join(r.sessionPath(plan.IncarnationID), repositoryGenerations), limit, purpose)
451+ if errors.Is(err, os.ErrNotExist) {
452+ state.manifestsDone = true
453+ break
454+ }
455+ if err != nil {
456+ return false, err
457+ }
458+ for i, entry := range entries {
459+ path := r.manifestPath(plan.IncarnationID, 0)
460+ path = filepath.Join(filepath.Dir(path), entry.name)
461+ info, statErr := r.stat(path)
462+ if statErr != nil && !errors.Is(statErr, os.ErrNotExist) {
463+ return false, statErr
464+ }
465+ size := uint64(0)
466+ if statErr == nil && info.Size() > 0 {
467+ size = uint64(info.Size())
468+ }
469+ if budget.Entries == 0 || size > budget.Bytes {
470+ r.requeueMaintenanceEntries(filepath.Dir(path), purpose, entries[i:])
471+ return false, nil
472+ }
473+ budget.Entries--
474+ budget.Bytes -= size
475+ generation, canonical := parseGenerationFilename(entry.name)
476+ if !entry.isDir && canonical {
477+ if _, retained := keep[generation]; !retained {
478+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
479+ return false, err
480+ }
481+ if err := r.syncDirectory(filepath.Dir(path)); err != nil {
482+ return false, err
483+ }
484+ }
485+ }
486+ }
487+ state.manifestsDone = done
488+ if len(entries) == 0 && !done {
489+ return false, nil
490+ }
491+ }
492+ if !state.manifestsDone {
493+ return false, nil
494+ }
495+
496+ objectsRoot := filepath.Join(r.sessionPath(plan.IncarnationID), repositoryObjectsDir)
497+ for budget.Entries > 0 && budget.Bytes > 0 {
498+ if state.objectShard != "" {
499+ shard := state.objectShard
500+ dir := filepath.Join(objectsRoot, shard)
501+ objectPurpose := "retention-objects:" + key + ":" + shard + ":" + token
502+ entries, done, err := r.readMaintenanceDir(dir, int(min(budget.Entries, uint64(maintenanceBatch))), objectPurpose)
503+ if errors.Is(err, os.ErrNotExist) {
504+ state.objectShard = ""
505+ continue
506+ }
507+ if err != nil {
508+ return false, err
509+ }
510+ for i, entry := range entries {
511+ path := filepath.Join(dir, entry.name)
512+ info, statErr := r.stat(path)
513+ if statErr != nil && !errors.Is(statErr, os.ErrNotExist) {
514+ return false, statErr
515+ }
516+ size := uint64(0)
517+ if statErr == nil && info.Size() > 0 {
518+ size = uint64(info.Size())
519+ }
520+ if budget.Entries == 0 || size > budget.Bytes {
521+ r.requeueMaintenanceEntries(dir, objectPurpose, entries[i:])
522+ return false, nil
523+ }
524+ budget.Entries--
525+ budget.Bytes -= size
526+ digest, canonical := parseObjectDigest(entry.name)
527+ if !entry.isDir && canonical {
528+ if _, retained := state.references[digest]; !retained {
529+ if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
530+ return false, err
531+ }
532+ if err := r.syncDirectory(dir); err != nil {
533+ return false, err
534+ }
535+ }
536+ }
537+ }
538+ if done {
539+ state.objectShard = ""
540+ } else if len(entries) == 0 {
541+ return false, nil
542+ }
543+ continue
544+ }
545+ if state.objectRootDone {
546+ delete(r.retentionSessions, key)
547+ return true, nil
548+ }
549+ entries, done, err := r.readMaintenanceDir(objectsRoot, 1, "retention-shards:"+key+":"+token)
550+ if errors.Is(err, os.ErrNotExist) {
551+ delete(r.retentionSessions, key)
552+ return true, nil
553+ }
554+ if err != nil {
555+ return false, err
556+ }
557+ state.objectRootDone = done
558+ if len(entries) != 0 {
559+ budget.Entries--
560+ if entries[0].isDir {
561+ state.objectShard = entries[0].name
562+ }
563+ }
564+ }
565+ return false, nil
566+}
567+
568+func (r *Repository) clearRetentionMaintenance(key string) {
569+ delete(r.retentionSessions, key)
570+ for id := range r.maintenanceCursors {
571+ if strings.Contains(id, "retention-") && strings.Contains(id, key) {
572+ delete(r.maintenanceCursors, id)
573+ }
574+ }
575+}
576+
577+func retentionToken(plan ports.RetentionPlan) string {
578+ var b strings.Builder
579+ b.WriteString(plan.IncarnationID.String())
580+ for _, ref := range plan.Keep {
581+ fmt.Fprintf(&b, ":%d:%x", ref.Generation, ref.ManifestDigest)
582+ }
583+ return b.String()
584+}
585+
586+// Reconcile scans top-level incarnation namespaces from an external seek
587+// cookie. Checkpoint-chain work receives a fresh budget for every directory.
588+func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRecord, cursor ports.ReconcileCursor, budget ports.MaintenanceBudget) (ports.ReconcileCursor, []ports.ReconcileFinding, error) {
589+ if err := ctx.Err(); err != nil {
590+ return cursor, nil, err
591+ }
592+ if budget.Entries == 0 || budget.Bytes == 0 {
593+ return cursor, nil, nil
594+ }
595+ offset := int64(cursor.DirectoryCookie)
596+ if offset < 0 {
597+ return cursor, nil, errors.New("snapshot: invalid reconciliation cursor")
598+ }
599+ byIncarnation := make(map[domain.IncarnationID]domain.CatalogueRecord, len(records))
600+ for _, record := range records {
601+ byIncarnation[record.IncarnationID] = record
602+ }
603+
604+ r.maintenanceMu.Lock()
605+ defer r.maintenanceMu.Unlock()
606+ root := filepath.Join(r.dir, repositorySessionsDir)
607+ state := &maintenanceCursor{offset: offset}
608+ findingCapacity := maintenanceBatch
609+ if budget.Entries < maintenanceBatch {
610+ findingCapacity = int(budget.Entries)
611+ }
612+ findings := make([]ports.ReconcileFinding, 0, findingCapacity)
613+ for scanned := uint64(0); scanned < budget.Entries; scanned++ {
614+ entries, done, err := r.readMaintenanceDirent(root, 1, state)
615+ if errors.Is(err, os.ErrNotExist) {
616+ return ports.ReconcileCursor{}, findings, nil
617+ }
618+ if err != nil {
619+ return cursor, nil, err
620+ }
621+ if len(entries) == 0 {
622+ if done {
623+ return ports.ReconcileCursor{}, findings, nil
624+ }
625+ continue
626+ }
627+ entry := entries[0]
628+ next := ports.ReconcileCursor{DirectoryCookie: uint64(state.offset)}
629+ if !entry.isDir || isQuarantine(entry.name) {
630+ if done {
631+ return ports.ReconcileCursor{}, findings, nil
632+ }
633+ continue
634+ }
635+ var id domain.IncarnationID
636+ if err := id.UnmarshalText([]byte(entry.name)); err != nil {
637+ findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name}, Cursor: next, Consumed: ports.MaintenanceBudget{Entries: 1}})
638+ } else if record, ok := byIncarnation[id]; !ok {
639+ findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name, IncarnationID: id}, Cursor: next, Consumed: ports.MaintenanceBudget{Entries: 1}})
640+ } else {
641+ finding := r.reconcileIncarnation(ctx, record, budget)
642+ finding.Cursor = next
643+ findings = append(findings, finding)
644+ }
645+ if done {
646+ return ports.ReconcileCursor{}, findings, nil
647+ }
648+ }
649+ return ports.ReconcileCursor{DirectoryCookie: uint64(state.offset)}, findings, nil
650+}
651+
652+func (r *Repository) reconcileIncarnation(ctx context.Context, record domain.CatalogueRecord, budget ports.MaintenanceBudget) ports.ReconcileFinding {
653+ finding := ports.ReconcileFinding{Kind: ports.ReconcileInvalidCandidate, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: record.Name, IncarnationID: record.IncarnationID}, Consumed: ports.MaintenanceBudget{Entries: 1}}
654+ if record.Committed == nil {
655+ return finding
656+ }
657+ generation, digest, err := r.readHead(record.IncarnationID)
658+ if err != nil {
659+ return finding
660+ }
661+ candidateRef := domain.CheckpointRef{Generation: generation, ManifestDigest: digest}
662+ finding.Candidate.Ref = candidateRef
663+ if generation <= record.Committed.Generation {
664+ finding.Status = ports.ReconcileValidated
665+ return finding
666+ }
667+ finding.Kind = ports.ReconcileForwardOrphan
668+
669+ candidate, exhausted, err := r.reconcileLoad(ctx, record, candidateRef, &finding.Consumed, budget)
670+ if exhausted {
671+ finding.Status = ports.ReconcileBudgetExhausted
672+ return finding
673+ }
674+ if err != nil {
675+ return finding
676+ }
677+ finding.Candidate.Parent = candidate.ParentCheckpoint
678+ parent := candidate.ParentCheckpoint
679+ for parent != nil && *parent != *record.Committed {
680+ ancestor, exhausted, err := r.reconcileLoad(ctx, record, *parent, &finding.Consumed, budget)
681+ if exhausted {
682+ finding.Status = ports.ReconcileBudgetExhausted
683+ return finding
684+ }
685+ if err != nil {
686+ return finding
687+ }
688+ finding.AncestorChain = append(finding.AncestorChain, ports.ValidatedCheckpoint{Ref: *parent, Parent: ancestor.ParentCheckpoint})
689+ parent = ancestor.ParentCheckpoint
690+ }
691+ if parent == nil {
692+ return finding
693+ }
694+ finding.Status = ports.ReconcileValidated
695+ return finding
696+}
697+
698+func (r *Repository) reconcileLoad(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef, consumed *ports.MaintenanceBudget, budget ports.MaintenanceBudget) (ports.SnapshotGeneration, bool, error) {
699+ if consumed.Entries >= budget.Entries {
700+ return ports.SnapshotGeneration{}, true, nil
701+ }
702+ generation, err := r.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)
703+ if err != nil {
704+ return ports.SnapshotGeneration{}, false, err
705+ }
706+ bytes := uint64(len(generation.Manifest))
707+ for _, object := range generation.Objects {
708+ bytes += uint64(len(object))
709+ }
710+ if bytes > budget.Bytes-consumed.Bytes {
711+ return ports.SnapshotGeneration{}, true, nil
712+ }
713+ consumed.Entries++
714+ consumed.Bytes += bytes
715+ return generation, false, nil
716+}
717+
349 func retainedReferences(marked map[uint64]manifestMaintenance, conservative bool) map[ports.SnapshotDigest]struct{} {718 func retainedReferences(marked map[uint64]manifestMaintenance, conservative bool) map[ports.SnapshotDigest]struct{} {
350 if conservative {719 if conservative {
351 references := make(map[ports.SnapshotDigest]struct{})720 references := make(map[ports.SnapshotDigest]struct{})
M internal/adapters/snapshot/maintenance_test.go +111 -0
@@ -15,6 +15,7 @@ import (
1515 "testing"
1616 "unsafe"
1717
18+ "github.com/bnema/vev/internal/domain"
1819 "github.com/bnema/vev/internal/ports"
1920 )
2021
@@ -1008,3 +1009,113 @@ func digestInShard(t *testing.T, shard string, n int) ports.SnapshotDigest {
10081009 t.Fatalf("did not find stale digest in shard %q", shard)
10091010 return ports.SnapshotDigest{}
10101011 }
1012+
1013+func TestRetentionSlots(t *testing.T) {
1014+ for _, tc := range []struct {
1015+ name string
1016+ refs []uint64
1017+ want []uint64
1018+ }{
1019+ {name: "first checkpoint", refs: []uint64{1}, want: []uint64{1}},
1020+ {name: "second checkpoint", refs: []uint64{2, 1}, want: []uint64{2, 1}},
1021+ {name: "committed and two predecessors", refs: []uint64{4, 3, 2}, want: []uint64{4, 3, 2}},
1022+ {name: "fallback hole", refs: []uint64{4, 2}, want: []uint64{4, 2}},
1023+ } {
1024+ t.Run(tc.name, func(t *testing.T) {
1025+ repo := NewRepository(privateDir(t))
1026+ publications := publishMaintenanceGenerations(t, repo, "work", 4)
1027+ plan := ports.RetentionPlan{IncarnationID: publications[0].IncarnationID}
1028+ for _, generation := range tc.refs {
1029+ plan.Keep = append(plan.Keep, checkpointRefForPublication(publications[generation-1]))
1030+ }
1031+ done, err := repo.MaintainSession(context.Background(), plan, ports.MaintenanceBudget{Entries: 64, Bytes: 8 << 20})
1032+ if err != nil {
1033+ t.Fatal(err)
1034+ }
1035+ if !done {
1036+ t.Fatal("MaintainSession() done = false with sufficient budget")
1037+ }
1038+ if got := remainingMaintenanceGenerations(t, repo, plan.IncarnationID); !slices.Equal(got, tc.want) {
1039+ t.Fatalf("remaining generations = %v, want %v", got, tc.want)
1040+ }
1041+ })
1042+ }
1043+}
1044+
1045+func TestUnresolvedDataPinned(t *testing.T) {
1046+ for _, state := range []string{"degraded", "deleting", "migration", "transaction"} {
1047+ t.Run(state, func(t *testing.T) {
1048+ repo := NewRepository(privateDir(t))
1049+ publications := publishMaintenanceGenerations(t, repo, state, 4)
1050+ done, err := repo.MaintainSession(context.Background(), ports.RetentionPlan{IncarnationID: publications[0].IncarnationID, PinAll: true}, ports.MaintenanceBudget{Entries: 1, Bytes: 1})
1051+ if err != nil {
1052+ t.Fatal(err)
1053+ }
1054+ if !done {
1055+ t.Fatal("pinned maintenance did not complete without traversing")
1056+ }
1057+ if got := remainingMaintenanceGenerations(t, repo, publications[0].IncarnationID); !slices.Equal(got, []uint64{4, 3, 2, 1}) {
1058+ t.Fatalf("pinned generations = %v", got)
1059+ }
1060+ })
1061+ }
1062+}
1063+
1064+func TestPerSessionBudgetIsolation(t *testing.T) {
1065+ repo := NewRepository(privateDir(t))
1066+ large := publishMaintenanceGenerations(t, repo, "large", 4)
1067+ small := publishMaintenanceGenerations(t, repo, "small", 2)
1068+ largePlan := ports.RetentionPlan{IncarnationID: large[0].IncarnationID, Keep: []ports.CheckpointRef{checkpointRefForPublication(large[3])}}
1069+ smallPlan := ports.RetentionPlan{IncarnationID: small[0].IncarnationID, Keep: []ports.CheckpointRef{checkpointRefForPublication(small[1])}}
1070+
1071+ done, err := repo.MaintainSession(context.Background(), largePlan, ports.MaintenanceBudget{Entries: 1, Bytes: 1})
1072+ if err != nil {
1073+ t.Fatal(err)
1074+ }
1075+ if done {
1076+ t.Fatal("large session unexpectedly exhausted no budget")
1077+ }
1078+ done, err = repo.MaintainSession(context.Background(), smallPlan, ports.MaintenanceBudget{Entries: 64, Bytes: 8 << 20})
1079+ if err != nil {
1080+ t.Fatal(err)
1081+ }
1082+ if !done {
1083+ t.Fatal("small session did not progress after unrelated exhaustion")
1084+ }
1085+ if got := remainingMaintenanceGenerations(t, repo, small[0].IncarnationID); !slices.Equal(got, []uint64{2}) {
1086+ t.Fatalf("small generations = %v, want [2]", got)
1087+ }
1088+}
1089+
1090+func publishMaintenanceGenerations(t *testing.T, repo *Repository, name string, count uint64) []ports.SnapshotPublication {
1091+ t.Helper()
1092+ publications := make([]ports.SnapshotPublication, 0, count)
1093+ for generation := uint64(1); generation <= count; generation++ {
1094+ publication := repositoryPublicationAfter(t, repo, name, generation, []byte{byte(generation)})
1095+ if err := repo.Publish(context.Background(), publication); err != nil {
1096+ t.Fatal(err)
1097+ }
1098+ publications = append(publications, publication)
1099+ }
1100+ return publications
1101+}
1102+
1103+func checkpointRefForPublication(publication ports.SnapshotPublication) domain.CheckpointRef {
1104+ return domain.CheckpointRef{Generation: publication.Generation, ManifestDigest: sha256.Sum256(publication.Manifest)}
1105+}
1106+
1107+func remainingMaintenanceGenerations(t *testing.T, repo *Repository, id domain.IncarnationID) []uint64 {
1108+ t.Helper()
1109+ entries, err := os.ReadDir(filepath.Join(repo.sessionPath(id), repositoryGenerations))
1110+ if err != nil {
1111+ t.Fatal(err)
1112+ }
1113+ generations := make([]uint64, 0, len(entries))
1114+ for _, entry := range entries {
1115+ if generation, ok := parseGenerationFilename(entry.Name()); ok {
1116+ generations = append(generations, generation)
1117+ }
1118+ }
1119+ slices.SortFunc(generations, func(a, b uint64) int { return int(b) - int(a) })
1120+ return generations
1121+}
@@ -15,6 +15,7 @@ import (
15 "testing"15 "testing"
16 "unsafe"16 "unsafe"
17 17
18+ "github.com/bnema/vev/internal/domain"
18 "github.com/bnema/vev/internal/ports"19 "github.com/bnema/vev/internal/ports"
19 )20 )
20 21
@@ -1008,3 +1009,113 @@ func digestInShard(t *testing.T, shard string, n int) ports.SnapshotDigest {
1008 t.Fatalf("did not find stale digest in shard %q", shard)1009 t.Fatalf("did not find stale digest in shard %q", shard)
1009 return ports.SnapshotDigest{}1010 return ports.SnapshotDigest{}
1010 }1011 }
1012+
1013+func TestRetentionSlots(t *testing.T) {
1014+ for _, tc := range []struct {
1015+ name string
1016+ refs []uint64
1017+ want []uint64
1018+ }{
1019+ {name: "first checkpoint", refs: []uint64{1}, want: []uint64{1}},
1020+ {name: "second checkpoint", refs: []uint64{2, 1}, want: []uint64{2, 1}},
1021+ {name: "committed and two predecessors", refs: []uint64{4, 3, 2}, want: []uint64{4, 3, 2}},
1022+ {name: "fallback hole", refs: []uint64{4, 2}, want: []uint64{4, 2}},
1023+ } {
1024+ t.Run(tc.name, func(t *testing.T) {
1025+ repo := NewRepository(privateDir(t))
1026+ publications := publishMaintenanceGenerations(t, repo, "work", 4)
1027+ plan := ports.RetentionPlan{IncarnationID: publications[0].IncarnationID}
1028+ for _, generation := range tc.refs {
1029+ plan.Keep = append(plan.Keep, checkpointRefForPublication(publications[generation-1]))
1030+ }
1031+ done, err := repo.MaintainSession(context.Background(), plan, ports.MaintenanceBudget{Entries: 64, Bytes: 8 << 20})
1032+ if err != nil {
1033+ t.Fatal(err)
1034+ }
1035+ if !done {
1036+ t.Fatal("MaintainSession() done = false with sufficient budget")
1037+ }
1038+ if got := remainingMaintenanceGenerations(t, repo, plan.IncarnationID); !slices.Equal(got, tc.want) {
1039+ t.Fatalf("remaining generations = %v, want %v", got, tc.want)
1040+ }
1041+ })
1042+ }
1043+}
1044+
1045+func TestUnresolvedDataPinned(t *testing.T) {
1046+ for _, state := range []string{"degraded", "deleting", "migration", "transaction"} {
1047+ t.Run(state, func(t *testing.T) {
1048+ repo := NewRepository(privateDir(t))
1049+ publications := publishMaintenanceGenerations(t, repo, state, 4)
1050+ done, err := repo.MaintainSession(context.Background(), ports.RetentionPlan{IncarnationID: publications[0].IncarnationID, PinAll: true}, ports.MaintenanceBudget{Entries: 1, Bytes: 1})
1051+ if err != nil {
1052+ t.Fatal(err)
1053+ }
1054+ if !done {
1055+ t.Fatal("pinned maintenance did not complete without traversing")
1056+ }
1057+ if got := remainingMaintenanceGenerations(t, repo, publications[0].IncarnationID); !slices.Equal(got, []uint64{4, 3, 2, 1}) {
1058+ t.Fatalf("pinned generations = %v", got)
1059+ }
1060+ })
1061+ }
1062+}
1063+
1064+func TestPerSessionBudgetIsolation(t *testing.T) {
1065+ repo := NewRepository(privateDir(t))
1066+ large := publishMaintenanceGenerations(t, repo, "large", 4)
1067+ small := publishMaintenanceGenerations(t, repo, "small", 2)
1068+ largePlan := ports.RetentionPlan{IncarnationID: large[0].IncarnationID, Keep: []ports.CheckpointRef{checkpointRefForPublication(large[3])}}
1069+ smallPlan := ports.RetentionPlan{IncarnationID: small[0].IncarnationID, Keep: []ports.CheckpointRef{checkpointRefForPublication(small[1])}}
1070+
1071+ done, err := repo.MaintainSession(context.Background(), largePlan, ports.MaintenanceBudget{Entries: 1, Bytes: 1})
1072+ if err != nil {
1073+ t.Fatal(err)
1074+ }
1075+ if done {
1076+ t.Fatal("large session unexpectedly exhausted no budget")
1077+ }
1078+ done, err = repo.MaintainSession(context.Background(), smallPlan, ports.MaintenanceBudget{Entries: 64, Bytes: 8 << 20})
1079+ if err != nil {
1080+ t.Fatal(err)
1081+ }
1082+ if !done {
1083+ t.Fatal("small session did not progress after unrelated exhaustion")
1084+ }
1085+ if got := remainingMaintenanceGenerations(t, repo, small[0].IncarnationID); !slices.Equal(got, []uint64{2}) {
1086+ t.Fatalf("small generations = %v, want [2]", got)
1087+ }
1088+}
1089+
1090+func publishMaintenanceGenerations(t *testing.T, repo *Repository, name string, count uint64) []ports.SnapshotPublication {
1091+ t.Helper()
1092+ publications := make([]ports.SnapshotPublication, 0, count)
1093+ for generation := uint64(1); generation <= count; generation++ {
1094+ publication := repositoryPublicationAfter(t, repo, name, generation, []byte{byte(generation)})
1095+ if err := repo.Publish(context.Background(), publication); err != nil {
1096+ t.Fatal(err)
1097+ }
1098+ publications = append(publications, publication)
1099+ }
1100+ return publications
1101+}
1102+
1103+func checkpointRefForPublication(publication ports.SnapshotPublication) domain.CheckpointRef {
1104+ return domain.CheckpointRef{Generation: publication.Generation, ManifestDigest: sha256.Sum256(publication.Manifest)}
1105+}
1106+
1107+func remainingMaintenanceGenerations(t *testing.T, repo *Repository, id domain.IncarnationID) []uint64 {
1108+ t.Helper()
1109+ entries, err := os.ReadDir(filepath.Join(repo.sessionPath(id), repositoryGenerations))
1110+ if err != nil {
1111+ t.Fatal(err)
1112+ }
1113+ generations := make([]uint64, 0, len(entries))
1114+ for _, entry := range entries {
1115+ if generation, ok := parseGenerationFilename(entry.Name()); ok {
1116+ generations = append(generations, generation)
1117+ }
1118+ }
1119+ slices.SortFunc(generations, func(a, b uint64) int { return int(b) - int(a) })
1120+ return generations
1121+}
A internal/adapters/snapshot/reconciliation_test.go +67 -0
@@ -0,0 +1,67 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "os"
6+ "path/filepath"
7+ "testing"
8+
9+ "github.com/bnema/vev/internal/domain"
10+ "github.com/bnema/vev/internal/ports"
11+ "github.com/stretchr/testify/require"
12+)
13+
14+func TestReconciliationCursor(t *testing.T) {
15+ repo := NewRepository(privateDir(t))
16+ publications := publishMaintenanceGenerations(t, repo, "work", 2)
17+ committed := checkpointRefForPublication(publications[0])
18+ record := domain.CatalogueRecord{
19+ Name: "work",
20+ IncarnationID: publications[0].IncarnationID,
21+ RecoveryState: domain.RecoveryHealthy,
22+ Committed: &committed,
23+ }
24+
25+ sessions := filepath.Join(repo.dir, repositorySessionsDir)
26+ for i := byte(20); i < 26; i++ {
27+ id := domain.IncarnationID{i}
28+ require.NoError(t, os.Mkdir(filepath.Join(sessions, id.String()), 0o700))
29+ }
30+
31+ cursor := ports.ReconcileCursor{}
32+ var findings []ports.ReconcileFinding
33+ calls := 0
34+ for {
35+ next, page, err := repo.Reconcile(context.Background(), []domain.CatalogueRecord{record}, cursor, ports.MaintenanceBudget{Entries: 2, Bytes: 8 << 20})
36+ require.NoError(t, err)
37+ for _, finding := range page {
38+ require.LessOrEqual(t, finding.Consumed.Entries, uint64(2))
39+ require.LessOrEqual(t, finding.Consumed.Bytes, uint64(8<<20))
40+ }
41+ findings = append(findings, page...)
42+ calls++
43+ if next.DirectoryCookie == 0 {
44+ break
45+ }
46+ require.NotEqual(t, cursor, next, "reconciliation cursor must advance")
47+ cursor = next
48+ require.Less(t, calls, 10)
49+ }
50+ require.Greater(t, calls, 1, "top-level traversal must resume across bounded calls")
51+
52+ unknown := 0
53+ adoptable := 0
54+ for _, finding := range findings {
55+ switch finding.Kind {
56+ case ports.ReconcileUnknownIncarnation:
57+ unknown++
58+ case ports.ReconcileForwardOrphan:
59+ require.Equal(t, ports.ReconcileValidated, finding.Status)
60+ require.Equal(t, uint64(2), finding.Candidate.Ref.Generation)
61+ require.Equal(t, committed, *finding.Candidate.Parent)
62+ adoptable++
63+ }
64+ }
65+ require.Equal(t, 6, unknown)
66+ require.Equal(t, 1, adoptable)
67+}
@@ -0,0 +1,67 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "os"
6+ "path/filepath"
7+ "testing"
8+
9+ "github.com/bnema/vev/internal/domain"
10+ "github.com/bnema/vev/internal/ports"
11+ "github.com/stretchr/testify/require"
12+)
13+
14+func TestReconciliationCursor(t *testing.T) {
15+ repo := NewRepository(privateDir(t))
16+ publications := publishMaintenanceGenerations(t, repo, "work", 2)
17+ committed := checkpointRefForPublication(publications[0])
18+ record := domain.CatalogueRecord{
19+ Name: "work",
20+ IncarnationID: publications[0].IncarnationID,
21+ RecoveryState: domain.RecoveryHealthy,
22+ Committed: &committed,
23+ }
24+
25+ sessions := filepath.Join(repo.dir, repositorySessionsDir)
26+ for i := byte(20); i < 26; i++ {
27+ id := domain.IncarnationID{i}
28+ require.NoError(t, os.Mkdir(filepath.Join(sessions, id.String()), 0o700))
29+ }
30+
31+ cursor := ports.ReconcileCursor{}
32+ var findings []ports.ReconcileFinding
33+ calls := 0
34+ for {
35+ next, page, err := repo.Reconcile(context.Background(), []domain.CatalogueRecord{record}, cursor, ports.MaintenanceBudget{Entries: 2, Bytes: 8 << 20})
36+ require.NoError(t, err)
37+ for _, finding := range page {
38+ require.LessOrEqual(t, finding.Consumed.Entries, uint64(2))
39+ require.LessOrEqual(t, finding.Consumed.Bytes, uint64(8<<20))
40+ }
41+ findings = append(findings, page...)
42+ calls++
43+ if next.DirectoryCookie == 0 {
44+ break
45+ }
46+ require.NotEqual(t, cursor, next, "reconciliation cursor must advance")
47+ cursor = next
48+ require.Less(t, calls, 10)
49+ }
50+ require.Greater(t, calls, 1, "top-level traversal must resume across bounded calls")
51+
52+ unknown := 0
53+ adoptable := 0
54+ for _, finding := range findings {
55+ switch finding.Kind {
56+ case ports.ReconcileUnknownIncarnation:
57+ unknown++
58+ case ports.ReconcileForwardOrphan:
59+ require.Equal(t, ports.ReconcileValidated, finding.Status)
60+ require.Equal(t, uint64(2), finding.Candidate.Ref.Generation)
61+ require.Equal(t, committed, *finding.Candidate.Parent)
62+ adoptable++
63+ }
64+ }
65+ require.Equal(t, 6, unknown)
66+ require.Equal(t, 1, adoptable)
67+}
M internal/adapters/snapshot/repository.go +2 -0
@@ -39,6 +39,7 @@ type Repository struct {
3939 maintenanceMu sync.Mutex
4040 maintenanceCursors map[string]*maintenanceCursor
4141 maintenanceSessions map[string]*sessionMaintenance
42+ retentionSessions map[string]*retentionMaintenance
4243 maintenanceQuarantine *quarantineMaintenance
4344
4445 // pendingLegacySync records an unlink whose root-directory sync failed.
@@ -101,6 +102,7 @@ func NewRepository(dir string) *Repository {
101102 storageEpochs: make(map[string]uint64),
102103 maintenanceCursors: make(map[string]*maintenanceCursor),
103104 maintenanceSessions: make(map[string]*sessionMaintenance),
105+ retentionSessions: make(map[string]*retentionMaintenance),
104106 }
105107 }
106108
@@ -39,6 +39,7 @@ type Repository struct {
39 maintenanceMu sync.Mutex39 maintenanceMu sync.Mutex
40 maintenanceCursors map[string]*maintenanceCursor40 maintenanceCursors map[string]*maintenanceCursor
41 maintenanceSessions map[string]*sessionMaintenance41 maintenanceSessions map[string]*sessionMaintenance
42+ retentionSessions map[string]*retentionMaintenance
42 maintenanceQuarantine *quarantineMaintenance43 maintenanceQuarantine *quarantineMaintenance
43 44
44 // pendingLegacySync records an unlink whose root-directory sync failed.45 // pendingLegacySync records an unlink whose root-directory sync failed.
@@ -101,6 +102,7 @@ func NewRepository(dir string) *Repository {
101 storageEpochs: make(map[string]uint64),102 storageEpochs: make(map[string]uint64),
102 maintenanceCursors: make(map[string]*maintenanceCursor),103 maintenanceCursors: make(map[string]*maintenanceCursor),
103 maintenanceSessions: make(map[string]*sessionMaintenance),104 maintenanceSessions: make(map[string]*sessionMaintenance),
105+ retentionSessions: make(map[string]*retentionMaintenance),
104 }106 }
105 }107 }
106 108
A internal/usecase/recovery/reconcile.go +145 -0
@@ -0,0 +1,145 @@
1+package recovery
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+
8+ "github.com/bnema/vev/internal/domain"
9+ "github.com/bnema/vev/internal/ports"
10+)
11+
12+// Reconciler runs bounded repository maintenance after catalogue-driven
13+// restoration. Its cursor is caller-owned and can be resumed by a later pass.
14+type Reconciler struct {
15+ coordinator *Coordinator
16+ catalogue ports.Catalogue
17+ repository ports.SnapshotRepository
18+ budget ports.MaintenanceBudget
19+}
20+
21+func NewReconciler(coordinator *Coordinator, catalogue ports.Catalogue, repository ports.SnapshotRepository, budget ports.MaintenanceBudget) *Reconciler {
22+ return &Reconciler{coordinator: coordinator, catalogue: catalogue, repository: repository, budget: budget}
23+}
24+
25+func (r *Reconciler) Step(ctx context.Context, cursor ports.ReconcileCursor) (ports.ReconcileCursor, []ports.ReconcileDecision, error) {
26+ if r == nil || r.coordinator == nil || r.catalogue == nil || r.repository == nil || r.budget.Entries == 0 || r.budget.Bytes == 0 {
27+ return cursor, nil, errors.New("recovery: incomplete reconciliation dependencies")
28+ }
29+ if err := ctx.Err(); err != nil {
30+ return cursor, nil, err
31+ }
32+ next, findings, err := r.repository.Reconcile(ctx, r.catalogue.Records(), cursor, r.budget)
33+ if err != nil {
34+ return cursor, nil, err
35+ }
36+ decisions := make([]ports.ReconcileDecision, 0, len(findings))
37+ for _, finding := range findings {
38+ if finding.Consumed.Entries > r.budget.Entries || finding.Consumed.Bytes > r.budget.Bytes {
39+ return cursor, nil, errors.New("recovery: repository exceeded reconciliation budget")
40+ }
41+ record, ok := r.catalogue.Record(finding.Candidate.Name)
42+ if !ok {
43+ record = domain.CatalogueRecord{}
44+ }
45+ decision := classifyFinding(record, finding)
46+ if decision.Kind == ports.ReconcileAdopt {
47+ if _, err := r.coordinator.PublishReconciledCheckpoint(ctx, finding.Candidate.Name, finding.Candidate, finding.AncestorChain); err != nil {
48+ return cursor, nil, err
49+ }
50+ }
51+ decisions = append(decisions, decision)
52+ }
53+ return next, decisions, nil
54+}
55+
56+func classifyFinding(record domain.CatalogueRecord, finding ports.ReconcileFinding) ports.ReconcileDecision {
57+ decision := ports.ReconcileDecision{Name: finding.Candidate.Name}
58+ candidate := finding.Candidate.Ref
59+ decision.Candidate = &candidate
60+ if finding.Status == ports.ReconcileBudgetExhausted {
61+ decision.Kind = ports.ReconcileDefer
62+ decision.ReasonCode = "budget-exhausted"
63+ return decision
64+ }
65+ if finding.Kind == ports.ReconcileUnknownIncarnation || record.IncarnationID == (domain.IncarnationID{}) {
66+ decision.Kind = ports.ReconcileKeepQuarantined
67+ decision.ReasonCode = "unknown-incarnation"
68+ return decision
69+ }
70+ if finding.Candidate.IncarnationID != record.IncarnationID {
71+ decision.Kind = ports.ReconcileKeepQuarantined
72+ decision.ReasonCode = "incarnation-mismatch"
73+ return decision
74+ }
75+ if finding.Kind != ports.ReconcileForwardOrphan || record.Committed == nil || finding.Candidate.Ref.Generation <= record.Committed.Generation {
76+ decision.Kind = ports.ReconcileKeepQuarantined
77+ decision.ReasonCode = "not-forward"
78+ return decision
79+ }
80+ if finding.Status != ports.ReconcileValidated || !validForwardChain(*record.Committed, finding.Candidate, finding.AncestorChain) {
81+ decision.Kind = ports.ReconcileKeepQuarantined
82+ decision.ReasonCode = "invalid-chain"
83+ return decision
84+ }
85+ decision.Kind = ports.ReconcileAdopt
86+ return decision
87+}
88+
89+func validForwardChain(committed domain.CheckpointRef, candidate ports.ReconcileCandidate, ancestors []ports.ValidatedCheckpoint) bool {
90+ if committed.Generation == 0 || candidate.Ref.Generation <= committed.Generation || candidate.Parent == nil {
91+ return false
92+ }
93+ if len(ancestors) == 0 {
94+ return *candidate.Parent == committed
95+ }
96+ if *candidate.Parent != ancestors[0].Ref || candidate.Ref.Generation <= ancestors[0].Ref.Generation {
97+ return false
98+ }
99+ for i, ancestor := range ancestors {
100+ if ancestor.Parent == nil || ancestor.Ref.Generation <= committed.Generation {
101+ return false
102+ }
103+ if i+1 < len(ancestors) {
104+ if ancestor.Ref.Generation <= ancestors[i+1].Ref.Generation || *ancestor.Parent != ancestors[i+1].Ref {
105+ return false
106+ }
107+ continue
108+ }
109+ if *ancestor.Parent != committed {
110+ return false
111+ }
112+ }
113+ return true
114+}
115+
116+// PublishReconciledCheckpoint rechecks the chain while holding the session
117+// transaction lock. Files are already durable and validated by the adapter.
118+func (c *Coordinator) PublishReconciledCheckpoint(ctx context.Context, name string, candidate ports.ReconcileCandidate, ancestors []ports.ValidatedCheckpoint) (domain.CatalogueRecord, error) {
119+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
120+ return domain.CatalogueRecord{}, errors.New("recovery: incomplete reconciled checkpoint dependencies")
121+ }
122+ unlock := c.locks.Lock([]string{name})
123+ defer unlock()
124+ if err := ctx.Err(); err != nil {
125+ return domain.CatalogueRecord{}, err
126+ }
127+ record, ok := c.catalogue.Record(name)
128+ if !ok {
129+ return domain.CatalogueRecord{}, ErrCheckpointRecordNotFound
130+ }
131+ if candidate.Name != name || candidate.IncarnationID != record.IncarnationID || record.Committed == nil || !validForwardChain(*record.Committed, candidate, ancestors) {
132+ return domain.CatalogueRecord{}, ErrCheckpointConflict
133+ }
134+ next := shiftedCheckpoint(record, candidate.Ref)
135+ if err := next.Validate(); err != nil {
136+ return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid reconciled checkpoint transition: %w", err)
137+ }
138+ if err := c.catalogue.Replace(name, next); err != nil {
139+ return domain.CatalogueRecord{}, err
140+ }
141+ if err := c.repository.RepairHEAD(ctx, record.IncarnationID, candidate.Ref); err != nil {
142+ return next, err
143+ }
144+ return next, nil
145+}
@@ -0,0 +1,145 @@
1+package recovery
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+
8+ "github.com/bnema/vev/internal/domain"
9+ "github.com/bnema/vev/internal/ports"
10+)
11+
12+// Reconciler runs bounded repository maintenance after catalogue-driven
13+// restoration. Its cursor is caller-owned and can be resumed by a later pass.
14+type Reconciler struct {
15+ coordinator *Coordinator
16+ catalogue ports.Catalogue
17+ repository ports.SnapshotRepository
18+ budget ports.MaintenanceBudget
19+}
20+
21+func NewReconciler(coordinator *Coordinator, catalogue ports.Catalogue, repository ports.SnapshotRepository, budget ports.MaintenanceBudget) *Reconciler {
22+ return &Reconciler{coordinator: coordinator, catalogue: catalogue, repository: repository, budget: budget}
23+}
24+
25+func (r *Reconciler) Step(ctx context.Context, cursor ports.ReconcileCursor) (ports.ReconcileCursor, []ports.ReconcileDecision, error) {
26+ if r == nil || r.coordinator == nil || r.catalogue == nil || r.repository == nil || r.budget.Entries == 0 || r.budget.Bytes == 0 {
27+ return cursor, nil, errors.New("recovery: incomplete reconciliation dependencies")
28+ }
29+ if err := ctx.Err(); err != nil {
30+ return cursor, nil, err
31+ }
32+ next, findings, err := r.repository.Reconcile(ctx, r.catalogue.Records(), cursor, r.budget)
33+ if err != nil {
34+ return cursor, nil, err
35+ }
36+ decisions := make([]ports.ReconcileDecision, 0, len(findings))
37+ for _, finding := range findings {
38+ if finding.Consumed.Entries > r.budget.Entries || finding.Consumed.Bytes > r.budget.Bytes {
39+ return cursor, nil, errors.New("recovery: repository exceeded reconciliation budget")
40+ }
41+ record, ok := r.catalogue.Record(finding.Candidate.Name)
42+ if !ok {
43+ record = domain.CatalogueRecord{}
44+ }
45+ decision := classifyFinding(record, finding)
46+ if decision.Kind == ports.ReconcileAdopt {
47+ if _, err := r.coordinator.PublishReconciledCheckpoint(ctx, finding.Candidate.Name, finding.Candidate, finding.AncestorChain); err != nil {
48+ return cursor, nil, err
49+ }
50+ }
51+ decisions = append(decisions, decision)
52+ }
53+ return next, decisions, nil
54+}
55+
56+func classifyFinding(record domain.CatalogueRecord, finding ports.ReconcileFinding) ports.ReconcileDecision {
57+ decision := ports.ReconcileDecision{Name: finding.Candidate.Name}
58+ candidate := finding.Candidate.Ref
59+ decision.Candidate = &candidate
60+ if finding.Status == ports.ReconcileBudgetExhausted {
61+ decision.Kind = ports.ReconcileDefer
62+ decision.ReasonCode = "budget-exhausted"
63+ return decision
64+ }
65+ if finding.Kind == ports.ReconcileUnknownIncarnation || record.IncarnationID == (domain.IncarnationID{}) {
66+ decision.Kind = ports.ReconcileKeepQuarantined
67+ decision.ReasonCode = "unknown-incarnation"
68+ return decision
69+ }
70+ if finding.Candidate.IncarnationID != record.IncarnationID {
71+ decision.Kind = ports.ReconcileKeepQuarantined
72+ decision.ReasonCode = "incarnation-mismatch"
73+ return decision
74+ }
75+ if finding.Kind != ports.ReconcileForwardOrphan || record.Committed == nil || finding.Candidate.Ref.Generation <= record.Committed.Generation {
76+ decision.Kind = ports.ReconcileKeepQuarantined
77+ decision.ReasonCode = "not-forward"
78+ return decision
79+ }
80+ if finding.Status != ports.ReconcileValidated || !validForwardChain(*record.Committed, finding.Candidate, finding.AncestorChain) {
81+ decision.Kind = ports.ReconcileKeepQuarantined
82+ decision.ReasonCode = "invalid-chain"
83+ return decision
84+ }
85+ decision.Kind = ports.ReconcileAdopt
86+ return decision
87+}
88+
89+func validForwardChain(committed domain.CheckpointRef, candidate ports.ReconcileCandidate, ancestors []ports.ValidatedCheckpoint) bool {
90+ if committed.Generation == 0 || candidate.Ref.Generation <= committed.Generation || candidate.Parent == nil {
91+ return false
92+ }
93+ if len(ancestors) == 0 {
94+ return *candidate.Parent == committed
95+ }
96+ if *candidate.Parent != ancestors[0].Ref || candidate.Ref.Generation <= ancestors[0].Ref.Generation {
97+ return false
98+ }
99+ for i, ancestor := range ancestors {
100+ if ancestor.Parent == nil || ancestor.Ref.Generation <= committed.Generation {
101+ return false
102+ }
103+ if i+1 < len(ancestors) {
104+ if ancestor.Ref.Generation <= ancestors[i+1].Ref.Generation || *ancestor.Parent != ancestors[i+1].Ref {
105+ return false
106+ }
107+ continue
108+ }
109+ if *ancestor.Parent != committed {
110+ return false
111+ }
112+ }
113+ return true
114+}
115+
116+// PublishReconciledCheckpoint rechecks the chain while holding the session
117+// transaction lock. Files are already durable and validated by the adapter.
118+func (c *Coordinator) PublishReconciledCheckpoint(ctx context.Context, name string, candidate ports.ReconcileCandidate, ancestors []ports.ValidatedCheckpoint) (domain.CatalogueRecord, error) {
119+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
120+ return domain.CatalogueRecord{}, errors.New("recovery: incomplete reconciled checkpoint dependencies")
121+ }
122+ unlock := c.locks.Lock([]string{name})
123+ defer unlock()
124+ if err := ctx.Err(); err != nil {
125+ return domain.CatalogueRecord{}, err
126+ }
127+ record, ok := c.catalogue.Record(name)
128+ if !ok {
129+ return domain.CatalogueRecord{}, ErrCheckpointRecordNotFound
130+ }
131+ if candidate.Name != name || candidate.IncarnationID != record.IncarnationID || record.Committed == nil || !validForwardChain(*record.Committed, candidate, ancestors) {
132+ return domain.CatalogueRecord{}, ErrCheckpointConflict
133+ }
134+ next := shiftedCheckpoint(record, candidate.Ref)
135+ if err := next.Validate(); err != nil {
136+ return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid reconciled checkpoint transition: %w", err)
137+ }
138+ if err := c.catalogue.Replace(name, next); err != nil {
139+ return domain.CatalogueRecord{}, err
140+ }
141+ if err := c.repository.RepairHEAD(ctx, record.IncarnationID, candidate.Ref); err != nil {
142+ return next, err
143+ }
144+ return next, nil
145+}
A internal/usecase/recovery/reconcile_test.go +152 -0
@@ -0,0 +1,152 @@
1+package recovery
2+
3+import (
4+ "context"
5+ "testing"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+ "github.com/stretchr/testify/require"
10+)
11+
12+type reconcileRepository struct {
13+ ports.SnapshotRepository
14+ pages map[uint64]reconcilePage
15+ seenCursors []ports.ReconcileCursor
16+ seenBudgets []ports.MaintenanceBudget
17+ repairs []domain.CheckpointRef
18+ events *[]string
19+}
20+
21+type reconcilePage struct {
22+ next ports.ReconcileCursor
23+ findings []ports.ReconcileFinding
24+ err error
25+}
26+
27+func (r *reconcileRepository) Reconcile(_ context.Context, _ []domain.CatalogueRecord, cursor ports.ReconcileCursor, budget ports.MaintenanceBudget) (ports.ReconcileCursor, []ports.ReconcileFinding, error) {
28+ r.seenCursors = append(r.seenCursors, cursor)
29+ r.seenBudgets = append(r.seenBudgets, budget)
30+ page := r.pages[cursor.DirectoryCookie]
31+ return page.next, page.findings, page.err
32+}
33+
34+func (r *reconcileRepository) RepairHEAD(_ context.Context, _ domain.IncarnationID, ref ports.CheckpointRef) error {
35+ if r.events != nil {
36+ *r.events = append(*r.events, "head")
37+ }
38+ r.repairs = append(r.repairs, ref)
39+ return nil
40+}
41+
42+func TestReconciliationCursor(t *testing.T) {
43+ record := healthyReconcileRecord()
44+ catalogue := &checkpointCatalogue{record: record}
45+ candidate := forwardCandidate(record, 2, record.Committed)
46+ repository := &reconcileRepository{pages: map[uint64]reconcilePage{
47+ 0: {next: ports.ReconcileCursor{DirectoryCookie: 12}, findings: []ports.ReconcileFinding{{Status: ports.ReconcileBudgetExhausted, Cursor: ports.ReconcileCursor{DirectoryCookie: 12}, Consumed: ports.MaintenanceBudget{Entries: 1, Bytes: 16}}}},
48+ 12: {next: ports.ReconcileCursor{DirectoryCookie: 24}, findings: []ports.ReconcileFinding{{Kind: ports.ReconcileForwardOrphan, Status: ports.ReconcileValidated, Candidate: candidate, Cursor: ports.ReconcileCursor{DirectoryCookie: 24}, Consumed: ports.MaintenanceBudget{Entries: 2, Bytes: 32}}}},
49+ }}
50+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
51+ reconciler := &Reconciler{coordinator: coordinator, catalogue: catalogue, repository: repository, budget: ports.MaintenanceBudget{Entries: 2, Bytes: 32}}
52+
53+ next, decisions, err := reconciler.Step(context.Background(), ports.ReconcileCursor{})
54+ require.NoError(t, err)
55+ require.Equal(t, uint64(12), next.DirectoryCookie)
56+ require.Equal(t, ports.ReconcileDefer, decisions[0].Kind)
57+ require.Equal(t, "budget-exhausted", decisions[0].ReasonCode)
58+
59+ next, decisions, err = reconciler.Step(context.Background(), next)
60+ require.NoError(t, err)
61+ require.Equal(t, uint64(24), next.DirectoryCookie)
62+ require.Equal(t, ports.ReconcileAdopt, decisions[0].Kind)
63+ require.Equal(t, []ports.ReconcileCursor{{}, {DirectoryCookie: 12}}, repository.seenCursors)
64+ require.Equal(t, []ports.MaintenanceBudget{{Entries: 2, Bytes: 32}, {Entries: 2, Bytes: 32}}, repository.seenBudgets)
65+}
66+
67+func TestForwardOrphanAdoption(t *testing.T) {
68+ for _, tc := range []struct {
69+ name string
70+ mutate func(domain.CatalogueRecord, *ports.ReconcileFinding)
71+ wantKind ports.ReconcileDecisionKind
72+ wantCause string
73+ }{
74+ {name: "direct child", wantKind: ports.ReconcileAdopt},
75+ {name: "two parent chain", mutate: addReconcileAncestor, wantKind: ports.ReconcileAdopt},
76+ {name: "wrong incarnation", mutate: func(record domain.CatalogueRecord, finding *ports.ReconcileFinding) {
77+ finding.Candidate.IncarnationID = domain.IncarnationID{9}
78+ }, wantKind: ports.ReconcileKeepQuarantined, wantCause: "incarnation-mismatch"},
79+ {name: "wrong digest", mutate: func(record domain.CatalogueRecord, finding *ports.ReconcileFinding) {
80+ finding.Candidate.Parent.ManifestDigest[0]++
81+ }, wantKind: ports.ReconcileKeepQuarantined, wantCause: "invalid-chain"},
82+ {name: "missing link", mutate: func(record domain.CatalogueRecord, finding *ports.ReconcileFinding) { finding.Candidate.Parent = nil }, wantKind: ports.ReconcileKeepQuarantined, wantCause: "invalid-chain"},
83+ {name: "generation reversal", mutate: func(record domain.CatalogueRecord, finding *ports.ReconcileFinding) {
84+ finding.Candidate.Ref.Generation = record.Committed.Generation
85+ }, wantKind: ports.ReconcileKeepQuarantined, wantCause: "not-forward"},
86+ {name: "not forward orphan", mutate: func(record domain.CatalogueRecord, finding *ports.ReconcileFinding) {
87+ finding.Kind = ports.ReconcileInvalidCandidate
88+ }, wantKind: ports.ReconcileKeepQuarantined, wantCause: "not-forward"},
89+ } {
90+ t.Run(tc.name, func(t *testing.T) {
91+ record := healthyReconcileRecord()
92+ finding := ports.ReconcileFinding{Kind: ports.ReconcileForwardOrphan, Status: ports.ReconcileValidated, Candidate: forwardCandidate(record, 2, record.Committed)}
93+ if tc.mutate != nil {
94+ tc.mutate(record, &finding)
95+ }
96+ decision := classifyFinding(record, finding)
97+ require.Equal(t, tc.wantKind, decision.Kind)
98+ require.Equal(t, tc.wantCause, decision.ReasonCode)
99+ })
100+ }
101+
102+ t.Run("catalogue replacement precedes head repair", func(t *testing.T) {
103+ record := healthyReconcileRecord()
104+ events := []string{}
105+ catalogue := &checkpointCatalogue{record: record, events: &events}
106+ repository := &reconcileRepository{events: &events}
107+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
108+ candidate := forwardCandidate(record, 2, record.Committed)
109+
110+ next, err := coordinator.PublishReconciledCheckpoint(context.Background(), record.Name, candidate, nil)
111+ require.NoError(t, err)
112+ require.Equal(t, candidate.Ref, *next.Committed)
113+ require.Equal(t, []string{"catalogue", "head"}, events)
114+ })
115+
116+ t.Run("stale decision after concurrent commit", func(t *testing.T) {
117+ record := healthyReconcileRecord()
118+ candidate := forwardCandidate(record, 2, record.Committed)
119+ concurrent := domain.CheckpointRef{Generation: 2, ManifestDigest: [32]byte{7}}
120+ record = shiftedCheckpoint(record, concurrent)
121+ catalogue := &checkpointCatalogue{record: record}
122+ repository := &reconcileRepository{}
123+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
124+
125+ _, err := coordinator.PublishReconciledCheckpoint(context.Background(), record.Name, candidate, nil)
126+ require.ErrorIs(t, err, ErrCheckpointConflict)
127+ require.Empty(t, repository.repairs)
128+ })
129+
130+ t.Run("unknown identity remains quarantined", func(t *testing.T) {
131+ finding := ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: "unknown", IncarnationID: domain.IncarnationID{9}}}
132+ decision := classifyFinding(domain.CatalogueRecord{}, finding)
133+ require.Equal(t, ports.ReconcileKeepQuarantined, decision.Kind)
134+ require.Equal(t, "unknown-incarnation", decision.ReasonCode)
135+ })
136+}
137+
138+func healthyReconcileRecord() domain.CatalogueRecord {
139+ ref := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
140+ return domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryHealthy, Committed: &ref}
141+}
142+
143+func forwardCandidate(record domain.CatalogueRecord, generation uint64, parent *domain.CheckpointRef) ports.ReconcileCandidate {
144+ parentCopy := *parent
145+ return ports.ReconcileCandidate{IncarnationID: record.IncarnationID, Name: record.Name, Ref: domain.CheckpointRef{Generation: generation, ManifestDigest: [32]byte{byte(generation)}}, Parent: &parentCopy}
146+}
147+
148+func addReconcileAncestor(record domain.CatalogueRecord, finding *ports.ReconcileFinding) {
149+ ancestor := ports.ValidatedCheckpoint{Ref: domain.CheckpointRef{Generation: 2, ManifestDigest: [32]byte{2}}, Parent: record.Committed}
150+ finding.Candidate = forwardCandidate(record, 3, &ancestor.Ref)
151+ finding.AncestorChain = []ports.ValidatedCheckpoint{ancestor}
152+}
@@ -0,0 +1,152 @@
1+package recovery
2+
3+import (
4+ "context"
5+ "testing"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+ "github.com/stretchr/testify/require"
10+)
11+
12+type reconcileRepository struct {
13+ ports.SnapshotRepository
14+ pages map[uint64]reconcilePage
15+ seenCursors []ports.ReconcileCursor
16+ seenBudgets []ports.MaintenanceBudget
17+ repairs []domain.CheckpointRef
18+ events *[]string
19+}
20+
21+type reconcilePage struct {
22+ next ports.ReconcileCursor
23+ findings []ports.ReconcileFinding
24+ err error
25+}
26+
27+func (r *reconcileRepository) Reconcile(_ context.Context, _ []domain.CatalogueRecord, cursor ports.ReconcileCursor, budget ports.MaintenanceBudget) (ports.ReconcileCursor, []ports.ReconcileFinding, error) {
28+ r.seenCursors = append(r.seenCursors, cursor)
29+ r.seenBudgets = append(r.seenBudgets, budget)
30+ page := r.pages[cursor.DirectoryCookie]
31+ return page.next, page.findings, page.err
32+}
33+
34+func (r *reconcileRepository) RepairHEAD(_ context.Context, _ domain.IncarnationID, ref ports.CheckpointRef) error {
35+ if r.events != nil {
36+ *r.events = append(*r.events, "head")
37+ }
38+ r.repairs = append(r.repairs, ref)
39+ return nil
40+}
41+
42+func TestReconciliationCursor(t *testing.T) {
43+ record := healthyReconcileRecord()
44+ catalogue := &checkpointCatalogue{record: record}
45+ candidate := forwardCandidate(record, 2, record.Committed)
46+ repository := &reconcileRepository{pages: map[uint64]reconcilePage{
47+ 0: {next: ports.ReconcileCursor{DirectoryCookie: 12}, findings: []ports.ReconcileFinding{{Status: ports.ReconcileBudgetExhausted, Cursor: ports.ReconcileCursor{DirectoryCookie: 12}, Consumed: ports.MaintenanceBudget{Entries: 1, Bytes: 16}}}},
48+ 12: {next: ports.ReconcileCursor{DirectoryCookie: 24}, findings: []ports.ReconcileFinding{{Kind: ports.ReconcileForwardOrphan, Status: ports.ReconcileValidated, Candidate: candidate, Cursor: ports.ReconcileCursor{DirectoryCookie: 24}, Consumed: ports.MaintenanceBudget{Entries: 2, Bytes: 32}}}},
49+ }}
50+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
51+ reconciler := &Reconciler{coordinator: coordinator, catalogue: catalogue, repository: repository, budget: ports.MaintenanceBudget{Entries: 2, Bytes: 32}}
52+
53+ next, decisions, err := reconciler.Step(context.Background(), ports.ReconcileCursor{})
54+ require.NoError(t, err)
55+ require.Equal(t, uint64(12), next.DirectoryCookie)
56+ require.Equal(t, ports.ReconcileDefer, decisions[0].Kind)
57+ require.Equal(t, "budget-exhausted", decisions[0].ReasonCode)
58+
59+ next, decisions, err = reconciler.Step(context.Background(), next)
60+ require.NoError(t, err)
61+ require.Equal(t, uint64(24), next.DirectoryCookie)
62+ require.Equal(t, ports.ReconcileAdopt, decisions[0].Kind)
63+ require.Equal(t, []ports.ReconcileCursor{{}, {DirectoryCookie: 12}}, repository.seenCursors)
64+ require.Equal(t, []ports.MaintenanceBudget{{Entries: 2, Bytes: 32}, {Entries: 2, Bytes: 32}}, repository.seenBudgets)
65+}
66+
67+func TestForwardOrphanAdoption(t *testing.T) {
68+ for _, tc := range []struct {
69+ name string
70+ mutate func(domain.CatalogueRecord, *ports.ReconcileFinding)
71+ wantKind ports.ReconcileDecisionKind
72+ wantCause string
73+ }{
74+ {name: "direct child", wantKind: ports.ReconcileAdopt},
75+ {name: "two parent chain", mutate: addReconcileAncestor, wantKind: ports.ReconcileAdopt},
76+ {name: "wrong incarnation", mutate: func(record domain.CatalogueRecord, finding *ports.ReconcileFinding) {
77+ finding.Candidate.IncarnationID = domain.IncarnationID{9}
78+ }, wantKind: ports.ReconcileKeepQuarantined, wantCause: "incarnation-mismatch"},
79+ {name: "wrong digest", mutate: func(record domain.CatalogueRecord, finding *ports.ReconcileFinding) {
80+ finding.Candidate.Parent.ManifestDigest[0]++
81+ }, wantKind: ports.ReconcileKeepQuarantined, wantCause: "invalid-chain"},
82+ {name: "missing link", mutate: func(record domain.CatalogueRecord, finding *ports.ReconcileFinding) { finding.Candidate.Parent = nil }, wantKind: ports.ReconcileKeepQuarantined, wantCause: "invalid-chain"},
83+ {name: "generation reversal", mutate: func(record domain.CatalogueRecord, finding *ports.ReconcileFinding) {
84+ finding.Candidate.Ref.Generation = record.Committed.Generation
85+ }, wantKind: ports.ReconcileKeepQuarantined, wantCause: "not-forward"},
86+ {name: "not forward orphan", mutate: func(record domain.CatalogueRecord, finding *ports.ReconcileFinding) {
87+ finding.Kind = ports.ReconcileInvalidCandidate
88+ }, wantKind: ports.ReconcileKeepQuarantined, wantCause: "not-forward"},
89+ } {
90+ t.Run(tc.name, func(t *testing.T) {
91+ record := healthyReconcileRecord()
92+ finding := ports.ReconcileFinding{Kind: ports.ReconcileForwardOrphan, Status: ports.ReconcileValidated, Candidate: forwardCandidate(record, 2, record.Committed)}
93+ if tc.mutate != nil {
94+ tc.mutate(record, &finding)
95+ }
96+ decision := classifyFinding(record, finding)
97+ require.Equal(t, tc.wantKind, decision.Kind)
98+ require.Equal(t, tc.wantCause, decision.ReasonCode)
99+ })
100+ }
101+
102+ t.Run("catalogue replacement precedes head repair", func(t *testing.T) {
103+ record := healthyReconcileRecord()
104+ events := []string{}
105+ catalogue := &checkpointCatalogue{record: record, events: &events}
106+ repository := &reconcileRepository{events: &events}
107+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
108+ candidate := forwardCandidate(record, 2, record.Committed)
109+
110+ next, err := coordinator.PublishReconciledCheckpoint(context.Background(), record.Name, candidate, nil)
111+ require.NoError(t, err)
112+ require.Equal(t, candidate.Ref, *next.Committed)
113+ require.Equal(t, []string{"catalogue", "head"}, events)
114+ })
115+
116+ t.Run("stale decision after concurrent commit", func(t *testing.T) {
117+ record := healthyReconcileRecord()
118+ candidate := forwardCandidate(record, 2, record.Committed)
119+ concurrent := domain.CheckpointRef{Generation: 2, ManifestDigest: [32]byte{7}}
120+ record = shiftedCheckpoint(record, concurrent)
121+ catalogue := &checkpointCatalogue{record: record}
122+ repository := &reconcileRepository{}
123+ coordinator := NewCoordinator(catalogue, repository, nil, nil)
124+
125+ _, err := coordinator.PublishReconciledCheckpoint(context.Background(), record.Name, candidate, nil)
126+ require.ErrorIs(t, err, ErrCheckpointConflict)
127+ require.Empty(t, repository.repairs)
128+ })
129+
130+ t.Run("unknown identity remains quarantined", func(t *testing.T) {
131+ finding := ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: "unknown", IncarnationID: domain.IncarnationID{9}}}
132+ decision := classifyFinding(domain.CatalogueRecord{}, finding)
133+ require.Equal(t, ports.ReconcileKeepQuarantined, decision.Kind)
134+ require.Equal(t, "unknown-incarnation", decision.ReasonCode)
135+ })
136+}
137+
138+func healthyReconcileRecord() domain.CatalogueRecord {
139+ ref := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
140+ return domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryHealthy, Committed: &ref}
141+}
142+
143+func forwardCandidate(record domain.CatalogueRecord, generation uint64, parent *domain.CheckpointRef) ports.ReconcileCandidate {
144+ parentCopy := *parent
145+ return ports.ReconcileCandidate{IncarnationID: record.IncarnationID, Name: record.Name, Ref: domain.CheckpointRef{Generation: generation, ManifestDigest: [32]byte{byte(generation)}}, Parent: &parentCopy}
146+}
147+
148+func addReconcileAncestor(record domain.CatalogueRecord, finding *ports.ReconcileFinding) {
149+ ancestor := ports.ValidatedCheckpoint{Ref: domain.CheckpointRef{Generation: 2, ManifestDigest: [32]byte{2}}, Parent: record.Committed}
150+ finding.Candidate = forwardCandidate(record, 3, &ancestor.Ref)
151+ finding.AncestorChain = []ports.ValidatedCheckpoint{ancestor}
152+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close