test(persistence): cover compaction failure boundaries

commit: 3faaee9126d085b9873aac84661bdc404f72fd71
author: brice · 2mo ago · 2026-07-25 20:08 UTC
parent: d6f7eb5
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

7 files · +231 -41

M internal/app/offline_kill_retry_test.go +4 -4


@@ -7,6 +7,7 @@ import (
7	7		"testing"
8	8		"time"
9	9
	10	+	"github.com/bnema/vev/internal/domain"
10	11		"github.com/bnema/vev/internal/persist"
11	12		"github.com/stretchr/testify/require"
12	13		)

@@ -54,10 +55,9 @@ func TestRunKillOfflineRetriesSnapshotDeletion(t *testing.T) {
54	55		t.Setenv("XDG_STATE_HOME", stateRoot)
55	56		t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
56	57
57		-	p, err := persist.Open(filepath.Join(stateRoot, "vev"))
58		-	require.NoError(t, err)
	58	+	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
59	59		now := time.Now().UnixNano()
60		-	require.NoError(t, p.Save(persist.Record{Name: "named", Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now}))
	60	+	require.NoError(t, p.Save(persist.Record{Name: "named", IncarnationID: domain.IncarnationID{1}, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}))
61	61		require.NoError(t, p.Close())
62	62
63	63		fake := &offlineSnapshotFake{deleteErr: tt.incrementalErr, deleteLegacyErr: tt.legacyErr}

@@ -65,7 +65,7 @@ func TestRunKillOfflineRetriesSnapshotDeletion(t *testing.T) {
65	65		newOfflineSnapshotRepository = func(string) offlineSnapshotSource { return fake }
66	66		t.Cleanup(func() { newOfflineSnapshotRepository = originalFactory })
67	67
68		-	err = runKill(context.Background(), "named", false, false)
	68	+	err := runKill(context.Background(), "named", false, false)
69	69		if tt.incrementalErr != nil {
70	70		require.ErrorIs(t, err, tt.incrementalErr)
71	71		}

 	"testing"
 	"time"
+	"github.com/bnema/vev/internal/domain"
 	"github.com/bnema/vev/internal/persist"
 	"github.com/stretchr/testify/require"
 )
 			t.Setenv("XDG_STATE_HOME", stateRoot)
 			t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
-			p, err := persist.Open(filepath.Join(stateRoot, "vev"))
+			p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
-			require.NoError(t, err)
 			now := time.Now().UnixNano()
-			require.NoError(t, p.Save(persist.Record{Name: "named", Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now}))
+			require.NoError(t, p.Save(persist.Record{Name: "named", IncarnationID: domain.IncarnationID{1}, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}))
 			require.NoError(t, p.Close())
 			fake := &offlineSnapshotFake{deleteErr: tt.incrementalErr, deleteLegacyErr: tt.legacyErr}
 			newOfflineSnapshotRepository = func(string) offlineSnapshotSource { return fake }
 			t.Cleanup(func() { newOfflineSnapshotRepository = originalFactory })
-			err = runKill(context.Background(), "named", false, false)
+			err := runKill(context.Background(), "named", false, false)
 			if tt.incrementalErr != nil {
 				require.ErrorIs(t, err, tt.incrementalErr)
 			}

M internal/app/run_test.go +11 -10


@@ -429,17 +429,21 @@ func TestPrintSessionsMarksEphemeral(t *testing.T) {
429	429		}
430	430		}
431	431
	432	+	func newTestPersister(t testing.T, stateDir string) persist.Persister {
	433	+	t.Helper()
	434	+	store, err := persist.OpenStore(persist.StorePath(stateDir))
	435	+	require.NoError(t, err)
	436	+	return persist.New(store)
	437	+	}
	438	+
432	439		func TestRunListReadsStoppedSessionsWithoutDaemon(t *testing.T) {
433	440		stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
434	441		t.Setenv("XDG_STATE_HOME", stateRoot)
435	442		t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
436	443
437		-	p, err := persist.Open(filepath.Join(stateRoot, "vev"))
438		-	if err != nil {
439		-	t.Fatalf("persist.Open error = %v", err)
440		-	}
	444	+	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
441	445		now := time.Now().UnixNano()
442		-	if err := p.Save(persist.Record{Name: "stored", Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now}); err != nil {
	446	+	if err := p.Save(persist.Record{Name: "stored", IncarnationID: domain.IncarnationID{1}, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}); err != nil {
443	447		t.Fatalf("Save error = %v", err)
444	448		}
445	449		if err := p.Close(); err != nil {

@@ -477,12 +481,9 @@ func TestRunKillDeletesStoppedSessionWithoutDaemon(t *testing.T) {
477	481		t.Setenv("XDG_STATE_HOME", stateRoot)
478	482		t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
479	483
480		-	p, err := persist.Open(filepath.Join(stateRoot, "vev"))
481		-	if err != nil {
482		-	t.Fatalf("persist.Open error = %v", err)
483		-	}
	484	+	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
484	485		now := time.Now().UnixNano()
485		-	if err := p.Save(persist.Record{Name: "stored", Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now}); err != nil {
	486	+	if err := p.Save(persist.Record{Name: "stored", IncarnationID: domain.IncarnationID{1}, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}); err != nil {
486	487		t.Fatalf("Save error = %v", err)
487	488		}
488	489		if err := p.Close(); err != nil {

 	}
 }
+func newTestPersister(t *testing.T, stateDir string) *persist.Persister {
+	t.Helper()
+	store, err := persist.OpenStore(persist.StorePath(stateDir))
+	require.NoError(t, err)
+	return persist.New(store)
+}
 func TestRunListReadsStoppedSessionsWithoutDaemon(t *testing.T) {
 	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
 	t.Setenv("XDG_STATE_HOME", stateRoot)
 	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
-	p, err := persist.Open(filepath.Join(stateRoot, "vev"))
+	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
-	if err != nil {
-		t.Fatalf("persist.Open error = %v", err)
-	}
 	now := time.Now().UnixNano()
-	if err := p.Save(persist.Record{Name: "stored", Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now}); err != nil {
+	if err := p.Save(persist.Record{Name: "stored", IncarnationID: domain.IncarnationID{1}, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}); err != nil {
 		t.Fatalf("Save error = %v", err)
 	}
 	if err := p.Close(); err != nil {
 	t.Setenv("XDG_STATE_HOME", stateRoot)
 	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
-	p, err := persist.Open(filepath.Join(stateRoot, "vev"))
+	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
-	if err != nil {
-		t.Fatalf("persist.Open error = %v", err)
-	}
 	now := time.Now().UnixNano()
-	if err := p.Save(persist.Record{Name: "stored", Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now}); err != nil {
+	if err := p.Save(persist.Record{Name: "stored", IncarnationID: domain.IncarnationID{1}, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}); err != nil {
 		t.Fatalf("Save error = %v", err)
 	}
 	if err := p.Close(); err != nil {

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


@@ -102,7 +102,7 @@ func LoadCatalogueReadOnly(dir string) ([]domain.CatalogueRecord, error) {
102	102		func (p *Persister) Save(record domain.CatalogueRecord) error { return p.Replace(record.Name, record) }
103	103		func (p Persister) Apply(records map[string]domain.CatalogueRecord) error {
104	104		if p == nil \|\| p.store == nil {
105		-	return errPersistenceUnavailable
	105	+	return nil
106	106		}
107	107		p.mu.Lock()
108	108		defer p.mu.Unlock()

@@ -165,6 +165,9 @@ func (p Persister) applyLocked(records map[string]domain.CatalogueRecord) erro
165	165		return p.store.Sync()
166	166		}
167	167		func (p *Persister) Rename(oldName string, next domain.CatalogueRecord) error {
	168	+	if p == nil \|\| p.store == nil {
	169	+	return nil
	170	+	}
168	171		if oldName == next.Name {
169	172		return errors.New("persist: rename requires distinct names")
170	173		}

@@ -177,7 +180,7 @@ func (p *Persister) Replace(name string, next domain.CatalogueRecord) error {
177	180		// Touch updates metadata while retaining the durable identity and recovery state.
178	181		func (p *Persister) Touch(name, cwd string, at int64) error {
179	182		if p == nil \|\| p.store == nil {
180		-	return errPersistenceUnavailable
	183	+	return nil
181	184		}
182	185		p.mu.Lock()
183	186		defer p.mu.Unlock()

@@ -198,7 +201,7 @@ func (p *Persister) Touch(name, cwd string, at int64) error {
198	201		}
199	202		func (p *Persister) TouchMRU(name string, sequence uint64) error {
200	203		if p == nil \|\| p.store == nil {
201		-	return errPersistenceUnavailable
	204	+	return nil
202	205		}
203	206		p.mu.Lock()
204	207		defer p.mu.Unlock()

@@ -223,7 +226,7 @@ func (p *Persister) Delete(name string) error {
223	226		func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }
224	227		func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) {
225	228		if p == nil \|\| p.store == nil {
226		-	return nil, errPersistenceUnavailable
	229	+	return []domain.CatalogueRecord{}, nil
227	230		}
228	231		p.mu.Lock()
229	232		defer p.mu.Unlock()

@@ -233,7 +236,7 @@ func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) {
233	236		}
234	237		func (p *Persister) Close() error {
235	238		if p == nil \|\| p.store == nil {
236		-	return errPersistenceUnavailable
	239	+	return nil
237	240		}
238	241		p.mu.Lock()
239	242		defer p.mu.Unlock()

 func (p *Persister) Save(record domain.CatalogueRecord) error { return p.Replace(record.Name, record) }
 func (p *Persister) Apply(records map[string]*domain.CatalogueRecord) error {
 	if p == nil || p.store == nil {
-		return errPersistenceUnavailable
+		return nil
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	return p.store.Sync()
 }
 func (p *Persister) Rename(oldName string, next domain.CatalogueRecord) error {
+	if p == nil || p.store == nil {
+		return nil
+	}
 	if oldName == next.Name {
 		return errors.New("persist: rename requires distinct names")
 	}
 // Touch updates metadata while retaining the durable identity and recovery state.
 func (p *Persister) Touch(name, cwd string, at int64) error {
 	if p == nil || p.store == nil {
-		return errPersistenceUnavailable
+		return nil
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
 }
 func (p *Persister) TouchMRU(name string, sequence uint64) error {
 	if p == nil || p.store == nil {
-		return errPersistenceUnavailable
+		return nil
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
 func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }
 func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) {
 	if p == nil || p.store == nil {
-		return nil, errPersistenceUnavailable
+		return []domain.CatalogueRecord{}, nil
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
 }
 func (p *Persister) Close() error {
 	if p == nil || p.store == nil {
-		return errPersistenceUnavailable
+		return nil
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()

M internal/persist/persist_test.go +13 -6


@@ -64,10 +64,17 @@ func TestCatalogueBatchFailureDoesNotSync(t *testing.T) {
64	64		require.ErrorIs(t, err, batchErr)
65	65		}
66	66
67		-	func TestCatalogueNilPersisterFailsClosed(t *testing.T) {
68		-	var p *Persister
69		-	require.ErrorIs(t, p.Save(validRecord("one", 1)), errPersistenceUnavailable)
70		-	_, err := p.LoadAll()
71		-	require.ErrorIs(t, err, errPersistenceUnavailable)
72		-	require.ErrorIs(t, p.Close(), errPersistenceUnavailable)
	67	+	func TestCatalogueNilStoreIsNoOp(t *testing.T) {
	68	+	persisters := []*Persister{nil, New(nil)}
	69	+	for _, p := range persisters {
	70	+	require.NoError(t, p.Save(validRecord("one", 1)))
	71	+	require.NoError(t, p.Apply(map[string]*domain.CatalogueRecord{"one": nil}))
	72	+	require.NoError(t, p.Touch("one", "/tmp", 1))
	73	+	require.NoError(t, p.TouchMRU("one", 1))
	74	+	require.NoError(t, p.Delete("one"))
	75	+	records, err := p.LoadAll()
	76	+	require.NoError(t, err)
	77	+	require.Empty(t, records)
	78	+	require.NoError(t, p.Close())
	79	+	}
73	80		}

 	require.ErrorIs(t, err, batchErr)
 }
-func TestCatalogueNilPersisterFailsClosed(t *testing.T) {
+func TestCatalogueNilStoreIsNoOp(t *testing.T) {
-	var p *Persister
+	persisters := []*Persister{nil, New(nil)}
-	require.ErrorIs(t, p.Save(validRecord("one", 1)), errPersistenceUnavailable)
+	for _, p := range persisters {
-	_, err := p.LoadAll()
+		require.NoError(t, p.Save(validRecord("one", 1)))
-	require.ErrorIs(t, err, errPersistenceUnavailable)
+		require.NoError(t, p.Apply(map[string]*domain.CatalogueRecord{"one": nil}))
-	require.ErrorIs(t, p.Close(), errPersistenceUnavailable)
+		require.NoError(t, p.Touch("one", "/tmp", 1))
+		require.NoError(t, p.TouchMRU("one", 1))
+		require.NoError(t, p.Delete("one"))
+		records, err := p.LoadAll()
+		require.NoError(t, err)
+		require.Empty(t, records)
+		require.NoError(t, p.Close())
+	}
 }

M internal/usecase/daemon/session_test.go +24 -1


@@ -703,22 +703,33 @@ func newMockStore(t testing.T) (portsmocks.MockStore, *mockStoreState) {
703	703		}).Maybe()
704	704		store.EXPECT().Batch(mock.Anything).RunAndReturn(func(changes []ports.StoreChange) error {
705	705		state.mu.Lock()
706		-	defer state.mu.Unlock()
	706	+	deleted := false
707	707		for _, change := range changes {
708	708		key := string(change.Key)
709	709		if change.Delete {
710	710		state.dels = append(state.dels, key)
711	711		if state.deleteErr != nil {
712	712		if err := state.deleteErr(key); err != nil {
	713	+	state.mu.Unlock()
713	714		return err
714	715		}
715	716		}
716	717		delete(state.data, key)
	718	+	deleted = true
717	719		continue
718	720		}
719	721		state.sets++
720	722		state.data[key] = append([]byte(nil), change.Value...)
721	723		}
	724	+	var deleteDone chan struct{}
	725	+	if deleted {
	726	+	deleteDone = state.deleteDone
	727	+	state.deleteDone = nil
	728	+	}
	729	+	state.mu.Unlock()
	730	+	if deleteDone != nil {
	731	+	close(deleteDone)
	732	+	}
722	733		return nil
723	734		}).Maybe()
724	735		store.EXPECT().Delete(mock.Anything).RunAndReturn(func(k []byte) error {

@@ -814,6 +825,18 @@ func (s newStaticStore) Range(fn func(k, v []byte) bool) {
814	825		func (s newStaticStore) Sync() error { return nil }
815	826		func (s newStaticStore) Close() error { return nil }
816	827
	828	+	func TestNamedSessionCreationIsAllowedWithNilStore(t *testing.T) {
	829	+	pty, release := newBlockingPTY(t)
	830	+	defer release()
	831	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
	832	+	WithStore(nil)(d)
	833	+
	834	+	sess, err := d.createSessionLocked("work", false, "/tmp/work", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
	835	+	require.NoError(t, err)
	836	+	require.Equal(t, "work", sess.name)
	837	+	require.False(t, d.persistEnabled)
	838	+	}
	839	+
817	840		func TestDaemonLoadsPersistedSessionsAsStopped(t *testing.T) {
818	841		store, _ := newMockStore(t)
819	842		seed := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithStore(store))

 	}).Maybe()
 	store.EXPECT().Batch(mock.Anything).RunAndReturn(func(changes []ports.StoreChange) error {
 		state.mu.Lock()
-		defer state.mu.Unlock()
+		deleted := false
 		for _, change := range changes {
 			key := string(change.Key)
 			if change.Delete {
 				state.dels = append(state.dels, key)
 				if state.deleteErr != nil {
 					if err := state.deleteErr(key); err != nil {
+						state.mu.Unlock()
 						return err
 					}
 				}
 				delete(state.data, key)
+				deleted = true
 				continue
 			}
 			state.sets++
 			state.data[key] = append([]byte(nil), change.Value...)
 		}
+		var deleteDone chan struct{}
+		if deleted {
+			deleteDone = state.deleteDone
+			state.deleteDone = nil
+		}
+		state.mu.Unlock()
+		if deleteDone != nil {
+			close(deleteDone)
+		}
 		return nil
 	}).Maybe()
 	store.EXPECT().Delete(mock.Anything).RunAndReturn(func(k []byte) error {
 func (s newStaticStore) Sync() error  { return nil }
 func (s newStaticStore) Close() error { return nil }
+func TestNamedSessionCreationIsAllowedWithNilStore(t *testing.T) {
+	pty, release := newBlockingPTY(t)
+	defer release()
+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
+	WithStore(nil)(d)
+	sess, err := d.createSessionLocked("work", false, "/tmp/work", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
+	require.NoError(t, err)
+	require.Equal(t, "work", sess.name)
+	require.False(t, d.persistEnabled)
+}
 func TestDaemonLoadsPersistedSessionsAsStopped(t *testing.T) {
 	store, _ := newMockStore(t)
 	seed := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithStore(store))

M pkg/kv/kv.go +26 -15


@@ -27,6 +27,9 @@ type ReplayResult struct {
27	27
28	28		type fsHooks struct {
29	29		Open func(string, int, os.FileMode) (*os.File, error)
	30	+	Write func(*os.File, []byte) (int, error)
	31	+	Sync func(*os.File) error
	32	+	Close func(*os.File) error
30	33		Remove func(string) error
31	34		Rename func(string, string) error
32	35		SyncDir func(string) error

@@ -34,7 +37,15 @@ type fsHooks struct {
34	37		}
35	38
36	39		func defaultFSHooks() fsHooks {
37		-	return fsHooks{Open: os.OpenFile, Remove: os.Remove, Rename: os.Rename, SyncDir: syncDir}
	40	+	return fsHooks{
	41	+	Open: os.OpenFile,
	42	+	Write: func(f *os.File, p []byte) (int, error) { return f.Write(p) },
	43	+	Sync: func(f *os.File) error { return f.Sync() },
	44	+	Close: func(f *os.File) error { return f.Close() },
	45	+	Remove: os.Remove,
	46	+	Rename: os.Rename,
	47	+	SyncDir: syncDir,
	48	+	}
38	49		}
39	50
40	51		// Store is an append-only WAL-backed key/value store.

@@ -263,11 +274,11 @@ func (s *Store) compactLocked() (retErr error) {
263	274		for _, key := range keys {
264	275		buf, err := encodeRecord(opSet, []byte(key), s.data[key])
265	276		if err != nil {
266		-	_ = f.Close()
	277	+	_ = s.hooks.Close(f)
267	278		return err
268	279		}
269		-	if n, err := f.Write(buf); err != nil \|\| n != len(buf) {
270		-	_ = f.Close()
	280	+	if n, err := s.hooks.Write(f, buf); err != nil \|\| n != len(buf) {
	281	+	_ = s.hooks.Close(f)
271	282		if err == nil {
272	283		err = io.ErrShortWrite
273	284		}

@@ -275,22 +286,22 @@ func (s *Store) compactLocked() (retErr error) {
275	286		}
276	287		}
277	288		if err := injectFault(s.hooks, "after-write-next"); err != nil {
278		-	_ = f.Close()
	289	+	_ = s.hooks.Close(f)
279	290		return err
280	291		}
281	292		if err := injectFault(s.hooks, "before-sync-next"); err != nil {
282		-	_ = f.Close()
	293	+	_ = s.hooks.Close(f)
283	294		return err
284	295		}
285		-	if err := f.Sync(); err != nil {
286		-	_ = f.Close()
	296	+	if err := s.hooks.Sync(f); err != nil {
	297	+	_ = s.hooks.Close(f)
287	298		return err
288	299		}
289	300		if err := injectFault(s.hooks, "after-sync-next"); err != nil {
290		-	_ = f.Close()
	301	+	_ = s.hooks.Close(f)
291	302		return err
292	303		}
293		-	if err := f.Close(); err != nil {
	304	+	if err := s.hooks.Close(f); err != nil {
294	305		return err
295	306		}
296	307		if err := validatePath(next); err != nil {

@@ -341,25 +352,25 @@ func (s *Store) compactLocked() (retErr error) {
341	352		return err
342	353		}
343	354		if err := injectFault(s.hooks, "after-reopen-current"); err != nil {
344		-	_ = current.Close()
	355	+	_ = s.hooks.Close(current)
345	356		return err
346	357		}
347	358		result, err := replayFile(current)
348	359		if err != nil \|\| result.TornTail {
349		-	_ = current.Close()
	360	+	_ = s.hooks.Close(current)
350	361		if err == nil {
351	362		err = ErrCorruptWAL
352	363		}
353	364		return err
354	365		}
355	366		if _, err := current.Seek(0, io.SeekEnd); err != nil {
356		-	_ = current.Close()
	367	+	_ = s.hooks.Close(current)
357	368		return err
358	369		}
359	370		old := s.file
360	371		s.file = current
361	372		s.data, s.entrySize, s.total, s.live = result.Data, result.EntrySize, result.Total, result.Live
362		-	if err := old.Close(); err != nil {
	373	+	if err := s.hooks.Close(old); err != nil {
363	374		return err
364	375		}
365	376		if err := removeIfExists(s.hooks, prev); err != nil {

@@ -386,7 +397,7 @@ func (s *Store) poisonLocked() error {
386	397		s.file, s.lockFile = nil, nil
387	398		var fileErr error
388	399		if file != nil {
389		-	fileErr = file.Close()
	400	+	fileErr = s.hooks.Close(file)
390	401		}
391	402		return errors.Join(fileErr, releaseLock(lockFile))
392	403		}

 type fsHooks struct {
 	Open    func(string, int, os.FileMode) (*os.File, error)
+	Write   func(*os.File, []byte) (int, error)
+	Sync    func(*os.File) error
+	Close   func(*os.File) error
 	Remove  func(string) error
 	Rename  func(string, string) error
 	SyncDir func(string) error
 }
 func defaultFSHooks() fsHooks {
-	return fsHooks{Open: os.OpenFile, Remove: os.Remove, Rename: os.Rename, SyncDir: syncDir}
+	return fsHooks{
+		Open:    os.OpenFile,
+		Write:   func(f *os.File, p []byte) (int, error) { return f.Write(p) },
+		Sync:    func(f *os.File) error { return f.Sync() },
+		Close:   func(f *os.File) error { return f.Close() },
+		Remove:  os.Remove,
+		Rename:  os.Rename,
+		SyncDir: syncDir,
+	}
 }
 // Store is an append-only WAL-backed key/value store.
 	for _, key := range keys {
 		buf, err := encodeRecord(opSet, []byte(key), s.data[key])
 		if err != nil {
-			_ = f.Close()
+			_ = s.hooks.Close(f)
 			return err
 		}
-		if n, err := f.Write(buf); err != nil || n != len(buf) {
+		if n, err := s.hooks.Write(f, buf); err != nil || n != len(buf) {
-			_ = f.Close()
+			_ = s.hooks.Close(f)
 			if err == nil {
 				err = io.ErrShortWrite
 			}
 		}
 	}
 	if err := injectFault(s.hooks, "after-write-next"); err != nil {
-		_ = f.Close()
+		_ = s.hooks.Close(f)
 		return err
 	}
 	if err := injectFault(s.hooks, "before-sync-next"); err != nil {
-		_ = f.Close()
+		_ = s.hooks.Close(f)
 		return err
 	}
-	if err := f.Sync(); err != nil {
+	if err := s.hooks.Sync(f); err != nil {
-		_ = f.Close()
+		_ = s.hooks.Close(f)
 		return err
 	}
 	if err := injectFault(s.hooks, "after-sync-next"); err != nil {
-		_ = f.Close()
+		_ = s.hooks.Close(f)
 		return err
 	}
-	if err := f.Close(); err != nil {
+	if err := s.hooks.Close(f); err != nil {
 		return err
 	}
 	if err := validatePath(next); err != nil {
 		return err
 	}
 	if err := injectFault(s.hooks, "after-reopen-current"); err != nil {
-		_ = current.Close()
+		_ = s.hooks.Close(current)
 		return err
 	}
 	result, err := replayFile(current)
 	if err != nil || result.TornTail {
-		_ = current.Close()
+		_ = s.hooks.Close(current)
 		if err == nil {
 			err = ErrCorruptWAL
 		}
 		return err
 	}
 	if _, err := current.Seek(0, io.SeekEnd); err != nil {
-		_ = current.Close()
+		_ = s.hooks.Close(current)
 		return err
 	}
 	old := s.file
 	s.file = current
 	s.data, s.entrySize, s.total, s.live = result.Data, result.EntrySize, result.Total, result.Live
-	if err := old.Close(); err != nil {
+	if err := s.hooks.Close(old); err != nil {
 		return err
 	}
 	if err := removeIfExists(s.hooks, prev); err != nil {
 	s.file, s.lockFile = nil, nil
 	var fileErr error
 	if file != nil {
-		fileErr = file.Close()
+		fileErr = s.hooks.Close(file)
 	}
 	return errors.Join(fileErr, releaseLock(lockFile))
 }

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


@@ -416,6 +416,151 @@ func TestCompactionRecoveryMatrix(t *testing.T) {
416	416		}
417	417		}
418	418
	419	+	func TestCompactionFilesystemOperationFailuresPoisonAndRecover(t *testing.T) {
	420	+	fault := errors.New("injected filesystem failure")
	421	+	tests := []struct {
	422	+	name string
	423	+	configure func(fsHooks, string) fsHooks
	424	+	}{
	425	+	{name: "write-next", configure: func(h fsHooks, _ string) fsHooks {
	426	+	h.Write = func(*os.File, []byte) (int, error) { return 0, fault }
	427	+	return h
	428	+	}},
	429	+	{name: "sync-next", configure: func(h fsHooks, _ string) fsHooks {
	430	+	h.Sync = func(*os.File) error { return fault }
	431	+	return h
	432	+	}},
	433	+	{name: "close-next", configure: func(h fsHooks, _ string) fsHooks {
	434	+	calls := 0
	435	+	h.Close = func(f *os.File) error {
	436	+	calls++
	437	+	if calls == 1 {
	438	+	return errors.Join(f.Close(), fault)
	439	+	}
	440	+	return f.Close()
	441	+	}
	442	+	return h
	443	+	}},
	444	+	{name: "remove-stale-prev", configure: failRemoveCall(fault, 1)},
	445	+	{name: "rename-current-prev", configure: failRenameCall(fault, 1)},
	446	+	{name: "sync-dir-next", configure: failSyncDirCall(fault, 1)},
	447	+	{name: "sync-dir-prev", configure: failSyncDirCall(fault, 2)},
	448	+	{name: "rename-next-current", configure: failRenameCall(fault, 2)},
	449	+	{name: "sync-dir-current", configure: failSyncDirCall(fault, 3)},
	450	+	{name: "reopen-current", configure: func(h fsHooks, path string) fsHooks {
	451	+	open := h.Open
	452	+	h.Open = func(name string, flags int, mode os.FileMode) (*os.File, error) {
	453	+	if name == path && flags == os.O_RDWR {
	454	+	return nil, fault
	455	+	}
	456	+	return open(name, flags, mode)
	457	+	}
	458	+	return h
	459	+	}},
	460	+	{name: "close-old-current", configure: func(h fsHooks, _ string) fsHooks {
	461	+	calls := 0
	462	+	h.Close = func(f *os.File) error {
	463	+	calls++
	464	+	if calls == 2 {
	465	+	return errors.Join(f.Close(), fault)
	466	+	}
	467	+	return f.Close()
	468	+	}
	469	+	return h
	470	+	}},
	471	+	{name: "remove-prev", configure: failRemoveCall(fault, 2)},
	472	+	{name: "sync-dir-final", configure: failSyncDirCall(fault, 4)},
	473	+	}
	474	+	oldRecord, err := encodeRecord(opSet, []byte("version"), []byte("old"))
	475	+	if err != nil {
	476	+	t.Fatal(err)
	477	+	}
	478	+	for _, tt := range tests {
	479	+	t.Run(tt.name, func(t *testing.T) {
	480	+	path := privatePath(t)
	481	+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
	482	+	t.Fatal(err)
	483	+	}
	484	+	if err := os.WriteFile(path, oldRecord, 0o600); err != nil {
	485	+	t.Fatal(err)
	486	+	}
	487	+	s, err := Open(path)
	488	+	if err != nil {
	489	+	t.Fatal(err)
	490	+	}
	491	+	s.data = map[string][]byte{"version": []byte("new")}
	492	+	s.hooks = tt.configure(s.hooks, path)
	493	+
	494	+	err = s.compactLocked()
	495	+	if !errors.Is(err, fault) {
	496	+	t.Fatalf("compaction error = %v, want injected fault", err)
	497	+	}
	498	+	beforeAppend := candidateBytes(t, path)
	499	+	if err := s.Set([]byte("later"), []byte("must-not-append")); !errors.Is(err, os.ErrClosed) {
	500	+	t.Fatalf("append after compaction fault = %v, want closed store", err)
	501	+	}
	502	+	requireCandidateBytes(t, path, beforeAppend)
	503	+
	504	+	if err := recoverCompaction(path, defaultFSHooks()); err != nil {
	505	+	t.Fatalf("recovery: %v", err)
	506	+	}
	507	+	got, err := Replay(path)
	508	+	if err != nil {
	509	+	t.Fatal(err)
	510	+	}
	511	+	version := string(got["version"])
	512	+	if (version != "old" && version != "new") \|\| len(got) != 1 {
	513	+	t.Fatalf("mixed or empty recovered map: %#v", got)
	514	+	}
	515	+	})
	516	+	}
	517	+	}
	518	+
	519	+	func failRemoveCall(fault error, target int) func(fsHooks, string) fsHooks {
	520	+	return func(h fsHooks, _ string) fsHooks {
	521	+	remove := h.Remove
	522	+	calls := 0
	523	+	h.Remove = func(path string) error {
	524	+	calls++
	525	+	if calls == target {
	526	+	return fault
	527	+	}
	528	+	return remove(path)
	529	+	}
	530	+	return h
	531	+	}
	532	+	}
	533	+
	534	+	func failRenameCall(fault error, target int) func(fsHooks, string) fsHooks {
	535	+	return func(h fsHooks, _ string) fsHooks {
	536	+	rename := h.Rename
	537	+	calls := 0
	538	+	h.Rename = func(oldPath, newPath string) error {
	539	+	calls++
	540	+	if calls == target {
	541	+	return fault
	542	+	}
	543	+	return rename(oldPath, newPath)
	544	+	}
	545	+	return h
	546	+	}
	547	+	}
	548	+
	549	+	func failSyncDirCall(fault error, target int) func(fsHooks, string) fsHooks {
	550	+	return func(h fsHooks, _ string) fsHooks {
	551	+	sync := h.SyncDir
	552	+	calls := 0
	553	+	h.SyncDir = func(path string) error {
	554	+	calls++
	555	+	if calls == target {
	556	+	return fault
	557	+	}
	558	+	return sync(path)
	559	+	}
	560	+	return h
	561	+	}
	562	+	}
	563	+
419	564		func candidateBytes(t *testing.T, path string) map[string][]byte {
420	565		t.Helper()
421	566		got := make(map[string][]byte, 3)


@@ -416,6 +416,151 @@ func TestCompactionRecoveryMatrix(t *testing.T) {
416	}	416		}
417	}	417		}
418		418
		419	+	func TestCompactionFilesystemOperationFailuresPoisonAndRecover(t *testing.T) {
		420	+	fault := errors.New("injected filesystem failure")
		421	+	tests := []struct {
		422	+	name string
		423	+	configure func(fsHooks, string) fsHooks
		424	+	}{
		425	+	{name: "write-next", configure: func(h fsHooks, _ string) fsHooks {
		426	+	h.Write = func(*os.File, []byte) (int, error) { return 0, fault }
		427	+	return h
		428	+	}},
		429	+	{name: "sync-next", configure: func(h fsHooks, _ string) fsHooks {
		430	+	h.Sync = func(*os.File) error { return fault }
		431	+	return h
		432	+	}},
		433	+	{name: "close-next", configure: func(h fsHooks, _ string) fsHooks {
		434	+	calls := 0
		435	+	h.Close = func(f *os.File) error {
		436	+	calls++
		437	+	if calls == 1 {
		438	+	return errors.Join(f.Close(), fault)
		439	+	}
		440	+	return f.Close()
		441	+	}
		442	+	return h
		443	+	}},
		444	+	{name: "remove-stale-prev", configure: failRemoveCall(fault, 1)},
		445	+	{name: "rename-current-prev", configure: failRenameCall(fault, 1)},
		446	+	{name: "sync-dir-next", configure: failSyncDirCall(fault, 1)},
		447	+	{name: "sync-dir-prev", configure: failSyncDirCall(fault, 2)},
		448	+	{name: "rename-next-current", configure: failRenameCall(fault, 2)},
		449	+	{name: "sync-dir-current", configure: failSyncDirCall(fault, 3)},
		450	+	{name: "reopen-current", configure: func(h fsHooks, path string) fsHooks {
		451	+	open := h.Open
		452	+	h.Open = func(name string, flags int, mode os.FileMode) (*os.File, error) {
		453	+	if name == path && flags == os.O_RDWR {
		454	+	return nil, fault
		455	+	}
		456	+	return open(name, flags, mode)
		457	+	}
		458	+	return h
		459	+	}},
		460	+	{name: "close-old-current", configure: func(h fsHooks, _ string) fsHooks {
		461	+	calls := 0
		462	+	h.Close = func(f *os.File) error {
		463	+	calls++
		464	+	if calls == 2 {
		465	+	return errors.Join(f.Close(), fault)
		466	+	}
		467	+	return f.Close()
		468	+	}
		469	+	return h
		470	+	}},
		471	+	{name: "remove-prev", configure: failRemoveCall(fault, 2)},
		472	+	{name: "sync-dir-final", configure: failSyncDirCall(fault, 4)},
		473	+	}
		474	+	oldRecord, err := encodeRecord(opSet, []byte("version"), []byte("old"))
		475	+	if err != nil {
		476	+	t.Fatal(err)
		477	+	}
		478	+	for _, tt := range tests {
		479	+	t.Run(tt.name, func(t *testing.T) {
		480	+	path := privatePath(t)
		481	+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		482	+	t.Fatal(err)
		483	+	}
		484	+	if err := os.WriteFile(path, oldRecord, 0o600); err != nil {
		485	+	t.Fatal(err)
		486	+	}
		487	+	s, err := Open(path)
		488	+	if err != nil {
		489	+	t.Fatal(err)
		490	+	}
		491	+	s.data = map[string][]byte{"version": []byte("new")}
		492	+	s.hooks = tt.configure(s.hooks, path)
		493	+
		494	+	err = s.compactLocked()
		495	+	if !errors.Is(err, fault) {
		496	+	t.Fatalf("compaction error = %v, want injected fault", err)
		497	+	}
		498	+	beforeAppend := candidateBytes(t, path)
		499	+	if err := s.Set([]byte("later"), []byte("must-not-append")); !errors.Is(err, os.ErrClosed) {
		500	+	t.Fatalf("append after compaction fault = %v, want closed store", err)
		501	+	}
		502	+	requireCandidateBytes(t, path, beforeAppend)
		503	+
		504	+	if err := recoverCompaction(path, defaultFSHooks()); err != nil {
		505	+	t.Fatalf("recovery: %v", err)
		506	+	}
		507	+	got, err := Replay(path)
		508	+	if err != nil {
		509	+	t.Fatal(err)
		510	+	}
		511	+	version := string(got["version"])
		512	+	if (version != "old" && version != "new") \|\| len(got) != 1 {
		513	+	t.Fatalf("mixed or empty recovered map: %#v", got)
		514	+	}
		515	+	})
		516	+	}
		517	+	}
		518	+
		519	+	func failRemoveCall(fault error, target int) func(fsHooks, string) fsHooks {
		520	+	return func(h fsHooks, _ string) fsHooks {
		521	+	remove := h.Remove
		522	+	calls := 0
		523	+	h.Remove = func(path string) error {
		524	+	calls++
		525	+	if calls == target {
		526	+	return fault
		527	+	}
		528	+	return remove(path)
		529	+	}
		530	+	return h
		531	+	}
		532	+	}
		533	+
		534	+	func failRenameCall(fault error, target int) func(fsHooks, string) fsHooks {
		535	+	return func(h fsHooks, _ string) fsHooks {
		536	+	rename := h.Rename
		537	+	calls := 0
		538	+	h.Rename = func(oldPath, newPath string) error {
		539	+	calls++
		540	+	if calls == target {
		541	+	return fault
		542	+	}
		543	+	return rename(oldPath, newPath)
		544	+	}
		545	+	return h
		546	+	}
		547	+	}
		548	+
		549	+	func failSyncDirCall(fault error, target int) func(fsHooks, string) fsHooks {
		550	+	return func(h fsHooks, _ string) fsHooks {
		551	+	sync := h.SyncDir
		552	+	calls := 0
		553	+	h.SyncDir = func(path string) error {
		554	+	calls++
		555	+	if calls == target {
		556	+	return fault
		557	+	}
		558	+	return sync(path)
		559	+	}
		560	+	return h
		561	+	}
		562	+	}
		563	+
419	func candidateBytes(t *testing.T, path string) map[string][]byte {	564		func candidateBytes(t *testing.T, path string) map[string][]byte {
420	t.Helper()	565		t.Helper()
421	got := make(map[string][]byte, 3)	566		got := make(map[string][]byte, 3)