2a1a401 perf(recovery): avoid reconciliation rereads · vev

M internal/adapters/snapshot/atomic.go +3 -0


@@ -344,6 +344,9 @@ func (r Repository) readBoundedRoot(root os.Root, path string) ([]byte, error)
344	344		if err != nil {
345	345		return nil, err
346	346		}
	347	+	if hook := r.hooks.beforePayloadRead; hook != nil {
	348	+	hook(path)
	349	+	}
347	350		return readBoundedFile(f)
348	351		}
349	352


@@ -344,6 +344,9 @@ func (r Repository) readBoundedRoot(root os.Root, path string) ([]byte, error)
344	if err != nil {	344		if err != nil {
345	return nil, err	345		return nil, err
346	}	346		}
		347	+	if hook := r.hooks.beforePayloadRead; hook != nil {
		348	+	hook(path)
		349	+	}
347	return readBoundedFile(f)	350		return readBoundedFile(f)
348	}	351		}
349		352

M internal/adapters/snapshot/maintenance_gc.go +53 -31


@@ -431,7 +431,7 @@ func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPl
431	431		if budget.Entries == 0 {
432	432		return false, nil
433	433		}
434		-	refs, consumedBytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, plan.IncarnationID, ref, budget.Bytes)
	434	+	validated, consumedBytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, plan.IncarnationID, ref, budget.Bytes)
435	435		if err != nil {
436	436		return false, fmt.Errorf("validate retained manifest %d: %w", ref.Generation, err)
437	437		}

@@ -440,7 +440,7 @@ func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPl
440	440		}
441	441		budget.Entries--
442	442		budget.Bytes -= consumedBytes
443		-	for digest := range refs {
	443	+	for digest := range validated.refs {
444	444		state.references[digest] = struct{}{}
445	445		}
446	446		state.validateIndex++

@@ -591,14 +591,19 @@ func retentionToken(plan ports.RetentionPlan) string {
591	591		// before reading it. It validates directly instead of calling LoadCheckpoint,
592	592		// which would materialize a complete generation before maintenance could
593	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) {
	594	+	type validatedMaintenanceCheckpoint struct {
	595	+	refs map[ports.SnapshotDigest]codec.ObjectRef
	596	+	parent *domain.CheckpointRef
	597	+	}
	598	+
	599	+	func (r *Repository) validateMaintenanceCheckpoint(ctx context.Context, id domain.IncarnationID, ref ports.CheckpointRef, byteBudget uint64) (validatedMaintenanceCheckpoint, uint64, bool, error) {
595	600		manifestPath := r.manifestPath(id, ref.Generation)
596	601		info, err := r.stat(manifestPath)
597	602		if err != nil {
598		-	return nil, 0, false, err
	603	+	return validatedMaintenanceCheckpoint{}, 0, false, err
599	604		}
600	605		if info.Size() < 0 \|\| uint64(info.Size()) > byteBudget {
601		-	return nil, 0, false, nil
	606	+	return validatedMaintenanceCheckpoint{}, 0, false, nil
602	607		}
603	608		consumed := uint64(info.Size())
604	609		if hook := r.hooks.beforeMaintenancePayloadRead; hook != nil {

@@ -606,27 +611,27 @@ func (r *Repository) validateMaintenanceCheckpoint(ctx context.Context, id domai
606	611		}
607	612		data, err := r.readBounded(manifestPath)
608	613		if err != nil {
609		-	return nil, 0, false, err
	614	+	return validatedMaintenanceCheckpoint{}, consumed, false, err
610	615		}
611	616		manifest, err := codec.UnmarshalManifest(data)
612	617		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")
	618	+	return validatedMaintenanceCheckpoint{}, consumed, false, errors.New("catalogue reference mismatch")
614	619		}
615	620		refs, valid := maintenanceManifestRefs(manifest)
616	621		if !valid \|\| !withinGenerationBudget(len(data), refs) {
617		-	return nil, 0, false, errors.New("invalid manifest references")
	622	+	return validatedMaintenanceCheckpoint{}, consumed, false, errors.New("invalid manifest references")
618	623		}
619	624		for digest, objectRef := range refs {
620	625		if err := ctx.Err(); err != nil {
621		-	return nil, 0, false, err
	626	+	return validatedMaintenanceCheckpoint{}, consumed, false, err
622	627		}
623	628		path := r.objectPath(id, digest)
624	629		info, err := r.stat(path)
625	630		if err != nil {
626		-	return nil, 0, false, err
	631	+	return validatedMaintenanceCheckpoint{}, consumed, false, err
627	632		}
628	633		if info.Size() < 0 \|\| uint64(info.Size()) > byteBudget-consumed {
629		-	return nil, 0, false, nil
	634	+	return validatedMaintenanceCheckpoint{}, consumed, false, nil
630	635		}
631	636		consumed += uint64(info.Size())
632	637		if hook := r.hooks.beforeMaintenancePayloadRead; hook != nil {

@@ -634,13 +639,13 @@ func (r *Repository) validateMaintenanceCheckpoint(ctx context.Context, id domai
634	639		}
635	640		object, err := r.readBounded(path)
636	641		if err != nil {
637		-	return nil, 0, false, err
	642	+	return validatedMaintenanceCheckpoint{}, consumed, false, err
638	643		}
639	644		if sha256.Sum256(object) != digest \|\| !validObject(object, objectRef) {
640		-	return nil, 0, false, errors.New("invalid retained object")
	645	+	return validatedMaintenanceCheckpoint{}, consumed, false, errors.New("invalid retained object")
641	646		}
642	647		}
643		-	return refs, consumed, true, nil
	648	+	return validatedMaintenanceCheckpoint{refs: refs, parent: manifest.ParentCheckpoint}, consumed, true, nil
644	649		}
645	650
646	651		// Reconcile scans top-level incarnation namespaces from an external seek

@@ -696,7 +701,8 @@ func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRe
696	701		findingCapacity = int(budget.Entries)
697	702		}
698	703		findings = make([]ports.ReconcileFinding, 0, findingCapacity)
699		-	for scanned := uint64(0); scanned < budget.Entries; scanned++ {
	704	+	stepConsumed := ports.MaintenanceBudget{}
	705	+	for scanned := uint64(0); scanned < budget.Entries && stepConsumed.Entries < budget.Entries; scanned++ {
700	706		entries, done, err := r.readMaintenanceDirent(root, 1, state)
701	707		if errors.Is(err, os.ErrNotExist) {
702	708		return ports.ReconcileCursor{}, findings, nil

@@ -718,16 +724,22 @@ func (r *Repository) Reconcile(ctx context.Context, records []domain.CatalogueRe
718	724		}
719	725		continue
720	726		}
	727	+	findingIndex := len(findings)
721	728		var id domain.IncarnationID
722	729		if err := id.UnmarshalText([]byte(entry.name)); err != nil {
723	730		findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name}, Cursor: next, Consumed: ports.MaintenanceBudget{Entries: 1}})
724	731		} else if record, ok := byIncarnation[id]; !ok {
725	732		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}})
726	733		} else {
727		-	finding := r.reconcileIncarnation(ctx, record, budget)
	734	+	remaining := ports.MaintenanceBudget{Entries: budget.Entries - stepConsumed.Entries, Bytes: budget.Bytes - stepConsumed.Bytes}
	735	+	finding := r.reconcileIncarnation(ctx, record, remaining)
728	736		finding.Cursor = next
729	737		findings = append(findings, finding)
730	738		}
	739	+	if len(findings) > findingIndex {
	740	+	stepConsumed.Entries += findings[findingIndex].Consumed.Entries
	741	+	stepConsumed.Bytes += findings[findingIndex].Consumed.Bytes
	742	+	}
731	743		if done {
732	744		return ports.ReconcileCursor{}, findings, nil
733	745		}

@@ -740,6 +752,20 @@ func (r *Repository) reconcileIncarnation(ctx context.Context, record domain.Cat
740	752		if record.Committed == nil {
741	753		return finding
742	754		}
	755	+	headPath := r.headPath(record.IncarnationID)
	756	+	info, err := r.stat(headPath)
	757	+	if err != nil \|\| info.Size() < 0 {
	758	+	return finding
	759	+	}
	760	+	headBytes := uint64(info.Size())
	761	+	if headBytes > budget.Bytes-finding.Consumed.Bytes {
	762	+	finding.Status = ports.ReconcileBudgetExhausted
	763	+	return finding
	764	+	}
	765	+	finding.Consumed.Bytes += headBytes
	766	+	if hook := r.hooks.beforeMaintenancePayloadRead; hook != nil {
	767	+	hook(headPath)
	768	+	}
743	769		generation, digest, err := r.readHead(record.IncarnationID)
744	770		if err != nil {
745	771		return finding

@@ -760,8 +786,8 @@ func (r *Repository) reconcileIncarnation(ctx context.Context, record domain.Cat
760	786		if err != nil {
761	787		return finding
762	788		}
763		-	finding.Candidate.Parent = candidate.ParentCheckpoint
764		-	parent := candidate.ParentCheckpoint
	789	+	finding.Candidate.Parent = candidate.parent
	790	+	parent := candidate.parent
765	791		for parent != nil && parent != record.Committed {
766	792		ancestor, exhausted, err := r.reconcileLoad(ctx, record, *parent, &finding.Consumed, budget)
767	793		if exhausted {

@@ -771,8 +797,8 @@ func (r *Repository) reconcileIncarnation(ctx context.Context, record domain.Cat
771	797		if err != nil {
772	798		return finding
773	799		}
774		-	finding.AncestorChain = append(finding.AncestorChain, ports.ValidatedCheckpoint{Ref: *parent, Parent: ancestor.ParentCheckpoint})
775		-	parent = ancestor.ParentCheckpoint
	800	+	finding.AncestorChain = append(finding.AncestorChain, ports.ValidatedCheckpoint{Ref: *parent, Parent: ancestor.parent})
	801	+	parent = ancestor.parent
776	802		}
777	803		if parent == nil {
778	804		return finding

@@ -781,24 +807,20 @@ func (r *Repository) reconcileIncarnation(ctx context.Context, record domain.Cat
781	807		return finding
782	808		}
783	809
784		-	func (r Repository) reconcileLoad(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef, consumed ports.MaintenanceBudget, budget ports.MaintenanceBudget) (ports.SnapshotGeneration, bool, error) {
	810	+	func (r Repository) reconcileLoad(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef, consumed ports.MaintenanceBudget, budget ports.MaintenanceBudget) (validatedMaintenanceCheckpoint, bool, error) {
785	811		if consumed.Entries >= budget.Entries \|\| consumed.Bytes >= budget.Bytes {
786		-	return ports.SnapshotGeneration{}, true, nil
	812	+	return validatedMaintenanceCheckpoint{}, true, nil
787	813		}
788		-	_, bytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, record.IncarnationID, ref, budget.Bytes-consumed.Bytes)
	814	+	validated, bytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, record.IncarnationID, ref, budget.Bytes-consumed.Bytes)
	815	+	consumed.Bytes += bytes
789	816		if err != nil {
790		-	return ports.SnapshotGeneration{}, false, err
	817	+	return validatedMaintenanceCheckpoint{}, false, err
791	818		}
792	819		if !admitted {
793		-	return ports.SnapshotGeneration{}, true, nil
794		-	}
795		-	generation, err := r.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)
796		-	if err != nil {
797		-	return ports.SnapshotGeneration{}, false, err
	820	+	return validatedMaintenanceCheckpoint{}, true, nil
798	821		}
799	822		consumed.Entries++
800		-	consumed.Bytes += bytes
801		-	return generation, false, nil
	823	+	return validated, false, nil
802	824		}
803	825
804	826		func retainedReferences(marked map[uint64]manifestMaintenance, conservative bool) map[ports.SnapshotDigest]struct{} {

 			if budget.Entries == 0 {
 				return false, nil
 			}
-			refs, consumedBytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, plan.IncarnationID, ref, budget.Bytes)
+			validated, consumedBytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, plan.IncarnationID, ref, budget.Bytes)
 			if err != nil {
 				return false, fmt.Errorf("validate retained manifest %d: %w", ref.Generation, err)
 			}
 			}
 			budget.Entries--
 			budget.Bytes -= consumedBytes
-			for digest := range refs {
+			for digest := range validated.refs {
 				state.references[digest] = struct{}{}
 			}
 			state.validateIndex++
 // before reading it. It validates directly instead of calling LoadCheckpoint,
 // which would materialize a complete generation before maintenance could
 // enforce its byte budget.
-func (r *Repository) validateMaintenanceCheckpoint(ctx context.Context, id domain.IncarnationID, ref ports.CheckpointRef, byteBudget uint64) (map[ports.SnapshotDigest]codec.ObjectRef, uint64, bool, error) {
+type validatedMaintenanceCheckpoint struct {
+	refs   map[ports.SnapshotDigest]codec.ObjectRef
+	parent *domain.CheckpointRef
+}
+func (r *Repository) validateMaintenanceCheckpoint(ctx context.Context, id domain.IncarnationID, ref ports.CheckpointRef, byteBudget uint64) (validatedMaintenanceCheckpoint, uint64, bool, error) {
 	manifestPath := r.manifestPath(id, ref.Generation)
 	info, err := r.stat(manifestPath)
 	if err != nil {
-		return nil, 0, false, err
+		return validatedMaintenanceCheckpoint{}, 0, false, err
 	}
 	if info.Size() < 0 || uint64(info.Size()) > byteBudget {
-		return nil, 0, false, nil
+		return validatedMaintenanceCheckpoint{}, 0, false, nil
 	}
 	consumed := uint64(info.Size())
 	if hook := r.hooks.beforeMaintenancePayloadRead; hook != nil {
 	}
 	data, err := r.readBounded(manifestPath)
 	if err != nil {
-		return nil, 0, false, err
+		return validatedMaintenanceCheckpoint{}, consumed, false, err
 	}
 	manifest, err := codec.UnmarshalManifest(data)
 	if err != nil || manifest.Generation != ref.Generation || manifest.IncarnationID != id || sha256.Sum256(data) != ref.ManifestDigest {
-		return nil, 0, false, errors.New("catalogue reference mismatch")
+		return validatedMaintenanceCheckpoint{}, consumed, false, errors.New("catalogue reference mismatch")
 	}
 	refs, valid := maintenanceManifestRefs(manifest)
 	if !valid || !withinGenerationBudget(len(data), refs) {
-		return nil, 0, false, errors.New("invalid manifest references")
+		return validatedMaintenanceCheckpoint{}, consumed, false, errors.New("invalid manifest references")
 	}
 	for digest, objectRef := range refs {
 		if err := ctx.Err(); err != nil {
-			return nil, 0, false, err
+			return validatedMaintenanceCheckpoint{}, consumed, false, err
 		}
 		path := r.objectPath(id, digest)
 		info, err := r.stat(path)
 		if err != nil {
-			return nil, 0, false, err
+			return validatedMaintenanceCheckpoint{}, consumed, false, err
 		}
 		if info.Size() < 0 || uint64(info.Size()) > byteBudget-consumed {
-			return nil, 0, false, nil
+			return validatedMaintenanceCheckpoint{}, consumed, false, nil
 		}
 		consumed += uint64(info.Size())
 		if hook := r.hooks.beforeMaintenancePayloadRead; hook != nil {
 		}
 		object, err := r.readBounded(path)
 		if err != nil {
-			return nil, 0, false, err
+			return validatedMaintenanceCheckpoint{}, consumed, false, err
 		}
 		if sha256.Sum256(object) != digest || !validObject(object, objectRef) {
-			return nil, 0, false, errors.New("invalid retained object")
+			return validatedMaintenanceCheckpoint{}, consumed, false, errors.New("invalid retained object")
 		}
 	}
-	return refs, consumed, true, nil
+	return validatedMaintenanceCheckpoint{refs: refs, parent: manifest.ParentCheckpoint}, consumed, true, nil
 }
 // Reconcile scans top-level incarnation namespaces from an external seek
 		findingCapacity = int(budget.Entries)
 	}
 	findings = make([]ports.ReconcileFinding, 0, findingCapacity)
-	for scanned := uint64(0); scanned < budget.Entries; scanned++ {
+	stepConsumed := ports.MaintenanceBudget{}
+	for scanned := uint64(0); scanned < budget.Entries && stepConsumed.Entries < budget.Entries; scanned++ {
 		entries, done, err := r.readMaintenanceDirent(root, 1, state)
 		if errors.Is(err, os.ErrNotExist) {
 			return ports.ReconcileCursor{}, findings, nil
 			}
 			continue
 		}
+		findingIndex := len(findings)
 		var id domain.IncarnationID
 		if err := id.UnmarshalText([]byte(entry.name)); err != nil {
 			findings = append(findings, ports.ReconcileFinding{Kind: ports.ReconcileUnknownIncarnation, Status: ports.ReconcileQuarantined, Candidate: ports.ReconcileCandidate{Name: entry.name}, Cursor: next, Consumed: ports.MaintenanceBudget{Entries: 1}})
 		} else if record, ok := byIncarnation[id]; !ok {
 			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}})
 		} else {
-			finding := r.reconcileIncarnation(ctx, record, budget)
+			remaining := ports.MaintenanceBudget{Entries: budget.Entries - stepConsumed.Entries, Bytes: budget.Bytes - stepConsumed.Bytes}
+			finding := r.reconcileIncarnation(ctx, record, remaining)
 			finding.Cursor = next
 			findings = append(findings, finding)
 		}
+		if len(findings) > findingIndex {
+			stepConsumed.Entries += findings[findingIndex].Consumed.Entries
+			stepConsumed.Bytes += findings[findingIndex].Consumed.Bytes
+		}
 		if done {
 			return ports.ReconcileCursor{}, findings, nil
 		}
 	if record.Committed == nil {
 		return finding
 	}
+	headPath := r.headPath(record.IncarnationID)
+	info, err := r.stat(headPath)
+	if err != nil || info.Size() < 0 {
+		return finding
+	}
+	headBytes := uint64(info.Size())
+	if headBytes > budget.Bytes-finding.Consumed.Bytes {
+		finding.Status = ports.ReconcileBudgetExhausted
+		return finding
+	}
+	finding.Consumed.Bytes += headBytes
+	if hook := r.hooks.beforeMaintenancePayloadRead; hook != nil {
+		hook(headPath)
+	}
 	generation, digest, err := r.readHead(record.IncarnationID)
 	if err != nil {
 		return finding
 	if err != nil {
 		return finding
 	}
-	finding.Candidate.Parent = candidate.ParentCheckpoint
+	finding.Candidate.Parent = candidate.parent
-	parent := candidate.ParentCheckpoint
+	parent := candidate.parent
 	for parent != nil && *parent != *record.Committed {
 		ancestor, exhausted, err := r.reconcileLoad(ctx, record, *parent, &finding.Consumed, budget)
 		if exhausted {
 		if err != nil {
 			return finding
 		}
-		finding.AncestorChain = append(finding.AncestorChain, ports.ValidatedCheckpoint{Ref: *parent, Parent: ancestor.ParentCheckpoint})
+		finding.AncestorChain = append(finding.AncestorChain, ports.ValidatedCheckpoint{Ref: *parent, Parent: ancestor.parent})
-		parent = ancestor.ParentCheckpoint
+		parent = ancestor.parent
 	}
 	if parent == nil {
 		return finding
 	return finding
 }
-func (r *Repository) reconcileLoad(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef, consumed *ports.MaintenanceBudget, budget ports.MaintenanceBudget) (ports.SnapshotGeneration, bool, error) {
+func (r *Repository) reconcileLoad(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef, consumed *ports.MaintenanceBudget, budget ports.MaintenanceBudget) (validatedMaintenanceCheckpoint, bool, error) {
 	if consumed.Entries >= budget.Entries || consumed.Bytes >= budget.Bytes {
-		return ports.SnapshotGeneration{}, true, nil
+		return validatedMaintenanceCheckpoint{}, true, nil
 	}
-	_, bytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, record.IncarnationID, ref, budget.Bytes-consumed.Bytes)
+	validated, bytes, admitted, err := r.validateMaintenanceCheckpoint(ctx, record.IncarnationID, ref, budget.Bytes-consumed.Bytes)
+	consumed.Bytes += bytes
 	if err != nil {
-		return ports.SnapshotGeneration{}, false, err
+		return validatedMaintenanceCheckpoint{}, false, err
 	}
 	if !admitted {
-		return ports.SnapshotGeneration{}, true, nil
+		return validatedMaintenanceCheckpoint{}, true, nil
-	}
-	generation, err := r.LoadCheckpoint(ctx, record.IncarnationID, record.Name, ref)
-	if err != nil {
-		return ports.SnapshotGeneration{}, false, err
 	}
 	consumed.Entries++
-	consumed.Bytes += bytes
+	return validated, false, nil
-	return generation, false, nil
 }
 func retainedReferences(marked map[uint64]manifestMaintenance, conservative bool) map[ports.SnapshotDigest]struct{} {

M internal/adapters/snapshot/reconciliation_test.go +83 -0


@@ -8,9 +8,92 @@ import (
8	8
9	9		"github.com/bnema/vev/internal/domain"
10	10		"github.com/bnema/vev/internal/ports"
	11	+	codec "github.com/bnema/vev/internal/usecase/snapshot"
11	12		"github.com/stretchr/testify/require"
12	13		)
13	14
	15	+	func TestReconciliationReadsEachCandidatePayloadOnceWithinBudget(t *testing.T) {
	16	+	repo := NewRepository(privateDir(t))
	17	+	first := publishMaintenanceGenerations(t, repo, "first", 2)
	18	+	second := publishMaintenanceGenerations(t, repo, "second", 2)
	19	+
	20	+	records := make([]domain.CatalogueRecord, 0, 2)
	21	+	expectedPaths := make(map[string]struct{})
	22	+	for _, publications := range [][]ports.SnapshotPublication{first, second} {
	23	+	committed := checkpointRefForPublication(publications[0])
	24	+	records = append(records, domain.CatalogueRecord{
	25	+	Name: publications[0].Name,
	26	+	IncarnationID: publications[0].IncarnationID,
	27	+	RecoveryState: domain.RecoveryHealthy,
	28	+	Committed: &committed,
	29	+	})
	30	+	candidate := checkpointRefForPublication(publications[1])
	31	+	expectedPaths[repo.headPath(publications[0].IncarnationID)] = struct{}{}
	32	+	expectedPaths[repo.manifestPath(publications[0].IncarnationID, candidate.Generation)] = struct{}{}
	33	+	manifest, err := codec.UnmarshalManifest(publications[1].Manifest)
	34	+	require.NoError(t, err)
	35	+	for digest := range manifestRefs(manifest) {
	36	+	expectedPaths[repo.objectPath(publications[0].IncarnationID, digest)] = struct{}{}
	37	+	}
	38	+	}
	39	+
	40	+	var budgetBytes uint64
	41	+	for path := range expectedPaths {
	42	+	info, err := repo.stat(path)
	43	+	require.NoError(t, err)
	44	+	budgetBytes += uint64(info.Size())
	45	+	}
	46	+	reads := make(map[string]int)
	47	+	var actualBytes uint64
	48	+	repo.hooks.beforePayloadRead = func(path string) {
	49	+	info, err := repo.stat(path)
	50	+	require.NoError(t, err)
	51	+	reads[path]++
	52	+	actualBytes += uint64(info.Size())
	53	+	}
	54	+
	55	+	next, findings, err := repo.Reconcile(context.Background(), records, ports.ReconcileCursor{}, ports.MaintenanceBudget{Entries: 16, Bytes: budgetBytes})
	56	+	require.NoError(t, err)
	57	+	require.Zero(t, next)
	58	+	require.Len(t, findings, 2)
	59	+	var consumedBytes uint64
	60	+	for _, finding := range findings {
	61	+	require.Equal(t, ports.ReconcileForwardOrphan, finding.Kind)
	62	+	require.Equal(t, ports.ReconcileValidated, finding.Status)
	63	+	consumedBytes += finding.Consumed.Bytes
	64	+	}
	65	+	require.Equal(t, consumedBytes, actualBytes, "every payload read must be charged")
	66	+	require.LessOrEqual(t, actualBytes, budgetBytes)
	67	+	require.Equal(t, expectedPaths, func() map[string]struct{} {
	68	+	got := make(map[string]struct{}, len(reads))
	69	+	for path := range reads {
	70	+	got[path] = struct{}{}
	71	+	}
	72	+	return got
	73	+	}())
	74	+	for path, count := range reads {
	75	+	require.Equalf(t, 1, count, "payload %q was reread", path)
	76	+	}
	77	+
	78	+	reads = make(map[string]int)
	79	+	actualBytes = 0
	80	+	limitedBudget := budgetBytes - 1
	81	+	_, findings, err = repo.Reconcile(context.Background(), records, ports.ReconcileCursor{}, ports.MaintenanceBudget{Entries: 16, Bytes: limitedBudget})
	82	+	require.NoError(t, err)
	83	+	consumedBytes = 0
	84	+	exhausted := false
	85	+	for _, finding := range findings {
	86	+	consumedBytes += finding.Consumed.Bytes
	87	+	exhausted = exhausted \|\| finding.Status == ports.ReconcileBudgetExhausted
	88	+	}
	89	+	require.True(t, exhausted, "reconciliation must yield rather than read past its budget")
	90	+	require.Equal(t, consumedBytes, actualBytes)
	91	+	require.LessOrEqual(t, actualBytes, limitedBudget)
	92	+	for path, count := range reads {
	93	+	require.Equalf(t, 1, count, "payload %q was reread in a budget-limited step", path)
	94	+	}
	95	+	}
	96	+
14	97		func TestReconciliationCursor(t *testing.T) {
15	98		repo := NewRepository(privateDir(t))
16	99		publications := publishMaintenanceGenerations(t, repo, "work", 2)


@@ -8,9 +8,92 @@ import (
8		8
9	"github.com/bnema/vev/internal/domain"	9		"github.com/bnema/vev/internal/domain"
10	"github.com/bnema/vev/internal/ports"	10		"github.com/bnema/vev/internal/ports"
		11	+	codec "github.com/bnema/vev/internal/usecase/snapshot"
11	"github.com/stretchr/testify/require"	12		"github.com/stretchr/testify/require"
12	)	13		)
13		14
		15	+	func TestReconciliationReadsEachCandidatePayloadOnceWithinBudget(t *testing.T) {
		16	+	repo := NewRepository(privateDir(t))
		17	+	first := publishMaintenanceGenerations(t, repo, "first", 2)
		18	+	second := publishMaintenanceGenerations(t, repo, "second", 2)
		19	+
		20	+	records := make([]domain.CatalogueRecord, 0, 2)
		21	+	expectedPaths := make(map[string]struct{})
		22	+	for _, publications := range [][]ports.SnapshotPublication{first, second} {
		23	+	committed := checkpointRefForPublication(publications[0])
		24	+	records = append(records, domain.CatalogueRecord{
		25	+	Name: publications[0].Name,
		26	+	IncarnationID: publications[0].IncarnationID,
		27	+	RecoveryState: domain.RecoveryHealthy,
		28	+	Committed: &committed,
		29	+	})
		30	+	candidate := checkpointRefForPublication(publications[1])
		31	+	expectedPaths[repo.headPath(publications[0].IncarnationID)] = struct{}{}
		32	+	expectedPaths[repo.manifestPath(publications[0].IncarnationID, candidate.Generation)] = struct{}{}
		33	+	manifest, err := codec.UnmarshalManifest(publications[1].Manifest)
		34	+	require.NoError(t, err)
		35	+	for digest := range manifestRefs(manifest) {
		36	+	expectedPaths[repo.objectPath(publications[0].IncarnationID, digest)] = struct{}{}
		37	+	}
		38	+	}
		39	+
		40	+	var budgetBytes uint64
		41	+	for path := range expectedPaths {
		42	+	info, err := repo.stat(path)
		43	+	require.NoError(t, err)
		44	+	budgetBytes += uint64(info.Size())
		45	+	}
		46	+	reads := make(map[string]int)
		47	+	var actualBytes uint64
		48	+	repo.hooks.beforePayloadRead = func(path string) {
		49	+	info, err := repo.stat(path)
		50	+	require.NoError(t, err)
		51	+	reads[path]++
		52	+	actualBytes += uint64(info.Size())
		53	+	}
		54	+
		55	+	next, findings, err := repo.Reconcile(context.Background(), records, ports.ReconcileCursor{}, ports.MaintenanceBudget{Entries: 16, Bytes: budgetBytes})
		56	+	require.NoError(t, err)
		57	+	require.Zero(t, next)
		58	+	require.Len(t, findings, 2)
		59	+	var consumedBytes uint64
		60	+	for _, finding := range findings {
		61	+	require.Equal(t, ports.ReconcileForwardOrphan, finding.Kind)
		62	+	require.Equal(t, ports.ReconcileValidated, finding.Status)
		63	+	consumedBytes += finding.Consumed.Bytes
		64	+	}
		65	+	require.Equal(t, consumedBytes, actualBytes, "every payload read must be charged")
		66	+	require.LessOrEqual(t, actualBytes, budgetBytes)
		67	+	require.Equal(t, expectedPaths, func() map[string]struct{} {
		68	+	got := make(map[string]struct{}, len(reads))
		69	+	for path := range reads {
		70	+	got[path] = struct{}{}
		71	+	}
		72	+	return got
		73	+	}())
		74	+	for path, count := range reads {
		75	+	require.Equalf(t, 1, count, "payload %q was reread", path)
		76	+	}
		77	+
		78	+	reads = make(map[string]int)
		79	+	actualBytes = 0
		80	+	limitedBudget := budgetBytes - 1
		81	+	_, findings, err = repo.Reconcile(context.Background(), records, ports.ReconcileCursor{}, ports.MaintenanceBudget{Entries: 16, Bytes: limitedBudget})
		82	+	require.NoError(t, err)
		83	+	consumedBytes = 0
		84	+	exhausted := false
		85	+	for _, finding := range findings {
		86	+	consumedBytes += finding.Consumed.Bytes
		87	+	exhausted = exhausted \|\| finding.Status == ports.ReconcileBudgetExhausted
		88	+	}
		89	+	require.True(t, exhausted, "reconciliation must yield rather than read past its budget")
		90	+	require.Equal(t, consumedBytes, actualBytes)
		91	+	require.LessOrEqual(t, actualBytes, limitedBudget)
		92	+	for path, count := range reads {
		93	+	require.Equalf(t, 1, count, "payload %q was reread in a budget-limited step", path)
		94	+	}
		95	+	}
		96	+
14	func TestReconciliationCursor(t *testing.T) {	97		func TestReconciliationCursor(t *testing.T) {
15	repo := NewRepository(privateDir(t))	98		repo := NewRepository(privateDir(t))
16	publications := publishMaintenanceGenerations(t, repo, "work", 2)	99		publications := publishMaintenanceGenerations(t, repo, "work", 2)

M internal/adapters/snapshot/repository.go +3 -0


@@ -63,6 +63,9 @@ type repositoryHooks struct {
63	63		// beforeMaintenancePayloadRead instruments payload reads after maintenance
64	64		// has admitted their stat-reported size against the current byte budget.
65	65		beforeMaintenancePayloadRead func(string)
	66	+	// beforePayloadRead observes every bounded repository payload read. It is
	67	+	// test-only accounting instrumentation and runs immediately before the read.
	68	+	beforePayloadRead func(string)
66	69		beforeObjectHash func([]byte)
67	70		beforeObjectCopy func([]byte)
68	71		beforeHeadRead func(string) error


@@ -63,6 +63,9 @@ type repositoryHooks struct {
63	// beforeMaintenancePayloadRead instruments payload reads after maintenance	63		// beforeMaintenancePayloadRead instruments payload reads after maintenance
64	// has admitted their stat-reported size against the current byte budget.	64		// has admitted their stat-reported size against the current byte budget.
65	beforeMaintenancePayloadRead func(string)	65		beforeMaintenancePayloadRead func(string)
		66	+	// beforePayloadRead observes every bounded repository payload read. It is
		67	+	// test-only accounting instrumentation and runs immediately before the read.
		68	+	beforePayloadRead func(string)
66	beforeObjectHash func([]byte)	69		beforeObjectHash func([]byte)
67	beforeObjectCopy func([]byte)	70		beforeObjectCopy func([]byte)
68	beforeHeadRead func(string) error	71		beforeHeadRead func(string) error

M internal/usecase/recovery/reconcile.go +2 -4


@@ -114,7 +114,8 @@ func validForwardChain(committed domain.CheckpointRef, candidate ports.Reconcile
114	114		}
115	115
116	116		// PublishReconciledCheckpoint rechecks the chain while holding the session
117		-	// transaction lock. Files are already durable and validated by the adapter.
	117	+	// transaction lock. The candidate is already the durable HEAD and its payloads
	118	+	// were validated by the adapter, so publication performs no repository reread.
118	119		func (c *Coordinator) PublishReconciledCheckpoint(ctx context.Context, name string, candidate ports.ReconcileCandidate, ancestors []ports.ValidatedCheckpoint) (domain.CatalogueRecord, error) {
119	120		if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil {
120	121		return domain.CatalogueRecord{}, errors.New("recovery: incomplete reconciled checkpoint dependencies")

@@ -138,8 +139,5 @@ func (c *Coordinator) PublishReconciledCheckpoint(ctx context.Context, name stri
138	139		if err := c.catalogue.Replace(name, next); err != nil {
139	140		return domain.CatalogueRecord{}, err
140	141		}
141		-	if err := c.repository.RepairHEAD(ctx, record.IncarnationID, candidate.Ref); err != nil {
142		-	return next, err
143		-	}
144	142		return next, nil
145	143		}

 }
 // PublishReconciledCheckpoint rechecks the chain while holding the session
-// transaction lock. Files are already durable and validated by the adapter.
+// transaction lock. The candidate is already the durable HEAD and its payloads
+// were validated by the adapter, so publication performs no repository reread.
 func (c *Coordinator) PublishReconciledCheckpoint(ctx context.Context, name string, candidate ports.ReconcileCandidate, ancestors []ports.ValidatedCheckpoint) (domain.CatalogueRecord, error) {
 	if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
 		return domain.CatalogueRecord{}, errors.New("recovery: incomplete reconciled checkpoint dependencies")
 	if err := c.catalogue.Replace(name, next); err != nil {
 		return domain.CatalogueRecord{}, err
 	}
-	if err := c.repository.RepairHEAD(ctx, record.IncarnationID, candidate.Ref); err != nil {
-		return next, err
-	}
 	return next, nil
 }

M internal/usecase/recovery/reconcile_test.go +3 -2


@@ -99,7 +99,7 @@ func TestForwardOrphanAdoption(t *testing.T) {
99	99		})
100	100		}
101	101
102		-	t.Run("catalogue replacement precedes head repair", func(t *testing.T) {
	102	+	t.Run("validated head publishes without repository reread", func(t *testing.T) {
103	103		record := healthyReconcileRecord()
104	104		events := []string{}
105	105		catalogue := &checkpointCatalogue{record: record, events: &events}

@@ -110,7 +110,8 @@ func TestForwardOrphanAdoption(t *testing.T) {
110	110		next, err := coordinator.PublishReconciledCheckpoint(context.Background(), record.Name, candidate, nil)
111	111		require.NoError(t, err)
112	112		require.Equal(t, candidate.Ref, *next.Committed)
113		-	require.Equal(t, []string{"catalogue", "head"}, events)
	113	+	require.Equal(t, []string{"catalogue"}, events)
	114	+	require.Empty(t, repository.repairs)
114	115		})
115	116
116	117		t.Run("stale decision after concurrent commit", func(t *testing.T) {

 		})
 	}
-	t.Run("catalogue replacement precedes head repair", func(t *testing.T) {
+	t.Run("validated head publishes without repository reread", func(t *testing.T) {
 		record := healthyReconcileRecord()
 		events := []string{}
 		catalogue := &checkpointCatalogue{record: record, events: &events}
 		next, err := coordinator.PublishReconciledCheckpoint(context.Background(), record.Name, candidate, nil)
 		require.NoError(t, err)
 		require.Equal(t, candidate.Ref, *next.Committed)
-		require.Equal(t, []string{"catalogue", "head"}, events)
+		require.Equal(t, []string{"catalogue"}, events)
+		require.Empty(t, repository.repairs)
 	})
 	t.Run("stale decision after concurrent commit", func(t *testing.T) {