836c233 fix(kv): abort fenced catalogue writes safely · vev

M internal/persist/persist.go +8 -14


@@ -256,14 +256,12 @@ func (p Persister) applyLocked(records map[string]domain.CatalogueRecord, dura
256	256		err = p.store.Set(change.key, change.value)
257	257		}
258	258		if err != nil {
259		-	p.restoreLocked(previous)
260		-	return p.fenceLocked(err)
	259	+	return p.fenceLocked(errors.Join(err, p.restoreLocked(previous)))
261	260		}
262	261		}
263	262		if durable {
264	263		if err := p.store.Sync(); err != nil {
265		-	p.restoreLocked(previous)
266		-	return p.fenceLocked(err)
	264	+	return p.fenceLocked(errors.Join(err, p.restoreLocked(previous)))
267	265		}
268	266		}
269	267		for name := range records {

@@ -288,14 +286,16 @@ type storedValue struct {
288	286
289	287		// restoreLocked restores only keys affected by the rejected identity mutation.
290	288		// It intentionally does not Sync: the Persister is fenced immediately after.
291		-	func (p *Persister) restoreLocked(previous map[string]storedValue) {
	289	+	func (p *Persister) restoreLocked(previous map[string]storedValue) error {
	290	+	var err error
292	291		for key, old := range previous {
293	292		if old.exists {
294		-	_ = p.store.Set([]byte(key), old.value)
	293	+	err = errors.Join(err, p.store.Set([]byte(key), old.value))
295	294		} else {
296		-	_ = p.store.Delete([]byte(key))
	295	+	err = errors.Join(err, p.store.Delete([]byte(key)))
297	296		}
298	297		}
	298	+	return err
299	299		}
300	300
301	301		func (p *Persister) fenceLocked(cause error) error {

@@ -445,13 +445,7 @@ func (p *Persister) Close() error {
445	445		p.mu.Lock()
446	446		defer p.mu.Unlock()
447	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 {
	448	+	if closeErr := p.store.CloseWithoutSync(); closeErr != nil {
455	449		return errors.Join(err, closeErr)
456	450		}
457	451		return err

 			err = p.store.Set(change.key, change.value)
 		}
 		if err != nil {
-			p.restoreLocked(previous)
+			return p.fenceLocked(errors.Join(err, p.restoreLocked(previous)))
-			return p.fenceLocked(err)
 		}
 	}
 	if durable {
 		if err := p.store.Sync(); err != nil {
-			p.restoreLocked(previous)
+			return p.fenceLocked(errors.Join(err, p.restoreLocked(previous)))
-			return p.fenceLocked(err)
 		}
 	}
 	for name := range records {
 // 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) {
+func (p *Persister) restoreLocked(previous map[string]storedValue) error {
+	var err error
 	for key, old := range previous {
 		if old.exists {
-			_ = p.store.Set([]byte(key), old.value)
+			err = errors.Join(err, p.store.Set([]byte(key), old.value))
 		} else {
-			_ = p.store.Delete([]byte(key))
+			err = errors.Join(err, p.store.Delete([]byte(key)))
 		}
 	}
+	return err
 }
 func (p *Persister) fenceLocked(cause error) error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	if err := p.terminalLocked(); err != nil {
-		if store, ok := p.store.(interface{ CloseWithoutSync() error }); ok {
+		if closeErr := p.store.CloseWithoutSync(); closeErr != nil {
-			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

M internal/persist/persist_test.go +51 -5


@@ -216,17 +216,26 @@ type failClosedStore struct {
216	216		data map[string][]byte
217	217		durable map[string][]byte
218	218
219		-	setFailures int
220		-	deleteFailures int
221		-	syncFailures int
222		-	failure error
	219	+	setFailures int
	220	+	deleteFailures int
	221	+	syncFailures int
	222	+	restoreSetFailures int
	223	+	restoreDeleteFailures int
	224	+	restoring bool
	225	+	failure error
	226	+	restoreFailure error
223	227
224	228		closeFlushed bool
225	229		closeAborted bool
226	230		}
227	231
228	232		func newFailClosedStore(data map[string][]byte) *failClosedStore {
229		-	return &failClosedStore{data: copyStoreData(data), durable: copyStoreData(data), failure: errors.New("store write failed")}
	233	+	return &failClosedStore{
	234	+	data: copyStoreData(data),
	235	+	durable: copyStoreData(data),
	236	+	failure: errors.New("store write failed"),
	237	+	restoreFailure: errors.New("store restore failed"),
	238	+	}
230	239		}
231	240
232	241		func (s *failClosedStore) Get(key []byte) ([]byte, bool) {

@@ -236,6 +245,10 @@ func (s *failClosedStore) Get(key []byte) ([]byte, bool) {
236	245
237	246		func (s *failClosedStore) Set(key, value []byte) error {
238	247		s.data[string(key)] = append([]byte(nil), value...)
	248	+	if s.restoring && s.restoreSetFailures > 0 {
	249	+	s.restoreSetFailures--
	250	+	return s.restoreFailure
	251	+	}
239	252		if s.setFailures > 0 {
240	253		s.setFailures--
241	254		return s.failure

@@ -245,6 +258,10 @@ func (s *failClosedStore) Set(key, value []byte) error {
245	258
246	259		func (s *failClosedStore) Delete(key []byte) error {
247	260		delete(s.data, string(key))
	261	+	if s.restoring && s.restoreDeleteFailures > 0 {
	262	+	s.restoreDeleteFailures--
	263	+	return s.restoreFailure
	264	+	}
248	265		if s.deleteFailures > 0 {
249	266		s.deleteFailures--
250	267		return s.failure

@@ -263,6 +280,7 @@ func (s *failClosedStore) Range(fn func([]byte, []byte) bool) {
263	280		func (s *failClosedStore) Sync() error {
264	281		if s.syncFailures > 0 {
265	282		s.syncFailures--
	283	+	s.restoring = true
266	284		return s.failure
267	285		}
268	286		s.durable = copyStoreData(s.data)

@@ -411,6 +429,34 @@ func TestRejectedIdentityWriteFencesCatalogue(t *testing.T) {
411	429		}
412	430		}
413	431
	432	+	func TestFailedSyncRetainsRepeatedRollbackFailuresAndAborts(t *testing.T) {
	433	+	t.Parallel()
	434	+	encoded, err := encodeRecordValue(validRecord("work", 1))
	435	+	require.NoError(t, err)
	436	+	store := newFailClosedStore(map[string][]byte{"work": encoded})
	437	+	store.syncFailures = 1
	438	+	store.restoreSetFailures = 1
	439	+	store.restoreDeleteFailures = 1
	440	+	p := New(store)
	441	+
	442	+	err = p.Rename("work", validRecord("renamed", 1))
	443	+	require.ErrorIs(t, err, ErrCatalogueDurability)
	444	+	require.ErrorIs(t, err, store.failure)
	445	+	require.ErrorIs(t, err, store.restoreFailure)
	446	+	require.Zero(t, store.restoreSetFailures, "rollback Set failure must be observed")
	447	+	require.Zero(t, store.restoreDeleteFailures, "rollback Delete failure must be observed")
	448	+
	449	+	terminal := p.Sync()
	450	+	require.Same(t, err, terminal, "the fenced terminal error must remain stable")
	451	+	require.ErrorIs(t, terminal, store.failure)
	452	+	require.ErrorIs(t, terminal, store.restoreFailure)
	453	+	require.ErrorIs(t, p.Close(), ErrCatalogueDurability)
	454	+	require.True(t, store.closeAborted)
	455	+	require.False(t, store.closeFlushed)
	456	+	require.Equal(t, map[string][]byte{"work": encoded}, store.durable,
	457	+	"aborting a fenced store must not persist dirty rejected state")
	458	+	}
	459	+
414	460		func TestRejectedIdentityWriteRestoresBufferedMetadata(t *testing.T) {
415	461		t.Parallel()
416	462		encoded, err := encodeRecordValue(validRecord("work", 1))


@@ -216,17 +216,26 @@ type failClosedStore struct {
216		data map[string][]byte	216		data map[string][]byte
217		durable map[string][]byte	217		durable map[string][]byte
218			218
219	-	setFailures int	219	+	setFailures int
220	-	deleteFailures int	220	+	deleteFailures int
221	-	syncFailures int	221	+	syncFailures int
222	-	failure error	222	+	restoreSetFailures int
			223	+	restoreDeleteFailures int
			224	+	restoring bool
			225	+	failure error
			226	+	restoreFailure error
223			227
224		closeFlushed bool	228		closeFlushed bool
225		closeAborted bool	229		closeAborted bool
226		}	230		}
227			231
228		func newFailClosedStore(data map[string][]byte) *failClosedStore {	232		func newFailClosedStore(data map[string][]byte) *failClosedStore {
229	-	return &failClosedStore{data: copyStoreData(data), durable: copyStoreData(data), failure: errors.New("store write failed")}	233	+	return &failClosedStore{
			234	+	data: copyStoreData(data),
			235	+	durable: copyStoreData(data),
			236	+	failure: errors.New("store write failed"),
			237	+	restoreFailure: errors.New("store restore failed"),
			238	+	}
230		}	239		}
231			240
232		func (s *failClosedStore) Get(key []byte) ([]byte, bool) {	241		func (s *failClosedStore) Get(key []byte) ([]byte, bool) {

@@ -236,6 +245,10 @@ func (s *failClosedStore) Get(key []byte) ([]byte, bool) {
236			245
237		func (s *failClosedStore) Set(key, value []byte) error {	246		func (s *failClosedStore) Set(key, value []byte) error {
238		s.data[string(key)] = append([]byte(nil), value...)	247		s.data[string(key)] = append([]byte(nil), value...)
			248	+	if s.restoring && s.restoreSetFailures > 0 {
			249	+	s.restoreSetFailures--
			250	+	return s.restoreFailure
			251	+	}
239		if s.setFailures > 0 {	252		if s.setFailures > 0 {
240		s.setFailures--	253		s.setFailures--
241		return s.failure	254		return s.failure

@@ -245,6 +258,10 @@ func (s *failClosedStore) Set(key, value []byte) error {
245			258
246		func (s *failClosedStore) Delete(key []byte) error {	259		func (s *failClosedStore) Delete(key []byte) error {
247		delete(s.data, string(key))	260		delete(s.data, string(key))
			261	+	if s.restoring && s.restoreDeleteFailures > 0 {
			262	+	s.restoreDeleteFailures--
			263	+	return s.restoreFailure
			264	+	}
248		if s.deleteFailures > 0 {	265		if s.deleteFailures > 0 {
249		s.deleteFailures--	266		s.deleteFailures--
250		return s.failure	267		return s.failure

@@ -263,6 +280,7 @@ func (s *failClosedStore) Range(fn func([]byte, []byte) bool) {
263		func (s *failClosedStore) Sync() error {	280		func (s *failClosedStore) Sync() error {
264		if s.syncFailures > 0 {	281		if s.syncFailures > 0 {
265		s.syncFailures--	282		s.syncFailures--
			283	+	s.restoring = true
266		return s.failure	284		return s.failure
267		}	285		}
268		s.durable = copyStoreData(s.data)	286		s.durable = copyStoreData(s.data)

@@ -411,6 +429,34 @@ func TestRejectedIdentityWriteFencesCatalogue(t *testing.T) {
411		}	429		}
412		}	430		}
413			431
			432	+	func TestFailedSyncRetainsRepeatedRollbackFailuresAndAborts(t *testing.T) {
			433	+	t.Parallel()
			434	+	encoded, err := encodeRecordValue(validRecord("work", 1))
			435	+	require.NoError(t, err)
			436	+	store := newFailClosedStore(map[string][]byte{"work": encoded})
			437	+	store.syncFailures = 1
			438	+	store.restoreSetFailures = 1
			439	+	store.restoreDeleteFailures = 1
			440	+	p := New(store)
			441	+
			442	+	err = p.Rename("work", validRecord("renamed", 1))
			443	+	require.ErrorIs(t, err, ErrCatalogueDurability)
			444	+	require.ErrorIs(t, err, store.failure)
			445	+	require.ErrorIs(t, err, store.restoreFailure)
			446	+	require.Zero(t, store.restoreSetFailures, "rollback Set failure must be observed")
			447	+	require.Zero(t, store.restoreDeleteFailures, "rollback Delete failure must be observed")
			448	+
			449	+	terminal := p.Sync()
			450	+	require.Same(t, err, terminal, "the fenced terminal error must remain stable")
			451	+	require.ErrorIs(t, terminal, store.failure)
			452	+	require.ErrorIs(t, terminal, store.restoreFailure)
			453	+	require.ErrorIs(t, p.Close(), ErrCatalogueDurability)
			454	+	require.True(t, store.closeAborted)
			455	+	require.False(t, store.closeFlushed)
			456	+	require.Equal(t, map[string][]byte{"work": encoded}, store.durable,
			457	+	"aborting a fenced store must not persist dirty rejected state")
			458	+	}
			459	+
414		func TestRejectedIdentityWriteRestoresBufferedMetadata(t *testing.T) {	460		func TestRejectedIdentityWriteRestoresBufferedMetadata(t *testing.T) {
415		t.Parallel()	461		t.Parallel()
416		encoded, err := encodeRecordValue(validRecord("work", 1))	462		encoded, err := encodeRecordValue(validRecord("work", 1))

M internal/ports/mocks/store_mock.go +44 -0


@@ -79,6 +79,50 @@ func (_c MockStore_Close_Call) RunAndReturn(run func() error) MockStore_Close_
79	79		return _c
80	80		}
81	81
	82	+	// CloseWithoutSync provides a mock function for the type MockStore
	83	+	func (_mock *MockStore) CloseWithoutSync() error {
	84	+	ret := _mock.Called()
	85	+
	86	+	if len(ret) == 0 {
	87	+	panic("no return value specified for CloseWithoutSync")
	88	+	}
	89	+
	90	+	var r0 error
	91	+	if returnFunc, ok := ret.Get(0).(func() error); ok {
	92	+	r0 = returnFunc()
	93	+	} else {
	94	+	r0 = ret.Error(0)
	95	+	}
	96	+	return r0
	97	+	}
	98	+
	99	+	// MockStore_CloseWithoutSync_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'CloseWithoutSync'
	100	+	type MockStore_CloseWithoutSync_Call struct {
	101	+	*mock.Call
	102	+	}
	103	+
	104	+	// CloseWithoutSync is a helper method to define mock.On call
	105	+	func (_e MockStore_Expecter) CloseWithoutSync() MockStore_CloseWithoutSync_Call {
	106	+	return &MockStore_CloseWithoutSync_Call{Call: _e.mock.On("CloseWithoutSync")}
	107	+	}
	108	+
	109	+	func (_c MockStore_CloseWithoutSync_Call) Run(run func()) MockStore_CloseWithoutSync_Call {
	110	+	_c.Call.Run(func(args mock.Arguments) {
	111	+	run()
	112	+	})
	113	+	return _c
	114	+	}
	115	+
	116	+	func (_c MockStore_CloseWithoutSync_Call) Return(err error) MockStore_CloseWithoutSync_Call {
	117	+	_c.Call.Return(err)
	118	+	return _c
	119	+	}
	120	+
	121	+	func (_c MockStore_CloseWithoutSync_Call) RunAndReturn(run func() error) MockStore_CloseWithoutSync_Call {
	122	+	_c.Call.Return(run)
	123	+	return _c
	124	+	}
	125	+
82	126		// Delete provides a mock function for the type MockStore
83	127		func (_mock *MockStore) Delete(key []byte) error {
84	128		ret := _mock.Called(key)


@@ -79,6 +79,50 @@ func (_c MockStore_Close_Call) RunAndReturn(run func() error) MockStore_Close_
79	return _c	79		return _c
80	}	80		}
81		81
		82	+	// CloseWithoutSync provides a mock function for the type MockStore
		83	+	func (_mock *MockStore) CloseWithoutSync() error {
		84	+	ret := _mock.Called()
		85	+
		86	+	if len(ret) == 0 {
		87	+	panic("no return value specified for CloseWithoutSync")
		88	+	}
		89	+
		90	+	var r0 error
		91	+	if returnFunc, ok := ret.Get(0).(func() error); ok {
		92	+	r0 = returnFunc()
		93	+	} else {
		94	+	r0 = ret.Error(0)
		95	+	}
		96	+	return r0
		97	+	}
		98	+
		99	+	// MockStore_CloseWithoutSync_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'CloseWithoutSync'
		100	+	type MockStore_CloseWithoutSync_Call struct {
		101	+	*mock.Call
		102	+	}
		103	+
		104	+	// CloseWithoutSync is a helper method to define mock.On call
		105	+	func (_e MockStore_Expecter) CloseWithoutSync() MockStore_CloseWithoutSync_Call {
		106	+	return &MockStore_CloseWithoutSync_Call{Call: _e.mock.On("CloseWithoutSync")}
		107	+	}
		108	+
		109	+	func (_c MockStore_CloseWithoutSync_Call) Run(run func()) MockStore_CloseWithoutSync_Call {
		110	+	_c.Call.Run(func(args mock.Arguments) {
		111	+	run()
		112	+	})
		113	+	return _c
		114	+	}
		115	+
		116	+	func (_c MockStore_CloseWithoutSync_Call) Return(err error) MockStore_CloseWithoutSync_Call {
		117	+	_c.Call.Return(err)
		118	+	return _c
		119	+	}
		120	+
		121	+	func (_c MockStore_CloseWithoutSync_Call) RunAndReturn(run func() error) MockStore_CloseWithoutSync_Call {
		122	+	_c.Call.Return(run)
		123	+	return _c
		124	+	}
		125	+
82	// Delete provides a mock function for the type MockStore	126		// Delete provides a mock function for the type MockStore
83	func (_mock *MockStore) Delete(key []byte) error {	127		func (_mock *MockStore) Delete(key []byte) error {
84	ret := _mock.Called(key)	128		ret := _mock.Called(key)

M internal/ports/store.go +3 -0


@@ -16,6 +16,9 @@ type Store interface {
16	16		// or mutate either slice after Range returns.
17	17		Range(fn func(k, v []byte) bool)
18	18		Sync() error
	19	+	// CloseWithoutSync releases the store without persisting buffered writes.
	20	+	// Callers use it after a rejected durable mutation.
	21	+	CloseWithoutSync() error
19	22		Close() error
20	23		}
21	24


@@ -16,6 +16,9 @@ type Store interface {
16	// or mutate either slice after Range returns.	16		// or mutate either slice after Range returns.
17	Range(fn func(k, v []byte) bool)	17		Range(fn func(k, v []byte) bool)
18	Sync() error	18		Sync() error
		19	+	// CloseWithoutSync releases the store without persisting buffered writes.
		20	+	// Callers use it after a rejected durable mutation.
		21	+	CloseWithoutSync() error
19	Close() error	22		Close() error
20	}	23		}
21		24

M internal/usecase/daemon/session_test.go +3 -2


@@ -787,8 +787,9 @@ func (s newStaticStore) Range(fn func(k, v []byte) bool) {
787	787		}
788	788		}
789	789		}
790		-	func (s newStaticStore) Sync() error { return nil }
791		-	func (s newStaticStore) Close() error { return nil }
	790	+	func (s newStaticStore) Sync() error { return nil }
	791	+	func (s newStaticStore) CloseWithoutSync() error { return nil }
	792	+	func (s newStaticStore) Close() error { return nil }
792	793
793	794		func TestNamedSessionCreationIsAllowedWithNilStore(t *testing.T) {
794	795		pty, release := newBlockingPTY(t)

 		}
 	}
 }
-func (s newStaticStore) Sync() error  { return nil }
+func (s newStaticStore) Sync() error             { return nil }
-func (s newStaticStore) Close() error { return nil }
+func (s newStaticStore) CloseWithoutSync() error { return nil }
+func (s newStaticStore) Close() error            { return nil }
 func TestNamedSessionCreationIsAllowedWithNilStore(t *testing.T) {
 	pty, release := newBlockingPTY(t)

M pkg/kv/kv_test.go +2 -1


@@ -191,13 +191,14 @@ func TestDecodeRejectsDuplicateKeysAndTrailingData(t *testing.T) {
191	191		require.True(t, errors.Is(err, ErrCorrupt))
192	192		}
193	193
194		-	func TestCloseWithoutSyncDiscardsDirtyStore(t *testing.T) {
	194	+	func TestCloseWithoutSyncPreventsLaterCloseFromSyncing(t *testing.T) {
195	195		t.Parallel()
196	196		path := privateStorePath(t)
197	197		store, err := Open(path)
198	198		require.NoError(t, err)
199	199		require.NoError(t, store.Set([]byte("rejected"), []byte("value")))
200	200		require.NoError(t, store.CloseWithoutSync())
	201	+	require.NoError(t, store.Close(), "a later normal Close must remain a no-op")
201	202
202	203		reopened, err := Open(path)
203	204		require.NoError(t, err)

 	require.True(t, errors.Is(err, ErrCorrupt))
 }
-func TestCloseWithoutSyncDiscardsDirtyStore(t *testing.T) {
+func TestCloseWithoutSyncPreventsLaterCloseFromSyncing(t *testing.T) {
 	t.Parallel()
 	path := privateStorePath(t)
 	store, err := Open(path)
 	require.NoError(t, err)
 	require.NoError(t, store.Set([]byte("rejected"), []byte("value")))
 	require.NoError(t, store.CloseWithoutSync())
+	require.NoError(t, store.Close(), "a later normal Close must remain a no-op")
 	reopened, err := Open(path)
 	require.NoError(t, err)