detached 9297910

fix(recovery): wire bounded production maintenance

commit
92979100d6f5dc1d8d857d59db197a113a318ec2
author
brice · · 2026-07-26 02:15 UTC
parent
2d93e51
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
11 files · +445 -58
Unified
UnifiedSplit
M internal/adapters/snapshot/maintenance_gc.go +103 -25
@@ -371,6 +371,7 @@ func keepSet(plan ports.RetentionPlan) map[uint64]ports.SnapshotDigest {
371371 // MaintainSession incrementally validates and retains only catalogue-indexed
372372 // checkpoints. Every incarnation owns its budget and continuation state.
373373 func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPlan, budget ports.MaintenanceBudget) (done bool, err error) {
374+ requested := budget
374375 if err := ctx.Err(); err != nil {
375376 return false, err
376377 }
@@ -381,9 +382,14 @@ func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPl
381382 defer func() {
382383 r.log.Info("snapshot_maintenance_progress",
383384 "incarnation", plan.IncarnationID.String(),
385+ "action", "retention",
384386 "retained", len(plan.Keep),
385387 "cursor", key,
388+ "consumed_entries", requested.Entries-budget.Entries,
389+ "consumed_bytes", requested.Bytes-budget.Bytes,
390+ "done", done,
386391 "budget_exhausted", !done && err == nil,
392+ "err", err,
387393 )
388394 }()
389395 if plan.PinAll {
@@ -422,29 +428,19 @@ func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPl
422428 if !state.validated {
423429 for state.validateIndex < len(state.validateQueue) {
424430 ref := state.validateQueue[state.validateIndex]
425- generation, digest := ref.Generation, ref.ManifestDigest
426- data, err := r.readBounded(r.manifestPath(plan.IncarnationID, generation))
427- if err != nil {
428- return false, fmt.Errorf("validate retained manifest %d: %w", generation, err)
429- }
430- manifest, err := codec.UnmarshalManifest(data)
431- if err != nil || manifest.Generation != generation || manifest.IncarnationID != plan.IncarnationID || sha256.Sum256(data) != digest {
432- return false, fmt.Errorf("validate retained manifest %d: catalogue reference mismatch", generation)
431+ if budget.Entries == 0 {
432+ return false, nil
433433 }
434- loaded, err := r.loadGeneration(ctx, manifest.Name, key, generation)
434+ refs, consumedBytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, plan.IncarnationID, ref, budget.Bytes)
435435 if err != nil {
436- return false, fmt.Errorf("validate retained manifest %d: %w", generation, err)
436+ return false, fmt.Errorf("validate retained manifest %d: %w", ref.Generation, err)
437437 }
438- consumedBytes := uint64(len(data))
439- for _, object := range loaded.Objects {
440- consumedBytes += uint64(len(object))
441- }
442- if budget.Entries == 0 || consumedBytes > budget.Bytes {
438+ if !admitted {
443439 return false, nil
444440 }
445441 budget.Entries--
446442 budget.Bytes -= consumedBytes
447- for digest := range loaded.Objects {
443+ for digest := range refs {
448444 state.references[digest] = struct{}{}
449445 }
450446 state.validateIndex++
@@ -591,9 +587,91 @@ func retentionToken(plan ports.RetentionPlan) string {
591587 return b.String()
592588 }
593589
590+// validateMaintenanceCheckpoint admits every payload by stat-reported size
591+// before reading it. It validates directly instead of calling LoadCheckpoint,
592+// which would materialize a complete generation before maintenance could
593+// enforce its byte budget.
594+func (r *Repository) validateMaintenanceCheckpoint(ctx context.Context, id domain.IncarnationID, ref ports.CheckpointRef, byteBudget uint64) (map[ports.SnapshotDigest]codec.ObjectRef, uint64, bool, error) {
595+ manifestPath := r.manifestPath(id, ref.Generation)
596+ info, err := r.stat(manifestPath)
597+ if err != nil {
598+ return nil, 0, false, err
599+ }
600+ if info.Size() < 0 || uint64(info.Size()) > byteBudget {
601+ return nil, 0, false, nil
602+ }
603+ consumed := uint64(info.Size())
604+ if hook := r.hooks.beforeMaintenancePayloadRead; hook != nil {
605+ hook(manifestPath)
606+ }
607+ data, err := r.readBounded(manifestPath)
608+ if err != nil {
609+ return nil, 0, false, err
610+ }
611+ manifest, err := codec.UnmarshalManifest(data)
612+ if err != nil || manifest.Generation != ref.Generation || manifest.IncarnationID != id || sha256.Sum256(data) != ref.ManifestDigest {
613+ return nil, 0, false, errors.New("catalogue reference mismatch")
614+ }
615+ refs, valid := maintenanceManifestRefs(manifest)
616+ if !valid || !withinGenerationBudget(len(data), refs) {
617+ return nil, 0, false, errors.New("invalid manifest references")
618+ }
619+ for digest, objectRef := range refs {
620+ if err := ctx.Err(); err != nil {
621+ return nil, 0, false, err
622+ }
623+ path := r.objectPath(id, digest)
624+ info, err := r.stat(path)
625+ if err != nil {
626+ return nil, 0, false, err
627+ }
628+ if info.Size() < 0 || uint64(info.Size()) > byteBudget-consumed {
629+ return nil, 0, false, nil
630+ }
631+ consumed += uint64(info.Size())
632+ if hook := r.hooks.beforeMaintenancePayloadRead; hook != nil {
633+ hook(path)
634+ }
635+ object, err := r.readBounded(path)
636+ if err != nil {
637+ return nil, 0, false, err
638+ }
639+ if sha256.Sum256(object) != digest || !validObject(object, objectRef) {
640+ return nil, 0, false, errors.New("invalid retained object")
641+ }
642+ }
643+ return refs, consumed, true, nil
644+}
645+
594646 // Reconcile scans top-level incarnation namespaces from an external seek
595647 // cookie. Checkpoint-chain work receives a fresh budget for every directory.
596-func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRecord, cursor ports.ReconcileCursor, budget ports.MaintenanceBudget) (ports.ReconcileCursor, []ports.ReconcileFinding, error) {
648+func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRecord, cursor ports.ReconcileCursor, budget ports.MaintenanceBudget) (nextCursor ports.ReconcileCursor, findings []ports.ReconcileFinding, retErr error) {
649+ requested := budget
650+ defer func() {
651+ consumed := ports.MaintenanceBudget{}
652+ for _, finding := range findings {
653+ consumed.Entries += finding.Consumed.Entries
654+ consumed.Bytes += finding.Consumed.Bytes
655+ r.log.Info("snapshot_reconciliation_progress",
656+ "incarnation", finding.Candidate.IncarnationID.String(),
657+ "action", finding.Kind,
658+ "status", finding.Status,
659+ "consumed_entries", finding.Consumed.Entries,
660+ "consumed_bytes", finding.Consumed.Bytes,
661+ "cursor", finding.Cursor.DirectoryCookie,
662+ )
663+ }
664+ r.log.Info("snapshot_reconciliation_complete",
665+ "action", "scan",
666+ "cursor", nextCursor.DirectoryCookie,
667+ "requested_entries", requested.Entries,
668+ "requested_bytes", requested.Bytes,
669+ "consumed_entries", consumed.Entries,
670+ "consumed_bytes", consumed.Bytes,
671+ "done", nextCursor.DirectoryCookie == 0 && retErr == nil,
672+ "err", retErr,
673+ )
674+ }()
597675 if err := ctx.Err(); err != nil {
598676 return cursor, nil, err
599677 }
@@ -617,7 +695,7 @@ func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRe
617695 if budget.Entries < maintenanceBatch {
618696 findingCapacity = int(budget.Entries)
619697 }
620- findings := make([]ports.ReconcileFinding, 0, findingCapacity)
698+ findings = make([]ports.ReconcileFinding, 0, findingCapacity)
621699 for scanned := uint64(0); scanned < budget.Entries; scanned++ {
622700 entries, done, err := r.readMaintenanceDirent(root, 1, state)
623701 if errors.Is(err, os.ErrNotExist) {
@@ -704,20 +782,20 @@ func (r *Repository) reconcileIncarnation(ctx context.Context, record domain.Cat
704782 }
705783
706784 func (r *Repository) reconcileLoad(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef, consumed *ports.MaintenanceBudget, budget ports.MaintenanceBudget) (ports.SnapshotGeneration, bool, error) {
707- if consumed.Entries >= budget.Entries {
785+ if consumed.Entries >= budget.Entries || consumed.Bytes >= budget.Bytes {
708786 return ports.SnapshotGeneration{}, true, nil
709787 }
710- generation, err := r.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)
788+ _, bytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, record.IncarnationID, ref, budget.Bytes-consumed.Bytes)
711789 if err != nil {
712790 return ports.SnapshotGeneration{}, false, err
713791 }
714- bytes := uint64(len(generation.Manifest))
715- for _, object := range generation.Objects {
716- bytes += uint64(len(object))
717- }
718- if bytes > budget.Bytes-consumed.Bytes {
792+ if !admitted {
719793 return ports.SnapshotGeneration{}, true, nil
720794 }
795+ generation, err := r.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)
796+ if err != nil {
797+ return ports.SnapshotGeneration{}, false, err
798+ }
721799 consumed.Entries++
722800 consumed.Bytes += bytes
723801 return generation, false, nil
@@ -371,6 +371,7 @@ func keepSet(plan ports.RetentionPlan) map[uint64]ports.SnapshotDigest {
371 // MaintainSession incrementally validates and retains only catalogue-indexed371 // MaintainSession incrementally validates and retains only catalogue-indexed
372 // checkpoints. Every incarnation owns its budget and continuation state.372 // checkpoints. Every incarnation owns its budget and continuation state.
373 func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPlan, budget ports.MaintenanceBudget) (done bool, err error) {373 func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPlan, budget ports.MaintenanceBudget) (done bool, err error) {
374+ requested := budget
374 if err := ctx.Err(); err != nil {375 if err := ctx.Err(); err != nil {
375 return false, err376 return false, err
376 }377 }
@@ -381,9 +382,14 @@ func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPl
381 defer func() {382 defer func() {
382 r.log.Info("snapshot_maintenance_progress",383 r.log.Info("snapshot_maintenance_progress",
383 "incarnation", plan.IncarnationID.String(),384 "incarnation", plan.IncarnationID.String(),
385+ "action", "retention",
384 "retained", len(plan.Keep),386 "retained", len(plan.Keep),
385 "cursor", key,387 "cursor", key,
388+ "consumed_entries", requested.Entries-budget.Entries,
389+ "consumed_bytes", requested.Bytes-budget.Bytes,
390+ "done", done,
386 "budget_exhausted", !done && err == nil,391 "budget_exhausted", !done && err == nil,
392+ "err", err,
387 )393 )
388 }()394 }()
389 if plan.PinAll {395 if plan.PinAll {
@@ -422,29 +428,19 @@ func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPl
422 if !state.validated {428 if !state.validated {
423 for state.validateIndex < len(state.validateQueue) {429 for state.validateIndex < len(state.validateQueue) {
424 ref := state.validateQueue[state.validateIndex]430 ref := state.validateQueue[state.validateIndex]
425- generation, digest := ref.Generation, ref.ManifestDigest431+ if budget.Entries == 0 {
426- data, err := r.readBounded(r.manifestPath(plan.IncarnationID, generation))432+ return false, nil
427- if err != nil {
428- return false, fmt.Errorf("validate retained manifest %d: %w", generation, err)
429- }
430- manifest, err := codec.UnmarshalManifest(data)
431- if err != nil || manifest.Generation != generation || manifest.IncarnationID != plan.IncarnationID || sha256.Sum256(data) != digest {
432- return false, fmt.Errorf("validate retained manifest %d: catalogue reference mismatch", generation)
433 }433 }
434- loaded, err := r.loadGeneration(ctx, manifest.Name, key, generation)434+ refs, consumedBytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, plan.IncarnationID, ref, budget.Bytes)
435 if err != nil {435 if err != nil {
436- return false, fmt.Errorf("validate retained manifest %d: %w", generation, err)436+ return false, fmt.Errorf("validate retained manifest %d: %w", ref.Generation, err)
437 }437 }
438- consumedBytes := uint64(len(data))438+ if !admitted {
439- for _, object := range loaded.Objects {
440- consumedBytes += uint64(len(object))
441- }
442- if budget.Entries == 0 || consumedBytes > budget.Bytes {
443 return false, nil439 return false, nil
444 }440 }
445 budget.Entries--441 budget.Entries--
446 budget.Bytes -= consumedBytes442 budget.Bytes -= consumedBytes
447- for digest := range loaded.Objects {443+ for digest := range refs {
448 state.references[digest] = struct{}{}444 state.references[digest] = struct{}{}
449 }445 }
450 state.validateIndex++446 state.validateIndex++
@@ -591,9 +587,91 @@ func retentionToken(plan ports.RetentionPlan) string {
591 return b.String()587 return b.String()
592 }588 }
593 589
590+// validateMaintenanceCheckpoint admits every payload by stat-reported size
591+// before reading it. It validates directly instead of calling LoadCheckpoint,
592+// which would materialize a complete generation before maintenance could
593+// enforce its byte budget.
594+func (r *Repository) validateMaintenanceCheckpoint(ctx context.Context, id domain.IncarnationID, ref ports.CheckpointRef, byteBudget uint64) (map[ports.SnapshotDigest]codec.ObjectRef, uint64, bool, error) {
595+ manifestPath := r.manifestPath(id, ref.Generation)
596+ info, err := r.stat(manifestPath)
597+ if err != nil {
598+ return nil, 0, false, err
599+ }
600+ if info.Size() < 0 || uint64(info.Size()) > byteBudget {
601+ return nil, 0, false, nil
602+ }
603+ consumed := uint64(info.Size())
604+ if hook := r.hooks.beforeMaintenancePayloadRead; hook != nil {
605+ hook(manifestPath)
606+ }
607+ data, err := r.readBounded(manifestPath)
608+ if err != nil {
609+ return nil, 0, false, err
610+ }
611+ manifest, err := codec.UnmarshalManifest(data)
612+ if err != nil || manifest.Generation != ref.Generation || manifest.IncarnationID != id || sha256.Sum256(data) != ref.ManifestDigest {
613+ return nil, 0, false, errors.New("catalogue reference mismatch")
614+ }
615+ refs, valid := maintenanceManifestRefs(manifest)
616+ if !valid || !withinGenerationBudget(len(data), refs) {
617+ return nil, 0, false, errors.New("invalid manifest references")
618+ }
619+ for digest, objectRef := range refs {
620+ if err := ctx.Err(); err != nil {
621+ return nil, 0, false, err
622+ }
623+ path := r.objectPath(id, digest)
624+ info, err := r.stat(path)
625+ if err != nil {
626+ return nil, 0, false, err
627+ }
628+ if info.Size() < 0 || uint64(info.Size()) > byteBudget-consumed {
629+ return nil, 0, false, nil
630+ }
631+ consumed += uint64(info.Size())
632+ if hook := r.hooks.beforeMaintenancePayloadRead; hook != nil {
633+ hook(path)
634+ }
635+ object, err := r.readBounded(path)
636+ if err != nil {
637+ return nil, 0, false, err
638+ }
639+ if sha256.Sum256(object) != digest || !validObject(object, objectRef) {
640+ return nil, 0, false, errors.New("invalid retained object")
641+ }
642+ }
643+ return refs, consumed, true, nil
644+}
645+
594 // Reconcile scans top-level incarnation namespaces from an external seek646 // Reconcile scans top-level incarnation namespaces from an external seek
595 // cookie. Checkpoint-chain work receives a fresh budget for every directory.647 // cookie. Checkpoint-chain work receives a fresh budget for every directory.
596-func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRecord, cursor ports.ReconcileCursor, budget ports.MaintenanceBudget) (ports.ReconcileCursor, []ports.ReconcileFinding, error) {648+func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRecord, cursor ports.ReconcileCursor, budget ports.MaintenanceBudget) (nextCursor ports.ReconcileCursor, findings []ports.ReconcileFinding, retErr error) {
649+ requested := budget
650+ defer func() {
651+ consumed := ports.MaintenanceBudget{}
652+ for _, finding := range findings {
653+ consumed.Entries += finding.Consumed.Entries
654+ consumed.Bytes += finding.Consumed.Bytes
655+ r.log.Info("snapshot_reconciliation_progress",
656+ "incarnation", finding.Candidate.IncarnationID.String(),
657+ "action", finding.Kind,
658+ "status", finding.Status,
659+ "consumed_entries", finding.Consumed.Entries,
660+ "consumed_bytes", finding.Consumed.Bytes,
661+ "cursor", finding.Cursor.DirectoryCookie,
662+ )
663+ }
664+ r.log.Info("snapshot_reconciliation_complete",
665+ "action", "scan",
666+ "cursor", nextCursor.DirectoryCookie,
667+ "requested_entries", requested.Entries,
668+ "requested_bytes", requested.Bytes,
669+ "consumed_entries", consumed.Entries,
670+ "consumed_bytes", consumed.Bytes,
671+ "done", nextCursor.DirectoryCookie == 0 && retErr == nil,
672+ "err", retErr,
673+ )
674+ }()
597 if err := ctx.Err(); err != nil {675 if err := ctx.Err(); err != nil {
598 return cursor, nil, err676 return cursor, nil, err
599 }677 }
@@ -617,7 +695,7 @@ func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRe
617 if budget.Entries < maintenanceBatch {695 if budget.Entries < maintenanceBatch {
618 findingCapacity = int(budget.Entries)696 findingCapacity = int(budget.Entries)
619 }697 }
620- findings := make([]ports.ReconcileFinding, 0, findingCapacity)698+ findings = make([]ports.ReconcileFinding, 0, findingCapacity)
621 for scanned := uint64(0); scanned < budget.Entries; scanned++ {699 for scanned := uint64(0); scanned < budget.Entries; scanned++ {
622 entries, done, err := r.readMaintenanceDirent(root, 1, state)700 entries, done, err := r.readMaintenanceDirent(root, 1, state)
623 if errors.Is(err, os.ErrNotExist) {701 if errors.Is(err, os.ErrNotExist) {
@@ -704,20 +782,20 @@ func (r *Repository) reconcileIncarnation(ctx context.Context, record domain.Cat
704 }782 }
705 783
706 func (r *Repository) reconcileLoad(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef, consumed *ports.MaintenanceBudget, budget ports.MaintenanceBudget) (ports.SnapshotGeneration, bool, error) {784 func (r *Repository) reconcileLoad(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef, consumed *ports.MaintenanceBudget, budget ports.MaintenanceBudget) (ports.SnapshotGeneration, bool, error) {
707- if consumed.Entries >= budget.Entries {785+ if consumed.Entries >= budget.Entries || consumed.Bytes >= budget.Bytes {
708 return ports.SnapshotGeneration{}, true, nil786 return ports.SnapshotGeneration{}, true, nil
709 }787 }
710- generation, err := r.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)788+ _, bytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, record.IncarnationID, ref, budget.Bytes-consumed.Bytes)
711 if err != nil {789 if err != nil {
712 return ports.SnapshotGeneration{}, false, err790 return ports.SnapshotGeneration{}, false, err
713 }791 }
714- bytes := uint64(len(generation.Manifest))792+ if !admitted {
715- for _, object := range generation.Objects {
716- bytes += uint64(len(object))
717- }
718- if bytes > budget.Bytes-consumed.Bytes {
719 return ports.SnapshotGeneration{}, true, nil793 return ports.SnapshotGeneration{}, true, nil
720 }794 }
795+ generation, err := r.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)
796+ if err != nil {
797+ return ports.SnapshotGeneration{}, false, err
798+ }
721 consumed.Entries++799 consumed.Entries++
722 consumed.Bytes += bytes800 consumed.Bytes += bytes
723 return generation, false, nil801 return generation, false, nil
M internal/adapters/snapshot/maintenance_test.go +33 -0
@@ -1061,6 +1061,39 @@ func TestUnresolvedDataPinned(t *testing.T) {
10611061 }
10621062 }
10631063
1064+func TestMaintenanceBudgetAdmitsPayloadsBeforeRead(t *testing.T) {
1065+ repo := NewRepository(privateDir(t))
1066+ publications := publishMaintenanceGenerations(t, repo, "budget", 1)
1067+ plan := ports.RetentionPlan{IncarnationID: publications[0].IncarnationID, Keep: []ports.CheckpointRef{checkpointRefForPublication(publications[0])}}
1068+
1069+ var reads []string
1070+ repo.hooks.beforeMaintenancePayloadRead = func(path string) { reads = append(reads, path) }
1071+ done, err := repo.MaintainSession(context.Background(), plan, ports.MaintenanceBudget{Entries: 1, Bytes: 1})
1072+ if err != nil {
1073+ t.Fatal(err)
1074+ }
1075+ if done || len(reads) != 0 {
1076+ t.Fatalf("undersized manifest budget: done=%v payload reads=%v", done, reads)
1077+ }
1078+
1079+ manifestPath := repo.manifestPath(plan.IncarnationID, 1)
1080+ info, err := repo.stat(manifestPath)
1081+ if err != nil {
1082+ t.Fatal(err)
1083+ }
1084+ reads = nil
1085+ done, err = repo.MaintainSession(context.Background(), plan, ports.MaintenanceBudget{Entries: 1, Bytes: uint64(info.Size())})
1086+ if err != nil {
1087+ t.Fatal(err)
1088+ }
1089+ if done {
1090+ t.Fatal("manifest-only budget unexpectedly admitted the generation objects")
1091+ }
1092+ if len(reads) != 1 || reads[0] != manifestPath {
1093+ t.Fatalf("payload reads = %v, want only admitted manifest %q", reads, manifestPath)
1094+ }
1095+}
1096+
10641097 func TestPerSessionBudgetIsolation(t *testing.T) {
10651098 repo := NewRepository(privateDir(t))
10661099 large := publishMaintenanceGenerations(t, repo, "large", 4)
@@ -1061,6 +1061,39 @@ func TestUnresolvedDataPinned(t *testing.T) {
1061 }1061 }
1062 }1062 }
1063 1063
1064+func TestMaintenanceBudgetAdmitsPayloadsBeforeRead(t *testing.T) {
1065+ repo := NewRepository(privateDir(t))
1066+ publications := publishMaintenanceGenerations(t, repo, "budget", 1)
1067+ plan := ports.RetentionPlan{IncarnationID: publications[0].IncarnationID, Keep: []ports.CheckpointRef{checkpointRefForPublication(publications[0])}}
1068+
1069+ var reads []string
1070+ repo.hooks.beforeMaintenancePayloadRead = func(path string) { reads = append(reads, path) }
1071+ done, err := repo.MaintainSession(context.Background(), plan, ports.MaintenanceBudget{Entries: 1, Bytes: 1})
1072+ if err != nil {
1073+ t.Fatal(err)
1074+ }
1075+ if done || len(reads) != 0 {
1076+ t.Fatalf("undersized manifest budget: done=%v payload reads=%v", done, reads)
1077+ }
1078+
1079+ manifestPath := repo.manifestPath(plan.IncarnationID, 1)
1080+ info, err := repo.stat(manifestPath)
1081+ if err != nil {
1082+ t.Fatal(err)
1083+ }
1084+ reads = nil
1085+ done, err = repo.MaintainSession(context.Background(), plan, ports.MaintenanceBudget{Entries: 1, Bytes: uint64(info.Size())})
1086+ if err != nil {
1087+ t.Fatal(err)
1088+ }
1089+ if done {
1090+ t.Fatal("manifest-only budget unexpectedly admitted the generation objects")
1091+ }
1092+ if len(reads) != 1 || reads[0] != manifestPath {
1093+ t.Fatalf("payload reads = %v, want only admitted manifest %q", reads, manifestPath)
1094+ }
1095+}
1096+
1064 func TestPerSessionBudgetIsolation(t *testing.T) {1097 func TestPerSessionBudgetIsolation(t *testing.T) {
1065 repo := NewRepository(privateDir(t))1098 repo := NewRepository(privateDir(t))
1066 large := publishMaintenanceGenerations(t, repo, "large", 4)1099 large := publishMaintenanceGenerations(t, repo, "large", 4)
M internal/adapters/snapshot/repository.go +4 -1
@@ -59,7 +59,10 @@ type repositoryHooks struct {
5959 // Object hooks instrument the publication path in package tests. They keep
6060 // the steady-state cost of retained history observable without exposing a
6161 // production metrics surface.
62- beforeObjectRead func(string)
62+ beforeObjectRead func(string)
63+ // beforeMaintenancePayloadRead instruments payload reads after maintenance
64+ // has admitted their stat-reported size against the current byte budget.
65+ beforeMaintenancePayloadRead func(string)
6366 beforeObjectHash func([]byte)
6467 beforeObjectCopy func([]byte)
6568 beforeHeadRead func(string) error
@@ -59,7 +59,10 @@ type repositoryHooks struct {
59 // Object hooks instrument the publication path in package tests. They keep59 // Object hooks instrument the publication path in package tests. They keep
60 // the steady-state cost of retained history observable without exposing a60 // the steady-state cost of retained history observable without exposing a
61 // production metrics surface.61 // production metrics surface.
62- beforeObjectRead func(string)62+ beforeObjectRead func(string)
63+ // beforeMaintenancePayloadRead instruments payload reads after maintenance
64+ // has admitted their stat-reported size against the current byte budget.
65+ beforeMaintenancePayloadRead func(string)
63 beforeObjectHash func([]byte)66 beforeObjectHash func([]byte)
64 beforeObjectCopy func([]byte)67 beforeObjectCopy func([]byte)
65 beforeHeadRead func(string) error68 beforeHeadRead func(string) error
M internal/app/daemon_integration_test.go +2 -2
@@ -533,13 +533,13 @@ func TestLifecycleOwnershipOutlivesSnapshotWriter(t *testing.T) {
533533 result := runBlockedSnapshotWriterShutdown(t)
534534 assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
535535 assertLifecycleResultPending(t, result.wrapperReturned, "lifecycle wrapper return while snapshot writer is blocked")
536- owner, err := lifecycle.TryAcquire(result.runtimeDir)
536+ _, err := lifecycle.TryAcquire(result.runtimeDir)
537537 require.ErrorIs(t, err, lifecycle.ErrBusy)
538538
539539 result.releaseWriter()
540540 awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
541541 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
542- owner, err = lifecycle.TryAcquire(result.runtimeDir)
542+ owner, err := lifecycle.TryAcquire(result.runtimeDir)
543543 require.NoError(t, err)
544544 require.NoError(t, owner.Release())
545545 }
@@ -533,13 +533,13 @@ func TestLifecycleOwnershipOutlivesSnapshotWriter(t *testing.T) {
533 result := runBlockedSnapshotWriterShutdown(t)533 result := runBlockedSnapshotWriterShutdown(t)
534 assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")534 assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
535 assertLifecycleResultPending(t, result.wrapperReturned, "lifecycle wrapper return while snapshot writer is blocked")535 assertLifecycleResultPending(t, result.wrapperReturned, "lifecycle wrapper return while snapshot writer is blocked")
536- owner, err := lifecycle.TryAcquire(result.runtimeDir)536+ _, err := lifecycle.TryAcquire(result.runtimeDir)
537 require.ErrorIs(t, err, lifecycle.ErrBusy)537 require.ErrorIs(t, err, lifecycle.ErrBusy)
538 538
539 result.releaseWriter()539 result.releaseWriter()
540 awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")540 awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
541 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))541 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
542- owner, err = lifecycle.TryAcquire(result.runtimeDir)542+ owner, err := lifecycle.TryAcquire(result.runtimeDir)
543 require.NoError(t, err)543 require.NoError(t, err)
544 require.NoError(t, owner.Release())544 require.NoError(t, owner.Release())
545 }545 }
M internal/app/run.go +16 -4
@@ -543,17 +543,20 @@ func pendingInterruptedRecoveries(ctx context.Context, records []domain.Catalogu
543543 }
544544 }
545545
546-func persistTransactionRecoveryNotices(store ports.NoticeStore, recovered []interruptedRecoveryIdentity, now time.Time) {
546+func persistTransactionRecoveryNotices(store ports.NoticeStore, recovered []interruptedRecoveryIdentity, now time.Time) error {
547547 for _, item := range recovered {
548- _ = store.Append(domain.Notification{
548+ if err := store.Append(domain.Notification{
549549 Code: domain.NoticeSnapshotRestore,
550550 Severity: domain.NoticeWarn,
551551 Message: "recovered an interrupted durable session transaction",
552552 Details: "session=" + item.name + " incarnation=" + item.id.String(),
553553 Time: now,
554554 SessionID: domain.SessionID(item.id.String()),
555- })
555+ }); err != nil {
556+ return fmt.Errorf("persist interrupted transaction recovery notice for session %q incarnation %s: %w", item.name, item.id.String(), err)
557+ }
556558 }
559+ return nil
557560 }
558561
559562 func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {
@@ -654,9 +657,18 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
654657 if err := coordinator.Recover(ctx); err != nil {
655658 return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
656659 }
657- persistTransactionRecoveryNotices(noticeStore, pendingRecoveries, clk.Now())
660+ if err := persistTransactionRecoveryNotices(noticeStore, pendingRecoveries, clk.Now()); err != nil {
661+ log.Error("transaction_recovery_notice_failed", "err", err)
662+ return errors.Join(fmt.Errorf("vev: persist durable transaction recovery notice: %w", err), opened.Catalogue.Close())
663+ }
658664 logTransactionRecovery(log, len(pendingRecoveries))
659665 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
666+ unresolved := make([]domain.IncarnationID, 0, len(pendingRecoveries))
667+ for _, pending := range pendingRecoveries {
668+ unresolved = append(unresolved, pending.id)
669+ }
670+ reconciler := recovery.NewReconciler(coordinator, opened.Catalogue, snapshotRepository, ports.MaintenanceBudget{Entries: 64, Bytes: 8 << 20})
671+ daemonOpts = append(daemonOpts, daemon.WithDurableMaintenance(opened.Catalogue, snapshotRepository, reconciler, unresolved))
660672 log.Info("session persistence enabled", "path", storePath)
661673 daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
662674
@@ -543,17 +543,20 @@ func pendingInterruptedRecoveries(ctx context.Context, records []domain.Catalogu
543 }543 }
544 }544 }
545 545
546-func persistTransactionRecoveryNotices(store ports.NoticeStore, recovered []interruptedRecoveryIdentity, now time.Time) {546+func persistTransactionRecoveryNotices(store ports.NoticeStore, recovered []interruptedRecoveryIdentity, now time.Time) error {
547 for _, item := range recovered {547 for _, item := range recovered {
548- _ = store.Append(domain.Notification{548+ if err := store.Append(domain.Notification{
549 Code: domain.NoticeSnapshotRestore,549 Code: domain.NoticeSnapshotRestore,
550 Severity: domain.NoticeWarn,550 Severity: domain.NoticeWarn,
551 Message: "recovered an interrupted durable session transaction",551 Message: "recovered an interrupted durable session transaction",
552 Details: "session=" + item.name + " incarnation=" + item.id.String(),552 Details: "session=" + item.name + " incarnation=" + item.id.String(),
553 Time: now,553 Time: now,
554 SessionID: domain.SessionID(item.id.String()),554 SessionID: domain.SessionID(item.id.String()),
555- })555+ }); err != nil {
556+ return fmt.Errorf("persist interrupted transaction recovery notice for session %q incarnation %s: %w", item.name, item.id.String(), err)
557+ }
556 }558 }
559+ return nil
557 }560 }
558 561
559 func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {562 func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {
@@ -654,9 +657,18 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
654 if err := coordinator.Recover(ctx); err != nil {657 if err := coordinator.Recover(ctx); err != nil {
655 return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())658 return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
656 }659 }
657- persistTransactionRecoveryNotices(noticeStore, pendingRecoveries, clk.Now())660+ if err := persistTransactionRecoveryNotices(noticeStore, pendingRecoveries, clk.Now()); err != nil {
661+ log.Error("transaction_recovery_notice_failed", "err", err)
662+ return errors.Join(fmt.Errorf("vev: persist durable transaction recovery notice: %w", err), opened.Catalogue.Close())
663+ }
658 logTransactionRecovery(log, len(pendingRecoveries))664 logTransactionRecovery(log, len(pendingRecoveries))
659 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))665 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
666+ unresolved := make([]domain.IncarnationID, 0, len(pendingRecoveries))
667+ for _, pending := range pendingRecoveries {
668+ unresolved = append(unresolved, pending.id)
669+ }
670+ reconciler := recovery.NewReconciler(coordinator, opened.Catalogue, snapshotRepository, ports.MaintenanceBudget{Entries: 64, Bytes: 8 << 20})
671+ daemonOpts = append(daemonOpts, daemon.WithDurableMaintenance(opened.Catalogue, snapshotRepository, reconciler, unresolved))
660 log.Info("session persistence enabled", "path", storePath)672 log.Info("session persistence enabled", "path", storePath)
661 daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))673 daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
662 674
M internal/app/run_test.go +9 -1
@@ -88,7 +88,7 @@ func TestRecoveryObservability(t *testing.T) {
8888 require.NoError(t, runWithLifecycleOwnerDeps(context.Background(), "/runtime/vev", "/state/vev", func(context.Context) error {
8989 logCatalogueRecovery(log, records, "current")
9090 logTransactionRecovery(log, 2)
91- persistTransactionRecoveryNotices(noticeStore, []interruptedRecoveryIdentity{{name: "work", id: domain.IncarnationID{9}}}, time.Unix(1, 0))
91+ require.NoError(t, persistTransactionRecoveryNotices(noticeStore, []interruptedRecoveryIdentity{{name: "work", id: domain.IncarnationID{9}}}, time.Unix(1, 0)))
9292 logStartupRecoveryCounts(log, records, 0)
9393 return nil
9494 }, deps))
@@ -112,6 +112,14 @@ func TestRecoveryObservability(t *testing.T) {
112112 require.NotContains(t, fmt.Sprint(recoveryNotice), "terminal contents")
113113 }
114114
115+func TestTransactionRecoveryNoticeFailurePropagates(t *testing.T) {
116+ store := portsmocks.NewMockNoticeStore(t)
117+ store.EXPECT().Append(mock.Anything).Return(errors.New("disk full")).Once()
118+ err := persistTransactionRecoveryNotices(store, []interruptedRecoveryIdentity{{name: "work", id: domain.IncarnationID{1}}}, time.Unix(1, 0))
119+ require.ErrorContains(t, err, "session \"work\"")
120+ require.ErrorContains(t, err, domain.IncarnationID{1}.String())
121+}
122+
115123 type fakeLifecycleOwnership struct {
116124 release func() error
117125 }
@@ -88,7 +88,7 @@ func TestRecoveryObservability(t *testing.T) {
88 require.NoError(t, runWithLifecycleOwnerDeps(context.Background(), "/runtime/vev", "/state/vev", func(context.Context) error {88 require.NoError(t, runWithLifecycleOwnerDeps(context.Background(), "/runtime/vev", "/state/vev", func(context.Context) error {
89 logCatalogueRecovery(log, records, "current")89 logCatalogueRecovery(log, records, "current")
90 logTransactionRecovery(log, 2)90 logTransactionRecovery(log, 2)
91- persistTransactionRecoveryNotices(noticeStore, []interruptedRecoveryIdentity{{name: "work", id: domain.IncarnationID{9}}}, time.Unix(1, 0))91+ require.NoError(t, persistTransactionRecoveryNotices(noticeStore, []interruptedRecoveryIdentity{{name: "work", id: domain.IncarnationID{9}}}, time.Unix(1, 0)))
92 logStartupRecoveryCounts(log, records, 0)92 logStartupRecoveryCounts(log, records, 0)
93 return nil93 return nil
94 }, deps))94 }, deps))
@@ -112,6 +112,14 @@ func TestRecoveryObservability(t *testing.T) {
112 require.NotContains(t, fmt.Sprint(recoveryNotice), "terminal contents")112 require.NotContains(t, fmt.Sprint(recoveryNotice), "terminal contents")
113 }113 }
114 114
115+func TestTransactionRecoveryNoticeFailurePropagates(t *testing.T) {
116+ store := portsmocks.NewMockNoticeStore(t)
117+ store.EXPECT().Append(mock.Anything).Return(errors.New("disk full")).Once()
118+ err := persistTransactionRecoveryNotices(store, []interruptedRecoveryIdentity{{name: "work", id: domain.IncarnationID{1}}}, time.Unix(1, 0))
119+ require.ErrorContains(t, err, "session \"work\"")
120+ require.ErrorContains(t, err, domain.IncarnationID{1}.String())
121+}
122+
115 type fakeLifecycleOwnership struct {123 type fakeLifecycleOwnership struct {
116 release func() error124 release func() error
117 }125 }
M internal/usecase/daemon/daemon.go +45 -3
@@ -128,6 +128,7 @@ type Daemon struct {
128128 checkpointRecovery ports.CheckpointCoordinator
129129 lifecycleRecovery ports.SessionLifecycleCoordinator
130130 degradedRecovery ports.DegradedRecoveryCoordinator
131+ maintenance maintenanceDependencies
131132 legacySnapshots ports.LegacySnapshotSource
132133 snapsEnabled bool
133134 noticeStore ports.NoticeStore
@@ -155,6 +156,8 @@ type Daemon struct {
155156 // producer or a repository worker.
156157 snapshotNoticeMu sync.Mutex
157158 snapshotActiveFailureSignature string
159+ shutdownNoticeMu sync.Mutex
160+ shutdownNoticedSessions map[string]struct{}
158161 restoreDone chan struct{}
159162 restoreOnce sync.Once
160163 procCwd func(int) (string, error)
@@ -319,6 +322,15 @@ func WithNoticeStore(store ports.NoticeStore) Option {
319322 return func(d *Daemon) { d.noticeStore = store }
320323 }
321324
325+// WithDurableMaintenance schedules bounded retention and reconciliation after
326+// catalogue-driven restoration. Each call receives an independent budget;
327+// unresolved incarnation IDs remain pinned for this daemon lifetime.
328+func WithDurableMaintenance(catalogue ports.Catalogue, repository ports.SnapshotRepository, reconciler maintenanceReconciler, unresolved []domain.IncarnationID) Option {
329+ return func(d *Daemon) {
330+ d.maintenance = newMaintenanceDependencies(catalogue, repository, reconciler, unresolved)
331+ }
332+}
333+
322334 // WithCwdReader overrides the process cwd reader used for persistence tests.
323335 func WithCwdReader(fn func(int) (string, error)) Option {
324336 return func(d *Daemon) {
@@ -533,6 +545,11 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
533545 } else {
534546 d.closeRestoreDone()
535547 }
548+ if d.maintenance.reconciler != nil {
549+ d.sessWg.Go(func() {
550+ d.runDurableMaintenance(d.serveCtx)
551+ })
552+ }
536553
537554 // Break the accept loop when either the parent context is cancelled or the
538555 // registry drains to empty: both close the listener, which fails Accept.
@@ -586,9 +603,15 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
586603 d.connWg.Wait()
587604 d.attachmentCleanupWg.Wait()
588605 // Forced terminal checkpoints run before shutdown stops producers or the
589- // worker. The deadline may abandon checkpoint success and cancel cooperative
590- // I/O, but ownership remains pinned until every writer has actually exited.
591- d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)
606+ // worker. StopDurableWriters reports every session whose final capture still
607+ // owns work when the shared checkpoint budget expires. Persist one notice per
608+ // affected session, then retain ownership until every writer actually exits.
609+ stopCtx, stopCancel := snapshotStopContext(snapshotDeadline)
610+ timedOutSessions := d.StopDurableWriters(stopCtx)
611+ stopCancel()
612+ for _, name := range timedOutSessions {
613+ d.persistShutdownSnapshotFailure(name, context.DeadlineExceeded)
614+ }
592615 d.WaitDurableWriters()
593616 d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
594617 d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
@@ -701,6 +724,25 @@ func (d *snapshotShutdownDeadline) Done() <-chan struct{} {
701724 return d.done
702725 }
703726
727+func snapshotStopContext(deadline *snapshotShutdownDeadline) (context.Context, context.CancelFunc) {
728+ ctx, cancel := context.WithCancel(context.Background())
729+ if deadline == nil {
730+ return ctx, cancel
731+ }
732+ stop := make(chan struct{})
733+ go func() {
734+ select {
735+ case <-deadline.Done():
736+ cancel()
737+ case <-stop:
738+ }
739+ }()
740+ return ctx, func() {
741+ close(stop)
742+ cancel()
743+ }
744+}
745+
704746 func (d *Daemon) sessionsSnapshotLocked() []*session {
705747 snapshot := make([]*session, 0, len(d.sessions))
706748 for _, s := range d.sessions {
@@ -128,6 +128,7 @@ type Daemon struct {
128 checkpointRecovery ports.CheckpointCoordinator128 checkpointRecovery ports.CheckpointCoordinator
129 lifecycleRecovery ports.SessionLifecycleCoordinator129 lifecycleRecovery ports.SessionLifecycleCoordinator
130 degradedRecovery ports.DegradedRecoveryCoordinator130 degradedRecovery ports.DegradedRecoveryCoordinator
131+ maintenance maintenanceDependencies
131 legacySnapshots ports.LegacySnapshotSource132 legacySnapshots ports.LegacySnapshotSource
132 snapsEnabled bool133 snapsEnabled bool
133 noticeStore ports.NoticeStore134 noticeStore ports.NoticeStore
@@ -155,6 +156,8 @@ type Daemon struct {
155 // producer or a repository worker.156 // producer or a repository worker.
156 snapshotNoticeMu sync.Mutex157 snapshotNoticeMu sync.Mutex
157 snapshotActiveFailureSignature string158 snapshotActiveFailureSignature string
159+ shutdownNoticeMu sync.Mutex
160+ shutdownNoticedSessions map[string]struct{}
158 restoreDone chan struct{}161 restoreDone chan struct{}
159 restoreOnce sync.Once162 restoreOnce sync.Once
160 procCwd func(int) (string, error)163 procCwd func(int) (string, error)
@@ -319,6 +322,15 @@ func WithNoticeStore(store ports.NoticeStore) Option {
319 return func(d *Daemon) { d.noticeStore = store }322 return func(d *Daemon) { d.noticeStore = store }
320 }323 }
321 324
325+// WithDurableMaintenance schedules bounded retention and reconciliation after
326+// catalogue-driven restoration. Each call receives an independent budget;
327+// unresolved incarnation IDs remain pinned for this daemon lifetime.
328+func WithDurableMaintenance(catalogue ports.Catalogue, repository ports.SnapshotRepository, reconciler maintenanceReconciler, unresolved []domain.IncarnationID) Option {
329+ return func(d *Daemon) {
330+ d.maintenance = newMaintenanceDependencies(catalogue, repository, reconciler, unresolved)
331+ }
332+}
333+
322 // WithCwdReader overrides the process cwd reader used for persistence tests.334 // WithCwdReader overrides the process cwd reader used for persistence tests.
323 func WithCwdReader(fn func(int) (string, error)) Option {335 func WithCwdReader(fn func(int) (string, error)) Option {
324 return func(d *Daemon) {336 return func(d *Daemon) {
@@ -533,6 +545,11 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
533 } else {545 } else {
534 d.closeRestoreDone()546 d.closeRestoreDone()
535 }547 }
548+ if d.maintenance.reconciler != nil {
549+ d.sessWg.Go(func() {
550+ d.runDurableMaintenance(d.serveCtx)
551+ })
552+ }
536 553
537 // Break the accept loop when either the parent context is cancelled or the554 // Break the accept loop when either the parent context is cancelled or the
538 // registry drains to empty: both close the listener, which fails Accept.555 // registry drains to empty: both close the listener, which fails Accept.
@@ -586,9 +603,15 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
586 d.connWg.Wait()603 d.connWg.Wait()
587 d.attachmentCleanupWg.Wait()604 d.attachmentCleanupWg.Wait()
588 // Forced terminal checkpoints run before shutdown stops producers or the605 // Forced terminal checkpoints run before shutdown stops producers or the
589- // worker. The deadline may abandon checkpoint success and cancel cooperative606+ // worker. StopDurableWriters reports every session whose final capture still
590- // I/O, but ownership remains pinned until every writer has actually exited.607+ // owns work when the shared checkpoint budget expires. Persist one notice per
591- d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)608+ // affected session, then retain ownership until every writer actually exits.
609+ stopCtx, stopCancel := snapshotStopContext(snapshotDeadline)
610+ timedOutSessions := d.StopDurableWriters(stopCtx)
611+ stopCancel()
612+ for _, name := range timedOutSessions {
613+ d.persistShutdownSnapshotFailure(name, context.DeadlineExceeded)
614+ }
592 d.WaitDurableWriters()615 d.WaitDurableWriters()
593 d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)616 d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
594 d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)617 d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
@@ -701,6 +724,25 @@ func (d *snapshotShutdownDeadline) Done() <-chan struct{} {
701 return d.done724 return d.done
702 }725 }
703 726
727+func snapshotStopContext(deadline *snapshotShutdownDeadline) (context.Context, context.CancelFunc) {
728+ ctx, cancel := context.WithCancel(context.Background())
729+ if deadline == nil {
730+ return ctx, cancel
731+ }
732+ stop := make(chan struct{})
733+ go func() {
734+ select {
735+ case <-deadline.Done():
736+ cancel()
737+ case <-stop:
738+ }
739+ }()
740+ return ctx, func() {
741+ close(stop)
742+ cancel()
743+ }
744+}
745+
704 func (d *Daemon) sessionsSnapshotLocked() []*session {746 func (d *Daemon) sessionsSnapshotLocked() []*session {
705 snapshot := make([]*session, 0, len(d.sessions))747 snapshot := make([]*session, 0, len(d.sessions))
706 for _, s := range d.sessions {748 for _, s := range d.sessions {
A internal/usecase/daemon/maintenance.go +143 -0
@@ -0,0 +1,143 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+)
10+
11+const (
12+ maintenanceInterval = time.Minute
13+ maintenanceEntries = 64
14+ maintenanceBytes = 8 << 20
15+)
16+
17+type maintenanceReconciler interface {
18+ Step(context.Context, ports.ReconcileCursor) (ports.ReconcileCursor, []ports.ReconcileDecision, error)
19+}
20+
21+type maintenanceDependencies struct {
22+ catalogue ports.Catalogue
23+ repository ports.SnapshotRepository
24+ reconciler maintenanceReconciler
25+ unresolved map[domain.IncarnationID]struct{}
26+ repairUncertain map[domain.IncarnationID]struct{}
27+ cursor ports.ReconcileCursor
28+}
29+
30+func newMaintenanceDependencies(catalogue ports.Catalogue, repository ports.SnapshotRepository, reconciler maintenanceReconciler, unresolved []domain.IncarnationID) maintenanceDependencies {
31+ deps := maintenanceDependencies{
32+ catalogue: catalogue, repository: repository, reconciler: reconciler,
33+ unresolved: make(map[domain.IncarnationID]struct{}, len(unresolved)),
34+ repairUncertain: make(map[domain.IncarnationID]struct{}),
35+ }
36+ for _, id := range unresolved {
37+ if id != (domain.IncarnationID{}) {
38+ deps.unresolved[id] = struct{}{}
39+ }
40+ }
41+ return deps
42+}
43+
44+func retentionPlan(record domain.CatalogueRecord, unresolved, restoring bool) ports.RetentionPlan {
45+ plan := ports.RetentionPlan{IncarnationID: record.IncarnationID}
46+ if unresolved || restoring || record.RecoveryState == domain.RecoveryDegraded || record.RecoveryState == domain.RecoveryDeleting {
47+ plan.PinAll = true
48+ return plan
49+ }
50+ if record.Committed != nil {
51+ plan.Keep = append(plan.Keep, *record.Committed)
52+ }
53+ for _, fallback := range record.Fallbacks {
54+ if fallback != nil {
55+ plan.Keep = append(plan.Keep, *fallback)
56+ }
57+ }
58+ return plan
59+}
60+
61+func (d *Daemon) runDurableMaintenance(ctx context.Context) {
62+ select {
63+ case <-ctx.Done():
64+ return
65+ case <-d.restoreDone:
66+ }
67+ for {
68+ d.runDurableMaintenanceTick(ctx)
69+ timer := d.clock.NewTimer(maintenanceInterval)
70+ select {
71+ case <-ctx.Done():
72+ timer.Stop()
73+ return
74+ case <-timer.C():
75+ }
76+ }
77+}
78+
79+func (d *Daemon) runDurableMaintenanceTick(ctx context.Context) {
80+ deps := &d.maintenance
81+ if deps.catalogue == nil || deps.repository == nil || deps.reconciler == nil {
82+ return
83+ }
84+
85+ // Reconcile before collection. Until a complete top-level scan reaches EOF,
86+ // an unseen forward orphan may still be repairable, so retention yields
87+ // without deleting any incarnation data.
88+ next, decisions, err := deps.reconciler.Step(ctx, deps.cursor)
89+ if err != nil {
90+ d.log.Warn("durable_reconciliation_tick", "action", "scan", "cursor", deps.cursor.DirectoryCookie, "done", false, "err", err)
91+ return
92+ }
93+ deps.cursor = next
94+ for _, decision := range decisions {
95+ record, ok := deps.catalogue.Record(decision.Name)
96+ if ok && decision.Kind != ports.ReconcileAdopt {
97+ deps.repairUncertain[record.IncarnationID] = struct{}{}
98+ }
99+ d.log.Info("durable_reconciliation_decision",
100+ "incarnation", record.IncarnationID.String(),
101+ "session", decision.Name,
102+ "action", decision.Kind,
103+ "reason_code", decision.ReasonCode,
104+ "cursor", next.DirectoryCookie,
105+ )
106+ }
107+ d.log.Info("durable_reconciliation_tick", "action", "scan", "cursor", next.DirectoryCookie, "done", next.DirectoryCookie == 0, "err", nil)
108+ if next.DirectoryCookie != 0 {
109+ return
110+ }
111+
112+ for _, record := range deps.catalogue.Records() {
113+ if err := ctx.Err(); err != nil {
114+ return
115+ }
116+ _, unresolved := deps.unresolved[record.IncarnationID]
117+ _, uncertainRepair := deps.repairUncertain[record.IncarnationID]
118+ restoring := d.incarnationRestoring(record.IncarnationID)
119+ plan := retentionPlan(record, unresolved || uncertainRepair, restoring)
120+ done, err := deps.repository.MaintainSession(ctx, plan, ports.MaintenanceBudget{Entries: maintenanceEntries, Bytes: maintenanceBytes})
121+ d.log.Info("durable_maintenance_session",
122+ "incarnation", record.IncarnationID.String(),
123+ "action", "retention",
124+ "pin_all", plan.PinAll,
125+ "requested_entries", maintenanceEntries,
126+ "requested_bytes", maintenanceBytes,
127+ "done", done,
128+ "err", err,
129+ )
130+ }
131+ clear(deps.repairUncertain)
132+}
133+
134+func (d *Daemon) incarnationRestoring(id domain.IncarnationID) bool {
135+ d.mu.Lock()
136+ defer d.mu.Unlock()
137+ for _, stopped := range d.stopped {
138+ if stopped.incarnation == id && stopped.state == runtimeRestoring {
139+ return true
140+ }
141+ }
142+ return false
143+}
@@ -0,0 +1,143 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+)
10+
11+const (
12+ maintenanceInterval = time.Minute
13+ maintenanceEntries = 64
14+ maintenanceBytes = 8 << 20
15+)
16+
17+type maintenanceReconciler interface {
18+ Step(context.Context, ports.ReconcileCursor) (ports.ReconcileCursor, []ports.ReconcileDecision, error)
19+}
20+
21+type maintenanceDependencies struct {
22+ catalogue ports.Catalogue
23+ repository ports.SnapshotRepository
24+ reconciler maintenanceReconciler
25+ unresolved map[domain.IncarnationID]struct{}
26+ repairUncertain map[domain.IncarnationID]struct{}
27+ cursor ports.ReconcileCursor
28+}
29+
30+func newMaintenanceDependencies(catalogue ports.Catalogue, repository ports.SnapshotRepository, reconciler maintenanceReconciler, unresolved []domain.IncarnationID) maintenanceDependencies {
31+ deps := maintenanceDependencies{
32+ catalogue: catalogue, repository: repository, reconciler: reconciler,
33+ unresolved: make(map[domain.IncarnationID]struct{}, len(unresolved)),
34+ repairUncertain: make(map[domain.IncarnationID]struct{}),
35+ }
36+ for _, id := range unresolved {
37+ if id != (domain.IncarnationID{}) {
38+ deps.unresolved[id] = struct{}{}
39+ }
40+ }
41+ return deps
42+}
43+
44+func retentionPlan(record domain.CatalogueRecord, unresolved, restoring bool) ports.RetentionPlan {
45+ plan := ports.RetentionPlan{IncarnationID: record.IncarnationID}
46+ if unresolved || restoring || record.RecoveryState == domain.RecoveryDegraded || record.RecoveryState == domain.RecoveryDeleting {
47+ plan.PinAll = true
48+ return plan
49+ }
50+ if record.Committed != nil {
51+ plan.Keep = append(plan.Keep, *record.Committed)
52+ }
53+ for _, fallback := range record.Fallbacks {
54+ if fallback != nil {
55+ plan.Keep = append(plan.Keep, *fallback)
56+ }
57+ }
58+ return plan
59+}
60+
61+func (d *Daemon) runDurableMaintenance(ctx context.Context) {
62+ select {
63+ case <-ctx.Done():
64+ return
65+ case <-d.restoreDone:
66+ }
67+ for {
68+ d.runDurableMaintenanceTick(ctx)
69+ timer := d.clock.NewTimer(maintenanceInterval)
70+ select {
71+ case <-ctx.Done():
72+ timer.Stop()
73+ return
74+ case <-timer.C():
75+ }
76+ }
77+}
78+
79+func (d *Daemon) runDurableMaintenanceTick(ctx context.Context) {
80+ deps := &d.maintenance
81+ if deps.catalogue == nil || deps.repository == nil || deps.reconciler == nil {
82+ return
83+ }
84+
85+ // Reconcile before collection. Until a complete top-level scan reaches EOF,
86+ // an unseen forward orphan may still be repairable, so retention yields
87+ // without deleting any incarnation data.
88+ next, decisions, err := deps.reconciler.Step(ctx, deps.cursor)
89+ if err != nil {
90+ d.log.Warn("durable_reconciliation_tick", "action", "scan", "cursor", deps.cursor.DirectoryCookie, "done", false, "err", err)
91+ return
92+ }
93+ deps.cursor = next
94+ for _, decision := range decisions {
95+ record, ok := deps.catalogue.Record(decision.Name)
96+ if ok && decision.Kind != ports.ReconcileAdopt {
97+ deps.repairUncertain[record.IncarnationID] = struct{}{}
98+ }
99+ d.log.Info("durable_reconciliation_decision",
100+ "incarnation", record.IncarnationID.String(),
101+ "session", decision.Name,
102+ "action", decision.Kind,
103+ "reason_code", decision.ReasonCode,
104+ "cursor", next.DirectoryCookie,
105+ )
106+ }
107+ d.log.Info("durable_reconciliation_tick", "action", "scan", "cursor", next.DirectoryCookie, "done", next.DirectoryCookie == 0, "err", nil)
108+ if next.DirectoryCookie != 0 {
109+ return
110+ }
111+
112+ for _, record := range deps.catalogue.Records() {
113+ if err := ctx.Err(); err != nil {
114+ return
115+ }
116+ _, unresolved := deps.unresolved[record.IncarnationID]
117+ _, uncertainRepair := deps.repairUncertain[record.IncarnationID]
118+ restoring := d.incarnationRestoring(record.IncarnationID)
119+ plan := retentionPlan(record, unresolved || uncertainRepair, restoring)
120+ done, err := deps.repository.MaintainSession(ctx, plan, ports.MaintenanceBudget{Entries: maintenanceEntries, Bytes: maintenanceBytes})
121+ d.log.Info("durable_maintenance_session",
122+ "incarnation", record.IncarnationID.String(),
123+ "action", "retention",
124+ "pin_all", plan.PinAll,
125+ "requested_entries", maintenanceEntries,
126+ "requested_bytes", maintenanceBytes,
127+ "done", done,
128+ "err", err,
129+ )
130+ }
131+ clear(deps.repairUncertain)
132+}
133+
134+func (d *Daemon) incarnationRestoring(id domain.IncarnationID) bool {
135+ d.mu.Lock()
136+ defer d.mu.Unlock()
137+ for _, stopped := range d.stopped {
138+ if stopped.incarnation == id && stopped.state == runtimeRestoring {
139+ return true
140+ }
141+ }
142+ return false
143+}
A internal/usecase/daemon/maintenance_test.go +76 -0
@@ -0,0 +1,76 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "testing"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
10+ "github.com/stretchr/testify/mock"
11+)
12+
13+type cursorMaintenanceReconciler struct {
14+ seen []ports.ReconcileCursor
15+}
16+
17+func (r *cursorMaintenanceReconciler) Step(_ context.Context, cursor ports.ReconcileCursor) (ports.ReconcileCursor, []ports.ReconcileDecision, error) {
18+ r.seen = append(r.seen, cursor)
19+ if cursor.DirectoryCookie == 0 {
20+ return ports.ReconcileCursor{DirectoryCookie: 7}, nil, nil
21+ }
22+ return ports.ReconcileCursor{}, nil, nil
23+}
24+
25+func TestProductionMaintenanceUsesCatalogueAndResumesCursor(t *testing.T) {
26+ record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryFresh}
27+ catalogue := portsmocks.NewMockCatalogue(t)
28+ catalogue.EXPECT().Records().Return([]domain.CatalogueRecord{record}).Once()
29+ repository := portsmocks.NewMockSnapshotRepository(t)
30+ repository.EXPECT().MaintainSession(mock.Anything, mock.MatchedBy(func(plan ports.RetentionPlan) bool {
31+ return plan.IncarnationID == record.IncarnationID && !plan.PinAll && len(plan.Keep) == 0
32+ }), ports.MaintenanceBudget{Entries: maintenanceEntries, Bytes: maintenanceBytes}).Return(true, nil).Once()
33+ reconciler := &cursorMaintenanceReconciler{}
34+ d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
35+ WithDurableMaintenance(catalogue, repository, reconciler, nil)(d)
36+
37+ d.runDurableMaintenanceTick(context.Background())
38+ d.runDurableMaintenanceTick(context.Background())
39+ if len(reconciler.seen) != 2 || reconciler.seen[0].DirectoryCookie != 0 || reconciler.seen[1].DirectoryCookie != 7 {
40+ t.Fatalf("reconciliation cursors = %v, want [0 7]", reconciler.seen)
41+ }
42+}
43+
44+func TestRetentionPlanPinsEveryUnresolvedRuntimeState(t *testing.T) {
45+ committed := domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}}
46+ fallback1 := domain.CheckpointRef{Generation: 2, ManifestDigest: [32]byte{2}}
47+ fallback2 := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
48+ healthy := domain.CatalogueRecord{
49+ Name: "work", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryHealthy,
50+ Committed: &committed, Fallbacks: [2]*domain.CheckpointRef{&fallback1, &fallback2},
51+ }
52+
53+ for _, tc := range []struct {
54+ name string
55+ record domain.CatalogueRecord
56+ unresolved bool
57+ restoring bool
58+ wantPin bool
59+ }{
60+ {name: "healthy", record: healthy},
61+ {name: "pending journal tombstone or repair", record: healthy, unresolved: true, wantPin: true},
62+ {name: "restoring", record: healthy, restoring: true, wantPin: true},
63+ {name: "degraded", record: func() domain.CatalogueRecord { r := healthy; r.RecoveryState = domain.RecoveryDegraded; return r }(), wantPin: true},
64+ {name: "deleting", record: func() domain.CatalogueRecord { r := healthy; r.RecoveryState = domain.RecoveryDeleting; return r }(), wantPin: true},
65+ } {
66+ t.Run(tc.name, func(t *testing.T) {
67+ plan := retentionPlan(tc.record, tc.unresolved, tc.restoring)
68+ if plan.PinAll != tc.wantPin {
69+ t.Fatalf("PinAll = %v, want %v", plan.PinAll, tc.wantPin)
70+ }
71+ if !tc.wantPin && len(plan.Keep) != 3 {
72+ t.Fatalf("healthy keep set = %v, want committed plus two fallbacks", plan.Keep)
73+ }
74+ })
75+ }
76+}
@@ -0,0 +1,76 @@
1+package daemon
2+
3+import (
4+ "context"
5+ "testing"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
10+ "github.com/stretchr/testify/mock"
11+)
12+
13+type cursorMaintenanceReconciler struct {
14+ seen []ports.ReconcileCursor
15+}
16+
17+func (r *cursorMaintenanceReconciler) Step(_ context.Context, cursor ports.ReconcileCursor) (ports.ReconcileCursor, []ports.ReconcileDecision, error) {
18+ r.seen = append(r.seen, cursor)
19+ if cursor.DirectoryCookie == 0 {
20+ return ports.ReconcileCursor{DirectoryCookie: 7}, nil, nil
21+ }
22+ return ports.ReconcileCursor{}, nil, nil
23+}
24+
25+func TestProductionMaintenanceUsesCatalogueAndResumesCursor(t *testing.T) {
26+ record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryFresh}
27+ catalogue := portsmocks.NewMockCatalogue(t)
28+ catalogue.EXPECT().Records().Return([]domain.CatalogueRecord{record}).Once()
29+ repository := portsmocks.NewMockSnapshotRepository(t)
30+ repository.EXPECT().MaintainSession(mock.Anything, mock.MatchedBy(func(plan ports.RetentionPlan) bool {
31+ return plan.IncarnationID == record.IncarnationID && !plan.PinAll && len(plan.Keep) == 0
32+ }), ports.MaintenanceBudget{Entries: maintenanceEntries, Bytes: maintenanceBytes}).Return(true, nil).Once()
33+ reconciler := &cursorMaintenanceReconciler{}
34+ d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
35+ WithDurableMaintenance(catalogue, repository, reconciler, nil)(d)
36+
37+ d.runDurableMaintenanceTick(context.Background())
38+ d.runDurableMaintenanceTick(context.Background())
39+ if len(reconciler.seen) != 2 || reconciler.seen[0].DirectoryCookie != 0 || reconciler.seen[1].DirectoryCookie != 7 {
40+ t.Fatalf("reconciliation cursors = %v, want [0 7]", reconciler.seen)
41+ }
42+}
43+
44+func TestRetentionPlanPinsEveryUnresolvedRuntimeState(t *testing.T) {
45+ committed := domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}}
46+ fallback1 := domain.CheckpointRef{Generation: 2, ManifestDigest: [32]byte{2}}
47+ fallback2 := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
48+ healthy := domain.CatalogueRecord{
49+ Name: "work", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryHealthy,
50+ Committed: &committed, Fallbacks: [2]*domain.CheckpointRef{&fallback1, &fallback2},
51+ }
52+
53+ for _, tc := range []struct {
54+ name string
55+ record domain.CatalogueRecord
56+ unresolved bool
57+ restoring bool
58+ wantPin bool
59+ }{
60+ {name: "healthy", record: healthy},
61+ {name: "pending journal tombstone or repair", record: healthy, unresolved: true, wantPin: true},
62+ {name: "restoring", record: healthy, restoring: true, wantPin: true},
63+ {name: "degraded", record: func() domain.CatalogueRecord { r := healthy; r.RecoveryState = domain.RecoveryDegraded; return r }(), wantPin: true},
64+ {name: "deleting", record: func() domain.CatalogueRecord { r := healthy; r.RecoveryState = domain.RecoveryDeleting; return r }(), wantPin: true},
65+ } {
66+ t.Run(tc.name, func(t *testing.T) {
67+ plan := retentionPlan(tc.record, tc.unresolved, tc.restoring)
68+ if plan.PinAll != tc.wantPin {
69+ t.Fatalf("PinAll = %v, want %v", plan.PinAll, tc.wantPin)
70+ }
71+ if !tc.wantPin && len(plan.Keep) != 3 {
72+ t.Fatalf("healthy keep set = %v, want committed plus two fallbacks", plan.Keep)
73+ }
74+ })
75+ }
76+}
M internal/usecase/daemon/notify.go +14 -1
@@ -198,6 +198,16 @@ func (d *Daemon) persistShutdownSnapshotFailure(name string, cause error) {
198198 if d.noticeStore == nil {
199199 return
200200 }
201+ d.shutdownNoticeMu.Lock()
202+ if d.shutdownNoticedSessions == nil {
203+ d.shutdownNoticedSessions = make(map[string]struct{})
204+ }
205+ if _, exists := d.shutdownNoticedSessions[name]; exists {
206+ d.shutdownNoticeMu.Unlock()
207+ return
208+ }
209+ d.shutdownNoticedSessions[name] = struct{}{}
210+ d.shutdownNoticeMu.Unlock()
201211 if err := d.noticeStore.Append(domain.Notification{
202212 Code: domain.NoticeSnapshotWrite,
203213 Severity: domain.NoticeError,
@@ -205,7 +215,10 @@ func (d *Daemon) persistShutdownSnapshotFailure(name string, cause error) {
205215 Details: noticeDetails(cause),
206216 Time: d.clock.Now(),
207217 }); err != nil {
208- d.log.Warn("persisting shutdown notice failed", "err", err, "session", name)
218+ d.shutdownNoticeMu.Lock()
219+ delete(d.shutdownNoticedSessions, name)
220+ d.shutdownNoticeMu.Unlock()
221+ d.log.Warn("persisting shutdown notice failed", "err", err, "session", name, "action", "final-checkpoint-timeout")
209222 }
210223 }
211224
@@ -198,6 +198,16 @@ func (d *Daemon) persistShutdownSnapshotFailure(name string, cause error) {
198 if d.noticeStore == nil {198 if d.noticeStore == nil {
199 return199 return
200 }200 }
201+ d.shutdownNoticeMu.Lock()
202+ if d.shutdownNoticedSessions == nil {
203+ d.shutdownNoticedSessions = make(map[string]struct{})
204+ }
205+ if _, exists := d.shutdownNoticedSessions[name]; exists {
206+ d.shutdownNoticeMu.Unlock()
207+ return
208+ }
209+ d.shutdownNoticedSessions[name] = struct{}{}
210+ d.shutdownNoticeMu.Unlock()
201 if err := d.noticeStore.Append(domain.Notification{211 if err := d.noticeStore.Append(domain.Notification{
202 Code: domain.NoticeSnapshotWrite,212 Code: domain.NoticeSnapshotWrite,
203 Severity: domain.NoticeError,213 Severity: domain.NoticeError,
@@ -205,7 +215,10 @@ func (d *Daemon) persistShutdownSnapshotFailure(name string, cause error) {
205 Details: noticeDetails(cause),215 Details: noticeDetails(cause),
206 Time: d.clock.Now(),216 Time: d.clock.Now(),
207 }); err != nil {217 }); err != nil {
208- d.log.Warn("persisting shutdown notice failed", "err", err, "session", name)218+ d.shutdownNoticeMu.Lock()
219+ delete(d.shutdownNoticedSessions, name)
220+ d.shutdownNoticeMu.Unlock()
221+ d.log.Warn("persisting shutdown notice failed", "err", err, "session", name, "action", "final-checkpoint-timeout")
209 }222 }
210 }223 }
211 224
M internal/usecase/daemon/snapshot_worker.go +0 -21
@@ -272,27 +272,6 @@ func (d *Daemon) durableWriterFailureNames() []string {
272272 return names
273273 }
274274
275-// stopSnapshotEncodeWorkerWithDeadline bounds the final checkpoint attempt,
276-// but deliberately does not bound the ownership join that follows it.
277-func (d *Daemon) stopSnapshotEncodeWorkerWithDeadline(deadline *snapshotShutdownDeadline) {
278- if d == nil {
279- return
280- }
281- cancel, done := d.requestDurableWriterStop()
282- if done == nil {
283- return
284- }
285- if deadline == nil {
286- <-done
287- return
288- }
289- select {
290- case <-done:
291- case <-deadline.Done():
292- cancel()
293- }
294-}
295-
296275 func (d *Daemon) finishStoppedSnapshotWorker(abandoned bool) {
297276 d.snapshotWorkerMu.Lock()
298277 inFlight := d.snapshotWorkerInFlight
@@ -272,27 +272,6 @@ func (d *Daemon) durableWriterFailureNames() []string {
272 return names272 return names
273 }273 }
274 274
275-// stopSnapshotEncodeWorkerWithDeadline bounds the final checkpoint attempt,
276-// but deliberately does not bound the ownership join that follows it.
277-func (d *Daemon) stopSnapshotEncodeWorkerWithDeadline(deadline *snapshotShutdownDeadline) {
278- if d == nil {
279- return
280- }
281- cancel, done := d.requestDurableWriterStop()
282- if done == nil {
283- return
284- }
285- if deadline == nil {
286- <-done
287- return
288- }
289- select {
290- case <-done:
291- case <-deadline.Done():
292- cancel()
293- }
294-}
295-
296 func (d *Daemon) finishStoppedSnapshotWorker(abandoned bool) {275 func (d *Daemon) finishStoppedSnapshotWorker(abandoned bool) {
297 d.snapshotWorkerMu.Lock()276 d.snapshotWorkerMu.Lock()
298 inFlight := d.snapshotWorkerInFlight277 inFlight := d.snapshotWorkerInFlight

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