fix(persist): fence failed identity writes

commit: e5d0dcf9df08c8926ce85e569b39304f84b847fb
author: brice · 2mo ago · 2026-07-26 13:10 UTC
parent: 692d04f
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +346 -15

M internal/persist/persist.go +81 -5


@@ -17,6 +17,11 @@ const filename = "sessions.kv"
17	17
18	18		var errPersistenceUnavailable = errors.New("persist: catalogue unavailable")
19	19
	20	+	// ErrCatalogueDurability reports a failed identity write whose durability is
	21	+	// ambiguous. A Persister remains fenced after this error so rejected state
	22	+	// cannot be made durable by a later flush.
	23	+	var ErrCatalogueDurability = errors.New("persist: catalogue durability rejected")
	24	+
20	25		// ErrCatalogueUnreadable reports durable state that exists but cannot be
21	26		// decoded. vev never repairs or erases it; the operator resets explicitly.
22	27		var ErrCatalogueUnreadable = errors.New("persist: catalogue unreadable")

@@ -37,6 +42,7 @@ type Persister struct {
37	42		incarnationOwners map[domain.IncarnationID]string
38	43		nameIncarnations map[string]domain.IncarnationID
39	44		incarnationIndex bool
	45	+	terminalErr error
40	46		}
41	47
42	48		// KVStore adapts the reusable whole-file store to the persistence port.

@@ -59,6 +65,8 @@ func (s *KVStore) Range(fn func(key, value []byte) bool) {
59	65		func (s *KVStore) Sync() error { return s.store.Sync() }
60	66		func (s *KVStore) Close() error { return s.store.Close() }
61	67
	68	+	func (s *KVStore) CloseWithoutSync() error { return s.store.CloseWithoutSync() }
	69	+
62	70		// Open opens an existing strict catalogue and never creates an unproven empty one.
63	71		func Open(dir string) (*Persister, error) {
64	72		p, _, err := openCurrentCatalogue(dir, false)

@@ -140,6 +148,9 @@ func (p *Persister) Create(record domain.CatalogueRecord) error {
140	148		}
141	149		p.mu.Lock()
142	150		defer p.mu.Unlock()
	151	+	if err := p.terminalLocked(); err != nil {
	152	+	return err
	153	+	}
143	154		if _, exists := p.store.Get([]byte(record.Name)); exists {
144	155		return errors.New("persist: session already exists")
145	156		}

@@ -153,6 +164,9 @@ func (p *Persister) Record(name string) (domain.CatalogueRecord, bool, error) {
153	164		}
154	165		p.mu.Lock()
155	166		defer p.mu.Unlock()
	167	+	if err := p.terminalLocked(); err != nil {
	168	+	return domain.CatalogueRecord{}, false, err
	169	+	}
156	170		value, ok := p.store.Get([]byte(name))
157	171		if !ok {
158	172		return domain.CatalogueRecord{}, false, nil

@@ -172,6 +186,9 @@ func (p Persister) Apply(records map[string]domain.CatalogueRecord) error {
172	186		}
173	187		p.mu.Lock()
174	188		defer p.mu.Unlock()
	189	+	if err := p.terminalLocked(); err != nil {
	190	+	return err
	191	+	}
175	192		return p.applyLocked(records, true)
176	193		}
177	194		func (p Persister) applyLocked(records map[string]domain.CatalogueRecord, durable bool) error {

@@ -226,6 +243,11 @@ func (p Persister) applyLocked(records map[string]domain.CatalogueRecord, dura
226	243		}
227	244		}
228	245		}
	246	+	previous := make(map[string]storedValue, len(changes))
	247	+	for _, change := range changes {
	248	+	value, ok := p.store.Get(change.key)
	249	+	previous[string(change.key)] = storedValue{value: value, exists: ok}
	250	+	}
229	251		for _, change := range changes {
230	252		var err error
231	253		if change.delete {

@@ -234,14 +256,14 @@ func (p Persister) applyLocked(records map[string]domain.CatalogueRecord, dura
234	256		err = p.store.Set(change.key, change.value)
235	257		}
236	258		if err != nil {
237		-	p.incarnationIndex = false
238		-	return err
	259	+	p.restoreLocked(previous)
	260	+	return p.fenceLocked(err)
239	261		}
240	262		}
241	263		if durable {
242	264		if err := p.store.Sync(); err != nil {
243		-	p.incarnationIndex = false
244		-	return err
	265	+	p.restoreLocked(previous)
	266	+	return p.fenceLocked(err)
245	267		}
246	268		}
247	269		for name := range records {

@@ -259,7 +281,37 @@ func (p Persister) applyLocked(records map[string]domain.CatalogueRecord, dura
259	281		return nil
260	282		}
261	283
	284	+	type storedValue struct {
	285	+	value []byte
	286	+	exists bool
	287	+	}
	288	+
	289	+	// restoreLocked restores only keys affected by the rejected identity mutation.
	290	+	// It intentionally does not Sync: the Persister is fenced immediately after.
	291	+	func (p *Persister) restoreLocked(previous map[string]storedValue) {
	292	+	for key, old := range previous {
	293	+	if old.exists {
	294	+	_ = p.store.Set([]byte(key), old.value)
	295	+	} else {
	296	+	_ = p.store.Delete([]byte(key))
	297	+	}
	298	+	}
	299	+	}
	300	+
	301	+	func (p *Persister) fenceLocked(cause error) error {
	302	+	if p.terminalErr == nil {
	303	+	p.terminalErr = fmt.Errorf("%w: %w", ErrCatalogueDurability, cause)
	304	+	}
	305	+	p.incarnationIndex = false
	306	+	return p.terminalErr
	307	+	}
	308	+
	309	+	func (p *Persister) terminalLocked() error { return p.terminalErr }
	310	+
262	311		func (p *Persister) ensureIncarnationIndexLocked() error {
	312	+	if err := p.terminalLocked(); err != nil {
	313	+	return err
	314	+	}
263	315		if p.incarnationIndex {
264	316		return nil
265	317		}

@@ -318,6 +370,9 @@ func (p *Persister) UpdateMetadata(update domain.CatalogueMetadataUpdate) error
318	370		}
319	371		p.mu.Lock()
320	372		defer p.mu.Unlock()
	373	+	if err := p.terminalLocked(); err != nil {
	374	+	return err
	375	+	}
321	376		value, ok := p.store.Get([]byte(update.Name))
322	377		if !ok {
323	378		return errors.New("persist: session not found")

@@ -354,7 +409,13 @@ func (p *Persister) Sync() error {
354	409		}
355	410		p.mu.Lock()
356	411		defer p.mu.Unlock()
357		-	return p.store.Sync()
	412	+	if err := p.terminalLocked(); err != nil {
	413	+	return err
	414	+	}
	415	+	if err := p.store.Sync(); err != nil {
	416	+	return p.fenceLocked(err)
	417	+	}
	418	+	return nil
358	419		}
359	420
360	421		func (p *Persister) Delete(name string) error {

@@ -370,6 +431,9 @@ func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) {
370	431		}
371	432		p.mu.Lock()
372	433		defer p.mu.Unlock()
	434	+	if err := p.terminalLocked(); err != nil {
	435	+	return nil, err
	436	+	}
373	437		data := make(map[string][]byte)
374	438		p.store.Range(func(key, value []byte) bool { data[string(key)] = append([]byte(nil), value...); return true })
375	439		return decodeAll(data)

@@ -380,6 +444,18 @@ func (p *Persister) Close() error {
380	444		}
381	445		p.mu.Lock()
382	446		defer p.mu.Unlock()
	447	+	if err := p.terminalLocked(); err != nil {
	448	+	if store, ok := p.store.(interface{ CloseWithoutSync() error }); ok {
	449	+	if closeErr := store.CloseWithoutSync(); closeErr != nil {
	450	+	return errors.Join(err, closeErr)
	451	+	}
	452	+	return err
	453	+	}
	454	+	if closeErr := p.store.Close(); closeErr != nil {
	455	+	return errors.Join(err, closeErr)
	456	+	}
	457	+	return err
	458	+	}
383	459		return p.store.Close()
384	460		}
385	461

 var errPersistenceUnavailable = errors.New("persist: catalogue unavailable")
+// ErrCatalogueDurability reports a failed identity write whose durability is
+// ambiguous. A Persister remains fenced after this error so rejected state
+// cannot be made durable by a later flush.
+var ErrCatalogueDurability = errors.New("persist: catalogue durability rejected")
 // ErrCatalogueUnreadable reports durable state that exists but cannot be
 // decoded. vev never repairs or erases it; the operator resets explicitly.
 var ErrCatalogueUnreadable = errors.New("persist: catalogue unreadable")
 	incarnationOwners map[domain.IncarnationID]string
 	nameIncarnations  map[string]domain.IncarnationID
 	incarnationIndex  bool
+	terminalErr       error
 }
 // KVStore adapts the reusable whole-file store to the persistence port.
 func (s *KVStore) Sync() error  { return s.store.Sync() }
 func (s *KVStore) Close() error { return s.store.Close() }
+func (s *KVStore) CloseWithoutSync() error { return s.store.CloseWithoutSync() }
 // Open opens an existing strict catalogue and never creates an unproven empty one.
 func Open(dir string) (*Persister, error) {
 	p, _, err := openCurrentCatalogue(dir, false)
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
+	if err := p.terminalLocked(); err != nil {
+		return err
+	}
 	if _, exists := p.store.Get([]byte(record.Name)); exists {
 		return errors.New("persist: session already exists")
 	}
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
+	if err := p.terminalLocked(); err != nil {
+		return domain.CatalogueRecord{}, false, err
+	}
 	value, ok := p.store.Get([]byte(name))
 	if !ok {
 		return domain.CatalogueRecord{}, false, nil
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
+	if err := p.terminalLocked(); err != nil {
+		return err
+	}
 	return p.applyLocked(records, true)
 }
 func (p *Persister) applyLocked(records map[string]*domain.CatalogueRecord, durable bool) error {
 			}
 		}
 	}
+	previous := make(map[string]storedValue, len(changes))
+	for _, change := range changes {
+		value, ok := p.store.Get(change.key)
+		previous[string(change.key)] = storedValue{value: value, exists: ok}
+	}
 	for _, change := range changes {
 		var err error
 		if change.delete {
 			err = p.store.Set(change.key, change.value)
 		}
 		if err != nil {
-			p.incarnationIndex = false
+			p.restoreLocked(previous)
-			return err
+			return p.fenceLocked(err)
 		}
 	}
 	if durable {
 		if err := p.store.Sync(); err != nil {
-			p.incarnationIndex = false
+			p.restoreLocked(previous)
-			return err
+			return p.fenceLocked(err)
 		}
 	}
 	for name := range records {
 	return nil
 }
+type storedValue struct {
+	value  []byte
+	exists bool
+}
+// restoreLocked restores only keys affected by the rejected identity mutation.
+// It intentionally does not Sync: the Persister is fenced immediately after.
+func (p *Persister) restoreLocked(previous map[string]storedValue) {
+	for key, old := range previous {
+		if old.exists {
+			_ = p.store.Set([]byte(key), old.value)
+		} else {
+			_ = p.store.Delete([]byte(key))
+		}
+	}
+}
+func (p *Persister) fenceLocked(cause error) error {
+	if p.terminalErr == nil {
+		p.terminalErr = fmt.Errorf("%w: %w", ErrCatalogueDurability, cause)
+	}
+	p.incarnationIndex = false
+	return p.terminalErr
+}
+func (p *Persister) terminalLocked() error { return p.terminalErr }
 func (p *Persister) ensureIncarnationIndexLocked() error {
+	if err := p.terminalLocked(); err != nil {
+		return err
+	}
 	if p.incarnationIndex {
 		return nil
 	}
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
+	if err := p.terminalLocked(); err != nil {
+		return err
+	}
 	value, ok := p.store.Get([]byte(update.Name))
 	if !ok {
 		return errors.New("persist: session not found")
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	return p.store.Sync()
+	if err := p.terminalLocked(); err != nil {
+		return err
+	}
+	if err := p.store.Sync(); err != nil {
+		return p.fenceLocked(err)
+	}
+	return nil
 }
 func (p *Persister) Delete(name string) error {
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
+	if err := p.terminalLocked(); err != nil {
+		return nil, err
+	}
 	data := make(map[string][]byte)
 	p.store.Range(func(key, value []byte) bool { data[string(key)] = append([]byte(nil), value...); return true })
 	return decodeAll(data)
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
+	if err := p.terminalLocked(); err != nil {
+		if store, ok := p.store.(interface{ CloseWithoutSync() error }); ok {
+			if closeErr := store.CloseWithoutSync(); closeErr != nil {
+				return errors.Join(err, closeErr)
+			}
+			return err
+		}
+		if closeErr := p.store.Close(); closeErr != nil {
+			return errors.Join(err, closeErr)
+		}
+		return err
+	}
 	return p.store.Close()
 }

M internal/persist/persist_test.go +226 -0


@@ -118,6 +118,10 @@ func TestCatalogueMutationsUseOneSync(t *testing.T) {
118	118		}
119	119		}
120	120		}).Once()
	121	+	store.EXPECT().Get(mock.Anything).RunAndReturn(func(key []byte) ([]byte, bool) {
	122	+	value, ok := state[string(key)]
	123	+	return append([]byte(nil), value...), ok
	124	+	}).Times(3)
121	125		store.EXPECT().Set(mock.Anything, mock.Anything).RunAndReturn(func(key, value []byte) error {
122	126		state[string(key)] = append([]byte(nil), value...)
123	127		return nil

@@ -146,7 +150,9 @@ func TestCatalogueMutationFailureDoesNotSync(t *testing.T) {
146	150		store := portsmocks.NewMockStore(t)
147	151		setErr := errors.New("set failed")
148	152		store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {})
	153	+	store.EXPECT().Get([]byte("one")).Return(nil, false).Once()
149	154		store.EXPECT().Set(mock.Anything, mock.Anything).Return(setErr)
	155	+	store.EXPECT().Delete([]byte("one")).Return(nil).Once()
150	156		p := New(store)
151	157		err := p.Replace("one", validRecord("one", 1))
152	158		require.ErrorIs(t, err, setErr)

@@ -205,3 +211,223 @@ func TestCatalogueRecordsPropagatesMalformedRecord(t *testing.T) {
205	211		require.Error(t, err)
206	212		require.Empty(t, records)
207	213		}
	214	+
	215	+	type failClosedStore struct {
	216	+	data map[string][]byte
	217	+	durable map[string][]byte
	218	+
	219	+	setFailures int
	220	+	deleteFailures int
	221	+	syncFailures int
	222	+	failure error
	223	+
	224	+	closeFlushed bool
	225	+	closeAborted bool
	226	+	}
	227	+
	228	+	func newFailClosedStore(data map[string][]byte) *failClosedStore {
	229	+	return &failClosedStore{data: copyStoreData(data), durable: copyStoreData(data), failure: errors.New("store write failed")}
	230	+	}
	231	+
	232	+	func (s *failClosedStore) Get(key []byte) ([]byte, bool) {
	233	+	value, ok := s.data[string(key)]
	234	+	return append([]byte(nil), value...), ok
	235	+	}
	236	+
	237	+	func (s *failClosedStore) Set(key, value []byte) error {
	238	+	s.data[string(key)] = append([]byte(nil), value...)
	239	+	if s.setFailures > 0 {
	240	+	s.setFailures--
	241	+	return s.failure
	242	+	}
	243	+	return nil
	244	+	}
	245	+
	246	+	func (s *failClosedStore) Delete(key []byte) error {
	247	+	delete(s.data, string(key))
	248	+	if s.deleteFailures > 0 {
	249	+	s.deleteFailures--
	250	+	return s.failure
	251	+	}
	252	+	return nil
	253	+	}
	254	+
	255	+	func (s *failClosedStore) Range(fn func([]byte, []byte) bool) {
	256	+	for key, value := range s.data {
	257	+	if !fn([]byte(key), append([]byte(nil), value...)) {
	258	+	return
	259	+	}
	260	+	}
	261	+	}
	262	+
	263	+	func (s *failClosedStore) Sync() error {
	264	+	if s.syncFailures > 0 {
	265	+	s.syncFailures--
	266	+	return s.failure
	267	+	}
	268	+	s.durable = copyStoreData(s.data)
	269	+	return nil
	270	+	}
	271	+
	272	+	func (s *failClosedStore) Close() error {
	273	+	s.closeFlushed = true
	274	+	s.durable = copyStoreData(s.data)
	275	+	return nil
	276	+	}
	277	+
	278	+	func (s *failClosedStore) CloseWithoutSync() error {
	279	+	s.closeAborted = true
	280	+	return nil
	281	+	}
	282	+
	283	+	func copyStoreData(data map[string][]byte) map[string][]byte {
	284	+	copy := make(map[string][]byte, len(data))
	285	+	for key, value := range data {
	286	+	copy[key] = append([]byte(nil), value...)
	287	+	}
	288	+	return copy
	289	+	}
	290	+
	291	+	func TestRejectedIdentityWriteFencesCatalogue(t *testing.T) {
	292	+	t.Parallel()
	293	+	one, err := encodeRecordValue(validRecord("work", 1))
	294	+	require.NoError(t, err)
	295	+
	296	+	tests := []struct {
	297	+	name string
	298	+	initial map[string][]byte
	299	+	setFailures int
	300	+	deleteFailures int
	301	+	syncFailures int
	302	+	mutate func(*Persister) error
	303	+	}{
	304	+	{
	305	+	name: "create set failure",
	306	+	initial: map[string][]byte{},
	307	+	setFailures: 1,
	308	+	mutate: func(p *Persister) error {
	309	+	return p.Create(validRecord("new", 2))
	310	+	},
	311	+	},
	312	+	{
	313	+	name: "create sync failure",
	314	+	initial: map[string][]byte{},
	315	+	syncFailures: 1,
	316	+	mutate: func(p *Persister) error {
	317	+	return p.Create(validRecord("new", 2))
	318	+	},
	319	+	},
	320	+	{
	321	+	name: "replace set failure",
	322	+	initial: map[string][]byte{"work": one},
	323	+	setFailures: 1,
	324	+	mutate: func(p *Persister) error {
	325	+	next := validRecord("work", 1)
	326	+	next.Cwd = "/next"
	327	+	return p.Replace("work", next)
	328	+	},
	329	+	},
	330	+	{
	331	+	name: "replace sync failure",
	332	+	initial: map[string][]byte{"work": one},
	333	+	syncFailures: 1,
	334	+	mutate: func(p *Persister) error {
	335	+	next := validRecord("work", 1)
	336	+	next.Cwd = "/next"
	337	+	return p.Replace("work", next)
	338	+	},
	339	+	},
	340	+	{
	341	+	name: "rename set failure",
	342	+	initial: map[string][]byte{"work": one},
	343	+	setFailures: 1,
	344	+	mutate: func(p *Persister) error {
	345	+	next := validRecord("renamed", 1)
	346	+	return p.Rename("work", next)
	347	+	},
	348	+	},
	349	+	{
	350	+	name: "rename delete failure",
	351	+	initial: map[string][]byte{"work": one},
	352	+	deleteFailures: 1,
	353	+	mutate: func(p *Persister) error {
	354	+	next := validRecord("renamed", 1)
	355	+	return p.Rename("work", next)
	356	+	},
	357	+	},
	358	+	{
	359	+	name: "rename sync failure",
	360	+	initial: map[string][]byte{"work": one},
	361	+	syncFailures: 1,
	362	+	mutate: func(p *Persister) error {
	363	+	next := validRecord("renamed", 1)
	364	+	return p.Rename("work", next)
	365	+	},
	366	+	},
	367	+	{
	368	+	name: "delete delete failure",
	369	+	initial: map[string][]byte{"work": one},
	370	+	deleteFailures: 1,
	371	+	mutate: func(p *Persister) error {
	372	+	return p.Delete("work")
	373	+	},
	374	+	},
	375	+	{
	376	+	name: "delete sync failure",
	377	+	initial: map[string][]byte{"work": one},
	378	+	syncFailures: 1,
	379	+	mutate: func(p *Persister) error {
	380	+	return p.Delete("work")
	381	+	},
	382	+	},
	383	+	}
	384	+
	385	+	for _, tt := range tests {
	386	+	t.Run(tt.name, func(t *testing.T) {
	387	+	t.Parallel()
	388	+	store := newFailClosedStore(tt.initial)
	389	+	store.setFailures = tt.setFailures
	390	+	store.deleteFailures = tt.deleteFailures
	391	+	store.syncFailures = tt.syncFailures
	392	+	p := New(store)
	393	+
	394	+	err := tt.mutate(p)
	395	+	require.ErrorIs(t, err, ErrCatalogueDurability)
	396	+	require.ErrorIs(t, err, store.failure)
	397	+	require.Equal(t, tt.initial, store.data, "rejected identity keys must be restored in memory")
	398	+	require.Equal(t, tt.initial, store.durable, "rejected identity write must not reach disk")
	399	+
	400	+	_, _, recordErr := p.Record("work")
	401	+	require.ErrorIs(t, recordErr, ErrCatalogueDurability)
	402	+	require.ErrorIs(t, p.Sync(), ErrCatalogueDurability)
	403	+	require.ErrorIs(t, p.Create(validRecord("later", 3)), ErrCatalogueDurability)
	404	+	require.Equal(t, tt.initial, store.durable, "later Sync must be fenced")
	405	+
	406	+	require.ErrorIs(t, p.Close(), ErrCatalogueDurability)
	407	+	require.True(t, store.closeAborted, "Close must release without flushing rejected state")
	408	+	require.False(t, store.closeFlushed)
	409	+	require.Equal(t, tt.initial, store.durable, "Close must not make rejected state durable")
	410	+	})
	411	+	}
	412	+	}
	413	+
	414	+	func TestRejectedIdentityWriteRestoresBufferedMetadata(t *testing.T) {
	415	+	t.Parallel()
	416	+	encoded, err := encodeRecordValue(validRecord("work", 1))
	417	+	require.NoError(t, err)
	418	+	store := newFailClosedStore(map[string][]byte{"work": encoded})
	419	+	p := New(store)
	420	+
	421	+	cwd := "/buffered"
	422	+	require.NoError(t, p.UpdateMetadata(domain.CatalogueMetadataUpdate{
	423	+	Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: &cwd,
	424	+	}))
	425	+	buffered := copyStoreData(store.data)
	426	+	store.syncFailures = 1
	427	+
	428	+	require.ErrorIs(t, p.Rename("work", validRecord("renamed", 1)), ErrCatalogueDurability)
	429	+	require.Equal(t, buffered, store.data, "rollback must preserve already-buffered metadata on the affected key")
	430	+	require.Equal(t, map[string][]byte{"work": encoded}, store.durable)
	431	+	require.ErrorIs(t, p.Close(), ErrCatalogueDurability)
	432	+	require.Equal(t, map[string][]byte{"work": encoded}, store.durable)
	433	+	}


@@ -118,6 +118,10 @@ func TestCatalogueMutationsUseOneSync(t *testing.T) {
118	}	118		}
119	}	119		}
120	}).Once()	120		}).Once()
		121	+	store.EXPECT().Get(mock.Anything).RunAndReturn(func(key []byte) ([]byte, bool) {
		122	+	value, ok := state[string(key)]
		123	+	return append([]byte(nil), value...), ok
		124	+	}).Times(3)
121	store.EXPECT().Set(mock.Anything, mock.Anything).RunAndReturn(func(key, value []byte) error {	125		store.EXPECT().Set(mock.Anything, mock.Anything).RunAndReturn(func(key, value []byte) error {
122	state[string(key)] = append([]byte(nil), value...)	126		state[string(key)] = append([]byte(nil), value...)
123	return nil	127		return nil

@@ -146,7 +150,9 @@ func TestCatalogueMutationFailureDoesNotSync(t *testing.T) {
146	store := portsmocks.NewMockStore(t)	150		store := portsmocks.NewMockStore(t)
147	setErr := errors.New("set failed")	151		setErr := errors.New("set failed")
148	store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {})	152		store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {})
		153	+	store.EXPECT().Get([]byte("one")).Return(nil, false).Once()
149	store.EXPECT().Set(mock.Anything, mock.Anything).Return(setErr)	154		store.EXPECT().Set(mock.Anything, mock.Anything).Return(setErr)
		155	+	store.EXPECT().Delete([]byte("one")).Return(nil).Once()
150	p := New(store)	156		p := New(store)
151	err := p.Replace("one", validRecord("one", 1))	157		err := p.Replace("one", validRecord("one", 1))
152	require.ErrorIs(t, err, setErr)	158		require.ErrorIs(t, err, setErr)

@@ -205,3 +211,223 @@ func TestCatalogueRecordsPropagatesMalformedRecord(t *testing.T) {
205	require.Error(t, err)	211		require.Error(t, err)
206	require.Empty(t, records)	212		require.Empty(t, records)
207	}	213		}
		214	+
		215	+	type failClosedStore struct {
		216	+	data map[string][]byte
		217	+	durable map[string][]byte
		218	+
		219	+	setFailures int
		220	+	deleteFailures int
		221	+	syncFailures int
		222	+	failure error
		223	+
		224	+	closeFlushed bool
		225	+	closeAborted bool
		226	+	}
		227	+
		228	+	func newFailClosedStore(data map[string][]byte) *failClosedStore {
		229	+	return &failClosedStore{data: copyStoreData(data), durable: copyStoreData(data), failure: errors.New("store write failed")}
		230	+	}
		231	+
		232	+	func (s *failClosedStore) Get(key []byte) ([]byte, bool) {
		233	+	value, ok := s.data[string(key)]
		234	+	return append([]byte(nil), value...), ok
		235	+	}
		236	+
		237	+	func (s *failClosedStore) Set(key, value []byte) error {
		238	+	s.data[string(key)] = append([]byte(nil), value...)
		239	+	if s.setFailures > 0 {
		240	+	s.setFailures--
		241	+	return s.failure
		242	+	}
		243	+	return nil
		244	+	}
		245	+
		246	+	func (s *failClosedStore) Delete(key []byte) error {
		247	+	delete(s.data, string(key))
		248	+	if s.deleteFailures > 0 {
		249	+	s.deleteFailures--
		250	+	return s.failure
		251	+	}
		252	+	return nil
		253	+	}
		254	+
		255	+	func (s *failClosedStore) Range(fn func([]byte, []byte) bool) {
		256	+	for key, value := range s.data {
		257	+	if !fn([]byte(key), append([]byte(nil), value...)) {
		258	+	return
		259	+	}
		260	+	}
		261	+	}
		262	+
		263	+	func (s *failClosedStore) Sync() error {
		264	+	if s.syncFailures > 0 {
		265	+	s.syncFailures--
		266	+	return s.failure
		267	+	}
		268	+	s.durable = copyStoreData(s.data)
		269	+	return nil
		270	+	}
		271	+
		272	+	func (s *failClosedStore) Close() error {
		273	+	s.closeFlushed = true
		274	+	s.durable = copyStoreData(s.data)
		275	+	return nil
		276	+	}
		277	+
		278	+	func (s *failClosedStore) CloseWithoutSync() error {
		279	+	s.closeAborted = true
		280	+	return nil
		281	+	}
		282	+
		283	+	func copyStoreData(data map[string][]byte) map[string][]byte {
		284	+	copy := make(map[string][]byte, len(data))
		285	+	for key, value := range data {
		286	+	copy[key] = append([]byte(nil), value...)
		287	+	}
		288	+	return copy
		289	+	}
		290	+
		291	+	func TestRejectedIdentityWriteFencesCatalogue(t *testing.T) {
		292	+	t.Parallel()
		293	+	one, err := encodeRecordValue(validRecord("work", 1))
		294	+	require.NoError(t, err)
		295	+
		296	+	tests := []struct {
		297	+	name string
		298	+	initial map[string][]byte
		299	+	setFailures int
		300	+	deleteFailures int
		301	+	syncFailures int
		302	+	mutate func(*Persister) error
		303	+	}{
		304	+	{
		305	+	name: "create set failure",
		306	+	initial: map[string][]byte{},
		307	+	setFailures: 1,
		308	+	mutate: func(p *Persister) error {
		309	+	return p.Create(validRecord("new", 2))
		310	+	},
		311	+	},
		312	+	{
		313	+	name: "create sync failure",
		314	+	initial: map[string][]byte{},
		315	+	syncFailures: 1,
		316	+	mutate: func(p *Persister) error {
		317	+	return p.Create(validRecord("new", 2))
		318	+	},
		319	+	},
		320	+	{
		321	+	name: "replace set failure",
		322	+	initial: map[string][]byte{"work": one},
		323	+	setFailures: 1,
		324	+	mutate: func(p *Persister) error {
		325	+	next := validRecord("work", 1)
		326	+	next.Cwd = "/next"
		327	+	return p.Replace("work", next)
		328	+	},
		329	+	},
		330	+	{
		331	+	name: "replace sync failure",
		332	+	initial: map[string][]byte{"work": one},
		333	+	syncFailures: 1,
		334	+	mutate: func(p *Persister) error {
		335	+	next := validRecord("work", 1)
		336	+	next.Cwd = "/next"
		337	+	return p.Replace("work", next)
		338	+	},
		339	+	},
		340	+	{
		341	+	name: "rename set failure",
		342	+	initial: map[string][]byte{"work": one},
		343	+	setFailures: 1,
		344	+	mutate: func(p *Persister) error {
		345	+	next := validRecord("renamed", 1)
		346	+	return p.Rename("work", next)
		347	+	},
		348	+	},
		349	+	{
		350	+	name: "rename delete failure",
		351	+	initial: map[string][]byte{"work": one},
		352	+	deleteFailures: 1,
		353	+	mutate: func(p *Persister) error {
		354	+	next := validRecord("renamed", 1)
		355	+	return p.Rename("work", next)
		356	+	},
		357	+	},
		358	+	{
		359	+	name: "rename sync failure",
		360	+	initial: map[string][]byte{"work": one},
		361	+	syncFailures: 1,
		362	+	mutate: func(p *Persister) error {
		363	+	next := validRecord("renamed", 1)
		364	+	return p.Rename("work", next)
		365	+	},
		366	+	},
		367	+	{
		368	+	name: "delete delete failure",
		369	+	initial: map[string][]byte{"work": one},
		370	+	deleteFailures: 1,
		371	+	mutate: func(p *Persister) error {
		372	+	return p.Delete("work")
		373	+	},
		374	+	},
		375	+	{
		376	+	name: "delete sync failure",
		377	+	initial: map[string][]byte{"work": one},
		378	+	syncFailures: 1,
		379	+	mutate: func(p *Persister) error {
		380	+	return p.Delete("work")
		381	+	},
		382	+	},
		383	+	}
		384	+
		385	+	for _, tt := range tests {
		386	+	t.Run(tt.name, func(t *testing.T) {
		387	+	t.Parallel()
		388	+	store := newFailClosedStore(tt.initial)
		389	+	store.setFailures = tt.setFailures
		390	+	store.deleteFailures = tt.deleteFailures
		391	+	store.syncFailures = tt.syncFailures
		392	+	p := New(store)
		393	+
		394	+	err := tt.mutate(p)
		395	+	require.ErrorIs(t, err, ErrCatalogueDurability)
		396	+	require.ErrorIs(t, err, store.failure)
		397	+	require.Equal(t, tt.initial, store.data, "rejected identity keys must be restored in memory")
		398	+	require.Equal(t, tt.initial, store.durable, "rejected identity write must not reach disk")
		399	+
		400	+	_, _, recordErr := p.Record("work")
		401	+	require.ErrorIs(t, recordErr, ErrCatalogueDurability)
		402	+	require.ErrorIs(t, p.Sync(), ErrCatalogueDurability)
		403	+	require.ErrorIs(t, p.Create(validRecord("later", 3)), ErrCatalogueDurability)
		404	+	require.Equal(t, tt.initial, store.durable, "later Sync must be fenced")
		405	+
		406	+	require.ErrorIs(t, p.Close(), ErrCatalogueDurability)
		407	+	require.True(t, store.closeAborted, "Close must release without flushing rejected state")
		408	+	require.False(t, store.closeFlushed)
		409	+	require.Equal(t, tt.initial, store.durable, "Close must not make rejected state durable")
		410	+	})
		411	+	}
		412	+	}
		413	+
		414	+	func TestRejectedIdentityWriteRestoresBufferedMetadata(t *testing.T) {
		415	+	t.Parallel()
		416	+	encoded, err := encodeRecordValue(validRecord("work", 1))
		417	+	require.NoError(t, err)
		418	+	store := newFailClosedStore(map[string][]byte{"work": encoded})
		419	+	p := New(store)
		420	+
		421	+	cwd := "/buffered"
		422	+	require.NoError(t, p.UpdateMetadata(domain.CatalogueMetadataUpdate{
		423	+	Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: &cwd,
		424	+	}))
		425	+	buffered := copyStoreData(store.data)
		426	+	store.syncFailures = 1
		427	+
		428	+	require.ErrorIs(t, p.Rename("work", validRecord("renamed", 1)), ErrCatalogueDurability)
		429	+	require.Equal(t, buffered, store.data, "rollback must preserve already-buffered metadata on the affected key")
		430	+	require.Equal(t, map[string][]byte{"work": encoded}, store.durable)
		431	+	require.ErrorIs(t, p.Close(), ErrCatalogueDurability)
		432	+	require.Equal(t, map[string][]byte{"work": encoded}, store.durable)
		433	+	}

M internal/usecase/daemon/session_purge_control_test.go +4 -3


@@ -7,10 +7,11 @@ import (
7	7		"github.com/stretchr/testify/require"
8	8
9	9		"github.com/bnema/vev/internal/domain"
	10	+	"github.com/bnema/vev/internal/persist"
10	11		portsmocks "github.com/bnema/vev/internal/ports/mocks"
11	12		)
12	13
13		-	func TestStoppedPurgeMetadataFailureRemainsFencedForRetry(t *testing.T) {
	14	+	func TestStoppedPurgeMetadataFailureFencesCatalogue(t *testing.T) {
14	15		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
15	16		repository := &retryablePurgeRepository{}
16	17		WithSnapshotRepository(repository)(d)

@@ -30,6 +31,6 @@ func TestStoppedPurgeMetadataFailureRemainsFencedForRetry(t *testing.T) {
30	31		state.mu.Lock()
31	32		state.deleteErr = nil
32	33		state.mu.Unlock()
33		-	require.NoError(t, d.retryStoppedPurge("work"))
34		-	require.Equal(t, []string{"delete incarnation"}, repository.calls)
	34	+	require.ErrorIs(t, d.retryStoppedPurge("work"), persist.ErrCatalogueDurability)
	35	+	require.Empty(t, repository.calls, "a fenced catalogue must not retry deletion")
35	36		}

 	"github.com/stretchr/testify/require"
 	"github.com/bnema/vev/internal/domain"
+	"github.com/bnema/vev/internal/persist"
 	portsmocks "github.com/bnema/vev/internal/ports/mocks"
 )
-func TestStoppedPurgeMetadataFailureRemainsFencedForRetry(t *testing.T) {
+func TestStoppedPurgeMetadataFailureFencesCatalogue(t *testing.T) {
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
 	repository := &retryablePurgeRepository{}
 	WithSnapshotRepository(repository)(d)
 	state.mu.Lock()
 	state.deleteErr = nil
 	state.mu.Unlock()
-	require.NoError(t, d.retryStoppedPurge("work"))
+	require.ErrorIs(t, d.retryStoppedPurge("work"), persist.ErrCatalogueDurability)
-	require.Equal(t, []string{"delete incarnation"}, repository.calls)
+	require.Empty(t, repository.calls, "a fenced catalogue must not retry deletion")
 }

M internal/usecase/daemon/session_test.go +8 -7


@@ -1184,19 +1184,22 @@ func TestEphemeralPromotionLifecycleFailuresLeaveStateRollbackSafe(t *testing.T)
1184	1184		defer release()
1185	1185		store := portsmocks.NewMockStore(t)
1186	1186		var attempted map[string][]byte
1187		-	store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {}).Times(3)
1188		-	store.EXPECT().Get([]byte("named")).Return(nil, false).Times(4)
	1187	+	store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {}).Times(2)
	1188	+	store.EXPECT().Get([]byte("named")).Return(nil, false).Times(3)
1189	1189		store.EXPECT().Set(mock.Anything, mock.Anything).RunAndReturn(func(key, value []byte) error {
1190	1190		attempted = map[string][]byte{string(key): append([]byte(nil), value...)}
1191	1191		return errors.New("disk full")
1192	1192		}).Once()
	1193	+	store.EXPECT().Delete([]byte("named")).Return(nil).Once()
1193	1194		clock := &lifecycleClock{nows: []time.Time{time.Unix(0, 100), time.Unix(0, 100)}}
1194	1195		d := newTestDaemon(t, newFactory(t, p), clock)
1195	1196		WithStore(t, store)(d)
1196	1197
1197	1198		sess, err := createSessionForTest(d, "0", true, "/tmp", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
1198	1199		require.NoError(t, err)
1199		-	require.EqualError(t, d.renameSession(sess, "named"), "disk full")
	1200	+	err = d.renameSession(sess, "named")
	1201	+	require.ErrorIs(t, err, persist.ErrCatalogueDurability)
	1202	+	require.ErrorContains(t, err, "disk full")
1200	1203		require.Equal(t, "0", sess.name)
1201	1204		require.True(t, sess.ephemeral)
1202	1205		require.Zero(t, sess.createdAt)

@@ -1206,10 +1209,8 @@ func TestEphemeralPromotionLifecycleFailuresLeaveStateRollbackSafe(t *testing.T)
1206	1209		require.Len(t, records, 1)
1207	1210		require.Equal(t, int64(100), records[0].CreatedAt, "a promotion must never attempt to persist a zero lifecycle identity")
1208	1211
1209		-	store.EXPECT().Set(mock.Anything, mock.Anything).Return(nil).Once()
1210		-	store.EXPECT().Sync().Return(nil).Once()
1211		-	require.NoError(t, d.renameSession(sess, "named"))
1212		-	require.Equal(t, int64(101), sess.createdAt, "the high-water mark must remain monotonic after a failed promotion")
	1212	+	require.ErrorIs(t, d.renameSession(sess, "named"), persist.ErrCatalogueDurability)
	1213	+	require.Equal(t, int64(101), d.lastAllocatedCreatedAt, "the lifecycle identity is allocated before the fenced catalogue rejects promotion")
1213	1214		})
1214	1215		}
1215	1216

 		defer release()
 		store := portsmocks.NewMockStore(t)
 		var attempted map[string][]byte
-		store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {}).Times(3)
+		store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {}).Times(2)
-		store.EXPECT().Get([]byte("named")).Return(nil, false).Times(4)
+		store.EXPECT().Get([]byte("named")).Return(nil, false).Times(3)
 		store.EXPECT().Set(mock.Anything, mock.Anything).RunAndReturn(func(key, value []byte) error {
 			attempted = map[string][]byte{string(key): append([]byte(nil), value...)}
 			return errors.New("disk full")
 		}).Once()
+		store.EXPECT().Delete([]byte("named")).Return(nil).Once()
 		clock := &lifecycleClock{nows: []time.Time{time.Unix(0, 100), time.Unix(0, 100)}}
 		d := newTestDaemon(t, newFactory(t, p), clock)
 		WithStore(t, store)(d)
 		sess, err := createSessionForTest(d, "0", true, "/tmp", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
 		require.NoError(t, err)
-		require.EqualError(t, d.renameSession(sess, "named"), "disk full")
+		err = d.renameSession(sess, "named")
+		require.ErrorIs(t, err, persist.ErrCatalogueDurability)
+		require.ErrorContains(t, err, "disk full")
 		require.Equal(t, "0", sess.name)
 		require.True(t, sess.ephemeral)
 		require.Zero(t, sess.createdAt)
 		require.Len(t, records, 1)
 		require.Equal(t, int64(100), records[0].CreatedAt, "a promotion must never attempt to persist a zero lifecycle identity")
-		store.EXPECT().Set(mock.Anything, mock.Anything).Return(nil).Once()
+		require.ErrorIs(t, d.renameSession(sess, "named"), persist.ErrCatalogueDurability)
-		store.EXPECT().Sync().Return(nil).Once()
+		require.Equal(t, int64(101), d.lastAllocatedCreatedAt, "the lifecycle identity is allocated before the fenced catalogue rejects promotion")
-		require.NoError(t, d.renameSession(sess, "named"))
-		require.Equal(t, int64(101), sess.createdAt, "the high-water mark must remain monotonic after a failed promotion")
 	})
 }

M pkg/kv/kv.go +12 -0


@@ -158,6 +158,18 @@ func (s *Store) Close() error {
158	158		return errors.Join(err, releaseLock(s.lockFile))
159	159		}
160	160
	161	+	// CloseWithoutSync releases the process lock without making buffered mutations
	162	+	// durable. It is used when a higher-level transaction rejected those mutations.
	163	+	func (s *Store) CloseWithoutSync() error {
	164	+	s.mu.Lock()
	165	+	defer s.mu.Unlock()
	166	+	if s.closed {
	167	+	return nil
	168	+	}
	169	+	s.closed = true
	170	+	return releaseLock(s.lockFile)
	171	+	}
	172	+
161	173		func writeAll(file *os.File, data []byte) error {
162	174		for len(data) > 0 {
163	175		n, err := file.Write(data)


@@ -158,6 +158,18 @@ func (s *Store) Close() error {
158	return errors.Join(err, releaseLock(s.lockFile))	158		return errors.Join(err, releaseLock(s.lockFile))
159	}	159		}
160		160
		161	+	// CloseWithoutSync releases the process lock without making buffered mutations
		162	+	// durable. It is used when a higher-level transaction rejected those mutations.
		163	+	func (s *Store) CloseWithoutSync() error {
		164	+	s.mu.Lock()
		165	+	defer s.mu.Unlock()
		166	+	if s.closed {
		167	+	return nil
		168	+	}
		169	+	s.closed = true
		170	+	return releaseLock(s.lockFile)
		171	+	}
		172	+
161	func writeAll(file *os.File, data []byte) error {	173		func writeAll(file *os.File, data []byte) error {
162	for len(data) > 0 {	174		for len(data) > 0 {
163	n, err := file.Write(data)	175		n, err := file.Write(data)

M pkg/kv/kv_test.go +15 -0


@@ -190,3 +190,18 @@ func TestDecodeRejectsDuplicateKeysAndTrailingData(t *testing.T) {
190	190		_, err = decodeFile(valid)
191	191		require.True(t, errors.Is(err, ErrCorrupt))
192	192		}
	193	+
	194	+	func TestCloseWithoutSyncDiscardsDirtyStore(t *testing.T) {
	195	+	t.Parallel()
	196	+	path := privateStorePath(t)
	197	+	store, err := Open(path)
	198	+	require.NoError(t, err)
	199	+	require.NoError(t, store.Set([]byte("rejected"), []byte("value")))
	200	+	require.NoError(t, store.CloseWithoutSync())
	201	+
	202	+	reopened, err := Open(path)
	203	+	require.NoError(t, err)
	204	+	t.Cleanup(func() { _ = reopened.Close() })
	205	+	_, ok := reopened.Get([]byte("rejected"))
	206	+	require.False(t, ok)
	207	+	}


@@ -190,3 +190,18 @@ func TestDecodeRejectsDuplicateKeysAndTrailingData(t *testing.T) {
190	_, err = decodeFile(valid)	190		_, err = decodeFile(valid)
191	require.True(t, errors.Is(err, ErrCorrupt))	191		require.True(t, errors.Is(err, ErrCorrupt))
192	}	192		}
		193	+
		194	+	func TestCloseWithoutSyncDiscardsDirtyStore(t *testing.T) {
		195	+	t.Parallel()
		196	+	path := privateStorePath(t)
		197	+	store, err := Open(path)
		198	+	require.NoError(t, err)
		199	+	require.NoError(t, store.Set([]byte("rejected"), []byte("value")))
		200	+	require.NoError(t, store.CloseWithoutSync())
		201	+
		202	+	reopened, err := Open(path)
		203	+	require.NoError(t, err)
		204	+	t.Cleanup(func() { _ = reopened.Close() })
		205	+	_, ok := reopened.Get([]byte("rejected"))
		206	+	require.False(t, ok)
		207	+	}