d6f7eb5 fix(persistence): complete strict catalogue integration · vev

M internal/app/run.go +1 -2


@@ -50,7 +50,6 @@ import (
50	50		"github.com/bnema/vev/internal/usecase/confirm"
51	51		"github.com/bnema/vev/internal/usecase/daemon"
52	52		pdgram "github.com/bnema/vev/pkg/dgram"
53		-	"github.com/bnema/vev/pkg/kv"
54	53		"github.com/bnema/vev/pkg/safedir"
55	54		)
56	55

@@ -519,7 +518,7 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
519	518		if !catalogueCandidate {
520	519		return fmt.Errorf("vev: durable session catalogue unavailable at %s", storePath)
521	520		}
522		-	store, err := kv.Open(storePath)
	521	+	store, err := persist.OpenStore(storePath)
523	522		if err != nil {
524	523		log.Error("opening session store failed", "path", storePath, "err", err)
525	524		return fmt.Errorf("vev: open durable session catalogue %s: %w", storePath, err)

 	"github.com/bnema/vev/internal/usecase/confirm"
 	"github.com/bnema/vev/internal/usecase/daemon"
 	pdgram "github.com/bnema/vev/pkg/dgram"
-	"github.com/bnema/vev/pkg/kv"
 	"github.com/bnema/vev/pkg/safedir"
 )
 	if !catalogueCandidate {
 		return fmt.Errorf("vev: durable session catalogue unavailable at %s", storePath)
 	}
-	store, err := kv.Open(storePath)
+	store, err := persist.OpenStore(storePath)
 	if err != nil {
 		log.Error("opening session store failed", "path", storePath, "err", err)
 		return fmt.Errorf("vev: open durable session catalogue %s: %w", storePath, err)

M internal/persist/persist.go +28 -1


@@ -24,6 +24,33 @@ type Persister struct {
24	24		mu sync.Mutex
25	25		}
26	26
	27	+	// KVStore adapts the reusable WAL implementation to the persistence port.
	28	+	type KVStore struct{ store *kv.Store }
	29	+
	30	+	func OpenStore(path string) (*KVStore, error) {
	31	+	store, err := kv.Open(path)
	32	+	if err != nil {
	33	+	return nil, err
	34	+	}
	35	+	return &KVStore{store: store}, nil
	36	+	}
	37	+
	38	+	func (s *KVStore) Get(key []byte) ([]byte, bool) { return s.store.Get(key) }
	39	+	func (s *KVStore) Set(key, value []byte) error { return s.store.Set(key, value) }
	40	+	func (s *KVStore) Delete(key []byte) error { return s.store.Delete(key) }
	41	+	func (s *KVStore) Range(fn func(key, value []byte) bool) {
	42	+	s.store.Range(fn)
	43	+	}
	44	+	func (s *KVStore) Sync() error { return s.store.Sync() }
	45	+	func (s *KVStore) Close() error { return s.store.Close() }
	46	+	func (s *KVStore) Batch(changes []ports.StoreChange) error {
	47	+	batch := make([]kv.BatchChange, len(changes))
	48	+	for i, change := range changes {
	49	+	batch[i] = kv.BatchChange{Key: change.Key, Value: change.Value, Delete: change.Delete}
	50	+	}
	51	+	return s.store.Batch(batch)
	52	+	}
	53	+
27	54		// Open opens an existing strict catalogue and never creates an unproven empty one.
28	55		func Open(dir string) (*Persister, error) {
29	56		p, _, err := openCurrentCatalogue(dir, false)

@@ -47,7 +74,7 @@ func openCurrentCatalogue(dir string, createProvenEmpty bool) (*Persister, []dom
47	74		if absent && !createProvenEmpty {
48	75		return nil, nil, fmt.Errorf("%w: no catalogue candidates", errPersistenceUnavailable)
49	76		}
50		-	store, err := kv.Open(path)
	77	+	store, err := OpenStore(path)
51	78		if err != nil {
52	79		return nil, nil, err
53	80		}


@@ -24,6 +24,33 @@ type Persister struct {
24		mu sync.Mutex	24		mu sync.Mutex
25		}	25		}
26			26
			27	+	// KVStore adapts the reusable WAL implementation to the persistence port.
			28	+	type KVStore struct{ store *kv.Store }
			29	+
			30	+	func OpenStore(path string) (*KVStore, error) {
			31	+	store, err := kv.Open(path)
			32	+	if err != nil {
			33	+	return nil, err
			34	+	}
			35	+	return &KVStore{store: store}, nil
			36	+	}
			37	+
			38	+	func (s *KVStore) Get(key []byte) ([]byte, bool) { return s.store.Get(key) }
			39	+	func (s *KVStore) Set(key, value []byte) error { return s.store.Set(key, value) }
			40	+	func (s *KVStore) Delete(key []byte) error { return s.store.Delete(key) }
			41	+	func (s *KVStore) Range(fn func(key, value []byte) bool) {
			42	+	s.store.Range(fn)
			43	+	}
			44	+	func (s *KVStore) Sync() error { return s.store.Sync() }
			45	+	func (s *KVStore) Close() error { return s.store.Close() }
			46	+	func (s *KVStore) Batch(changes []ports.StoreChange) error {
			47	+	batch := make([]kv.BatchChange, len(changes))
			48	+	for i, change := range changes {
			49	+	batch[i] = kv.BatchChange{Key: change.Key, Value: change.Value, Delete: change.Delete}
			50	+	}
			51	+	return s.store.Batch(batch)
			52	+	}
			53	+
27		// Open opens an existing strict catalogue and never creates an unproven empty one.	54		// Open opens an existing strict catalogue and never creates an unproven empty one.
28		func Open(dir string) (*Persister, error) {	55		func Open(dir string) (*Persister, error) {
29		p, _, err := openCurrentCatalogue(dir, false)	56		p, _, err := openCurrentCatalogue(dir, false)

@@ -47,7 +74,7 @@ func openCurrentCatalogue(dir string, createProvenEmpty bool) (*Persister, []dom
47		if absent && !createProvenEmpty {	74		if absent && !createProvenEmpty {
48		return nil, nil, fmt.Errorf("%w: no catalogue candidates", errPersistenceUnavailable)	75		return nil, nil, fmt.Errorf("%w: no catalogue candidates", errPersistenceUnavailable)
49		}	76		}
50	-	store, err := kv.Open(path)	77	+	store, err := OpenStore(path)
51		if err != nil {	78		if err != nil {
52		return nil, nil, err	79		return nil, nil, err
53		}	80		}

M internal/ports/mocks/store_mock.go +8 -8


@@ -39,8 +39,8 @@ func (_m MockStore) EXPECT() MockStore_Expecter {
39	39		}
40	40
41	41		// Batch provides a mock function for the type MockStore
42		-	func (_mock *MockStore) Batch(vs []ports.StoreChange) error {
43		-	ret := _mock.Called(vs)
	42	+	func (_mock *MockStore) Batch(storeChanges []ports.StoreChange) error {
	43	+	ret := _mock.Called(storeChanges)
44	44
45	45		if len(ret) == 0 {
46	46		panic("no return value specified for Batch")

@@ -48,7 +48,7 @@ func (_mock *MockStore) Batch(vs []ports.StoreChange) error {
48	48
49	49		var r0 error
50	50		if returnFunc, ok := ret.Get(0).(func([]ports.StoreChange) error); ok {
51		-	r0 = returnFunc(vs)
	51	+	r0 = returnFunc(storeChanges)
52	52		} else {
53	53		r0 = ret.Error(0)
54	54		}

@@ -61,12 +61,12 @@ type MockStore_Batch_Call struct {
61	61		}
62	62
63	63		// Batch is a helper method to define mock.On call
64		-	// - vs []ports.StoreChange
65		-	func (_e MockStore_Expecter) Batch(vs any) MockStore_Batch_Call {
66		-	return &MockStore_Batch_Call{Call: _e.mock.On("Batch", vs)}
	64	+	// - storeChanges []ports.StoreChange
	65	+	func (_e MockStore_Expecter) Batch(storeChanges any) MockStore_Batch_Call {
	66	+	return &MockStore_Batch_Call{Call: _e.mock.On("Batch", storeChanges)}
67	67		}
68	68
69		-	func (_c MockStore_Batch_Call) Run(run func(vs []ports.StoreChange)) MockStore_Batch_Call {
	69	+	func (_c MockStore_Batch_Call) Run(run func(storeChanges []ports.StoreChange)) MockStore_Batch_Call {
70	70		_c.Call.Run(func(args mock.Arguments) {
71	71		var arg0 []ports.StoreChange
72	72		if args[0] != nil {

@@ -84,7 +84,7 @@ func (_c MockStore_Batch_Call) Return(err error) MockStore_Batch_Call {
84	84		return _c
85	85		}
86	86
87		-	func (_c MockStore_Batch_Call) RunAndReturn(run func(vs []ports.StoreChange) error) MockStore_Batch_Call {
	87	+	func (_c MockStore_Batch_Call) RunAndReturn(run func(storeChanges []ports.StoreChange) error) MockStore_Batch_Call {
88	88		_c.Call.Return(run)
89	89		return _c
90	90		}


@@ -39,8 +39,8 @@ func (_m MockStore) EXPECT() MockStore_Expecter {
39		}	39		}
40			40
41		// Batch provides a mock function for the type MockStore	41		// Batch provides a mock function for the type MockStore
42	-	func (_mock *MockStore) Batch(vs []ports.StoreChange) error {	42	+	func (_mock *MockStore) Batch(storeChanges []ports.StoreChange) error {
43	-	ret := _mock.Called(vs)	43	+	ret := _mock.Called(storeChanges)
44			44
45		if len(ret) == 0 {	45		if len(ret) == 0 {
46		panic("no return value specified for Batch")	46		panic("no return value specified for Batch")

@@ -48,7 +48,7 @@ func (_mock *MockStore) Batch(vs []ports.StoreChange) error {
48			48
49		var r0 error	49		var r0 error
50		if returnFunc, ok := ret.Get(0).(func([]ports.StoreChange) error); ok {	50		if returnFunc, ok := ret.Get(0).(func([]ports.StoreChange) error); ok {
51	-	r0 = returnFunc(vs)	51	+	r0 = returnFunc(storeChanges)
52		} else {	52		} else {
53		r0 = ret.Error(0)	53		r0 = ret.Error(0)
54		}	54		}

@@ -61,12 +61,12 @@ type MockStore_Batch_Call struct {
61		}	61		}
62			62
63		// Batch is a helper method to define mock.On call	63		// Batch is a helper method to define mock.On call
64	-	// - vs []ports.StoreChange	64	+	// - storeChanges []ports.StoreChange
65	-	func (_e MockStore_Expecter) Batch(vs any) MockStore_Batch_Call {	65	+	func (_e MockStore_Expecter) Batch(storeChanges any) MockStore_Batch_Call {
66	-	return &MockStore_Batch_Call{Call: _e.mock.On("Batch", vs)}	66	+	return &MockStore_Batch_Call{Call: _e.mock.On("Batch", storeChanges)}
67		}	67		}
68			68
69	-	func (_c MockStore_Batch_Call) Run(run func(vs []ports.StoreChange)) MockStore_Batch_Call {	69	+	func (_c MockStore_Batch_Call) Run(run func(storeChanges []ports.StoreChange)) MockStore_Batch_Call {
70		_c.Call.Run(func(args mock.Arguments) {	70		_c.Call.Run(func(args mock.Arguments) {
71		var arg0 []ports.StoreChange	71		var arg0 []ports.StoreChange
72		if args[0] != nil {	72		if args[0] != nil {

@@ -84,7 +84,7 @@ func (_c MockStore_Batch_Call) Return(err error) MockStore_Batch_Call {
84		return _c	84		return _c
85		}	85		}
86			86
87	-	func (_c MockStore_Batch_Call) RunAndReturn(run func(vs []ports.StoreChange) error) MockStore_Batch_Call {	87	+	func (_c MockStore_Batch_Call) RunAndReturn(run func(storeChanges []ports.StoreChange) error) MockStore_Batch_Call {
88		_c.Call.Return(run)	88		_c.Call.Return(run)
89		return _c	89		return _c
90		}	90		}

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


@@ -3,12 +3,14 @@ package ports
3	3		import (
4	4		"context"
5	5		"crypto/sha256"
6		-
7		-	"github.com/bnema/vev/pkg/kv"
8	6		)
9	7
10	8		// StoreChange is one key mutation in an atomic store batch.
11		-	type StoreChange = kv.BatchChange
	9	+	type StoreChange struct {
	10	+	Key []byte
	11	+	Value []byte
	12	+	Delete bool
	13	+	}
12	14
13	15		// Store is a small byte-key/value persistence port.
14	16		//

 import (
 	"context"
 	"crypto/sha256"
-	"github.com/bnema/vev/pkg/kv"
 )
 // StoreChange is one key mutation in an atomic store batch.
-type StoreChange = kv.BatchChange
+type StoreChange struct {
+	Key    []byte
+	Value  []byte
+	Delete bool
+}
 // Store is a small byte-key/value persistence port.
 //

M internal/usecase/daemon/daemon.go +2 -1


@@ -212,6 +212,7 @@ type stoppedSession struct {
212	212		name string
213	213		cwd string
214	214		createdAt int64
	215	+	incarnation domain.IncarnationID
215	216		lastUsedSeq uint64
216	217		tabNames []string
217	218		purging bool

@@ -419,7 +420,7 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
419	420		var maxCreatedAt int64
420	421		hasCreatedAt := false
421	422		for _, r := range records {
422		-	d.stopped[r.Name] = stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, lastUsedSeq: r.LastUsedSeq, tabNames: r.TabNames}
	423	+	d.stopped[r.Name] = stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: r.TabNames}
423	424		if !hasCreatedAt \|\| r.CreatedAt > maxCreatedAt {
424	425		maxCreatedAt = r.CreatedAt
425	426		hasCreatedAt = true

 	name        string
 	cwd         string
 	createdAt   int64
+	incarnation domain.IncarnationID
 	lastUsedSeq uint64
 	tabNames    []string
 	purging     bool
 		var maxCreatedAt int64
 		hasCreatedAt := false
 		for _, r := range records {
-			d.stopped[r.Name] = stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, lastUsedSeq: r.LastUsedSeq, tabNames: r.TabNames}
+			d.stopped[r.Name] = stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: r.TabNames}
 			if !hasCreatedAt || r.CreatedAt > maxCreatedAt {
 				maxCreatedAt = r.CreatedAt
 				hasCreatedAt = true

M internal/usecase/daemon/picker_test.go +10 -2


@@ -28,8 +28,16 @@ import (
28	28		// generated mock's default Delete expectation.
29	29		type failingDeleteStore struct{ err error }
30	30
31		-	func (failingDeleteStore) Get([]byte) ([]byte, bool) { return nil, false }
32		-	func (failingDeleteStore) Set([]byte, []byte) error { return nil }
	31	+	func (failingDeleteStore) Get([]byte) ([]byte, bool) { return nil, false }
	32	+	func (failingDeleteStore) Set([]byte, []byte) error { return nil }
	33	+	func (s failingDeleteStore) Batch(changes []ports.StoreChange) error {
	34	+	for _, change := range changes {
	35	+	if change.Delete {
	36	+	return s.err
	37	+	}
	38	+	}
	39	+	return nil
	40	+	}
33	41		func (s failingDeleteStore) Delete([]byte) error { return s.err }
34	42		func (failingDeleteStore) Range(func(k, v []byte) bool) {}
35	43		func (failingDeleteStore) Sync() error { return nil }

 // generated mock's default Delete expectation.
 type failingDeleteStore struct{ err error }
-func (failingDeleteStore) Get([]byte) ([]byte, bool)    { return nil, false }
+func (failingDeleteStore) Get([]byte) ([]byte, bool) { return nil, false }
-func (failingDeleteStore) Set([]byte, []byte) error     { return nil }
+func (failingDeleteStore) Set([]byte, []byte) error  { return nil }
+func (s failingDeleteStore) Batch(changes []ports.StoreChange) error {
+	for _, change := range changes {
+		if change.Delete {
+			return s.err
+		}
+	}
+	return nil
+}
 func (s failingDeleteStore) Delete([]byte) error        { return s.err }
 func (failingDeleteStore) Range(func(k, v []byte) bool) {}
 func (failingDeleteStore) Sync() error                  { return nil }

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


@@ -3,6 +3,7 @@ package daemon
3	3
4	4		import (
5	5		"context"
	6	+	"crypto/rand"
6	7		"errors"
7	8		"fmt"
8	9		"os"

@@ -66,6 +67,7 @@ type session struct {
66	67		// It is guarded by mu and is always copied on ingress and egress.
67	68		env []string
68	69		createdAt int64
	70	+	incarnation domain.IncarnationID
69	71		mruAt atomic.Uint64
70	72		snapDirty atomic.Bool
71	73		snapEligible atomic.Bool

@@ -193,11 +195,13 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz
193	195		env = copyEnvironment(env)
194	196		stopped, resuming := d.stopped[name]
195	197		var createdAt int64
	198	+	var incarnation domain.IncarnationID
196	199		if !ephemeral {
197	200		if resuming {
198	201		// A stopped session is the same lifecycle when resumed; it must retain
199	202		// the persisted identity rather than receive a fresh timestamp.
200	203		createdAt = stopped.createdAt
	204	+	incarnation = stopped.incarnation
201	205		if createdAt > d.lastAllocatedCreatedAt {
202	206		d.lastAllocatedCreatedAt = createdAt
203	207		}

@@ -208,6 +212,13 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz
208	212		return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)
209	213		}
210	214		}
	215	+	if incarnation == (domain.IncarnationID{}) {
	216	+	var err error
	217	+	incarnation, err = domain.NewIncarnationID(rand.Reader)
	218	+	if err != nil {
	219	+	return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", fmt.Errorf("generate durable identity: %w", err))
	220	+	}
	221	+	}
211	222		}
212	223
213	224		tbSize := tabSize(sz)

@@ -258,6 +269,7 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz
258	269		terminal: term,
259	270		env: env,
260	271		createdAt: createdAt,
	272	+	incarnation: incarnation,
261	273		snapshotWake: d.snapshotWake,
262	274		}
263	275		if !ephemeral && name != "" {

@@ -268,7 +280,7 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz
268	280		}
269	281		sess.snapEligible.Store(!ephemeral && name != "")
270	282		if !ephemeral {
271		-	if err := d.persist.Save(persist.Record{Name: name, Cwd: cwd, CreatedAt: createdAt, UpdatedAt: createdAt, LastUsedSeq: lastUsedSeq, TabNames: names}); err != nil {
	283	+	if err := d.persist.Save(persist.Record{Name: name, IncarnationID: incarnation, Cwd: cwd, CreatedAt: createdAt, UpdatedAt: createdAt, LastUsedSeq: lastUsedSeq, TabNames: names, RecoveryState: domain.RecoveryFresh}); err != nil {
272	284		closeTabs(tabs)
273	285		cancel()
274	286		return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)

@@ -643,6 +655,12 @@ func (d Daemon) renameSession(sess session, name string) error {
643	655		d.mu.Unlock()
644	656		return err
645	657		}
	658	+	sess.incarnation, err = domain.NewIncarnationID(rand.Reader)
	659	+	if err != nil {
	660	+	sess.mu.Unlock()
	661	+	d.mu.Unlock()
	662	+	return fmt.Errorf("generate durable identity: %w", err)
	663	+	}
646	664		}
647	665		lastUsedSeq := sess.mruAt.Load()
648	666		record := sess.persistRecordLocked(d.nowUnixNano())

@@ -764,7 +782,7 @@ func (s *session) persistRecordLocked(updatedAt int64) persist.Record {
764	782		} else {
765	783		tabNames = tabNames[:lastCustom+1]
766	784		}
767		-	return persist.Record{Name: s.name, Cwd: s.cwd, CreatedAt: createdAt, UpdatedAt: updatedAt, LastUsedSeq: s.mruAt.Load(), TabNames: tabNames}
	785	+	return persist.Record{Name: s.name, IncarnationID: s.incarnation, Cwd: s.cwd, CreatedAt: createdAt, UpdatedAt: updatedAt, LastUsedSeq: s.mruAt.Load(), TabNames: tabNames, RecoveryState: domain.RecoveryFresh}
768	786		}
769	787
770	788		func (d Daemon) closeTab(sess session, tb *tab, repaint bool) error {

@@ -1145,7 +1163,7 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
1145	1163		ephemeral := sess.ephemeral
1146	1164		sess.mu.Unlock()
1147	1165		if !ephemeral {
1148		-	stopped := stoppedSession{name: stoppedName, cwd: stoppedCwd, createdAt: createdAt, lastUsedSeq: sess.mruAt.Load(), tabNames: tabNames, purging: purge}
	1166	+	stopped := stoppedSession{name: stoppedName, cwd: stoppedCwd, createdAt: createdAt, incarnation: sess.incarnation, lastUsedSeq: sess.mruAt.Load(), tabNames: tabNames, purging: purge}
1149	1167		d.stopped[stoppedName] = stopped
1150	1168		}
1151	1169		empty := len(d.sessions) == 0

 import (
 	"context"
+	"crypto/rand"
 	"errors"
 	"fmt"
 	"os"
 	// It is guarded by mu and is always copied on ingress and egress.
 	env          []string
 	createdAt    int64
+	incarnation  domain.IncarnationID
 	mruAt        atomic.Uint64
 	snapDirty    atomic.Bool
 	snapEligible atomic.Bool
 	env = copyEnvironment(env)
 	stopped, resuming := d.stopped[name]
 	var createdAt int64
+	var incarnation domain.IncarnationID
 	if !ephemeral {
 		if resuming {
 			// A stopped session is the same lifecycle when resumed; it must retain
 			// the persisted identity rather than receive a fresh timestamp.
 			createdAt = stopped.createdAt
+			incarnation = stopped.incarnation
 			if createdAt > d.lastAllocatedCreatedAt {
 				d.lastAllocatedCreatedAt = createdAt
 			}
 				return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)
 			}
 		}
+		if incarnation == (domain.IncarnationID{}) {
+			var err error
+			incarnation, err = domain.NewIncarnationID(rand.Reader)
+			if err != nil {
+				return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", fmt.Errorf("generate durable identity: %w", err))
+			}
+		}
 	}
 	tbSize := tabSize(sz)
 		terminal:     term,
 		env:          env,
 		createdAt:    createdAt,
+		incarnation:  incarnation,
 		snapshotWake: d.snapshotWake,
 	}
 	if !ephemeral && name != "" {
 	}
 	sess.snapEligible.Store(!ephemeral && name != "")
 	if !ephemeral {
-		if err := d.persist.Save(persist.Record{Name: name, Cwd: cwd, CreatedAt: createdAt, UpdatedAt: createdAt, LastUsedSeq: lastUsedSeq, TabNames: names}); err != nil {
+		if err := d.persist.Save(persist.Record{Name: name, IncarnationID: incarnation, Cwd: cwd, CreatedAt: createdAt, UpdatedAt: createdAt, LastUsedSeq: lastUsedSeq, TabNames: names, RecoveryState: domain.RecoveryFresh}); err != nil {
 			closeTabs(tabs)
 			cancel()
 			return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)
 			d.mu.Unlock()
 			return err
 		}
+		sess.incarnation, err = domain.NewIncarnationID(rand.Reader)
+		if err != nil {
+			sess.mu.Unlock()
+			d.mu.Unlock()
+			return fmt.Errorf("generate durable identity: %w", err)
+		}
 	}
 	lastUsedSeq := sess.mruAt.Load()
 	record := sess.persistRecordLocked(d.nowUnixNano())
 	} else {
 		tabNames = tabNames[:lastCustom+1]
 	}
-	return persist.Record{Name: s.name, Cwd: s.cwd, CreatedAt: createdAt, UpdatedAt: updatedAt, LastUsedSeq: s.mruAt.Load(), TabNames: tabNames}
+	return persist.Record{Name: s.name, IncarnationID: s.incarnation, Cwd: s.cwd, CreatedAt: createdAt, UpdatedAt: updatedAt, LastUsedSeq: s.mruAt.Load(), TabNames: tabNames, RecoveryState: domain.RecoveryFresh}
 }
 func (d *Daemon) closeTab(sess *session, tb *tab, repaint bool) error {
 	ephemeral := sess.ephemeral
 	sess.mu.Unlock()
 	if !ephemeral {
-		stopped := stoppedSession{name: stoppedName, cwd: stoppedCwd, createdAt: createdAt, lastUsedSeq: sess.mruAt.Load(), tabNames: tabNames, purging: purge}
+		stopped := stoppedSession{name: stoppedName, cwd: stoppedCwd, createdAt: createdAt, incarnation: sess.incarnation, lastUsedSeq: sess.mruAt.Load(), tabNames: tabNames, purging: purge}
 		d.stopped[stoppedName] = stopped
 	}
 	empty := len(d.sessions) == 0

M internal/usecase/daemon/session_test.go +53 -11


@@ -701,6 +701,26 @@ func newMockStore(t testing.T) (portsmocks.MockStore, *mockStoreState) {
701	701		state.data[string(k)] = append([]byte(nil), v...)
702	702		return nil
703	703		}).Maybe()
	704	+	store.EXPECT().Batch(mock.Anything).RunAndReturn(func(changes []ports.StoreChange) error {
	705	+	state.mu.Lock()
	706	+	defer state.mu.Unlock()
	707	+	for _, change := range changes {
	708	+	key := string(change.Key)
	709	+	if change.Delete {
	710	+	state.dels = append(state.dels, key)
	711	+	if state.deleteErr != nil {
	712	+	if err := state.deleteErr(key); err != nil {
	713	+	return err
	714	+	}
	715	+	}
	716	+	delete(state.data, key)
	717	+	continue
	718	+	}
	719	+	state.sets++
	720	+	state.data[key] = append([]byte(nil), change.Value...)
	721	+	}
	722	+	return nil
	723	+	}).Maybe()
704	724		store.EXPECT().Delete(mock.Anything).RunAndReturn(func(k []byte) error {
705	725		state.mu.Lock()
706	726		key := string(k)

@@ -774,6 +794,16 @@ func (s newStaticStore) Get(key []byte) ([]byte, bool) {
774	794		}
775	795		func (s newStaticStore) Set(_, _ []byte) error { return nil }
776	796		func (s newStaticStore) Delete(_ []byte) error { return nil }
	797	+	func (s newStaticStore) Batch(changes []ports.StoreChange) error {
	798	+	for _, change := range changes {
	799	+	if change.Delete {
	800	+	delete(s, string(change.Key))
	801	+	} else {
	802	+	s[string(change.Key)] = append([]byte(nil), change.Value...)
	803	+	}
	804	+	}
	805	+	return nil
	806	+	}
777	807		func (s newStaticStore) Range(fn func(k, v []byte) bool) {
778	808		for k, v := range s {
779	809		if !fn([]byte(k), append([]byte(nil), v...)) {

@@ -787,13 +817,14 @@ func (s newStaticStore) Close() error { return nil }
787	817		func TestDaemonLoadsPersistedSessionsAsStopped(t *testing.T) {
788	818		store, _ := newMockStore(t)
789	819		seed := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithStore(store))
790		-	require.NoError(t, seed.persist.Save(persist.Record{Name: "work", Cwd: "/tmp/work", CreatedAt: 7, UpdatedAt: 8, LastUsedSeq: 9}))
	820	+	incarnation := domain.IncarnationID{1}
	821	+	require.NoError(t, seed.persist.Save(persist.Record{Name: "work", IncarnationID: incarnation, Cwd: "/tmp/work", CreatedAt: 7, UpdatedAt: 8, LastUsedSeq: 9, RecoveryState: domain.RecoveryFresh}))
791	822
792	823		d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithStore(store))
793	824		d.mu.Lock()
794	825		stopped := d.stopped["work"]
795	826		d.mu.Unlock()
796		-	require.Equal(t, stoppedSession{name: "work", cwd: "/tmp/work", createdAt: 7, lastUsedSeq: 9}, stopped)
	827	+	require.Equal(t, stoppedSession{name: "work", cwd: "/tmp/work", createdAt: 7, incarnation: incarnation, lastUsedSeq: 9}, stopped)
797	828		require.Equal(t, uint64(9), d.mruSeq.Load())
798	829		}
799	830

@@ -846,8 +877,8 @@ func TestTouchMRUPersistsNamedButNotEphemeral(t *testing.T) {
846	877		store, state := newMockStore(t)
847	878		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
848	879		WithStore(store)(d)
849		-	named := &session{name: "work", tabs: []*tab{{}}, createdAt: 1}
850		-	require.NoError(t, d.persist.Save(persist.Record{Name: "work", Cwd: "/work", CreatedAt: 1, UpdatedAt: 1}))
	880	+	named := &session{name: "work", tabs: []*tab{{}}, createdAt: 1, incarnation: domain.IncarnationID{1}}
	881	+	require.NoError(t, d.persist.Save(persist.Record{Name: "work", IncarnationID: named.incarnation, Cwd: "/work", CreatedAt: 1, UpdatedAt: 1, RecoveryState: domain.RecoveryFresh}))
851	882
852	883		d.touchMRU(named)
853	884		require.Equal(t, named.mruAt.Load(), state.record(t, "work").LastUsedSeq)

@@ -888,6 +919,10 @@ func TestCreateRenameKillPersistenceLifecycle(t *testing.T) {
888	919		sess, err := d.createSessionLocked("work", false, "/tmp/work", sz, terminalEnv{}, d.baseEnv)
889	920		require.NoError(t, err)
890	921		require.True(t, state.has("work"))
	922	+	created := state.record(t, "work")
	923	+	require.NotEqual(t, domain.IncarnationID{}, created.IncarnationID)
	924	+	require.Equal(t, domain.RecoveryFresh, created.RecoveryState)
	925	+	require.NoError(t, created.Validate())
891	926		require.NoError(t, d.renameTab(sess, sess.tabs[0], "shell"))
892	927		require.NoError(t, d.createTab(sess, sz))
893	928		require.NoError(t, d.renameTab(sess, sess.tabs[1], "logs"))

@@ -895,7 +930,10 @@ func TestCreateRenameKillPersistenceLifecycle(t *testing.T) {
895	930		require.NoError(t, d.renameSession(sess, "renamed"))
896	931		require.False(t, state.has("work"))
897	932		require.True(t, state.has("renamed"))
898		-	require.Equal(t, []string{"shell", "logs"}, state.record(t, "renamed").TabNames)
	933	+	renamed := state.record(t, "renamed")
	934	+	require.Equal(t, []string{"shell", "logs"}, renamed.TabNames)
	935	+	require.Equal(t, created.IncarnationID, renamed.IncarnationID)
	936	+	require.NoError(t, renamed.Validate())
899	937
900	938		require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, true))
901	939		require.False(t, state.has("renamed"))

@@ -1014,7 +1052,7 @@ func TestAttachRestoresPersistedTabNames(t *testing.T) {
1014	1052		store, _ := newMockStore(t)
1015	1053		d := newTestDaemon(t, newFactorySeq(t, p1, p2), stubClock{})
1016	1054		WithStore(store)(d)
1017		-	require.NoError(t, d.persist.Save(persist.Record{Name: "work", Cwd: "/tmp/work", CreatedAt: 7, UpdatedAt: 8, TabNames: []string{"shell", "logs"}}))
	1055	+	require.NoError(t, d.persist.Save(persist.Record{Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp/work", CreatedAt: 7, UpdatedAt: 8, TabNames: []string{"shell", "logs"}, RecoveryState: domain.RecoveryFresh}))
1018	1056		d.stopped["work"] = stoppedSession{name: "work", cwd: "/tmp/work", createdAt: 7, tabNames: []string{"shell", "logs"}}
1019	1057		tr := portsmocks.NewMockTransport(t)
1020	1058		tr.EXPECT().Send(mock.Anything).Return(nil).Maybe()

@@ -1123,8 +1161,12 @@ func TestEphemeralPromotionLifecycleFailuresLeaveStateRollbackSafe(t *testing.T)
1123	1161		defer release()
1124	1162		store := portsmocks.NewMockStore(t)
1125	1163		var attempted map[string][]byte
1126		-	store.EXPECT().Set(mock.Anything, mock.Anything).RunAndReturn(func(key, value []byte) error {
1127		-	attempted = map[string][]byte{string(key): append([]byte(nil), value...)}
	1164	+	store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {}).Twice()
	1165	+	store.EXPECT().Batch(mock.Anything).RunAndReturn(func(changes []ports.StoreChange) error {
	1166	+	attempted = make(map[string][]byte, len(changes))
	1167	+	for _, change := range changes {
	1168	+	attempted[string(change.Key)] = append([]byte(nil), change.Value...)
	1169	+	}
1128	1170		return errors.New("disk full")
1129	1171		}).Once()
1130	1172		clock := &lifecycleClock{nows: []time.Time{time.Unix(0, 100), time.Unix(0, 100)}}

@@ -1143,7 +1185,7 @@ func TestEphemeralPromotionLifecycleFailuresLeaveStateRollbackSafe(t *testing.T)
1143	1185		require.Len(t, records, 1)
1144	1186		require.Equal(t, int64(100), records[0].CreatedAt, "a promotion must never attempt to persist a zero lifecycle identity")
1145	1187
1146		-	store.EXPECT().Set(mock.Anything, mock.Anything).Return(nil).Once()
	1188	+	store.EXPECT().Batch(mock.Anything).Return(nil).Once()
1147	1189		store.EXPECT().Sync().Return(nil).Once()
1148	1190		require.NoError(t, d.renameSession(sess, "named"))
1149	1191		require.Equal(t, int64(101), sess.createdAt, "the high-water mark must remain monotonic after a failed promotion")

@@ -1223,7 +1265,7 @@ func TestPickerStoppedTargetKillPurges(t *testing.T) {
1223	1265		store, state := newMockStore(t)
1224	1266		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
1225	1267		WithStore(store)(d)
1226		-	require.NoError(t, d.persist.Save(persist.Record{Name: "old", Cwd: "/tmp", CreatedAt: 1, UpdatedAt: 1}))
	1268	+	require.NoError(t, d.persist.Save(persist.Record{Name: "old", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp", CreatedAt: 1, UpdatedAt: 1, RecoveryState: domain.RecoveryFresh}))
1227	1269		d.stopped["old"] = stoppedSession{name: "old", cwd: "/tmp", createdAt: 1}
1228	1270		require.NoError(t, d.killPickerTarget(picker.Target{Name: "old", Stopped: true}))
1229	1271		require.False(t, state.has("old"))

@@ -1703,7 +1745,7 @@ func TestNamedSessionLifecycleExhaustionDoesNotMutateSessionState(t *testing.T)
1703	1745		func TestNamedSessionLifecycleTimestampStartsAfterPersistedHighWaterMark(t *testing.T) {
1704	1746		store, _ := newMockStore(t)
1705	1747		seed := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithStore(store))
1706		-	require.NoError(t, seed.persist.Save(persist.Record{Name: "old", Cwd: "/tmp", CreatedAt: 900, UpdatedAt: 900}))
	1748	+	require.NoError(t, seed.persist.Save(persist.Record{Name: "old", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp", CreatedAt: 900, UpdatedAt: 900, RecoveryState: domain.RecoveryFresh}))
1707	1749
1708	1750		p, release := newBlockingPTY(t)
1709	1751		defer release()

 		state.data[string(k)] = append([]byte(nil), v...)
 		return nil
 	}).Maybe()
+	store.EXPECT().Batch(mock.Anything).RunAndReturn(func(changes []ports.StoreChange) error {
+		state.mu.Lock()
+		defer state.mu.Unlock()
+		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 {
+						return err
+					}
+				}
+				delete(state.data, key)
+				continue
+			}
+			state.sets++
+			state.data[key] = append([]byte(nil), change.Value...)
+		}
+		return nil
+	}).Maybe()
 	store.EXPECT().Delete(mock.Anything).RunAndReturn(func(k []byte) error {
 		state.mu.Lock()
 		key := string(k)
 }
 func (s newStaticStore) Set(_, _ []byte) error { return nil }
 func (s newStaticStore) Delete(_ []byte) error { return nil }
+func (s newStaticStore) Batch(changes []ports.StoreChange) error {
+	for _, change := range changes {
+		if change.Delete {
+			delete(s, string(change.Key))
+		} else {
+			s[string(change.Key)] = append([]byte(nil), change.Value...)
+		}
+	}
+	return nil
+}
 func (s newStaticStore) Range(fn func(k, v []byte) bool) {
 	for k, v := range s {
 		if !fn([]byte(k), append([]byte(nil), v...)) {
 func TestDaemonLoadsPersistedSessionsAsStopped(t *testing.T) {
 	store, _ := newMockStore(t)
 	seed := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithStore(store))
-	require.NoError(t, seed.persist.Save(persist.Record{Name: "work", Cwd: "/tmp/work", CreatedAt: 7, UpdatedAt: 8, LastUsedSeq: 9}))
+	incarnation := domain.IncarnationID{1}
+	require.NoError(t, seed.persist.Save(persist.Record{Name: "work", IncarnationID: incarnation, Cwd: "/tmp/work", CreatedAt: 7, UpdatedAt: 8, LastUsedSeq: 9, RecoveryState: domain.RecoveryFresh}))
 	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithStore(store))
 	d.mu.Lock()
 	stopped := d.stopped["work"]
 	d.mu.Unlock()
-	require.Equal(t, stoppedSession{name: "work", cwd: "/tmp/work", createdAt: 7, lastUsedSeq: 9}, stopped)
+	require.Equal(t, stoppedSession{name: "work", cwd: "/tmp/work", createdAt: 7, incarnation: incarnation, lastUsedSeq: 9}, stopped)
 	require.Equal(t, uint64(9), d.mruSeq.Load())
 }
 	store, state := newMockStore(t)
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
 	WithStore(store)(d)
-	named := &session{name: "work", tabs: []*tab{{}}, createdAt: 1}
+	named := &session{name: "work", tabs: []*tab{{}}, createdAt: 1, incarnation: domain.IncarnationID{1}}
-	require.NoError(t, d.persist.Save(persist.Record{Name: "work", Cwd: "/work", CreatedAt: 1, UpdatedAt: 1}))
+	require.NoError(t, d.persist.Save(persist.Record{Name: "work", IncarnationID: named.incarnation, Cwd: "/work", CreatedAt: 1, UpdatedAt: 1, RecoveryState: domain.RecoveryFresh}))
 	d.touchMRU(named)
 	require.Equal(t, named.mruAt.Load(), state.record(t, "work").LastUsedSeq)
 	sess, err := d.createSessionLocked("work", false, "/tmp/work", sz, terminalEnv{}, d.baseEnv)
 	require.NoError(t, err)
 	require.True(t, state.has("work"))
+	created := state.record(t, "work")
+	require.NotEqual(t, domain.IncarnationID{}, created.IncarnationID)
+	require.Equal(t, domain.RecoveryFresh, created.RecoveryState)
+	require.NoError(t, created.Validate())
 	require.NoError(t, d.renameTab(sess, sess.tabs[0], "shell"))
 	require.NoError(t, d.createTab(sess, sz))
 	require.NoError(t, d.renameTab(sess, sess.tabs[1], "logs"))
 	require.NoError(t, d.renameSession(sess, "renamed"))
 	require.False(t, state.has("work"))
 	require.True(t, state.has("renamed"))
-	require.Equal(t, []string{"shell", "logs"}, state.record(t, "renamed").TabNames)
+	renamed := state.record(t, "renamed")
+	require.Equal(t, []string{"shell", "logs"}, renamed.TabNames)
+	require.Equal(t, created.IncarnationID, renamed.IncarnationID)
+	require.NoError(t, renamed.Validate())
 	require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, true))
 	require.False(t, state.has("renamed"))
 	store, _ := newMockStore(t)
 	d := newTestDaemon(t, newFactorySeq(t, p1, p2), stubClock{})
 	WithStore(store)(d)
-	require.NoError(t, d.persist.Save(persist.Record{Name: "work", Cwd: "/tmp/work", CreatedAt: 7, UpdatedAt: 8, TabNames: []string{"shell", "logs"}}))
+	require.NoError(t, d.persist.Save(persist.Record{Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp/work", CreatedAt: 7, UpdatedAt: 8, TabNames: []string{"shell", "logs"}, RecoveryState: domain.RecoveryFresh}))
 	d.stopped["work"] = stoppedSession{name: "work", cwd: "/tmp/work", createdAt: 7, tabNames: []string{"shell", "logs"}}
 	tr := portsmocks.NewMockTransport(t)
 	tr.EXPECT().Send(mock.Anything).Return(nil).Maybe()
 		defer release()
 		store := portsmocks.NewMockStore(t)
 		var attempted map[string][]byte
-		store.EXPECT().Set(mock.Anything, mock.Anything).RunAndReturn(func(key, value []byte) error {
+		store.EXPECT().Range(mock.Anything).Run(func(func([]byte, []byte) bool) {}).Twice()
-			attempted = map[string][]byte{string(key): append([]byte(nil), value...)}
+		store.EXPECT().Batch(mock.Anything).RunAndReturn(func(changes []ports.StoreChange) error {
+			attempted = make(map[string][]byte, len(changes))
+			for _, change := range changes {
+				attempted[string(change.Key)] = append([]byte(nil), change.Value...)
+			}
 			return errors.New("disk full")
 		}).Once()
 		clock := &lifecycleClock{nows: []time.Time{time.Unix(0, 100), time.Unix(0, 100)}}
 		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()
+		store.EXPECT().Batch(mock.Anything).Return(nil).Once()
 		store.EXPECT().Sync().Return(nil).Once()
 		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")
 	store, state := newMockStore(t)
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
 	WithStore(store)(d)
-	require.NoError(t, d.persist.Save(persist.Record{Name: "old", Cwd: "/tmp", CreatedAt: 1, UpdatedAt: 1}))
+	require.NoError(t, d.persist.Save(persist.Record{Name: "old", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp", CreatedAt: 1, UpdatedAt: 1, RecoveryState: domain.RecoveryFresh}))
 	d.stopped["old"] = stoppedSession{name: "old", cwd: "/tmp", createdAt: 1}
 	require.NoError(t, d.killPickerTarget(picker.Target{Name: "old", Stopped: true}))
 	require.False(t, state.has("old"))
 func TestNamedSessionLifecycleTimestampStartsAfterPersistedHighWaterMark(t *testing.T) {
 	store, _ := newMockStore(t)
 	seed := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithStore(store))
-	require.NoError(t, seed.persist.Save(persist.Record{Name: "old", Cwd: "/tmp", CreatedAt: 900, UpdatedAt: 900}))
+	require.NoError(t, seed.persist.Save(persist.Record{Name: "old", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp", CreatedAt: 900, UpdatedAt: 900, RecoveryState: domain.RecoveryFresh}))
 	p, release := newBlockingPTY(t)
 	defer release()

M internal/usecase/daemon/snapshot_restore.go +14 -0


@@ -2,6 +2,7 @@ package daemon
2	2
3	3		import (
4	4		"context"
	5	+	"crypto/rand"
5	6		"fmt"
6	7		"math"
7	8

@@ -253,6 +254,19 @@ func (d *Daemon) newRestoredSession(snap snapcodec.Session, sctx context.Context
253	254		// persistAndRegisterRestoredSession atomically transfers a successfully
254	255		// persisted session to the daemon. The caller retains cleanup on false/error.
255	256		func (d Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess session) (bool, error) {
	257	+	if sess.incarnation == (domain.IncarnationID{}) {
	258	+	d.mu.Lock()
	259	+	stopped := d.stopped[sess.name]
	260	+	d.mu.Unlock()
	261	+	sess.incarnation = stopped.incarnation
	262	+	if sess.incarnation == (domain.IncarnationID{}) {
	263	+	var err error
	264	+	sess.incarnation, err = domain.NewIncarnationID(rand.Reader)
	265	+	if err != nil {
	266	+	return false, fmt.Errorf("snapshot: generate durable identity: %w", err)
	267	+	}
	268	+	}
	269	+	}
256	270		if err := d.persist.Save(sess.persistRecordLocked(sess.createdAt)); err != nil {
257	271		return false, err
258	272		}


@@ -2,6 +2,7 @@ package daemon
2		2
3	import (	3		import (
4	"context"	4		"context"
		5	+	"crypto/rand"
5	"fmt"	6		"fmt"
6	"math"	7		"math"
7		8

@@ -253,6 +254,19 @@ func (d *Daemon) newRestoredSession(snap snapcodec.Session, sctx context.Context
253	// persistAndRegisterRestoredSession atomically transfers a successfully	254		// persistAndRegisterRestoredSession atomically transfers a successfully
254	// persisted session to the daemon. The caller retains cleanup on false/error.	255		// persisted session to the daemon. The caller retains cleanup on false/error.
255	func (d Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess session) (bool, error) {	256		func (d Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess session) (bool, error) {
		257	+	if sess.incarnation == (domain.IncarnationID{}) {
		258	+	d.mu.Lock()
		259	+	stopped := d.stopped[sess.name]
		260	+	d.mu.Unlock()
		261	+	sess.incarnation = stopped.incarnation
		262	+	if sess.incarnation == (domain.IncarnationID{}) {
		263	+	var err error
		264	+	sess.incarnation, err = domain.NewIncarnationID(rand.Reader)
		265	+	if err != nil {
		266	+	return false, fmt.Errorf("snapshot: generate durable identity: %w", err)
		267	+	}
		268	+	}
		269	+	}
256	if err := d.persist.Save(sess.persistRecordLocked(sess.createdAt)); err != nil {	270		if err := d.persist.Save(sess.persistRecordLocked(sess.createdAt)); err != nil {
257	return false, err	271		return false, err
258	}	272		}

M internal/usecase/daemon/snapshot_test.go +3 -0


@@ -46,6 +46,9 @@ func newSnapshotTestSession(t *testing.T, name string, ephemeral bool, cwd strin
46	46		p.screen.Write([]byte("hello"))
47	47		appendHistoryRow(t, p.history, []renderer.Cell{{Rune: 'h'}, {Rune: 'i'}})
48	48		sess := &session{id: domain.SessionID("sess-" + name), name: name, ephemeral: ephemeral, ctx: context.Background(), cancel: func() {}, tabs: []*tab{tb}, active: 0, cwd: cwd, createdAt: 42}
	49	+	if !ephemeral {
	50	+	sess.incarnation = domain.IncarnationID{1}
	51	+	}
49	52		sess.snapEligible.Store(!ephemeral && name != "")
50	53		return sess
51	54		}


@@ -46,6 +46,9 @@ func newSnapshotTestSession(t *testing.T, name string, ephemeral bool, cwd strin
46	p.screen.Write([]byte("hello"))	46		p.screen.Write([]byte("hello"))
47	appendHistoryRow(t, p.history, []renderer.Cell{{Rune: 'h'}, {Rune: 'i'}})	47		appendHistoryRow(t, p.history, []renderer.Cell{{Rune: 'h'}, {Rune: 'i'}})
48	sess := &session{id: domain.SessionID("sess-" + name), name: name, ephemeral: ephemeral, ctx: context.Background(), cancel: func() {}, tabs: []*tab{tb}, active: 0, cwd: cwd, createdAt: 42}	48		sess := &session{id: domain.SessionID("sess-" + name), name: name, ephemeral: ephemeral, ctx: context.Background(), cancel: func() {}, tabs: []*tab{tb}, active: 0, cwd: cwd, createdAt: 42}
		49	+	if !ephemeral {
		50	+	sess.incarnation = domain.IncarnationID{1}
		51	+	}
49	sess.snapEligible.Store(!ephemeral && name != "")	52		sess.snapEligible.Store(!ephemeral && name != "")
50	return sess	53		return sess
51	}	54		}

M pkg/kv/kv.go +23 -1


@@ -241,7 +241,12 @@ func (s *Store) shouldCompact() bool {
241	241		return s.total > compactThreshold && float64(s.total-s.live)/float64(s.total) > compactWasteRatio
242	242		}
243	243
244		-	func (s *Store) compactLocked() error {
	244	+	func (s *Store) compactLocked() (retErr error) {
	245	+	defer func() {
	246	+	if retErr != nil {
	247	+	retErr = errors.Join(retErr, s.poisonLocked())
	248	+	}
	249	+	}()
245	250		next, prev := s.path+".next", s.path+".prev"
246	251		if err := injectFault(s.hooks, "before-write-next"); err != nil {
247	252		return err

@@ -369,6 +374,23 @@ func (s *Store) compactLocked() error {
369	374		return injectFault(s.hooks, "after-final-sync-dir")
370	375		}
371	376
	377	+	// poisonLocked permanently disables a store whose compaction did not complete.
	378	+	// The caller holds s.mu, so no later append can retain a descriptor for a
	379	+	// transitional pathname.
	380	+	func (s *Store) poisonLocked() error {
	381	+	if s.closed {
	382	+	return nil
	383	+	}
	384	+	s.closed = true
	385	+	file, lockFile := s.file, s.lockFile
	386	+	s.file, s.lockFile = nil, nil
	387	+	var fileErr error
	388	+	if file != nil {
	389	+	fileErr = file.Close()
	390	+	}
	391	+	return errors.Join(fileErr, releaseLock(lockFile))
	392	+	}
	393	+
372	394		func replayFile(f *os.File) (ReplayResult, error) {
373	395		info, err := f.Stat()
374	396		if err != nil {


@@ -241,7 +241,12 @@ func (s *Store) shouldCompact() bool {
241		return s.total > compactThreshold && float64(s.total-s.live)/float64(s.total) > compactWasteRatio	241		return s.total > compactThreshold && float64(s.total-s.live)/float64(s.total) > compactWasteRatio
242		}	242		}
243			243
244	-	func (s *Store) compactLocked() error {	244	+	func (s *Store) compactLocked() (retErr error) {
			245	+	defer func() {
			246	+	if retErr != nil {
			247	+	retErr = errors.Join(retErr, s.poisonLocked())
			248	+	}
			249	+	}()
245		next, prev := s.path+".next", s.path+".prev"	250		next, prev := s.path+".next", s.path+".prev"
246		if err := injectFault(s.hooks, "before-write-next"); err != nil {	251		if err := injectFault(s.hooks, "before-write-next"); err != nil {
247		return err	252		return err

@@ -369,6 +374,23 @@ func (s *Store) compactLocked() error {
369		return injectFault(s.hooks, "after-final-sync-dir")	374		return injectFault(s.hooks, "after-final-sync-dir")
370		}	375		}
371			376
			377	+	// poisonLocked permanently disables a store whose compaction did not complete.
			378	+	// The caller holds s.mu, so no later append can retain a descriptor for a
			379	+	// transitional pathname.
			380	+	func (s *Store) poisonLocked() error {
			381	+	if s.closed {
			382	+	return nil
			383	+	}
			384	+	s.closed = true
			385	+	file, lockFile := s.file, s.lockFile
			386	+	s.file, s.lockFile = nil, nil
			387	+	var fileErr error
			388	+	if file != nil {
			389	+	fileErr = file.Close()
			390	+	}
			391	+	return errors.Join(fileErr, releaseLock(lockFile))
			392	+	}
			393	+
372		func replayFile(f *os.File) (ReplayResult, error) {	394		func replayFile(f *os.File) (ReplayResult, error) {
373		info, err := f.Stat()	395		info, err := f.Stat()
374		if err != nil {	396		if err != nil {

M pkg/kv/kv_test.go +29 -3


@@ -7,6 +7,7 @@ import (
7	7		"hash/crc32"
8	8		"os"
9	9		"path/filepath"
	10	+	"reflect"
10	11		"testing"
11	12		)
12	13

@@ -395,10 +396,11 @@ func TestCompactionRecoveryMatrix(t *testing.T) {
395	396		if err := s.compactLocked(); err == nil {
396	397		t.Fatalf("fault %q did not interrupt compaction", point)
397	398		}
398		-	if s.file != nil {
399		-	_ = s.file.Close()
	399	+	beforeAppend := candidateBytes(t, path)
	400	+	if err := s.Set([]byte("later"), []byte("must-not-append")); !errors.Is(err, os.ErrClosed) {
	401	+	t.Fatalf("append after compaction fault = %v, want closed store", err)
400	402		}
401		-	_ = releaseLock(s.lockFile)
	403	+	requireCandidateBytes(t, path, beforeAppend)
402	404		if err := recoverCompaction(path, defaultFSHooks()); err != nil {
403	405		t.Fatalf("recovery: %v", err)
404	406		}

@@ -414,6 +416,30 @@ func TestCompactionRecoveryMatrix(t *testing.T) {
414	416		}
415	417		}
416	418
	419	+	func candidateBytes(t *testing.T, path string) map[string][]byte {
	420	+	t.Helper()
	421	+	got := make(map[string][]byte, 3)
	422	+	for _, suffix := range []string{"", ".next", ".prev"} {
	423	+	data, err := os.ReadFile(path + suffix)
	424	+	if errors.Is(err, os.ErrNotExist) {
	425	+	continue
	426	+	}
	427	+	if err != nil {
	428	+	t.Fatal(err)
	429	+	}
	430	+	got[suffix] = data
	431	+	}
	432	+	return got
	433	+	}
	434	+
	435	+	func requireCandidateBytes(t *testing.T, path string, want map[string][]byte) {
	436	+	t.Helper()
	437	+	got := candidateBytes(t, path)
	438	+	if !reflect.DeepEqual(got, want) {
	439	+	t.Fatalf("catalogue candidates changed after poisoned append: got %#v, want %#v", got, want)
	440	+	}
	441	+	}
	442	+
417	443		func corruptRecord(buf []byte, offset int) []byte { buf[offset] ^= 0xff; return buf }
418	444		func rawRecord(payload []byte) []byte {
419	445		buf := make([]byte, headerLen+len(payload))


@@ -7,6 +7,7 @@ import (
7		"hash/crc32"	7		"hash/crc32"
8		"os"	8		"os"
9		"path/filepath"	9		"path/filepath"
			10	+	"reflect"
10		"testing"	11		"testing"
11		)	12		)
12			13

@@ -395,10 +396,11 @@ func TestCompactionRecoveryMatrix(t *testing.T) {
395		if err := s.compactLocked(); err == nil {	396		if err := s.compactLocked(); err == nil {
396		t.Fatalf("fault %q did not interrupt compaction", point)	397		t.Fatalf("fault %q did not interrupt compaction", point)
397		}	398		}
398	-	if s.file != nil {	399	+	beforeAppend := candidateBytes(t, path)
399	-	_ = s.file.Close()	400	+	if err := s.Set([]byte("later"), []byte("must-not-append")); !errors.Is(err, os.ErrClosed) {
			401	+	t.Fatalf("append after compaction fault = %v, want closed store", err)
400		}	402		}
401	-	_ = releaseLock(s.lockFile)	403	+	requireCandidateBytes(t, path, beforeAppend)
402		if err := recoverCompaction(path, defaultFSHooks()); err != nil {	404		if err := recoverCompaction(path, defaultFSHooks()); err != nil {
403		t.Fatalf("recovery: %v", err)	405		t.Fatalf("recovery: %v", err)
404		}	406		}

@@ -414,6 +416,30 @@ func TestCompactionRecoveryMatrix(t *testing.T) {
414		}	416		}
415		}	417		}
416			418
			419	+	func candidateBytes(t *testing.T, path string) map[string][]byte {
			420	+	t.Helper()
			421	+	got := make(map[string][]byte, 3)
			422	+	for _, suffix := range []string{"", ".next", ".prev"} {
			423	+	data, err := os.ReadFile(path + suffix)
			424	+	if errors.Is(err, os.ErrNotExist) {
			425	+	continue
			426	+	}
			427	+	if err != nil {
			428	+	t.Fatal(err)
			429	+	}
			430	+	got[suffix] = data
			431	+	}
			432	+	return got
			433	+	}
			434	+
			435	+	func requireCandidateBytes(t *testing.T, path string, want map[string][]byte) {
			436	+	t.Helper()
			437	+	got := candidateBytes(t, path)
			438	+	if !reflect.DeepEqual(got, want) {
			439	+	t.Fatalf("catalogue candidates changed after poisoned append: got %#v, want %#v", got, want)
			440	+	}
			441	+	}
			442	+
417		func corruptRecord(buf []byte, offset int) []byte { buf[offset] ^= 0xff; return buf }	443		func corruptRecord(buf []byte, offset int) []byte { buf[offset] ^= 0xff; return buf }
418		func rawRecord(payload []byte) []byte {	444		func rawRecord(payload []byte) []byte {
419		buf := make([]byte, headerLen+len(payload))	445		buf := make([]byte, headerLen+len(payload))