2b0e9f7 fix(recovery): tighten maintenance and restore fencing · vev

M cmd/vev-perf-harness/hangup_test.go +11 -3


@@ -198,13 +198,21 @@ func TestClientGracefulShutdownDrainsTracedStdioDescendant(t *testing.T) {
198	198		switch pm.Role {
199	199		case "daemon":
200	200		daemon = process
201		-	if err := process.(interface{ WaitReady() error }).WaitReady(); err != nil {
	201	+	readyProcess, ok := process.(interface{ WaitReady() error })
	202	+	if !ok {
	203	+	t.Fatalf("daemon process type %T does not support readiness", process)
	204	+	}
	205	+	if err := readyProcess.WaitReady(); err != nil {
202	206		t.Fatalf("daemon readiness: %v", err)
203	207		}
204	208		case "ssh_stdio_peer":
205	209		peer, peerMapping = process, pm
206	210		case "client":
207		-	client = process.(*cliProcess)
	211	+	var ok bool
	212	+	client, ok = process.(*cliProcess)
	213	+	if !ok {
	214	+	t.Fatalf("client process type = %T, want *cliProcess", process)
	215	+	}
208	216		}
209	217		}
210	218		if daemon == nil \|\| peer == nil \|\| client == nil {

@@ -234,7 +242,7 @@ func TestClientGracefulShutdownDrainsTracedStdioDescendant(t *testing.T) {
234	242		if _, err := mergeProcessTraces([]processMapping{peerMapping}); err != nil {
235	243		t.Fatalf("stdio descendant trace did not drain: %v", err)
236	244		}
237		-	if err := waitForProcessGroupGone(processGroup, time.Second); err != nil {
	245	+	if err := waitForProcessGroupGone(processGroup, 10*time.Second); err != nil {
238	246		t.Fatal(err)
239	247		}
240	248		}

 		switch pm.Role {
 		case "daemon":
 			daemon = process
-			if err := process.(interface{ WaitReady() error }).WaitReady(); err != nil {
+			readyProcess, ok := process.(interface{ WaitReady() error })
+			if !ok {
+				t.Fatalf("daemon process type %T does not support readiness", process)
+			}
+			if err := readyProcess.WaitReady(); err != nil {
 				t.Fatalf("daemon readiness: %v", err)
 			}
 		case "ssh_stdio_peer":
 			peer, peerMapping = process, pm
 		case "client":
-			client = process.(*cliProcess)
+			var ok bool
+			client, ok = process.(*cliProcess)
+			if !ok {
+				t.Fatalf("client process type = %T, want *cliProcess", process)
+			}
 		}
 	}
 	if daemon == nil || peer == nil || client == nil {
 	if _, err := mergeProcessTraces([]processMapping{peerMapping}); err != nil {
 		t.Fatalf("stdio descendant trace did not drain: %v", err)
 	}
-	if err := waitForProcessGroupGone(processGroup, time.Second); err != nil {
+	if err := waitForProcessGroupGone(processGroup, 10*time.Second); err != nil {
 		t.Fatal(err)
 	}
 }

M cmd/vev-perf-harness/launcher_test.go +4 -1


@@ -267,7 +267,10 @@ func TestCLILauncherIsolatesRoleXDGEnvironmentAcrossRepetitions(t *testing.T) {
267	267		}
268	268		// The fixture exits on its own. Synchronize with that exit instead of
269	269		// letting Close race its SIGTERM against the environment capture.
270		-	proc := p.(*cliProcess)
	270	+	proc, ok := p.(*cliProcess)
	271	+	if !ok {
	272	+	t.Fatalf("run %d %s process type = %T, want *cliProcess", run, role, p)
	273	+	}
271	274		exitErr := <-proc.waitErr
272	275		proc.waitErr <- exitErr
273	276		runtimeDir := launcher.runtimes[runDir]

 			}
 			// The fixture exits on its own. Synchronize with that exit instead of
 			// letting Close race its SIGTERM against the environment capture.
-			proc := p.(*cliProcess)
+			proc, ok := p.(*cliProcess)
+			if !ok {
+				t.Fatalf("run %d %s process type = %T, want *cliProcess", run, role, p)
+			}
 			exitErr := <-proc.waitErr
 			proc.waitErr <- exitErr
 			runtimeDir := launcher.runtimes[runDir]

M internal/adapters/recoveryfs/journal_test.go +4 -3


@@ -9,6 +9,7 @@ import (
9	9		"testing"
10	10
11	11		"github.com/bnema/vev/internal/domain"
	12	+	"github.com/bnema/vev/pkg/safedir"
12	13		"github.com/stretchr/testify/require"
13	14		)
14	15

@@ -80,7 +81,7 @@ func TestRecoveryJournalRoundTripAndIdempotentDelete(t *testing.T) {
80	81
81	82		func TestRecoveryJournalIgnoresAtomicLeftoversWhenCountingDiscards(t *testing.T) {
82	83		j := New(t.TempDir())
83		-	require.NoError(t, os.MkdirAll(j.dir, 0o700))
	84	+	require.NoError(t, safedir.EnsurePrivate(j.dir))
84	85		for i := range maxDiscardIntents + 1 {
85	86		require.NoError(t, os.WriteFile(filepath.Join(j.dir, fmt.Sprintf(".intent-leftover-%04d", i)), nil, 0o600))
86	87		}

@@ -101,7 +102,7 @@ func TestRecoveryJournalRejectsMalformedIntent(t *testing.T) {
101	102		func TestRecoveryJournalBoundsEnumerationAndIntentReads(t *testing.T) {
102	103		state := t.TempDir()
103	104		j := New(state)
104		-	require.NoError(t, os.MkdirAll(j.dir, 0o700))
	105	+	require.NoError(t, safedir.EnsurePrivate(j.dir))
105	106		for i := range maxDiscardIntents + 1 {
106	107		require.NoError(t, os.WriteFile(filepath.Join(j.dir, fmt.Sprintf("discard-extra-%04d", i)), nil, 0o600))
107	108		}

@@ -109,7 +110,7 @@ func TestRecoveryJournalBoundsEnumerationAndIntentReads(t *testing.T) {
109	110		require.ErrorContains(t, err, "entry limit exceeded")
110	111
111	112		require.NoError(t, os.RemoveAll(j.dir))
112		-	require.NoError(t, os.MkdirAll(j.dir, 0o700))
	113	+	require.NoError(t, safedir.EnsurePrivate(j.dir))
113	114		require.NoError(t, os.WriteFile(filepath.Join(j.dir, "discard-oversized"), make([]byte, maxDiscardIntentSize+1), 0o600))
114	115		_, err = j.ListDiscards(context.Background())
115	116		require.ErrorContains(t, err, "size limit exceeded")

 	"testing"
 	"github.com/bnema/vev/internal/domain"
+	"github.com/bnema/vev/pkg/safedir"
 	"github.com/stretchr/testify/require"
 )
 func TestRecoveryJournalIgnoresAtomicLeftoversWhenCountingDiscards(t *testing.T) {
 	j := New(t.TempDir())
-	require.NoError(t, os.MkdirAll(j.dir, 0o700))
+	require.NoError(t, safedir.EnsurePrivate(j.dir))
 	for i := range maxDiscardIntents + 1 {
 		require.NoError(t, os.WriteFile(filepath.Join(j.dir, fmt.Sprintf(".intent-leftover-%04d", i)), nil, 0o600))
 	}
 func TestRecoveryJournalBoundsEnumerationAndIntentReads(t *testing.T) {
 	state := t.TempDir()
 	j := New(state)
-	require.NoError(t, os.MkdirAll(j.dir, 0o700))
+	require.NoError(t, safedir.EnsurePrivate(j.dir))
 	for i := range maxDiscardIntents + 1 {
 		require.NoError(t, os.WriteFile(filepath.Join(j.dir, fmt.Sprintf("discard-extra-%04d", i)), nil, 0o600))
 	}
 	require.ErrorContains(t, err, "entry limit exceeded")
 	require.NoError(t, os.RemoveAll(j.dir))
-	require.NoError(t, os.MkdirAll(j.dir, 0o700))
+	require.NoError(t, safedir.EnsurePrivate(j.dir))
 	require.NoError(t, os.WriteFile(filepath.Join(j.dir, "discard-oversized"), make([]byte, maxDiscardIntentSize+1), 0o600))
 	_, err = j.ListDiscards(context.Background())
 	require.ErrorContains(t, err, "size limit exceeded")

M internal/adapters/snapshot/maintenance_delete_test.go +9 -1


@@ -11,6 +11,7 @@ import (
11	11		"reflect"
12	12		"slices"
13	13		"testing"
	14	+	"time"
14	15
15	16		"github.com/bnema/vev/internal/domain"
16	17		"github.com/bnema/vev/internal/ports"

@@ -184,12 +185,19 @@ func TestQuarantineDeletionSourcesSerializesConcurrentPublication(t *testing.T)
184	185		go func() { quarantineDone <- repo.QuarantineDeletionSources(context.Background(), tombstone, false) }()
185	186		<-quarantined
186	187
	188	+	publishAttempted := make(chan struct{})
	189	+	repo.hooks.beforeSessionLock = func(string) { close(publishAttempted) }
187	190		publishDone := make(chan error, 1)
188	191		go func() { publishDone <- repo.Publish(context.Background(), publication) }()
189	192		select {
	193	+	case <-publishAttempted:
	194	+	case <-time.After(time.Second):
	195	+	t.Fatal("Publish did not attempt the incarnation lock")
	196	+	}
	197	+	select {
190	198		case err := <-publishDone:
191	199		t.Fatalf("Publish completed while quarantine held the incarnation lock: %v", err)
192		-	default:
	200	+	case <-time.After(25 * time.Millisecond):
193	201		}
194	202		close(release)
195	203		if err := <-quarantineDone; err != nil {

 	"reflect"
 	"slices"
 	"testing"
+	"time"
 	"github.com/bnema/vev/internal/domain"
 	"github.com/bnema/vev/internal/ports"
 	go func() { quarantineDone <- repo.QuarantineDeletionSources(context.Background(), tombstone, false) }()
 	<-quarantined
+	publishAttempted := make(chan struct{})
+	repo.hooks.beforeSessionLock = func(string) { close(publishAttempted) }
 	publishDone := make(chan error, 1)
 	go func() { publishDone <- repo.Publish(context.Background(), publication) }()
 	select {
+	case <-publishAttempted:
+	case <-time.After(time.Second):
+		t.Fatal("Publish did not attempt the incarnation lock")
+	}
+	select {
 	case err := <-publishDone:
 		t.Fatalf("Publish completed while quarantine held the incarnation lock: %v", err)
-	default:
+	case <-time.After(25 * time.Millisecond):
 	}
 	close(release)
 	if err := <-quarantineDone; err != nil {

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


@@ -9,6 +9,7 @@ import (
9	9		"path/filepath"
10	10		"sort"
11	11		"strings"
	12	+	"sync/atomic"
12	13
13	14		"github.com/bnema/vev/internal/domain"
14	15		"github.com/bnema/vev/internal/ports"

@@ -16,6 +17,7 @@ import (
16	17		)
17	18
18	19		type retentionMaintenance struct {
	20	+	invalidated atomic.Bool
19	21		token string
20	22		cursors map[string]*maintenanceCursor
21	23		references map[ports.SnapshotDigest]struct{}

@@ -404,8 +406,21 @@ func keepSet(plan ports.RetentionPlan) map[uint64]ports.SnapshotDigest {
404	406		// checkpoints. Every incarnation owns its budget and continuation state.
405	407		// processRetentionEntries charges stat-reported size for every entry, requeues
406	408		// the unadmitted suffix, and removes only canonical entries rejected by policy.
407		-	func (r Repository) processRetentionEntries(ctx context.Context, state retentionMaintenance, dir, purpose string, entries []maintenanceDirEntry, budget *ports.MaintenanceBudget, consumedBefore bool, policy retentionEntryPolicy) (bool, error) {
	409	+	func (r Repository) processRetentionEntries(ctx context.Context, state retentionMaintenance, dir, purpose string, entries []maintenanceDirEntry, budget *ports.MaintenanceBudget, consumedBefore bool, policy retentionEntryPolicy) (admitted bool, err error) {
	410	+	removed := false
	411	+	defer func() {
	412	+	if !removed {
	413	+	return
	414	+	}
	415	+	if syncErr := r.syncDirectory(dir); syncErr != nil {
	416	+	admitted = false
	417	+	err = errors.Join(err, syncErr)
	418	+	}
	419	+	}()
408	420		for i, entry := range entries {
	421	+	if state.invalidated.Load() {
	422	+	return false, nil
	423	+	}
409	424		if err := ctx.Err(); err != nil {
410	425		return false, err
411	426		}

@@ -430,11 +445,12 @@ func (r Repository) processRetentionEntries(ctx context.Context, state retenti
430	445		if !policy.delete(entry) {
431	446		continue
432	447		}
433		-	if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
434		-	return false, err
435		-	}
436		-	if err := r.syncDirectory(dir); err != nil {
437		-	return false, err
	448	+	if err := r.remove(path); err != nil {
	449	+	if !errors.Is(err, os.ErrNotExist) {
	450	+	return false, err
	451	+	}
	452	+	} else {
	453	+	removed = true
438	454		}
439	455		}
440	456		return true, nil

@@ -463,23 +479,21 @@ func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPl
463	479		)
464	480		}()
465	481		if plan.PinAll {
466		-	// Lock order is maintenanceMu -> session mutex. Pinning the same
467		-	// incarnation waits for its active retention pass, while unrelated
468		-	// incarnations never wait for that pass's payload I/O.
469	482		r.maintenanceMu.Lock()
470		-	lock := r.lockSession(key)
	483	+	if state := r.retentionSessions[key]; state != nil {
	484	+	state.invalidated.Store(true)
	485	+	}
471	486		r.clearRetentionMaintenance(key)
472		-	r.unlockSession(lock)
473	487		r.maintenanceMu.Unlock()
474	488		return true, nil
475	489		}
476	490		if budget.Entries == 0 \|\| budget.Bytes == 0 {
477		-	return false, nil
	491	+	return false, ErrMaintenanceBudgetTooSmall
478	492		}
479	493
480	494		// Establish per-incarnation state under maintenanceMu, then retain only the
481	495		// session mutex across payload I/O. Never reacquire maintenanceMu while the
482		-	// session mutex is held; PinAll relies on maintenanceMu -> session ordering.
	496	+	// session mutex is held; PinAll invalidates active state without waiting.
483	497		r.maintenanceMu.Lock()
484	498		lock := r.lockSession(key)
485	499		if r.retentionSessions == nil {

@@ -665,7 +679,10 @@ func (r Repository) requeueRetentionEntries(state retentionMaintenance, dir, p
665	679		cursor = &maintenanceCursor{done: true}
666	680		state.cursors[id] = cursor
667	681		}
668		-	cursor.pending = append(entries, cursor.pending...)
	682	+	pending := make([]maintenanceDirEntry, 0, len(entries)+len(cursor.pending))
	683	+	pending = append(pending, entries...)
	684	+	pending = append(pending, cursor.pending...)
	685	+	cursor.pending = pending
669	686		}
670	687
671	688		func retentionToken(plan ports.RetentionPlan) string {

 	"path/filepath"
 	"sort"
 	"strings"
+	"sync/atomic"
 	"github.com/bnema/vev/internal/domain"
 	"github.com/bnema/vev/internal/ports"
 )
 type retentionMaintenance struct {
+	invalidated    atomic.Bool
 	token          string
 	cursors        map[string]*maintenanceCursor
 	references     map[ports.SnapshotDigest]struct{}
 // checkpoints. Every incarnation owns its budget and continuation state.
 // processRetentionEntries charges stat-reported size for every entry, requeues
 // the unadmitted suffix, and removes only canonical entries rejected by policy.
-func (r *Repository) processRetentionEntries(ctx context.Context, state *retentionMaintenance, dir, purpose string, entries []maintenanceDirEntry, budget *ports.MaintenanceBudget, consumedBefore bool, policy retentionEntryPolicy) (bool, error) {
+func (r *Repository) processRetentionEntries(ctx context.Context, state *retentionMaintenance, dir, purpose string, entries []maintenanceDirEntry, budget *ports.MaintenanceBudget, consumedBefore bool, policy retentionEntryPolicy) (admitted bool, err error) {
+	removed := false
+	defer func() {
+		if !removed {
+			return
+		}
+		if syncErr := r.syncDirectory(dir); syncErr != nil {
+			admitted = false
+			err = errors.Join(err, syncErr)
+		}
+	}()
 	for i, entry := range entries {
+		if state.invalidated.Load() {
+			return false, nil
+		}
 		if err := ctx.Err(); err != nil {
 			return false, err
 		}
 		if !policy.delete(entry) {
 			continue
 		}
-		if err := r.remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
+		if err := r.remove(path); err != nil {
-			return false, err
+			if !errors.Is(err, os.ErrNotExist) {
-		}
+				return false, err
-		if err := r.syncDirectory(dir); err != nil {
+			}
-			return false, err
+		} else {
+			removed = true
 		}
 	}
 	return true, nil
 		)
 	}()
 	if plan.PinAll {
-		// Lock order is maintenanceMu -> session mutex. Pinning the same
-		// incarnation waits for its active retention pass, while unrelated
-		// incarnations never wait for that pass's payload I/O.
 		r.maintenanceMu.Lock()
-		lock := r.lockSession(key)
+		if state := r.retentionSessions[key]; state != nil {
+			state.invalidated.Store(true)
+		}
 		r.clearRetentionMaintenance(key)
-		r.unlockSession(lock)
 		r.maintenanceMu.Unlock()
 		return true, nil
 	}
 	if budget.Entries == 0 || budget.Bytes == 0 {
-		return false, nil
+		return false, ErrMaintenanceBudgetTooSmall
 	}
 	// Establish per-incarnation state under maintenanceMu, then retain only the
 	// session mutex across payload I/O. Never reacquire maintenanceMu while the
-	// session mutex is held; PinAll relies on maintenanceMu -> session ordering.
+	// session mutex is held; PinAll invalidates active state without waiting.
 	r.maintenanceMu.Lock()
 	lock := r.lockSession(key)
 	if r.retentionSessions == nil {
 		cursor = &maintenanceCursor{done: true}
 		state.cursors[id] = cursor
 	}
-	cursor.pending = append(entries, cursor.pending...)
+	pending := make([]maintenanceDirEntry, 0, len(entries)+len(cursor.pending))
+	pending = append(pending, entries...)
+	pending = append(pending, cursor.pending...)
+	cursor.pending = pending
 }
 func retentionToken(plan ports.RetentionPlan) string {

M internal/adapters/snapshot/maintenance_test.go +98 -6


@@ -20,6 +20,7 @@ import (
20	20
21	21		"github.com/bnema/vev/internal/domain"
22	22		"github.com/bnema/vev/internal/ports"
	23	+	"github.com/bnema/vev/pkg/safedir"
23	24		)
24	25
25	26		func TestReadMaintenanceDirentPreservesCursorAndCloseFailuresWithoutPaths(t *testing.T) {

@@ -1048,7 +1049,7 @@ func TestRetentionSlots(t *testing.T) {
1048	1049		func TestUnresolvedDataPinned(t *testing.T) {
1049	1050		repo := NewRepository(privateDir(t))
1050	1051		publications := publishMaintenanceGenerations(t, repo, "unresolved", 4)
1051		-	done, err := repo.MaintainSession(context.Background(), ports.RetentionPlan{IncarnationID: publications[0].IncarnationID, PinAll: true}, ports.MaintenanceBudget{Entries: 1, Bytes: 1})
	1052	+	done, err := repo.MaintainSession(context.Background(), ports.RetentionPlan{IncarnationID: publications[0].IncarnationID, PinAll: true}, ports.MaintenanceBudget{})
1052	1053		if err != nil {
1053	1054		t.Fatal(err)
1054	1055		}

@@ -1060,6 +1061,100 @@ func TestUnresolvedDataPinned(t *testing.T) {
1060	1061		}
1061	1062		}
1062	1063
	1064	+	func TestPinAllDoesNotWaitForSameSessionRetention(t *testing.T) {
	1065	+	repo := NewRepository(privateDir(t))
	1066	+	publications := publishMaintenanceGenerations(t, repo, "pin-active", 2)
	1067	+	publication := publications[0]
	1068	+	plan := ports.RetentionPlan{IncarnationID: publication.IncarnationID, Keep: []ports.CheckpointRef{checkpointRefForPublication(publication)}}
	1069	+
	1070	+	entered := make(chan struct{})
	1071	+	release := make(chan struct{})
	1072	+	var enteredOnce sync.Once
	1073	+	repo.hooks.beforeMaintenancePayloadRead = func(string) {
	1074	+	enteredOnce.Do(func() { close(entered) })
	1075	+	<-release
	1076	+	}
	1077	+	activeDone := make(chan error, 1)
	1078	+	go func() {
	1079	+	_, err := repo.MaintainSession(context.Background(), plan, ports.MaintenanceBudget{Entries: 64, Bytes: 8 << 20})
	1080	+	activeDone <- err
	1081	+	}()
	1082	+	<-entered
	1083	+
	1084	+	pinDone := make(chan error, 1)
	1085	+	go func() {
	1086	+	_, err := repo.MaintainSession(context.Background(), ports.RetentionPlan{IncarnationID: publication.IncarnationID, PinAll: true}, ports.MaintenanceBudget{})
	1087	+	pinDone <- err
	1088	+	}()
	1089	+	select {
	1090	+	case err := <-pinDone:
	1091	+	if err != nil {
	1092	+	t.Fatal(err)
	1093	+	}
	1094	+	case <-time.After(remainingTestTime(t) / 2):
	1095	+	close(release)
	1096	+	t.Fatal("PinAll waited for active retention on the same incarnation")
	1097	+	}
	1098	+	close(release)
	1099	+	if err := <-activeDone; err != nil {
	1100	+	t.Fatal(err)
	1101	+	}
	1102	+	if got := remainingMaintenanceGenerations(t, repo, publication.IncarnationID); !slices.Equal(got, []uint64{2, 1}) {
	1103	+	t.Fatalf("PinAll allowed active retention to delete generations: %v", got)
	1104	+	}
	1105	+	}
	1106	+
	1107	+	func TestMaintainSessionRejectsZeroBudget(t *testing.T) {
	1108	+	repo := NewRepository(privateDir(t))
	1109	+	id := domain.IncarnationID{1}
	1110	+	done, err := repo.MaintainSession(context.Background(), ports.RetentionPlan{IncarnationID: id}, ports.MaintenanceBudget{})
	1111	+	if done \|\| !errors.Is(err, ErrMaintenanceBudgetTooSmall) {
	1112	+	t.Fatalf("zero budget: done=%v err=%v", done, err)
	1113	+	}
	1114	+	}
	1115	+
	1116	+	func TestProcessRetentionEntriesSyncsRemovalBatchOnce(t *testing.T) {
	1117	+	repo := NewRepository(privateDir(t))
	1118	+	dir := filepath.Join(repo.dir, "batch")
	1119	+	if err := safedir.EnsurePrivate(dir); err != nil {
	1120	+	t.Fatal(err)
	1121	+	}
	1122	+	entries := []maintenanceDirEntry{{name: generationFilename(1), cookie: 1}, {name: generationFilename(2), cookie: 2}}
	1123	+	for _, entry := range entries {
	1124	+	if err := os.WriteFile(filepath.Join(dir, entry.name), []byte("x"), 0o600); err != nil {
	1125	+	t.Fatal(err)
	1126	+	}
	1127	+	}
	1128	+	syncs := 0
	1129	+	repo.hooks.syncDirectory = func(got string) error {
	1130	+	if got == dir {
	1131	+	syncs++
	1132	+	}
	1133	+	return nil
	1134	+	}
	1135	+	state := &retentionMaintenance{cursors: make(map[string]*maintenanceCursor)}
	1136	+	budget := ports.MaintenanceBudget{Entries: 2, Bytes: 2}
	1137	+	admitted, err := repo.processRetentionEntries(context.Background(), state, dir, "batch", entries, &budget, false, retentionEntryPolicy{kind: retentionManifest})
	1138	+	if err != nil \|\| !admitted {
	1139	+	t.Fatalf("batch processing: admitted=%v err=%v", admitted, err)
	1140	+	}
	1141	+	if syncs != 1 {
	1142	+	t.Fatalf("directory syncs = %d, want 1", syncs)
	1143	+	}
	1144	+	}
	1145	+
	1146	+	func TestRequeueRetentionEntriesDoesNotAliasInput(t *testing.T) {
	1147	+	repo := NewRepository(privateDir(t))
	1148	+	state := &retentionMaintenance{cursors: make(map[string]*maintenanceCursor)}
	1149	+	entries := []maintenanceDirEntry{{name: "one", cookie: 1}}
	1150	+	repo.requeueRetentionEntries(state, "/dir", "purpose", entries)
	1151	+	entries[0].name = "mutated"
	1152	+	pending := state.cursors[maintenanceCursorID("/dir", "purpose")].pending
	1153	+	if pending[0].name != "one" {
	1154	+	t.Fatalf("queued entry aliased caller input: %q", pending[0].name)
	1155	+	}
	1156	+	}
	1157	+
1063	1158		func TestMaintenanceBudgetAdmitsPayloadsBeforeRead(t *testing.T) {
1064	1159		repo := NewRepository(privateDir(t))
1065	1160		publications := publishMaintenanceGenerations(t, repo, "budget", 1)

@@ -1132,12 +1227,9 @@ func TestReconcileDoesNotHoldMaintenanceLockDuringPayloadRead(t *testing.T) {
1132	1227		release := make(chan struct{})
1133	1228		var releaseOnce sync.Once
1134	1229		defer releaseOnce.Do(func() { close(release) })
	1230	+	var enteredOnce sync.Once
1135	1231		repo.hooks.beforeMaintenancePayloadRead = func(string) {
1136		-	select {
1137		-	case <-entered:
1138		-	default:
1139		-	close(entered)
1140		-	}
	1232	+	enteredOnce.Do(func() { close(entered) })
1141	1233		<-release
1142	1234		}
1143	1235		reconcileDone := make(chan error, 1)


@@ -20,6 +20,7 @@ import (
20			20
21		"github.com/bnema/vev/internal/domain"	21		"github.com/bnema/vev/internal/domain"
22		"github.com/bnema/vev/internal/ports"	22		"github.com/bnema/vev/internal/ports"
			23	+	"github.com/bnema/vev/pkg/safedir"
23		)	24		)
24			25
25		func TestReadMaintenanceDirentPreservesCursorAndCloseFailuresWithoutPaths(t *testing.T) {	26		func TestReadMaintenanceDirentPreservesCursorAndCloseFailuresWithoutPaths(t *testing.T) {

@@ -1048,7 +1049,7 @@ func TestRetentionSlots(t *testing.T) {
1048		func TestUnresolvedDataPinned(t *testing.T) {	1049		func TestUnresolvedDataPinned(t *testing.T) {
1049		repo := NewRepository(privateDir(t))	1050		repo := NewRepository(privateDir(t))
1050		publications := publishMaintenanceGenerations(t, repo, "unresolved", 4)	1051		publications := publishMaintenanceGenerations(t, repo, "unresolved", 4)
1051	-	done, err := repo.MaintainSession(context.Background(), ports.RetentionPlan{IncarnationID: publications[0].IncarnationID, PinAll: true}, ports.MaintenanceBudget{Entries: 1, Bytes: 1})	1052	+	done, err := repo.MaintainSession(context.Background(), ports.RetentionPlan{IncarnationID: publications[0].IncarnationID, PinAll: true}, ports.MaintenanceBudget{})
1052		if err != nil {	1053		if err != nil {
1053		t.Fatal(err)	1054		t.Fatal(err)
1054		}	1055		}

@@ -1060,6 +1061,100 @@ func TestUnresolvedDataPinned(t *testing.T) {
1060		}	1061		}
1061		}	1062		}
1062			1063
			1064	+	func TestPinAllDoesNotWaitForSameSessionRetention(t *testing.T) {
			1065	+	repo := NewRepository(privateDir(t))
			1066	+	publications := publishMaintenanceGenerations(t, repo, "pin-active", 2)
			1067	+	publication := publications[0]
			1068	+	plan := ports.RetentionPlan{IncarnationID: publication.IncarnationID, Keep: []ports.CheckpointRef{checkpointRefForPublication(publication)}}
			1069	+
			1070	+	entered := make(chan struct{})
			1071	+	release := make(chan struct{})
			1072	+	var enteredOnce sync.Once
			1073	+	repo.hooks.beforeMaintenancePayloadRead = func(string) {
			1074	+	enteredOnce.Do(func() { close(entered) })
			1075	+	<-release
			1076	+	}
			1077	+	activeDone := make(chan error, 1)
			1078	+	go func() {
			1079	+	_, err := repo.MaintainSession(context.Background(), plan, ports.MaintenanceBudget{Entries: 64, Bytes: 8 << 20})
			1080	+	activeDone <- err
			1081	+	}()
			1082	+	<-entered
			1083	+
			1084	+	pinDone := make(chan error, 1)
			1085	+	go func() {
			1086	+	_, err := repo.MaintainSession(context.Background(), ports.RetentionPlan{IncarnationID: publication.IncarnationID, PinAll: true}, ports.MaintenanceBudget{})
			1087	+	pinDone <- err
			1088	+	}()
			1089	+	select {
			1090	+	case err := <-pinDone:
			1091	+	if err != nil {
			1092	+	t.Fatal(err)
			1093	+	}
			1094	+	case <-time.After(remainingTestTime(t) / 2):
			1095	+	close(release)
			1096	+	t.Fatal("PinAll waited for active retention on the same incarnation")
			1097	+	}
			1098	+	close(release)
			1099	+	if err := <-activeDone; err != nil {
			1100	+	t.Fatal(err)
			1101	+	}
			1102	+	if got := remainingMaintenanceGenerations(t, repo, publication.IncarnationID); !slices.Equal(got, []uint64{2, 1}) {
			1103	+	t.Fatalf("PinAll allowed active retention to delete generations: %v", got)
			1104	+	}
			1105	+	}
			1106	+
			1107	+	func TestMaintainSessionRejectsZeroBudget(t *testing.T) {
			1108	+	repo := NewRepository(privateDir(t))
			1109	+	id := domain.IncarnationID{1}
			1110	+	done, err := repo.MaintainSession(context.Background(), ports.RetentionPlan{IncarnationID: id}, ports.MaintenanceBudget{})
			1111	+	if done \|\| !errors.Is(err, ErrMaintenanceBudgetTooSmall) {
			1112	+	t.Fatalf("zero budget: done=%v err=%v", done, err)
			1113	+	}
			1114	+	}
			1115	+
			1116	+	func TestProcessRetentionEntriesSyncsRemovalBatchOnce(t *testing.T) {
			1117	+	repo := NewRepository(privateDir(t))
			1118	+	dir := filepath.Join(repo.dir, "batch")
			1119	+	if err := safedir.EnsurePrivate(dir); err != nil {
			1120	+	t.Fatal(err)
			1121	+	}
			1122	+	entries := []maintenanceDirEntry{{name: generationFilename(1), cookie: 1}, {name: generationFilename(2), cookie: 2}}
			1123	+	for _, entry := range entries {
			1124	+	if err := os.WriteFile(filepath.Join(dir, entry.name), []byte("x"), 0o600); err != nil {
			1125	+	t.Fatal(err)
			1126	+	}
			1127	+	}
			1128	+	syncs := 0
			1129	+	repo.hooks.syncDirectory = func(got string) error {
			1130	+	if got == dir {
			1131	+	syncs++
			1132	+	}
			1133	+	return nil
			1134	+	}
			1135	+	state := &retentionMaintenance{cursors: make(map[string]*maintenanceCursor)}
			1136	+	budget := ports.MaintenanceBudget{Entries: 2, Bytes: 2}
			1137	+	admitted, err := repo.processRetentionEntries(context.Background(), state, dir, "batch", entries, &budget, false, retentionEntryPolicy{kind: retentionManifest})
			1138	+	if err != nil \|\| !admitted {
			1139	+	t.Fatalf("batch processing: admitted=%v err=%v", admitted, err)
			1140	+	}
			1141	+	if syncs != 1 {
			1142	+	t.Fatalf("directory syncs = %d, want 1", syncs)
			1143	+	}
			1144	+	}
			1145	+
			1146	+	func TestRequeueRetentionEntriesDoesNotAliasInput(t *testing.T) {
			1147	+	repo := NewRepository(privateDir(t))
			1148	+	state := &retentionMaintenance{cursors: make(map[string]*maintenanceCursor)}
			1149	+	entries := []maintenanceDirEntry{{name: "one", cookie: 1}}
			1150	+	repo.requeueRetentionEntries(state, "/dir", "purpose", entries)
			1151	+	entries[0].name = "mutated"
			1152	+	pending := state.cursors[maintenanceCursorID("/dir", "purpose")].pending
			1153	+	if pending[0].name != "one" {
			1154	+	t.Fatalf("queued entry aliased caller input: %q", pending[0].name)
			1155	+	}
			1156	+	}
			1157	+
1063		func TestMaintenanceBudgetAdmitsPayloadsBeforeRead(t *testing.T) {	1158		func TestMaintenanceBudgetAdmitsPayloadsBeforeRead(t *testing.T) {
1064		repo := NewRepository(privateDir(t))	1159		repo := NewRepository(privateDir(t))
1065		publications := publishMaintenanceGenerations(t, repo, "budget", 1)	1160		publications := publishMaintenanceGenerations(t, repo, "budget", 1)

@@ -1132,12 +1227,9 @@ func TestReconcileDoesNotHoldMaintenanceLockDuringPayloadRead(t *testing.T) {
1132		release := make(chan struct{})	1227		release := make(chan struct{})
1133		var releaseOnce sync.Once	1228		var releaseOnce sync.Once
1134		defer releaseOnce.Do(func() { close(release) })	1229		defer releaseOnce.Do(func() { close(release) })
			1230	+	var enteredOnce sync.Once
1135		repo.hooks.beforeMaintenancePayloadRead = func(string) {	1231		repo.hooks.beforeMaintenancePayloadRead = func(string) {
1136	-	select {	1232	+	enteredOnce.Do(func() { close(entered) })
1137	-	case <-entered:
1138	-	default:
1139	-	close(entered)
1140	-	}
1141		<-release	1233		<-release
1142		}	1234		}
1143		reconcileDone := make(chan error, 1)	1235		reconcileDone := make(chan error, 1)

M internal/adapters/snapshot/recovery_observability_test.go +7 -2


@@ -19,6 +19,7 @@ func TestGCBudgetDiagnostic(t *testing.T) {
19	19		arrange func(testing.T, Repository) ports.RetentionPlan
20	20		budget ports.MaintenanceBudget
21	21		wantDone bool
	22	+	wantErr error
22	23		wantExhausted bool
23	24		wantConsumption bool
24	25		}{

@@ -27,7 +28,7 @@ func TestGCBudgetDiagnostic(t *testing.T) {
27	28		arrange: func(_ testing.T, _ Repository) ports.RetentionPlan {
28	29		return ports.RetentionPlan{IncarnationID: domain.IncarnationID{1}, Keep: []ports.CheckpointRef{{Generation: 3}, {Generation: 2}, {Generation: 1}}}
29	30		},
30		-	wantExhausted: true,
	31	+	wantErr: ErrMaintenanceBudgetTooSmall,
31	32		},
32	33		{
33	34		name: "sufficient",

@@ -48,7 +49,11 @@ func TestGCBudgetDiagnostic(t *testing.T) {
48	49		buffer.Reset()
49	50
50	51		done, err := repository.MaintainSession(context.Background(), plan, tc.budget)
51		-	require.NoError(t, err)
	52	+	if tc.wantErr == nil {
	53	+	require.NoError(t, err)
	54	+	} else {
	55	+	require.ErrorIs(t, err, tc.wantErr)
	56	+	}
52	57		require.Equal(t, tc.wantDone, done)
53	58
54	59		scanner := bufio.NewScanner(bytes.NewReader(buffer.Bytes()))

 		arrange         func(*testing.T, *Repository) ports.RetentionPlan
 		budget          ports.MaintenanceBudget
 		wantDone        bool
+		wantErr         error
 		wantExhausted   bool
 		wantConsumption bool
 	}{
 			arrange: func(_ *testing.T, _ *Repository) ports.RetentionPlan {
 				return ports.RetentionPlan{IncarnationID: domain.IncarnationID{1}, Keep: []ports.CheckpointRef{{Generation: 3}, {Generation: 2}, {Generation: 1}}}
 			},
-			wantExhausted: true,
+			wantErr: ErrMaintenanceBudgetTooSmall,
 		},
 		{
 			name: "sufficient",
 			buffer.Reset()
 			done, err := repository.MaintainSession(context.Background(), plan, tc.budget)
-			require.NoError(t, err)
+			if tc.wantErr == nil {
+				require.NoError(t, err)
+			} else {
+				require.ErrorIs(t, err, tc.wantErr)
+			}
 			require.Equal(t, tc.wantDone, done)
 			scanner := bufio.NewScanner(bytes.NewReader(buffer.Bytes()))

M internal/app/durable_crash_matrix_test.go +14 -7


@@ -132,7 +132,8 @@ func TestDurableTransactionCrashMatrix(t *testing.T) {
132	132		require.NoError(t, catalogue.Close())
133	133
134	134		h.recoverTwice(ctx, t, func(catalogue persist.Persister, _ snapshot.Repository, _ *recoveryfs.Journal) {
135		-	record, exists, _ := catalogue.Record("work")
	135	+	record, exists, err := catalogue.Record("work")
	136	+	require.NoError(t, err)
136	137		require.True(t, exists)
137	138		require.Equal(t, domain.IncarnationID{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1}, record.IncarnationID)
138	139		require.Equal(t, domain.RecoveryFresh, record.RecoveryState)

@@ -160,7 +161,8 @@ func TestDurableTransactionCrashMatrix(t *testing.T) {
160	161
161	162		ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(publication.Manifest)}
162	163		h.recoverTwice(ctx, t, func(catalogue persist.Persister, repository snapshot.Repository, _ *recoveryfs.Journal) {
163		-	stored, exists, _ := catalogue.Record(record.Name)
	164	+	stored, exists, err := catalogue.Record(record.Name)
	165	+	require.NoError(t, err)
164	166		require.True(t, exists)
165	167		if boundaryName == "snapshot-publish" {
166	168		require.Equal(t, domain.RecoveryFresh, stored.RecoveryState)

@@ -189,8 +191,10 @@ func TestDurableTransactionCrashMatrix(t *testing.T) {
189	191		require.NoError(t, catalogue.Close())
190	192
191	193		h.recoverTwice(ctx, t, func(catalogue persist.Persister, _ snapshot.Repository, _ *recoveryfs.Journal) {
192		-	_, oldExists, _ := catalogue.Record("old")
193		-	renamed, newExists, _ := catalogue.Record("new")
	194	+	_, oldExists, err := catalogue.Record("old")
	195	+	require.NoError(t, err)
	196	+	renamed, newExists, err := catalogue.Record("new")
	197	+	require.NoError(t, err)
194	198		require.False(t, oldExists)
195	199		require.True(t, newExists)
196	200		require.Equal(t, record.IncarnationID, renamed.IncarnationID)

@@ -228,7 +232,8 @@ func TestDurableTransactionCrashMatrix(t *testing.T) {
228	232		require.NoError(t, catalogue.Close())
229	233
230	234		h.recoverTwice(ctx, t, func(catalogue persist.Persister, repository snapshot.Repository, _ *recoveryfs.Journal) {
231		-	_, exists, _ := catalogue.Record(record.Name)
	235	+	_, exists, err := catalogue.Record(record.Name)
	236	+	require.NoError(t, err)
232	237		require.False(t, exists)
233	238		assertNoTombstones(ctx, t, repository)
234	239		require.NoDirExists(t, filepath.Join(h.snapshotDir, "sessions", record.IncarnationID.String()))

@@ -275,7 +280,8 @@ func TestDurableDiscardCrashMatrix(t *testing.T) {
275	280		require.NoError(t, catalogue.Close())
276	281
277	282		h.recoverTwice(ctx, t, func(catalogue persist.Persister, _ snapshot.Repository, journal *recoveryfs.Journal) {
278		-	got, exists, _ := catalogue.Record(old.Name)
	283	+	got, exists, err := catalogue.Record(old.Name)
	284	+	require.NoError(t, err)
279	285		require.True(t, exists)
280	286		require.Equal(t, newID, got.IncarnationID)
281	287		require.Equal(t, domain.RecoveryFresh, got.RecoveryState)

@@ -312,7 +318,8 @@ func TestDeletionRecoveryProtectsReusedNameFilesystemBytes(t *testing.T) {
312	318		require.NoError(t, catalogue.Close())
313	319
314	320		h.recoverTwice(ctx, t, func(catalogue persist.Persister, repository snapshot.Repository, _ *recoveryfs.Journal) {
315		-	got, exists, _ := catalogue.Record("work")
	321	+	got, exists, err := catalogue.Record("work")
	322	+	require.NoError(t, err)
316	323		require.True(t, exists)
317	324		require.Equal(t, replacement, got)
318	325		require.Equal(t, replacementBytes, readTree(t, replacementPath))

 		require.NoError(t, catalogue.Close())
 		h.recoverTwice(ctx, t, func(catalogue *persist.Persister, _ *snapshot.Repository, _ *recoveryfs.Journal) {
-			record, exists, _ := catalogue.Record("work")
+			record, exists, err := catalogue.Record("work")
+			require.NoError(t, err)
 			require.True(t, exists)
 			require.Equal(t, domain.IncarnationID{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1}, record.IncarnationID)
 			require.Equal(t, domain.RecoveryFresh, record.RecoveryState)
 			ref := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(publication.Manifest)}
 			h.recoverTwice(ctx, t, func(catalogue *persist.Persister, repository *snapshot.Repository, _ *recoveryfs.Journal) {
-				stored, exists, _ := catalogue.Record(record.Name)
+				stored, exists, err := catalogue.Record(record.Name)
+				require.NoError(t, err)
 				require.True(t, exists)
 				if boundaryName == "snapshot-publish" {
 					require.Equal(t, domain.RecoveryFresh, stored.RecoveryState)
 		require.NoError(t, catalogue.Close())
 		h.recoverTwice(ctx, t, func(catalogue *persist.Persister, _ *snapshot.Repository, _ *recoveryfs.Journal) {
-			_, oldExists, _ := catalogue.Record("old")
+			_, oldExists, err := catalogue.Record("old")
-			renamed, newExists, _ := catalogue.Record("new")
+			require.NoError(t, err)
+			renamed, newExists, err := catalogue.Record("new")
+			require.NoError(t, err)
 			require.False(t, oldExists)
 			require.True(t, newExists)
 			require.Equal(t, record.IncarnationID, renamed.IncarnationID)
 			require.NoError(t, catalogue.Close())
 			h.recoverTwice(ctx, t, func(catalogue *persist.Persister, repository *snapshot.Repository, _ *recoveryfs.Journal) {
-				_, exists, _ := catalogue.Record(record.Name)
+				_, exists, err := catalogue.Record(record.Name)
+				require.NoError(t, err)
 				require.False(t, exists)
 				assertNoTombstones(ctx, t, repository)
 				require.NoDirExists(t, filepath.Join(h.snapshotDir, "sessions", record.IncarnationID.String()))
 			require.NoError(t, catalogue.Close())
 			h.recoverTwice(ctx, t, func(catalogue *persist.Persister, _ *snapshot.Repository, journal *recoveryfs.Journal) {
-				got, exists, _ := catalogue.Record(old.Name)
+				got, exists, err := catalogue.Record(old.Name)
+				require.NoError(t, err)
 				require.True(t, exists)
 				require.Equal(t, newID, got.IncarnationID)
 				require.Equal(t, domain.RecoveryFresh, got.RecoveryState)
 	require.NoError(t, catalogue.Close())
 	h.recoverTwice(ctx, t, func(catalogue *persist.Persister, repository *snapshot.Repository, _ *recoveryfs.Journal) {
-		got, exists, _ := catalogue.Record("work")
+		got, exists, err := catalogue.Record("work")
+		require.NoError(t, err)
 		require.True(t, exists)
 		require.Equal(t, replacement, got)
 		require.Equal(t, replacementBytes, readTree(t, replacementPath))

M internal/app/run_test.go +88 -34


@@ -15,6 +15,7 @@ import (
15	15		"os/exec"
16	16		"path/filepath"
17	17		"strings"
	18	+	"sync"
18	19		"testing"
19	20		"time"
20	21

@@ -142,35 +143,99 @@ func TestStartupReloadsCatalogueAfterRolledForwardDeletion(t *testing.T) {
142	143		}
143	144
144	145		func TestFencedRecoveryObservability(t *testing.T) {
145		-	ctx := context.Background()
146		-	stateDir := filepath.Join(t.TempDir(), "vev")
	146	+	// Keep the runtime path short enough for Darwin's smaller unix socket limit.
	147	+	runtimeRoot, err := os.MkdirTemp("/tmp", "vev-r-")
	148	+	require.NoError(t, err)
	149	+	t.Cleanup(func() { require.NoError(t, os.RemoveAll(runtimeRoot)) })
	150	+	stateRoot := t.TempDir()
	151	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
	152	+	t.Setenv("XDG_STATE_HOME", stateRoot)
	153	+	t.Setenv("XDG_CONFIG_HOME", t.TempDir())
	154	+	t.Setenv("VEV_PPROF_ADDR", "")
	155	+
	156	+	stateDir := filepath.Join(stateRoot, "vev")
147	157		repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
148	158		catalogue := newTestPersister(t, stateDir)
149		-	t.Cleanup(func() { _ = catalogue.Close() })
150		-
151	159		live := domain.CatalogueRecord{
152	160		Name: "work", IncarnationID: domain.IncarnationID{7}, Cwd: t.TempDir(),
153	161		CreatedAt: 1, UpdatedAt: 1, RecoveryState: domain.RecoveryHealthy,
154	162		Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}},
155	163		}
156	164		require.NoError(t, catalogue.Save(live))
157		-	require.NoError(t, repository.WriteDeletionTombstone(ctx, domain.DeletionTombstone{Name: live.Name, IncarnationID: live.IncarnationID}))
158		-
159		-	coordinator := recovery.NewCoordinator(catalogue, repository, recoveryfs.New(stateDir), rand.Reader)
160		-	require.NoError(t, coordinator.Recover(ctx), "a session-scoped conflict must not abort startup")
161		-	fenced, err := fencedRecoveries(coordinator.Conflicts(), catalogue)
162		-	require.NoError(t, err)
163		-	require.Equal(t, []fencedRecoveryIdentity{{
164		-	name: live.Name, id: live.IncarnationID, kind: "deletion-tombstone", reasonCode: "deletion-tombstone-conflict",
165		-	}}, fenced)
	165	+	require.NoError(t, catalogue.Close(), "startup must reopen the on-disk catalogue")
	166	+	require.NoError(t, repository.WriteDeletionTombstone(t.Context(), domain.DeletionTombstone{
	167	+	Name: live.Name, IncarnationID: live.IncarnationID,
	168	+	}))
	169	+
	170	+	// Pause immediately before socket publication so the test can reopen and
	171	+	// claim the notice before the daemon's restore worker consumes it.
	172	+	originalListenDaemon := listenDaemon
	173	+	listenEntered := make(chan struct{})
	174	+	allowListen := make(chan struct{})
	175	+	var allowListenOnce sync.Once
	176	+	releaseListen := func() { allowListenOnce.Do(func() { close(allowListen) }) }
	177	+	listenDaemon = func(dir string, observer ports.SerializedRuntimeObserver) (ports.Listener, error) {
	178	+	close(listenEntered)
	179	+	<-allowListen
	180	+	return originalListenDaemon(dir, observer)
	181	+	}
	182	+	t.Cleanup(func() {
	183	+	releaseListen()
	184	+	listenDaemon = originalListenDaemon
	185	+	})
166	186
167	187		var logBuffer bytes.Buffer
168	188		log := slog.New(slog.NewJSONHandler(&logBuffer, nil))
169		-	logFencedRecoveries(log, fenced)
170		-	logTransactionRecovery(log, 1, len(fenced))
171		-	records, err := catalogue.Records()
	189	+	ctx, cancel := context.WithCancel(t.Context())
	190	+	started := make(chan error, 1)
	191	+	go func() { started <- runDaemonOwnedWithLogger(ctx, log) }()
	192	+
	193	+	var shutdownErr error
	194	+	var shutdownOnce sync.Once
	195	+	shutdown := func() error {
	196	+	shutdownOnce.Do(func() {
	197	+	releaseListen()
	198	+	cancel()
	199	+	select {
	200	+	case shutdownErr = <-started:
	201	+	case <-time.After(5 * time.Second):
	202	+	shutdownErr = errors.New("daemon did not stop after cancellation")
	203	+	}
	204	+	})
	205	+	return shutdownErr
	206	+	}
	207	+	t.Cleanup(func() { require.NoError(t, shutdown()) })
	208	+
	209	+	select {
	210	+	case <-listenEntered:
	211	+	case <-time.After(5 * time.Second):
	212	+	t.Fatal("startup did not reach socket publication")
	213	+	}
	214	+	reopenedNotices := noticefile.New(stateDir)
	215	+	pending, err := reopenedNotices.Claim()
172	216		require.NoError(t, err)
173		-	logStartupRecoveryCounts(log, records, 0)
	217	+	require.Len(t, pending, 2, "startup persists both the interrupted and fenced recovery notices")
	218	+	var fencedNotice *domain.Notification
	219	+	for i := range pending {
	220	+	if strings.Contains(pending[i].Details, "reason_code=deletion-tombstone-conflict") {
	221	+	fencedNotice = &pending[i]
	222	+	}
	223	+	}
	224	+	require.NotNil(t, fencedNotice)
	225	+	require.Equal(t, domain.NoticeSnapshotRestore, fencedNotice.Code)
	226	+	require.Equal(t, domain.NoticeWarn, fencedNotice.Severity)
	227	+	require.Equal(t, domain.SessionID(live.IncarnationID.String()), fencedNotice.SessionID)
	228	+	require.Contains(t, fencedNotice.Details, "session="+live.Name)
	229	+	require.NotContains(t, fmt.Sprint(fencedNotice), "conflicts with catalogue")
	230	+	require.NoError(t, reopenedNotices.Ack(), "notice claim lock must be released before daemon startup resumes")
	231	+
	232	+	releaseListen()
	233	+	socketPath := filepath.Join(runtimeRoot, "vev", "daemon.sock")
	234	+	require.Eventually(t, func() bool {
	235	+	_, statErr := os.Stat(socketPath)
	236	+	return statErr == nil
	237	+	}, 5*time.Second, time.Millisecond, "fenced recovery must not prevent socket publication")
	238	+	require.NoError(t, shutdown(), "fenced recovery must allow clean daemon startup and shutdown")
174	239
175	240		entries := decodeJSONLogs(t, logBuffer.Bytes())
176	241		event := requireEvent(t, entries, "interrupted_transaction_fenced")

@@ -179,27 +244,16 @@ func TestFencedRecoveryObservability(t *testing.T) {
179	244		require.Equal(t, live.IncarnationID.String(), event["incarnation"])
180	245		require.Equal(t, "deletion-tombstone", event["kind"])
181	246		require.Equal(t, "deletion-tombstone-conflict", event["reason_code"])
182		-	require.NotContains(t, fmt.Sprint(event), "conflicts with catalogue", "raw cause text must not be logged")
183	247		recovered := requireEvent(t, entries, "interrupted_transaction_recovery_complete")
184	248		require.EqualValues(t, 1, recovered["sessions_fenced"])
185	249		require.Equal(t, "fenced", recovered["outcome"])
186		-	require.Contains(t, eventNames(entries), "daemon_startup_complete")
	250	+	startup := requireEvent(t, entries, "daemon_startup_complete")
	251	+	require.EqualValues(t, 1, startup["degraded"])
	252	+	require.NotContains(t, logBuffer.String(), "conflicts with catalogue", "raw conflict text must not be logged")
187	253
188		-	store := noticefile.New(stateDir)
189		-	require.NoError(t, persistFencedRecoveryNotices(store, fenced, time.Unix(1, 0)))
190		-	reopened := noticefile.New(stateDir)
191		-	pending, err := reopened.Claim()
192		-	require.NoError(t, err)
193		-	require.Len(t, pending, 1)
194		-	notice := pending[0]
195		-	require.Equal(t, domain.NoticeSnapshotRestore, notice.Code)
196		-	require.Equal(t, domain.NoticeWarn, notice.Severity)
197		-	require.Equal(t, domain.SessionID(live.IncarnationID.String()), notice.SessionID)
198		-	require.Contains(t, notice.Details, "session="+live.Name)
199		-	require.Contains(t, notice.Details, "reason_code=deletion-tombstone-conflict")
200		-	require.NoError(t, reopened.Ack())
201		-
202		-	degraded, ok, err := catalogue.Record(live.Name)
	254	+	reopenedCatalogue := newTestPersister(t, stateDir)
	255	+	t.Cleanup(func() { require.NoError(t, reopenedCatalogue.Close()) })
	256	+	degraded, ok, err := reopenedCatalogue.Record(live.Name)
203	257		require.NoError(t, err)
204	258		require.True(t, ok)
205	259		require.Equal(t, domain.RecoveryDegraded, degraded.RecoveryState)


@@ -15,6 +15,7 @@ import (
15		"os/exec"	15		"os/exec"
16		"path/filepath"	16		"path/filepath"
17		"strings"	17		"strings"
			18	+	"sync"
18		"testing"	19		"testing"
19		"time"	20		"time"
20			21

@@ -142,35 +143,99 @@ func TestStartupReloadsCatalogueAfterRolledForwardDeletion(t *testing.T) {
142		}	143		}
143			144
144		func TestFencedRecoveryObservability(t *testing.T) {	145		func TestFencedRecoveryObservability(t *testing.T) {
145	-	ctx := context.Background()	146	+	// Keep the runtime path short enough for Darwin's smaller unix socket limit.
146	-	stateDir := filepath.Join(t.TempDir(), "vev")	147	+	runtimeRoot, err := os.MkdirTemp("/tmp", "vev-r-")
			148	+	require.NoError(t, err)
			149	+	t.Cleanup(func() { require.NoError(t, os.RemoveAll(runtimeRoot)) })
			150	+	stateRoot := t.TempDir()
			151	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
			152	+	t.Setenv("XDG_STATE_HOME", stateRoot)
			153	+	t.Setenv("XDG_CONFIG_HOME", t.TempDir())
			154	+	t.Setenv("VEV_PPROF_ADDR", "")
			155	+
			156	+	stateDir := filepath.Join(stateRoot, "vev")
147		repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))	157		repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
148		catalogue := newTestPersister(t, stateDir)	158		catalogue := newTestPersister(t, stateDir)
149	-	t.Cleanup(func() { _ = catalogue.Close() })
150	-
151		live := domain.CatalogueRecord{	159		live := domain.CatalogueRecord{
152		Name: "work", IncarnationID: domain.IncarnationID{7}, Cwd: t.TempDir(),	160		Name: "work", IncarnationID: domain.IncarnationID{7}, Cwd: t.TempDir(),
153		CreatedAt: 1, UpdatedAt: 1, RecoveryState: domain.RecoveryHealthy,	161		CreatedAt: 1, UpdatedAt: 1, RecoveryState: domain.RecoveryHealthy,
154		Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}},	162		Committed: &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}},
155		}	163		}
156		require.NoError(t, catalogue.Save(live))	164		require.NoError(t, catalogue.Save(live))
157	-	require.NoError(t, repository.WriteDeletionTombstone(ctx, domain.DeletionTombstone{Name: live.Name, IncarnationID: live.IncarnationID}))	165	+	require.NoError(t, catalogue.Close(), "startup must reopen the on-disk catalogue")
158	-		166	+	require.NoError(t, repository.WriteDeletionTombstone(t.Context(), domain.DeletionTombstone{
159	-	coordinator := recovery.NewCoordinator(catalogue, repository, recoveryfs.New(stateDir), rand.Reader)	167	+	Name: live.Name, IncarnationID: live.IncarnationID,
160	-	require.NoError(t, coordinator.Recover(ctx), "a session-scoped conflict must not abort startup")	168	+	}))
161	-	fenced, err := fencedRecoveries(coordinator.Conflicts(), catalogue)	169	+
162	-	require.NoError(t, err)	170	+	// Pause immediately before socket publication so the test can reopen and
163	-	require.Equal(t, []fencedRecoveryIdentity{{	171	+	// claim the notice before the daemon's restore worker consumes it.
164	-	name: live.Name, id: live.IncarnationID, kind: "deletion-tombstone", reasonCode: "deletion-tombstone-conflict",	172	+	originalListenDaemon := listenDaemon
165	-	}}, fenced)	173	+	listenEntered := make(chan struct{})
			174	+	allowListen := make(chan struct{})
			175	+	var allowListenOnce sync.Once
			176	+	releaseListen := func() { allowListenOnce.Do(func() { close(allowListen) }) }
			177	+	listenDaemon = func(dir string, observer ports.SerializedRuntimeObserver) (ports.Listener, error) {
			178	+	close(listenEntered)
			179	+	<-allowListen
			180	+	return originalListenDaemon(dir, observer)
			181	+	}
			182	+	t.Cleanup(func() {
			183	+	releaseListen()
			184	+	listenDaemon = originalListenDaemon
			185	+	})
166			186
167		var logBuffer bytes.Buffer	187		var logBuffer bytes.Buffer
168		log := slog.New(slog.NewJSONHandler(&logBuffer, nil))	188		log := slog.New(slog.NewJSONHandler(&logBuffer, nil))
169	-	logFencedRecoveries(log, fenced)	189	+	ctx, cancel := context.WithCancel(t.Context())
170	-	logTransactionRecovery(log, 1, len(fenced))	190	+	started := make(chan error, 1)
171	-	records, err := catalogue.Records()	191	+	go func() { started <- runDaemonOwnedWithLogger(ctx, log) }()
			192	+
			193	+	var shutdownErr error
			194	+	var shutdownOnce sync.Once
			195	+	shutdown := func() error {
			196	+	shutdownOnce.Do(func() {
			197	+	releaseListen()
			198	+	cancel()
			199	+	select {
			200	+	case shutdownErr = <-started:
			201	+	case <-time.After(5 * time.Second):
			202	+	shutdownErr = errors.New("daemon did not stop after cancellation")
			203	+	}
			204	+	})
			205	+	return shutdownErr
			206	+	}
			207	+	t.Cleanup(func() { require.NoError(t, shutdown()) })
			208	+
			209	+	select {
			210	+	case <-listenEntered:
			211	+	case <-time.After(5 * time.Second):
			212	+	t.Fatal("startup did not reach socket publication")
			213	+	}
			214	+	reopenedNotices := noticefile.New(stateDir)
			215	+	pending, err := reopenedNotices.Claim()
172		require.NoError(t, err)	216		require.NoError(t, err)
173	-	logStartupRecoveryCounts(log, records, 0)	217	+	require.Len(t, pending, 2, "startup persists both the interrupted and fenced recovery notices")
			218	+	var fencedNotice *domain.Notification
			219	+	for i := range pending {
			220	+	if strings.Contains(pending[i].Details, "reason_code=deletion-tombstone-conflict") {
			221	+	fencedNotice = &pending[i]
			222	+	}
			223	+	}
			224	+	require.NotNil(t, fencedNotice)
			225	+	require.Equal(t, domain.NoticeSnapshotRestore, fencedNotice.Code)
			226	+	require.Equal(t, domain.NoticeWarn, fencedNotice.Severity)
			227	+	require.Equal(t, domain.SessionID(live.IncarnationID.String()), fencedNotice.SessionID)
			228	+	require.Contains(t, fencedNotice.Details, "session="+live.Name)
			229	+	require.NotContains(t, fmt.Sprint(fencedNotice), "conflicts with catalogue")
			230	+	require.NoError(t, reopenedNotices.Ack(), "notice claim lock must be released before daemon startup resumes")
			231	+
			232	+	releaseListen()
			233	+	socketPath := filepath.Join(runtimeRoot, "vev", "daemon.sock")
			234	+	require.Eventually(t, func() bool {
			235	+	_, statErr := os.Stat(socketPath)
			236	+	return statErr == nil
			237	+	}, 5*time.Second, time.Millisecond, "fenced recovery must not prevent socket publication")
			238	+	require.NoError(t, shutdown(), "fenced recovery must allow clean daemon startup and shutdown")
174			239
175		entries := decodeJSONLogs(t, logBuffer.Bytes())	240		entries := decodeJSONLogs(t, logBuffer.Bytes())
176		event := requireEvent(t, entries, "interrupted_transaction_fenced")	241		event := requireEvent(t, entries, "interrupted_transaction_fenced")

@@ -179,27 +244,16 @@ func TestFencedRecoveryObservability(t *testing.T) {
179		require.Equal(t, live.IncarnationID.String(), event["incarnation"])	244		require.Equal(t, live.IncarnationID.String(), event["incarnation"])
180		require.Equal(t, "deletion-tombstone", event["kind"])	245		require.Equal(t, "deletion-tombstone", event["kind"])
181		require.Equal(t, "deletion-tombstone-conflict", event["reason_code"])	246		require.Equal(t, "deletion-tombstone-conflict", event["reason_code"])
182	-	require.NotContains(t, fmt.Sprint(event), "conflicts with catalogue", "raw cause text must not be logged")
183		recovered := requireEvent(t, entries, "interrupted_transaction_recovery_complete")	247		recovered := requireEvent(t, entries, "interrupted_transaction_recovery_complete")
184		require.EqualValues(t, 1, recovered["sessions_fenced"])	248		require.EqualValues(t, 1, recovered["sessions_fenced"])
185		require.Equal(t, "fenced", recovered["outcome"])	249		require.Equal(t, "fenced", recovered["outcome"])
186	-	require.Contains(t, eventNames(entries), "daemon_startup_complete")	250	+	startup := requireEvent(t, entries, "daemon_startup_complete")
			251	+	require.EqualValues(t, 1, startup["degraded"])
			252	+	require.NotContains(t, logBuffer.String(), "conflicts with catalogue", "raw conflict text must not be logged")
187			253
188	-	store := noticefile.New(stateDir)	254	+	reopenedCatalogue := newTestPersister(t, stateDir)
189	-	require.NoError(t, persistFencedRecoveryNotices(store, fenced, time.Unix(1, 0)))	255	+	t.Cleanup(func() { require.NoError(t, reopenedCatalogue.Close()) })
190	-	reopened := noticefile.New(stateDir)	256	+	degraded, ok, err := reopenedCatalogue.Record(live.Name)
191	-	pending, err := reopened.Claim()
192	-	require.NoError(t, err)
193	-	require.Len(t, pending, 1)
194	-	notice := pending[0]
195	-	require.Equal(t, domain.NoticeSnapshotRestore, notice.Code)
196	-	require.Equal(t, domain.NoticeWarn, notice.Severity)
197	-	require.Equal(t, domain.SessionID(live.IncarnationID.String()), notice.SessionID)
198	-	require.Contains(t, notice.Details, "session="+live.Name)
199	-	require.Contains(t, notice.Details, "reason_code=deletion-tombstone-conflict")
200	-	require.NoError(t, reopened.Ack())
201	-
202	-	degraded, ok, err := catalogue.Record(live.Name)
203		require.NoError(t, err)	257		require.NoError(t, err)
204		require.True(t, ok)	258		require.True(t, ok)
205		require.Equal(t, domain.RecoveryDegraded, degraded.RecoveryState)	259		require.Equal(t, domain.RecoveryDegraded, degraded.RecoveryState)

M internal/usecase/daemon/durable_recovery_test.go +77 -1


@@ -31,6 +31,15 @@ type durableRecoveryCatalogue struct {
31	31		metadataUpdates []domain.CatalogueMetadataUpdate
32	32		}
33	33
	34	+	type recordErrorCatalogue struct {
	35	+	*durableRecoveryCatalogue
	36	+	err error
	37	+	}
	38	+
	39	+	func (c recordErrorCatalogue) Record(string) (domain.CatalogueRecord, bool, error) {
	40	+	return domain.CatalogueRecord{}, false, c.err
	41	+	}
	42	+
34	43		func newDurableRecoveryCatalogue(records []domain.CatalogueRecord) *durableRecoveryCatalogue {
35	44		catalogue := &durableRecoveryCatalogue{records: make(map[string]domain.CatalogueRecord, len(records))}
36	45		for _, record := range records {

@@ -224,7 +233,10 @@ func TestFinalCheckpointFailurePreservesCatalogue(t *testing.T) {
224	233		got, ok, _ := catalogue.Record(prior.Name)
225	234		require.True(t, ok)
226	235		require.Equal(t, prior, got, "a timed-out final checkpoint must not change authoritative bytes")
227		-	require.Equal(t, priorRef, *sess.snapshotPublishedCheckpoint)
	236	+	sess.snapshotMu.Lock()
	237	+	published := sess.snapshotPublishedCheckpoint
	238	+	sess.snapshotMu.Unlock()
	239	+	require.Equal(t, priorRef, *published)
228	240		}
229	241
230	242		func TestCatalogueRestoreIndependent(t *testing.T) {

@@ -486,6 +498,7 @@ func TestExplicitRecoveryRemainsRetryableAfterRepeatedTransientFailures(t *testi
486	498		for range 2 {
487	499		err := exec.restoreExplicitRecovery(t.Context())
488	500		require.ErrorIs(t, err, errRetryableRestoreLoad)
	501	+	require.ErrorIs(t, err, ports.ErrBudgetExhausted)
489	502
490	503		d.mu.Lock()
491	504		entry := d.stopped[record.Name]

@@ -524,6 +537,69 @@ func TestExplicitRecoveryRemainsRetryableAfterRepeatedTransientFailures(t *testi
524	537		repository.mu.Unlock()
525	538		}
526	539
	540	+	func TestFinishRecordRestoreDoesNotDegradeStaleCatalogueAuthority(t *testing.T) {
	541	+	for _, tt := range []struct {
	542	+	name string
	543	+	setup func(Daemon, durableRecoveryCatalogue, domain.CatalogueRecord)
	544	+	}{
	545	+	{
	546	+	name: "missing record",
	547	+	setup: func(_ Daemon, c durableRecoveryCatalogue, record domain.CatalogueRecord) {
	548	+	c.mu.Lock()
	549	+	delete(c.records, record.Name)
	550	+	c.mu.Unlock()
	551	+	},
	552	+	},
	553	+	{
	554	+	name: "replacement incarnation",
	555	+	setup: func(_ Daemon, c durableRecoveryCatalogue, record domain.CatalogueRecord) {
	556	+	record.IncarnationID = domain.IncarnationID{99}
	557	+	c.mu.Lock()
	558	+	c.records[record.Name] = record
	559	+	c.mu.Unlock()
	560	+	},
	561	+	},
	562	+	{
	563	+	name: "catalogue read error",
	564	+	setup: func(d Daemon, c durableRecoveryCatalogue, _ domain.CatalogueRecord) {
	565	+	d.catalogue = recordErrorCatalogue{durableRecoveryCatalogue: c, err: errors.New("read failed")}
	566	+	},
	567	+	},
	568	+	} {
	569	+	t.Run(tt.name, func(t *testing.T) {
	570	+	record := durableRecoveryRecord(0)
	571	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
	572	+	tt.setup(d, catalogue, record)
	573	+	d.mu.Lock()
	574	+	before := d.stopped[record.Name]
	575	+	d.mu.Unlock()
	576	+
	577	+	d.finishRecordRestore(record, errors.New("invalid checkpoint"), before.restoreDone)
	578	+
	579	+	d.mu.Lock()
	580	+	after := d.stopped[record.Name]
	581	+	d.mu.Unlock()
	582	+	require.Equal(t, before.record, after.record)
	583	+	require.Equal(t, before.state, after.state)
	584	+	})
	585	+	}
	586	+	}
	587	+
	588	+	func TestPersistAndRegisterRestoredSessionRejectsReplacementIncarnation(t *testing.T) {
	589	+	record := durableRecoveryRecord(0)
	590	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
	591	+	replacement := record
	592	+	replacement.IncarnationID = domain.IncarnationID{99}
	593	+	require.NoError(t, catalogue.Replace(record.Name, replacement))
	594	+	sess := newSnapshotTestSession(t, record.Name, false, record.Cwd)
	595	+	sess.incarnation = record.IncarnationID
	596	+
	597	+	registered, err := d.persistAndRegisterRestoredSession(t.Context(), sess)
	598	+	require.ErrorContains(t, err, "incarnation")
	599	+	require.False(t, registered)
	600	+	require.Empty(t, d.sessions)
	601	+	}
	602	+
527	603		func TestExplicitRecoveryRejectsConcurrentRestore(t *testing.T) {
528	604		record := durableRecoveryRecord(0)
529	605		record.RecoveryState = domain.RecoveryDegraded


@@ -31,6 +31,15 @@ type durableRecoveryCatalogue struct {
31		metadataUpdates []domain.CatalogueMetadataUpdate	31		metadataUpdates []domain.CatalogueMetadataUpdate
32		}	32		}
33			33
			34	+	type recordErrorCatalogue struct {
			35	+	*durableRecoveryCatalogue
			36	+	err error
			37	+	}
			38	+
			39	+	func (c recordErrorCatalogue) Record(string) (domain.CatalogueRecord, bool, error) {
			40	+	return domain.CatalogueRecord{}, false, c.err
			41	+	}
			42	+
34		func newDurableRecoveryCatalogue(records []domain.CatalogueRecord) *durableRecoveryCatalogue {	43		func newDurableRecoveryCatalogue(records []domain.CatalogueRecord) *durableRecoveryCatalogue {
35		catalogue := &durableRecoveryCatalogue{records: make(map[string]domain.CatalogueRecord, len(records))}	44		catalogue := &durableRecoveryCatalogue{records: make(map[string]domain.CatalogueRecord, len(records))}
36		for _, record := range records {	45		for _, record := range records {

@@ -224,7 +233,10 @@ func TestFinalCheckpointFailurePreservesCatalogue(t *testing.T) {
224		got, ok, _ := catalogue.Record(prior.Name)	233		got, ok, _ := catalogue.Record(prior.Name)
225		require.True(t, ok)	234		require.True(t, ok)
226		require.Equal(t, prior, got, "a timed-out final checkpoint must not change authoritative bytes")	235		require.Equal(t, prior, got, "a timed-out final checkpoint must not change authoritative bytes")
227	-	require.Equal(t, priorRef, *sess.snapshotPublishedCheckpoint)	236	+	sess.snapshotMu.Lock()
			237	+	published := sess.snapshotPublishedCheckpoint
			238	+	sess.snapshotMu.Unlock()
			239	+	require.Equal(t, priorRef, *published)
228		}	240		}
229			241
230		func TestCatalogueRestoreIndependent(t *testing.T) {	242		func TestCatalogueRestoreIndependent(t *testing.T) {

@@ -486,6 +498,7 @@ func TestExplicitRecoveryRemainsRetryableAfterRepeatedTransientFailures(t *testi
486		for range 2 {	498		for range 2 {
487		err := exec.restoreExplicitRecovery(t.Context())	499		err := exec.restoreExplicitRecovery(t.Context())
488		require.ErrorIs(t, err, errRetryableRestoreLoad)	500		require.ErrorIs(t, err, errRetryableRestoreLoad)
			501	+	require.ErrorIs(t, err, ports.ErrBudgetExhausted)
489			502
490		d.mu.Lock()	503		d.mu.Lock()
491		entry := d.stopped[record.Name]	504		entry := d.stopped[record.Name]

@@ -524,6 +537,69 @@ func TestExplicitRecoveryRemainsRetryableAfterRepeatedTransientFailures(t *testi
524		repository.mu.Unlock()	537		repository.mu.Unlock()
525		}	538		}
526			539
			540	+	func TestFinishRecordRestoreDoesNotDegradeStaleCatalogueAuthority(t *testing.T) {
			541	+	for _, tt := range []struct {
			542	+	name string
			543	+	setup func(Daemon, durableRecoveryCatalogue, domain.CatalogueRecord)
			544	+	}{
			545	+	{
			546	+	name: "missing record",
			547	+	setup: func(_ Daemon, c durableRecoveryCatalogue, record domain.CatalogueRecord) {
			548	+	c.mu.Lock()
			549	+	delete(c.records, record.Name)
			550	+	c.mu.Unlock()
			551	+	},
			552	+	},
			553	+	{
			554	+	name: "replacement incarnation",
			555	+	setup: func(_ Daemon, c durableRecoveryCatalogue, record domain.CatalogueRecord) {
			556	+	record.IncarnationID = domain.IncarnationID{99}
			557	+	c.mu.Lock()
			558	+	c.records[record.Name] = record
			559	+	c.mu.Unlock()
			560	+	},
			561	+	},
			562	+	{
			563	+	name: "catalogue read error",
			564	+	setup: func(d Daemon, c durableRecoveryCatalogue, _ domain.CatalogueRecord) {
			565	+	d.catalogue = recordErrorCatalogue{durableRecoveryCatalogue: c, err: errors.New("read failed")}
			566	+	},
			567	+	},
			568	+	} {
			569	+	t.Run(tt.name, func(t *testing.T) {
			570	+	record := durableRecoveryRecord(0)
			571	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
			572	+	tt.setup(d, catalogue, record)
			573	+	d.mu.Lock()
			574	+	before := d.stopped[record.Name]
			575	+	d.mu.Unlock()
			576	+
			577	+	d.finishRecordRestore(record, errors.New("invalid checkpoint"), before.restoreDone)
			578	+
			579	+	d.mu.Lock()
			580	+	after := d.stopped[record.Name]
			581	+	d.mu.Unlock()
			582	+	require.Equal(t, before.record, after.record)
			583	+	require.Equal(t, before.state, after.state)
			584	+	})
			585	+	}
			586	+	}
			587	+
			588	+	func TestPersistAndRegisterRestoredSessionRejectsReplacementIncarnation(t *testing.T) {
			589	+	record := durableRecoveryRecord(0)
			590	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
			591	+	replacement := record
			592	+	replacement.IncarnationID = domain.IncarnationID{99}
			593	+	require.NoError(t, catalogue.Replace(record.Name, replacement))
			594	+	sess := newSnapshotTestSession(t, record.Name, false, record.Cwd)
			595	+	sess.incarnation = record.IncarnationID
			596	+
			597	+	registered, err := d.persistAndRegisterRestoredSession(t.Context(), sess)
			598	+	require.ErrorContains(t, err, "incarnation")
			599	+	require.False(t, registered)
			600	+	require.Empty(t, d.sessions)
			601	+	}
			602	+
527		func TestExplicitRecoveryRejectsConcurrentRestore(t *testing.T) {	603		func TestExplicitRecoveryRejectsConcurrentRestore(t *testing.T) {
528		record := durableRecoveryRecord(0)	604		record := durableRecoveryRecord(0)
529		record.RecoveryState = domain.RecoveryDegraded	605		record.RecoveryState = domain.RecoveryDegraded

M internal/usecase/daemon/metadata_order_test.go +6 -1


@@ -42,7 +42,9 @@ func (c *blockingMetadataCatalogue) UpdateMetadata(update domain.CatalogueMetada
42	42		}
43	43
44	44		func newBlockingCatalogue(sess session, firstErr error) (durableRecoveryCatalogue, *blockingMetadataCatalogue) {
	45	+	sess.mu.Lock()
45	46		record := sess.persistRecordLocked(sess.createdAt)
	47	+	sess.mu.Unlock()
46	48		catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
47	49		return catalogue, &blockingMetadataCatalogue{
48	50		Catalogue: catalogue,

@@ -166,7 +168,10 @@ func TestConsecutiveMetadataFailuresRestoreDurableState(t *testing.T) {
166	168		func TestNewerMetadataFailureDoesNotSkipOlderSnapshot(t *testing.T) {
167	169		failure := errors.New("newer metadata write failed")
168	170		sess := newSnapshotTestSession(t, "work", false, "/work")
169		-	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{sess.persistRecordLocked(sess.createdAt)})
	171	+	sess.mu.Lock()
	172	+	record := sess.persistRecordLocked(sess.createdAt)
	173	+	sess.mu.Unlock()
	174	+	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
170	175		blocking := &blockingMetadataCatalogue{
171	176		Catalogue: catalogue,
172	177		firstStarted: make(chan struct{}),

 }
 func newBlockingCatalogue(sess *session, firstErr error) (*durableRecoveryCatalogue, *blockingMetadataCatalogue) {
+	sess.mu.Lock()
 	record := sess.persistRecordLocked(sess.createdAt)
+	sess.mu.Unlock()
 	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
 	return catalogue, &blockingMetadataCatalogue{
 		Catalogue:    catalogue,
 func TestNewerMetadataFailureDoesNotSkipOlderSnapshot(t *testing.T) {
 	failure := errors.New("newer metadata write failed")
 	sess := newSnapshotTestSession(t, "work", false, "/work")
-	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{sess.persistRecordLocked(sess.createdAt)})
+	sess.mu.Lock()
+	record := sess.persistRecordLocked(sess.createdAt)
+	sess.mu.Unlock()
+	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
 	blocking := &blockingMetadataCatalogue{
 		Catalogue:    catalogue,
 		firstStarted: make(chan struct{}),

M internal/usecase/daemon/recovery_observability.go +12 -7


@@ -26,15 +26,10 @@ func (d *Daemon) logStartupRecoveryCounts(restoring int) {
26	26		records, err = d.catalogue.Records()
27	27		if err != nil {
28	28		d.log.Error("daemon_startup_catalogue_read_failed", "err", err)
29		-	return
	29	+	records = d.stoppedRecords()
30	30		}
31	31		} else {
32		-	d.mu.Lock()
33		-	records = make([]domain.CatalogueRecord, 0, len(d.stopped))
34		-	for _, entry := range d.stopped {
35		-	records = append(records, entry.record)
36		-	}
37		-	d.mu.Unlock()
	32	+	records = d.stoppedRecords()
38	33		}
39	34		healthy, fresh, degraded := 0, 0, 0
40	35		for _, record := range records {

@@ -54,3 +49,13 @@ func (d *Daemon) logStartupRecoveryCounts(restoring int) {
54	49		"degraded", degraded,
55	50		)
56	51		}
	52	+
	53	+	func (d *Daemon) stoppedRecords() []domain.CatalogueRecord {
	54	+	d.mu.Lock()
	55	+	defer d.mu.Unlock()
	56	+	records := make([]domain.CatalogueRecord, 0, len(d.stopped))
	57	+	for _, entry := range d.stopped {
	58	+	records = append(records, entry.record)
	59	+	}
	60	+	return records
	61	+	}

 		records, err = d.catalogue.Records()
 		if err != nil {
 			d.log.Error("daemon_startup_catalogue_read_failed", "err", err)
-			return
+			records = d.stoppedRecords()
 		}
 	} else {
-		d.mu.Lock()
+		records = d.stoppedRecords()
-		records = make([]domain.CatalogueRecord, 0, len(d.stopped))
-		for _, entry := range d.stopped {
-			records = append(records, entry.record)
-		}
-		d.mu.Unlock()
 	}
 	healthy, fresh, degraded := 0, 0, 0
 	for _, record := range records {
 		"degraded", degraded,
 	)
 }
+func (d *Daemon) stoppedRecords() []domain.CatalogueRecord {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	records := make([]domain.CatalogueRecord, 0, len(d.stopped))
+	for _, entry := range d.stopped {
+		records = append(records, entry.record)
+	}
+	return records
+}

M internal/usecase/daemon/recovery_observability_test.go +26 -0


@@ -4,6 +4,7 @@ import (
4	4		"bufio"
5	5		"bytes"
6	6		"encoding/json"
	7	+	"errors"
7	8		"log/slog"
8	9		"testing"
9	10

@@ -40,6 +41,31 @@ func TestStartupRecoveryCounts(t *testing.T) {
40	41		require.EqualValues(t, 1, startup["degraded"])
41	42		}
42	43
	44	+	type failingRecordsCatalogue struct {
	45	+	*durableRecoveryCatalogue
	46	+	err error
	47	+	}
	48	+
	49	+	func (c failingRecordsCatalogue) Records() ([]domain.CatalogueRecord, error) {
	50	+	return nil, c.err
	51	+	}
	52	+
	53	+	func TestStartupRecoveryCountsFallsBackWhenCatalogueReadFails(t *testing.T) {
	54	+	var buffer bytes.Buffer
	55	+	record := domain.CatalogueRecord{Name: "broken", IncarnationID: domain.IncarnationID{4}, RecoveryState: domain.RecoveryDegraded, DegradedReason: "invalid"}
	56	+	d := New(nil, stubClock{}, slog.New(slog.NewJSONHandler(&buffer, nil)))
	57	+	d.persistEnabled = true
	58	+	d.catalogue = failingRecordsCatalogue{durableRecoveryCatalogue: newDurableRecoveryCatalogue(nil), err: errors.New("read failed")}
	59	+	d.stopped[record.Name] = stoppedSession{record: record}
	60	+
	61	+	d.logStartupRecoveryCounts(0)
	62	+
	63	+	entries := daemonJSONLogs(t, buffer.Bytes())
	64	+	daemonRequireEvent(t, entries, "daemon_startup_catalogue_read_failed")
	65	+	startup := daemonRequireEvent(t, entries, "daemon_startup_complete")
	66	+	require.EqualValues(t, 1, startup["degraded"])
	67	+	}
	68	+
43	69		func daemonJSONLogs(t *testing.T, data []byte) []map[string]any {
44	70		t.Helper()
45	71		var entries []map[string]any


@@ -4,6 +4,7 @@ import (
4	"bufio"	4		"bufio"
5	"bytes"	5		"bytes"
6	"encoding/json"	6		"encoding/json"
		7	+	"errors"
7	"log/slog"	8		"log/slog"
8	"testing"	9		"testing"
9		10

@@ -40,6 +41,31 @@ func TestStartupRecoveryCounts(t *testing.T) {
40	require.EqualValues(t, 1, startup["degraded"])	41		require.EqualValues(t, 1, startup["degraded"])
41	}	42		}
42		43
		44	+	type failingRecordsCatalogue struct {
		45	+	*durableRecoveryCatalogue
		46	+	err error
		47	+	}
		48	+
		49	+	func (c failingRecordsCatalogue) Records() ([]domain.CatalogueRecord, error) {
		50	+	return nil, c.err
		51	+	}
		52	+
		53	+	func TestStartupRecoveryCountsFallsBackWhenCatalogueReadFails(t *testing.T) {
		54	+	var buffer bytes.Buffer
		55	+	record := domain.CatalogueRecord{Name: "broken", IncarnationID: domain.IncarnationID{4}, RecoveryState: domain.RecoveryDegraded, DegradedReason: "invalid"}
		56	+	d := New(nil, stubClock{}, slog.New(slog.NewJSONHandler(&buffer, nil)))
		57	+	d.persistEnabled = true
		58	+	d.catalogue = failingRecordsCatalogue{durableRecoveryCatalogue: newDurableRecoveryCatalogue(nil), err: errors.New("read failed")}
		59	+	d.stopped[record.Name] = stoppedSession{record: record}
		60	+
		61	+	d.logStartupRecoveryCounts(0)
		62	+
		63	+	entries := daemonJSONLogs(t, buffer.Bytes())
		64	+	daemonRequireEvent(t, entries, "daemon_startup_catalogue_read_failed")
		65	+	startup := daemonRequireEvent(t, entries, "daemon_startup_complete")
		66	+	require.EqualValues(t, 1, startup["degraded"])
		67	+	}
		68	+
43	func daemonJSONLogs(t *testing.T, data []byte) []map[string]any {	69		func daemonJSONLogs(t *testing.T, data []byte) []map[string]any {
44	t.Helper()	70		t.Helper()
45	var entries []map[string]any	71		var entries []map[string]any

M internal/usecase/daemon/session.go +1 -3


@@ -316,9 +316,7 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz
316	316		if !ephemeral && name != "" {
317	317		sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)
318	318		}
319		-	if lastUsedSeq > 0 {
320		-	sess.mruAt.Store(lastUsedSeq)
321		-	}
	319	+	sess.mruAt.Store(lastUsedSeq)
322	320		sess.snapEligible.Store(!ephemeral && name != "")
323	321		if !ephemeral {
324	322		if d.persistEnabled {

 	if !ephemeral && name != "" {
 		sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)
 	}
-	if lastUsedSeq > 0 {
+	sess.mruAt.Store(lastUsedSeq)
-		sess.mruAt.Store(lastUsedSeq)
-	}
 	sess.snapEligible.Store(!ephemeral && name != "")
 	if !ephemeral {
 		if d.persistEnabled {

M internal/usecase/daemon/session_rename_purge_test.go +4 -1


@@ -27,7 +27,10 @@ func TestRenamePreservesIncarnationSnapshotSources(t *testing.T) {
27	27		WithStore(t, store)(d)
28	28		sess := newSnapshotTestSession(t, "old", false, "/work")
29	29		d.sessions = map[domain.SessionID]*session{sess.id: sess}
30		-	require.NoError(t, testPersister(t, d).Save(sess.persistRecordLocked(1)))
	30	+	sess.mu.Lock()
	31	+	record := sess.persistRecordLocked(1)
	32	+	sess.mu.Unlock()
	33	+	require.NoError(t, testPersister(t, d).Save(record))
31	34
32	35		require.NoError(t, d.renameSession(sess, "new"))
33	36		require.Empty(t, repository.calls, "rename must not invoke source deletion, including a failing legacy source")

 			WithStore(t, store)(d)
 			sess := newSnapshotTestSession(t, "old", false, "/work")
 			d.sessions = map[domain.SessionID]*session{sess.id: sess}
-			require.NoError(t, testPersister(t, d).Save(sess.persistRecordLocked(1)))
+			sess.mu.Lock()
+			record := sess.persistRecordLocked(1)
+			sess.mu.Unlock()
+			require.NoError(t, testPersister(t, d).Save(record))
 			require.NoError(t, d.renameSession(sess, "new"))
 			require.Empty(t, repository.calls, "rename must not invoke source deletion, including a failing legacy source")

M internal/usecase/daemon/snapshot_restore.go +4 -1


@@ -311,13 +311,16 @@ func (d Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess se
311	311		// Restoration is catalogue-authorized. Missing or unreadable authority is a
312	312		// hard failure; migration creates catalogue records before daemon startup.
313	313		if d.persistEnabled {
314		-	_, ok, err := d.catalogueRecord(sess.name)
	314	+	record, ok, err := d.catalogueRecord(sess.name)
315	315		if err != nil {
316	316		return false, err
317	317		}
318	318		if !ok {
319	319		return false, errors.New("snapshot: restored session is absent from catalogue")
320	320		}
	321	+	if record.IncarnationID != sess.incarnation {
	322	+	return false, errors.New("snapshot: restored session incarnation differs from catalogue")
	323	+	}
321	324		}
322	325		d.mu.Lock()
323	326		defer d.mu.Unlock()

 	// Restoration is catalogue-authorized. Missing or unreadable authority is a
 	// hard failure; migration creates catalogue records before daemon startup.
 	if d.persistEnabled {
-		_, ok, err := d.catalogueRecord(sess.name)
+		record, ok, err := d.catalogueRecord(sess.name)
 		if err != nil {
 			return false, err
 		}
 		if !ok {
 			return false, errors.New("snapshot: restored session is absent from catalogue")
 		}
+		if record.IncarnationID != sess.incarnation {
+			return false, errors.New("snapshot: restored session incarnation differs from catalogue")
+		}
 	}
 	d.mu.Lock()
 	defer d.mu.Unlock()

M internal/usecase/daemon/snapshot_restore_incremental.go +16 -19


@@ -144,26 +144,23 @@ func (d *Daemon) finishExplicitRecordRestore(record domain.CatalogueRecord, rest
144	144
145	145		func (d *Daemon) finishRecordRestoreAttempt(record domain.CatalogueRecord, restoreErr error, done chan struct{}, resetRetryable bool) {
146	146		retryable := errors.Is(restoreErr, errRetryableRestoreLoad)
147		-	if restoreErr != nil && !retryable {
148		-	catalogueReadable := true
149		-	if d.catalogue != nil {
150		-	current, ok, readErr := d.catalogue.Record(record.Name)
151		-	if readErr != nil {
152		-	catalogueReadable = false
153		-	d.log.Error("reading catalogue before degradation failed", "session", record.Name, "err", readErr)
154		-	} else if ok && current.IncarnationID == record.IncarnationID {
155		-	record = current
	147	+	degraded := false
	148	+	if restoreErr != nil && !retryable && d.catalogue != nil {
	149	+	current, ok, readErr := d.catalogue.Record(record.Name)
	150	+	switch {
	151	+	case readErr != nil:
	152	+	d.log.Error("reading catalogue before degradation failed", "session", record.Name, "err", readErr)
	153	+	case ok && current.IncarnationID == record.IncarnationID:
	154	+	record = current
	155	+	record.RecoveryState = domain.RecoveryDegraded
	156	+	record.DegradedReason = "checkpoint validation failed"
	157	+	if errors.Is(restoreErr, context.Canceled) \|\| errors.Is(restoreErr, context.DeadlineExceeded) {
	158	+	record.DegradedReason = "restore interrupted"
156	159		}
157		-	}
158		-	record.RecoveryState = domain.RecoveryDegraded
159		-	record.DegradedReason = "checkpoint validation failed"
160		-	if errors.Is(restoreErr, context.Canceled) \|\| errors.Is(restoreErr, context.DeadlineExceeded) {
161		-	record.DegradedReason = "restore interrupted"
162		-	}
163		-	if d.catalogue != nil && catalogueReadable {
164	160		if err := d.catalogue.Replace(record.Name, record); err != nil {
165	161		d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
166	162		}
	163	+	degraded = true
167	164		}
168	165		}
169	166

@@ -171,7 +168,7 @@ func (d *Daemon) finishRecordRestoreAttempt(record domain.CatalogueRecord, resto
171	168		entry, ok := d.stopped[record.Name]
172	169		if ok {
173	170		switch {
174		-	case restoreErr != nil && !retryable:
	171	+	case degraded:
175	172		entry.record = record
176	173		entry.state = runtimeDegraded
177	174		entry.cwd = record.Cwd

@@ -187,7 +184,7 @@ func (d *Daemon) finishRecordRestoreAttempt(record domain.CatalogueRecord, resto
187	184		}
188	185		closeRuntimeRestoreDoneLocked(done)
189	186		d.mu.Unlock()
190		-	if restoreErr != nil && !retryable {
	187	+	if degraded {
191	188		reasonCode := "checkpoint-invalid"
192	189		if errors.Is(restoreErr, context.Canceled) \|\| errors.Is(restoreErr, context.DeadlineExceeded) {
193	190		reasonCode = "restore-interrupted"

@@ -263,7 +260,7 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
263	260		}
264	261		if selectedIndex < 0 {
265	262		if retryableLoadErr != nil {
266		-	return fmt.Errorf("%w: %v", errRetryableRestoreLoad, retryableLoadErr)
	263	+	return fmt.Errorf("%w: %w", errRetryableRestoreLoad, retryableLoadErr)
267	264		}
268	265		return errors.New("snapshot: no catalogue checkpoint validated")
269	266		}

 func (d *Daemon) finishRecordRestoreAttempt(record domain.CatalogueRecord, restoreErr error, done chan struct{}, resetRetryable bool) {
 	retryable := errors.Is(restoreErr, errRetryableRestoreLoad)
-	if restoreErr != nil && !retryable {
+	degraded := false
-		catalogueReadable := true
+	if restoreErr != nil && !retryable && d.catalogue != nil {
-		if d.catalogue != nil {
+		current, ok, readErr := d.catalogue.Record(record.Name)
-			current, ok, readErr := d.catalogue.Record(record.Name)
+		switch {
-			if readErr != nil {
+		case readErr != nil:
-				catalogueReadable = false
+			d.log.Error("reading catalogue before degradation failed", "session", record.Name, "err", readErr)
-				d.log.Error("reading catalogue before degradation failed", "session", record.Name, "err", readErr)
+		case ok && current.IncarnationID == record.IncarnationID:
-			} else if ok && current.IncarnationID == record.IncarnationID {
+			record = current
-				record = current
+			record.RecoveryState = domain.RecoveryDegraded
+			record.DegradedReason = "checkpoint validation failed"
+			if errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded) {
+				record.DegradedReason = "restore interrupted"
 			}
-		}
-		record.RecoveryState = domain.RecoveryDegraded
-		record.DegradedReason = "checkpoint validation failed"
-		if errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded) {
-			record.DegradedReason = "restore interrupted"
-		}
-		if d.catalogue != nil && catalogueReadable {
 			if err := d.catalogue.Replace(record.Name, record); err != nil {
 				d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
 			}
+			degraded = true
 		}
 	}
 	entry, ok := d.stopped[record.Name]
 	if ok {
 		switch {
-		case restoreErr != nil && !retryable:
+		case degraded:
 			entry.record = record
 			entry.state = runtimeDegraded
 			entry.cwd = record.Cwd
 	}
 	closeRuntimeRestoreDoneLocked(done)
 	d.mu.Unlock()
-	if restoreErr != nil && !retryable {
+	if degraded {
 		reasonCode := "checkpoint-invalid"
 		if errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded) {
 			reasonCode = "restore-interrupted"
 	}
 	if selectedIndex < 0 {
 		if retryableLoadErr != nil {
-			return fmt.Errorf("%w: %v", errRetryableRestoreLoad, retryableLoadErr)
+			return fmt.Errorf("%w: %w", errRetryableRestoreLoad, retryableLoadErr)
 		}
 		return errors.New("snapshot: no catalogue checkpoint validated")
 	}

M internal/usecase/recovery/coordinator.go +3 -0


@@ -212,6 +212,9 @@ func (c *Coordinator) Delete(ctx context.Context, name string) error {
212	212		if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil {
213	213		return errors.New("recovery: incomplete delete dependencies")
214	214		}
	215	+	if err := ctx.Err(); err != nil {
	216	+	return err
	217	+	}
215	218		unlock := c.locks.Lock([]string{name})
216	219		defer unlock()
217	220		record, ok, err := c.catalogue.Record(name)


@@ -212,6 +212,9 @@ func (c *Coordinator) Delete(ctx context.Context, name string) error {
212	if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil {	212		if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil {
213	return errors.New("recovery: incomplete delete dependencies")	213		return errors.New("recovery: incomplete delete dependencies")
214	}	214		}
		215	+	if err := ctx.Err(); err != nil {
		216	+	return err
		217	+	}
215	unlock := c.locks.Lock([]string{name})	218		unlock := c.locks.Lock([]string{name})
216	defer unlock()	219		defer unlock()
217	record, ok, err := c.catalogue.Record(name)	220		record, ok, err := c.catalogue.Record(name)

M internal/usecase/recovery/coordinator_test.go +10 -0


@@ -22,6 +22,16 @@ func (j journalStub) ListDiscards(context.Context) ([]domain.DiscardIntent, erro
22	22		}
23	23		func (j journalStub) DeleteDiscard(context.Context, domain.IncarnationID) error { return nil }
24	24
	25	+	func TestDeleteChecksCancellationBeforeCatalogueAccess(t *testing.T) {
	26	+	ctx, cancel := context.WithCancel(t.Context())
	27	+	cancel()
	28	+	catalogue := portsmocks.NewMockCatalogue(t)
	29	+	repository := portsmocks.NewMockSnapshotRepository(t)
	30	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	31	+
	32	+	require.ErrorIs(t, coordinator.Delete(ctx, "work"), context.Canceled)
	33	+	}
	34	+
25	35		func TestCoordinatorRecoverRollsPendingDiscardForward(t *testing.T) {
26	36		ctx := context.Background()
27	37		old := domain.CatalogueRecord{


@@ -22,6 +22,16 @@ func (j journalStub) ListDiscards(context.Context) ([]domain.DiscardIntent, erro
22	}	22		}
23	func (j journalStub) DeleteDiscard(context.Context, domain.IncarnationID) error { return nil }	23		func (j journalStub) DeleteDiscard(context.Context, domain.IncarnationID) error { return nil }
24		24
		25	+	func TestDeleteChecksCancellationBeforeCatalogueAccess(t *testing.T) {
		26	+	ctx, cancel := context.WithCancel(t.Context())
		27	+	cancel()
		28	+	catalogue := portsmocks.NewMockCatalogue(t)
		29	+	repository := portsmocks.NewMockSnapshotRepository(t)
		30	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
		31	+
		32	+	require.ErrorIs(t, coordinator.Delete(ctx, "work"), context.Canceled)
		33	+	}
		34	+
25	func TestCoordinatorRecoverRollsPendingDiscardForward(t *testing.T) {	35		func TestCoordinatorRecoverRollsPendingDiscardForward(t *testing.T) {
26	ctx := context.Background()	36		ctx := context.Background()
27	old := domain.CatalogueRecord{	37		old := domain.CatalogueRecord{

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


@@ -2,6 +2,7 @@ package recovery
2	2
3	3		import (
4	4		"context"
	5	+	"fmt"
5	6		"testing"
6	7
7	8		"github.com/bnema/vev/internal/domain"

@@ -192,7 +193,7 @@ func TestForwardOrphanAdoption(t *testing.T) {
192	193		})
193	194
194	195		for _, state := range []domain.RecoveryState{domain.RecoveryDegraded, domain.RecoveryDeleting} {
195		-	t.Run("non-healthy record is not reconciled", func(t *testing.T) {
	196	+	t.Run(fmt.Sprintf("non-healthy record %d is not reconciled", state), func(t *testing.T) {
196	197		record := healthyReconcileRecord()
197	198		record.RecoveryState = state
198	199		if state == domain.RecoveryDegraded {

 import (
 	"context"
+	"fmt"
 	"testing"
 	"github.com/bnema/vev/internal/domain"
 	})
 	for _, state := range []domain.RecoveryState{domain.RecoveryDegraded, domain.RecoveryDeleting} {
-		t.Run("non-healthy record is not reconciled", func(t *testing.T) {
+		t.Run(fmt.Sprintf("non-healthy record %d is not reconciled", state), func(t *testing.T) {
 			record := healthyReconcileRecord()
 			record.RecoveryState = state
 			if state == domain.RecoveryDegraded {

M internal/usecase/recovery/transactions_test.go +2 -2


@@ -281,8 +281,8 @@ func TestDegradedExportIsReadOnly(t *testing.T) {
281	281		require.Empty(t, catalogue.events)
282	282		}
283	283
284		-	// A third incarnation still fails closed for that session: its intent and
285		-	// record are retained untouched, but startup is no longer aborted.
	284	+	// An absent catalogue record means the discard no longer has durable name
	285	+	// authority to replace, so recovery removes the completed intent.
286	286		func TestDiscardRecoveryRemovesIntentWhenCatalogueRecordIsAbsent(t *testing.T) {
287	287		old := degradedTransactionRecord()
288	288		intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}


@@ -281,8 +281,8 @@ func TestDegradedExportIsReadOnly(t *testing.T) {
281		require.Empty(t, catalogue.events)	281		require.Empty(t, catalogue.events)
282		}	282		}
283			283
284	-	// A third incarnation still fails closed for that session: its intent and	284	+	// An absent catalogue record means the discard no longer has durable name
285	-	// record are retained untouched, but startup is no longer aborted.	285	+	// authority to replace, so recovery removes the completed intent.
286		func TestDiscardRecoveryRemovesIntentWhenCatalogueRecordIsAbsent(t *testing.T) {	286		func TestDiscardRecoveryRemovesIntentWhenCatalogueRecordIsAbsent(t *testing.T) {
287		old := degradedTransactionRecord()	287		old := degradedTransactionRecord()
288		intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}	288		intent := domain.DiscardIntent{OldRecord: old, OldIncarnation: old.IncarnationID, NewIncarnation: domain.IncarnationID{2}, SessionName: old.Name, Reason: "discard"}