refactor(daemon): defer metadata writes and drop versioned rollback

commit: 70770e75ca761fed6ad2e26ef198c6bdbc01b346
author: brice · 2mo ago · 2026-07-26 12:41 UTC
parent: 4776aab
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

12 files · +245 -612

M internal/persist/catalogue_test.go +37 -0


@@ -17,6 +17,7 @@ func TestCatalogue(t *testing.T) {
17	17		t.Run("open-fails-closed", testCatalogueOpenFailsClosed)
18	18		t.Run("apply-rename-replace", testCatalogueApplyRenameReplace)
19	19		t.Run("metadata-update-preserves-authority", testCatalogueMetadataUpdatePreservesAuthority)
	20	+	t.Run("metadata-update-is-deferred", testCatalogueMetadataUpdateIsDeferred)
20	21		t.Run("read-only-malformed", testCatalogueLoadReadOnlyRejectsMalformedValue)
21	22		t.Run("read-only-fresh-install", testCatalogueLoadReadOnlyFreshInstallHasNoSessions)
22	23		t.Run("read-only-ignores-stray-tmp", testCatalogueLoadReadOnlyIgnoresStrayTmp)

@@ -160,6 +161,42 @@ func testCatalogueMetadataUpdatePreservesAuthority(t *testing.T) {
160	161		require.Equal(t, expected, unchanged)
161	162		}
162	163
	164	+	func testCatalogueMetadataUpdateIsDeferred(t *testing.T) {
	165	+	dir := privateDir(t)
	166	+	p, _, err := openCurrentCatalogue(dir, true)
	167	+	require.NoError(t, err)
	168	+	defer func() { require.NoError(t, p.Close()) }()
	169	+
	170	+	record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/old"}
	171	+	require.NoError(t, p.Create(record))
	172	+	before, err := os.ReadFile(StorePath(dir))
	173	+	require.NoError(t, err)
	174	+
	175	+	cwd := "/buffered"
	176	+	require.NoError(t, p.UpdateMetadata(domain.CatalogueMetadataUpdate{
	177	+	Name: record.Name, IncarnationID: record.IncarnationID, Cwd: &cwd,
	178	+	}))
	179	+	afterBuffered, err := os.ReadFile(StorePath(dir))
	180	+	require.NoError(t, err)
	181	+	require.Equal(t, before, afterBuffered, "metadata must remain buffered before Sync")
	182	+
	183	+	require.NoError(t, p.Sync())
	184	+	afterSync, err := os.ReadFile(StorePath(dir))
	185	+	require.NoError(t, err)
	186	+	require.NotEqual(t, before, afterSync)
	187	+
	188	+	cwd = "/next-identity"
	189	+	require.NoError(t, p.UpdateMetadata(domain.CatalogueMetadataUpdate{
	190	+	Name: record.Name, IncarnationID: record.IncarnationID, Cwd: &cwd,
	191	+	}))
	192	+	record.Cwd = cwd
	193	+	record.UpdatedAt = 2
	194	+	require.NoError(t, p.Replace(record.Name, record))
	195	+	afterIdentity, err := os.ReadFile(StorePath(dir))
	196	+	require.NoError(t, err)
	197	+	require.NotEqual(t, afterSync, afterIdentity, "identity writes must sync before returning")
	198	+	}
	199	+
163	200		func testCatalogueLoadReadOnlyRejectsMalformedValue(t *testing.T) {
164	201		dir := privateDir(t)
165	202		store, err := kv.Open(filepath.Join(dir, filename))


@@ -17,6 +17,7 @@ func TestCatalogue(t *testing.T) {
17	t.Run("open-fails-closed", testCatalogueOpenFailsClosed)	17		t.Run("open-fails-closed", testCatalogueOpenFailsClosed)
18	t.Run("apply-rename-replace", testCatalogueApplyRenameReplace)	18		t.Run("apply-rename-replace", testCatalogueApplyRenameReplace)
19	t.Run("metadata-update-preserves-authority", testCatalogueMetadataUpdatePreservesAuthority)	19		t.Run("metadata-update-preserves-authority", testCatalogueMetadataUpdatePreservesAuthority)
		20	+	t.Run("metadata-update-is-deferred", testCatalogueMetadataUpdateIsDeferred)
20	t.Run("read-only-malformed", testCatalogueLoadReadOnlyRejectsMalformedValue)	21		t.Run("read-only-malformed", testCatalogueLoadReadOnlyRejectsMalformedValue)
21	t.Run("read-only-fresh-install", testCatalogueLoadReadOnlyFreshInstallHasNoSessions)	22		t.Run("read-only-fresh-install", testCatalogueLoadReadOnlyFreshInstallHasNoSessions)
22	t.Run("read-only-ignores-stray-tmp", testCatalogueLoadReadOnlyIgnoresStrayTmp)	23		t.Run("read-only-ignores-stray-tmp", testCatalogueLoadReadOnlyIgnoresStrayTmp)

@@ -160,6 +161,42 @@ func testCatalogueMetadataUpdatePreservesAuthority(t *testing.T) {
160	require.Equal(t, expected, unchanged)	161		require.Equal(t, expected, unchanged)
161	}	162		}
162		163
		164	+	func testCatalogueMetadataUpdateIsDeferred(t *testing.T) {
		165	+	dir := privateDir(t)
		166	+	p, _, err := openCurrentCatalogue(dir, true)
		167	+	require.NoError(t, err)
		168	+	defer func() { require.NoError(t, p.Close()) }()
		169	+
		170	+	record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/old"}
		171	+	require.NoError(t, p.Create(record))
		172	+	before, err := os.ReadFile(StorePath(dir))
		173	+	require.NoError(t, err)
		174	+
		175	+	cwd := "/buffered"
		176	+	require.NoError(t, p.UpdateMetadata(domain.CatalogueMetadataUpdate{
		177	+	Name: record.Name, IncarnationID: record.IncarnationID, Cwd: &cwd,
		178	+	}))
		179	+	afterBuffered, err := os.ReadFile(StorePath(dir))
		180	+	require.NoError(t, err)
		181	+	require.Equal(t, before, afterBuffered, "metadata must remain buffered before Sync")
		182	+
		183	+	require.NoError(t, p.Sync())
		184	+	afterSync, err := os.ReadFile(StorePath(dir))
		185	+	require.NoError(t, err)
		186	+	require.NotEqual(t, before, afterSync)
		187	+
		188	+	cwd = "/next-identity"
		189	+	require.NoError(t, p.UpdateMetadata(domain.CatalogueMetadataUpdate{
		190	+	Name: record.Name, IncarnationID: record.IncarnationID, Cwd: &cwd,
		191	+	}))
		192	+	record.Cwd = cwd
		193	+	record.UpdatedAt = 2
		194	+	require.NoError(t, p.Replace(record.Name, record))
		195	+	afterIdentity, err := os.ReadFile(StorePath(dir))
		196	+	require.NoError(t, err)
		197	+	require.NotEqual(t, afterSync, afterIdentity, "identity writes must sync before returning")
		198	+	}
		199	+
163	func testCatalogueLoadReadOnlyRejectsMalformedValue(t *testing.T) {	200		func testCatalogueLoadReadOnlyRejectsMalformedValue(t *testing.T) {
164	dir := privateDir(t)	201		dir := privateDir(t)
165	store, err := kv.Open(filepath.Join(dir, filename))	202		store, err := kv.Open(filepath.Join(dir, filename))

M internal/persist/persist.go +26 -10


@@ -143,7 +143,7 @@ func (p *Persister) Create(record domain.CatalogueRecord) error {
143	143		if _, exists := p.store.Get([]byte(record.Name)); exists {
144	144		return errors.New("persist: session already exists")
145	145		}
146		-	return p.applyLocked(map[string]*domain.CatalogueRecord{record.Name: &record})
	146	+	return p.applyLocked(map[string]*domain.CatalogueRecord{record.Name: &record}, true)
147	147		}
148	148
149	149		func (p *Persister) Records() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }

@@ -172,9 +172,9 @@ func (p Persister) Apply(records map[string]domain.CatalogueRecord) error {
172	172		}
173	173		p.mu.Lock()
174	174		defer p.mu.Unlock()
175		-	return p.applyLocked(records)
	175	+	return p.applyLocked(records, true)
176	176		}
177		-	func (p Persister) applyLocked(records map[string]domain.CatalogueRecord) error {
	177	+	func (p Persister) applyLocked(records map[string]domain.CatalogueRecord, durable bool) error {
178	178		if len(records) == 0 {
179	179		return errors.New("persist: empty catalogue batch")
180	180		}

@@ -238,9 +238,11 @@ func (p Persister) applyLocked(records map[string]domain.CatalogueRecord) erro
238	238		return err
239	239		}
240	240		}
241		-	if err := p.store.Sync(); err != nil {
242		-	p.incarnationIndex = false
243		-	return err
	241	+	if durable {
	242	+	if err := p.store.Sync(); err != nil {
	243	+	p.incarnationIndex = false
	244	+	return err
	245	+	}
244	246		}
245	247		for name := range records {
246	248		if id, ok := p.nameIncarnations[name]; ok {

@@ -303,9 +305,10 @@ func (p *Persister) Replace(name string, next domain.CatalogueRecord) error {
303	305		return p.Apply(map[string]*domain.CatalogueRecord{name: &next})
304	306		}
305	307
306		-	// UpdateMetadata atomically updates ordinary session metadata on the existing
307		-	// authoritative incarnation. Recovery and transaction fields are retained from
308		-	// the stored record rather than accepted from a potentially stale runtime copy.
	308	+	// UpdateMetadata buffers ordinary session metadata on the existing authoritative
	309	+	// incarnation. Recovery and transaction fields are retained from the stored
	310	+	// record rather than accepted from a potentially stale runtime copy. Sync or the
	311	+	// next identity operation makes the update durable.
309	312		func (p *Persister) UpdateMetadata(update domain.CatalogueMetadataUpdate) error {
310	313		if p == nil {
311	314		return nil

@@ -338,7 +341,20 @@ func (p *Persister) UpdateMetadata(update domain.CatalogueMetadataUpdate) error
338	341		if update.TabNames != nil {
339	342		current.TabNames = append([]string(nil), (*update.TabNames)...)
340	343		}
341		-	return p.applyLocked(map[string]*domain.CatalogueRecord{current.Name: &current})
	344	+	return p.applyLocked(map[string]*domain.CatalogueRecord{current.Name: &current}, false)
	345	+	}
	346	+
	347	+	// Sync makes every buffered catalogue mutation durable.
	348	+	func (p *Persister) Sync() error {
	349	+	if p == nil {
	350	+	return nil
	351	+	}
	352	+	if p.store == nil {
	353	+	return errPersistenceUnavailable
	354	+	}
	355	+	p.mu.Lock()
	356	+	defer p.mu.Unlock()
	357	+	return p.store.Sync()
342	358		}
343	359
344	360		func (p *Persister) Delete(name string) error {

 	if _, exists := p.store.Get([]byte(record.Name)); exists {
 		return errors.New("persist: session already exists")
 	}
-	return p.applyLocked(map[string]*domain.CatalogueRecord{record.Name: &record})
+	return p.applyLocked(map[string]*domain.CatalogueRecord{record.Name: &record}, true)
 }
 func (p *Persister) Records() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	return p.applyLocked(records)
+	return p.applyLocked(records, true)
 }
-func (p *Persister) applyLocked(records map[string]*domain.CatalogueRecord) error {
+func (p *Persister) applyLocked(records map[string]*domain.CatalogueRecord, durable bool) error {
 	if len(records) == 0 {
 		return errors.New("persist: empty catalogue batch")
 	}
 			return err
 		}
 	}
-	if err := p.store.Sync(); err != nil {
+	if durable {
-		p.incarnationIndex = false
+		if err := p.store.Sync(); err != nil {
-		return err
+			p.incarnationIndex = false
+			return err
+		}
 	}
 	for name := range records {
 		if id, ok := p.nameIncarnations[name]; ok {
 	return p.Apply(map[string]*domain.CatalogueRecord{name: &next})
 }
-// UpdateMetadata atomically updates ordinary session metadata on the existing
+// UpdateMetadata buffers ordinary session metadata on the existing authoritative
-// authoritative incarnation. Recovery and transaction fields are retained from
+// incarnation. Recovery and transaction fields are retained from the stored
-// the stored record rather than accepted from a potentially stale runtime copy.
+// record rather than accepted from a potentially stale runtime copy. Sync or the
+// next identity operation makes the update durable.
 func (p *Persister) UpdateMetadata(update domain.CatalogueMetadataUpdate) error {
 	if p == nil {
 		return nil
 	if update.TabNames != nil {
 		current.TabNames = append([]string(nil), (*update.TabNames)...)
 	}
-	return p.applyLocked(map[string]*domain.CatalogueRecord{current.Name: &current})
+	return p.applyLocked(map[string]*domain.CatalogueRecord{current.Name: &current}, false)
+}
+// Sync makes every buffered catalogue mutation durable.
+func (p *Persister) Sync() error {
+	if p == nil {
+		return nil
+	}
+	if p.store == nil {
+		return errPersistenceUnavailable
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.store.Sync()
 }
 func (p *Persister) Delete(name string) error {

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


@@ -420,6 +420,50 @@ func (_c *MockCatalogue_Replace_Call) RunAndReturn(run func(s string, catalogueR
420	420		return _c
421	421		}
422	422
	423	+	// Sync provides a mock function for the type MockCatalogue
	424	+	func (_mock *MockCatalogue) Sync() error {
	425	+	ret := _mock.Called()
	426	+
	427	+	if len(ret) == 0 {
	428	+	panic("no return value specified for Sync")
	429	+	}
	430	+
	431	+	var r0 error
	432	+	if returnFunc, ok := ret.Get(0).(func() error); ok {
	433	+	r0 = returnFunc()
	434	+	} else {
	435	+	r0 = ret.Error(0)
	436	+	}
	437	+	return r0
	438	+	}
	439	+
	440	+	// MockCatalogue_Sync_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Sync'
	441	+	type MockCatalogue_Sync_Call struct {
	442	+	*mock.Call
	443	+	}
	444	+
	445	+	// Sync is a helper method to define mock.On call
	446	+	func (_e MockCatalogue_Expecter) Sync() MockCatalogue_Sync_Call {
	447	+	return &MockCatalogue_Sync_Call{Call: _e.mock.On("Sync")}
	448	+	}
	449	+
	450	+	func (_c MockCatalogue_Sync_Call) Run(run func()) MockCatalogue_Sync_Call {
	451	+	_c.Call.Run(func(args mock.Arguments) {
	452	+	run()
	453	+	})
	454	+	return _c
	455	+	}
	456	+
	457	+	func (_c MockCatalogue_Sync_Call) Return(err error) MockCatalogue_Sync_Call {
	458	+	_c.Call.Return(err)
	459	+	return _c
	460	+	}
	461	+
	462	+	func (_c MockCatalogue_Sync_Call) RunAndReturn(run func() error) MockCatalogue_Sync_Call {
	463	+	_c.Call.Return(run)
	464	+	return _c
	465	+	}
	466	+
423	467		// UpdateMetadata provides a mock function for the type MockCatalogue
424	468		func (_mock *MockCatalogue) UpdateMetadata(catalogueMetadataUpdate domain.CatalogueMetadataUpdate) error {
425	469		ret := _mock.Called(catalogueMetadataUpdate)


@@ -420,6 +420,50 @@ func (_c *MockCatalogue_Replace_Call) RunAndReturn(run func(s string, catalogueR
420	return _c	420		return _c
421	}	421		}
422		422
		423	+	// Sync provides a mock function for the type MockCatalogue
		424	+	func (_mock *MockCatalogue) Sync() error {
		425	+	ret := _mock.Called()
		426	+
		427	+	if len(ret) == 0 {
		428	+	panic("no return value specified for Sync")
		429	+	}
		430	+
		431	+	var r0 error
		432	+	if returnFunc, ok := ret.Get(0).(func() error); ok {
		433	+	r0 = returnFunc()
		434	+	} else {
		435	+	r0 = ret.Error(0)
		436	+	}
		437	+	return r0
		438	+	}
		439	+
		440	+	// MockCatalogue_Sync_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Sync'
		441	+	type MockCatalogue_Sync_Call struct {
		442	+	*mock.Call
		443	+	}
		444	+
		445	+	// Sync is a helper method to define mock.On call
		446	+	func (_e MockCatalogue_Expecter) Sync() MockCatalogue_Sync_Call {
		447	+	return &MockCatalogue_Sync_Call{Call: _e.mock.On("Sync")}
		448	+	}
		449	+
		450	+	func (_c MockCatalogue_Sync_Call) Run(run func()) MockCatalogue_Sync_Call {
		451	+	_c.Call.Run(func(args mock.Arguments) {
		452	+	run()
		453	+	})
		454	+	return _c
		455	+	}
		456	+
		457	+	func (_c MockCatalogue_Sync_Call) Return(err error) MockCatalogue_Sync_Call {
		458	+	_c.Call.Return(err)
		459	+	return _c
		460	+	}
		461	+
		462	+	func (_c MockCatalogue_Sync_Call) RunAndReturn(run func() error) MockCatalogue_Sync_Call {
		463	+	_c.Call.Return(run)
		464	+	return _c
		465	+	}
		466	+
423	// UpdateMetadata provides a mock function for the type MockCatalogue	467		// UpdateMetadata provides a mock function for the type MockCatalogue
424	func (_mock *MockCatalogue) UpdateMetadata(catalogueMetadataUpdate domain.CatalogueMetadataUpdate) error {	468		func (_mock *MockCatalogue) UpdateMetadata(catalogueMetadataUpdate domain.CatalogueMetadataUpdate) error {
425	ret := _mock.Called(catalogueMetadataUpdate)	469		ret := _mock.Called(catalogueMetadataUpdate)

M internal/ports/recovery.go +2 -0


@@ -34,6 +34,8 @@ type Catalogue interface {
34	34		Replace(string, domain.CatalogueRecord) error
35	35		Rename(string, domain.CatalogueRecord) error
36	36		Delete(string) error
	37	+	// Sync makes buffered metadata durable. Identity operations sync internally.
	38	+	Sync() error
37	39		Close() error
38	40		}
39	41


@@ -34,6 +34,8 @@ type Catalogue interface {
34	Replace(string, domain.CatalogueRecord) error	34		Replace(string, domain.CatalogueRecord) error
35	Rename(string, domain.CatalogueRecord) error	35		Rename(string, domain.CatalogueRecord) error
36	Delete(string) error	36		Delete(string) error
		37	+	// Sync makes buffered metadata durable. Identity operations sync internally.
		38	+	Sync() error
37	Close() error	39		Close() error
38	}	40		}
39		41

D internal/usecase/daemon/catalogue.go +0 -82

@@ -1,82 +0,0 @@
1	-	package daemon
2	-
3	-	import (
4	-	"github.com/bnema/vev/internal/domain"
5	-	)
6	-
7	-	// Catalogue helpers keep persistence optional for ephemeral and focused unit
8	-	// tests while ensuring every durable named-session mutation uses the authority
9	-	// port when one is configured.
10	-	func (d *Daemon) catalogueRecord(name string) (domain.CatalogueRecord, bool, error) {
11	-	if d == nil \|\| d.catalogue == nil {
12	-	return domain.CatalogueRecord{}, false, nil
13	-	}
14	-	return d.catalogue.Record(name)
15	-	}
16	-
17	-	func (d *Daemon) updateCatalogueMetadata(update domain.CatalogueMetadataUpdate) error {
18	-	if d == nil \|\| d.catalogue == nil {
19	-	return nil
20	-	}
21	-	return d.catalogue.UpdateMetadata(update)
22	-	}
23	-
24	-	// persistSessionMetadata serializes authority writes without holding daemon or
25	-	// session state locks during I/O. Only a successfully committed newer revision
26	-	// subsumes an older caller. Failed revisions retain their rollback until later
27	-	// failed mutations have unwound, keeping live metadata at the durable boundary.
28	-	func (d Daemon) persistSessionMetadata(sess session, version uint64, update domain.CatalogueMetadataUpdate, rollback func() bool) (bool, error) {
29	-	if sess == nil {
30	-	return false, nil
31	-	}
32	-	sess.metadataPersistMu.Lock()
33	-	defer sess.metadataPersistMu.Unlock()
34	-
35	-	if sess.metadataDurableVersion >= version {
36	-	return false, nil
37	-	}
38	-	if err := d.updateCatalogueMetadata(update); err != nil {
39	-	if sess.metadataFailedRollbacks == nil {
40	-	sess.metadataFailedRollbacks = make(map[uint64]func() bool)
41	-	}
42	-	sess.metadataFailedRollbacks[version] = rollback
43	-	// A later mutation may have been reserved while authority I/O blocked.
44	-	// It must settle before an older failed mutation can be unwound.
45	-	sess.mu.Lock()
46	-	current := sess.metadataVersion
47	-	sess.mu.Unlock()
48	-	if current > sess.metadataLiveVersion {
49	-	sess.metadataLiveVersion = current
50	-	}
51	-	rollbackRejected := sess.reconcileFailedMetadataLocked()
52	-	return rollbackRejected, err
53	-	}
54	-
55	-	sess.metadataDurableVersion = version
56	-	for failedVersion := range sess.metadataFailedRollbacks {
57	-	if failedVersion <= version {
58	-	delete(sess.metadataFailedRollbacks, failedVersion)
59	-	}
60	-	}
61	-	return false, nil
62	-	}
63	-
64	-	// reconcileFailedMetadataLocked unwinds only a contiguous failed suffix. An
65	-	// older failure remains pending while a newer mutation is still writing.
66	-	// Rollbacks run only after authority I/O has returned and acquire state locks in
67	-	// the daemon-before-session order.
68	-	func (s *session) reconcileFailedMetadataLocked() bool {
69	-	for s.metadataLiveVersion > s.metadataDurableVersion {
70	-	version := s.metadataLiveVersion
71	-	rollback, failed := s.metadataFailedRollbacks[version]
72	-	if !failed {
73	-	return false
74	-	}
75	-	if rollback != nil && !rollback() {
76	-	return true
77	-	}
78	-	delete(s.metadataFailedRollbacks, version)
79	-	s.metadataLiveVersion--
80	-	}
81	-	return false
82	-	}

@@ -1,82 +0,0 @@
1	-	package daemon
2	-
3	-	import (
4	-	"github.com/bnema/vev/internal/domain"
5	-	)
6	-
7	-	// Catalogue helpers keep persistence optional for ephemeral and focused unit
8	-	// tests while ensuring every durable named-session mutation uses the authority
9	-	// port when one is configured.
10	-	func (d *Daemon) catalogueRecord(name string) (domain.CatalogueRecord, bool, error) {
11	-	if d == nil \|\| d.catalogue == nil {
12	-	return domain.CatalogueRecord{}, false, nil
13	-	}
14	-	return d.catalogue.Record(name)
15	-	}
16	-
17	-	func (d *Daemon) updateCatalogueMetadata(update domain.CatalogueMetadataUpdate) error {
18	-	if d == nil \|\| d.catalogue == nil {
19	-	return nil
20	-	}
21	-	return d.catalogue.UpdateMetadata(update)
22	-	}
23	-
24	-	// persistSessionMetadata serializes authority writes without holding daemon or
25	-	// session state locks during I/O. Only a successfully committed newer revision
26	-	// subsumes an older caller. Failed revisions retain their rollback until later
27	-	// failed mutations have unwound, keeping live metadata at the durable boundary.
28	-	func (d Daemon) persistSessionMetadata(sess session, version uint64, update domain.CatalogueMetadataUpdate, rollback func() bool) (bool, error) {
29	-	if sess == nil {
30	-	return false, nil
31	-	}
32	-	sess.metadataPersistMu.Lock()
33	-	defer sess.metadataPersistMu.Unlock()
34	-
35	-	if sess.metadataDurableVersion >= version {
36	-	return false, nil
37	-	}
38	-	if err := d.updateCatalogueMetadata(update); err != nil {
39	-	if sess.metadataFailedRollbacks == nil {
40	-	sess.metadataFailedRollbacks = make(map[uint64]func() bool)
41	-	}
42	-	sess.metadataFailedRollbacks[version] = rollback
43	-	// A later mutation may have been reserved while authority I/O blocked.
44	-	// It must settle before an older failed mutation can be unwound.
45	-	sess.mu.Lock()
46	-	current := sess.metadataVersion
47	-	sess.mu.Unlock()
48	-	if current > sess.metadataLiveVersion {
49	-	sess.metadataLiveVersion = current
50	-	}
51	-	rollbackRejected := sess.reconcileFailedMetadataLocked()
52	-	return rollbackRejected, err
53	-	}
54	-
55	-	sess.metadataDurableVersion = version
56	-	for failedVersion := range sess.metadataFailedRollbacks {
57	-	if failedVersion <= version {
58	-	delete(sess.metadataFailedRollbacks, failedVersion)
59	-	}
60	-	}
61	-	return false, nil
62	-	}
63	-
64	-	// reconcileFailedMetadataLocked unwinds only a contiguous failed suffix. An
65	-	// older failure remains pending while a newer mutation is still writing.
66	-	// Rollbacks run only after authority I/O has returned and acquire state locks in
67	-	// the daemon-before-session order.
68	-	func (s *session) reconcileFailedMetadataLocked() bool {
69	-	for s.metadataLiveVersion > s.metadataDurableVersion {
70	-	version := s.metadataLiveVersion
71	-	rollback, failed := s.metadataFailedRollbacks[version]
72	-	if !failed {
73	-	return false
74	-	}
75	-	if rollback != nil && !rollback() {
76	-	return true
77	-	}
78	-	delete(s.metadataFailedRollbacks, version)
79	-	s.metadataLiveVersion--
80	-	}
81	-	return false
82	-	}

M internal/usecase/daemon/daemon.go +32 -6


@@ -265,6 +265,31 @@ func WithCatalogue(catalogue ports.Catalogue, records []domain.CatalogueRecord)
265	265		}
266	266		}
267	267
	268	+	func (d *Daemon) catalogueRecord(name string) (domain.CatalogueRecord, bool, error) {
	269	+	if d == nil \|\| d.catalogue == nil {
	270	+	return domain.CatalogueRecord{}, false, nil
	271	+	}
	272	+	return d.catalogue.Record(name)
	273	+	}
	274	+
	275	+	// markCatalogueDirty buffers metadata. The periodic catalogue timer, the next
	276	+	// identity write, or shutdown provides the durability barrier.
	277	+	func (d *Daemon) markCatalogueDirty(update domain.CatalogueMetadataUpdate) {
	278	+	if d == nil \|\| d.catalogue == nil {
	279	+	return
	280	+	}
	281	+	if err := d.catalogue.UpdateMetadata(update); err != nil {
	282	+	d.log.Warn("buffering session metadata failed", "err", err, "session", update.Name)
	283	+	}
	284	+	}
	285	+
	286	+	func (d *Daemon) flushCatalogue() error {
	287	+	if d == nil \|\| d.catalogue == nil {
	288	+	return nil
	289	+	}
	290	+	return d.catalogue.Sync()
	291	+	}
	292	+
268	293		// WithRecoveryCoordinator installs the durable recovery coordinator.
269	294		func WithRecoveryCoordinator(coordinator *recoveryusecase.Coordinator) Option {
270	295		return func(d *Daemon) {

@@ -510,7 +535,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
510	535		d.sessWg.Go(func() {
511	536		d.barScriptPoller(d.serveCtx)
512	537		})
513		-	if d.persistEnabled && d.procCwd != nil {
	538	+	if d.persistEnabled {
514	539		d.sessWg.Go(func() {
515	540		d.cwdSampler(d.serveCtx)
516	541		})

@@ -589,12 +614,13 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
589	614		d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
590	615		d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
591	616		d.waitNotifies()
592		-	var closeErr error
593		-	if d.catalogue != nil {
594		-	closeErr = d.catalogue.Close()
	617	+	if err := d.flushCatalogue(); err != nil {
	618	+	d.log.Warn("flushing session catalogue at shutdown failed", "err", err)
595	619		}
596		-	if closeErr != nil {
597		-	d.log.Warn("closing session catalogue failed", "err", closeErr)
	620	+	if d.catalogue != nil {
	621	+	if err := d.catalogue.Close(); err != nil {
	622	+	d.log.Warn("closing session catalogue failed", "err", err)
	623	+	}
598	624		}
599	625		return nil
600	626		}


@@ -265,6 +265,31 @@ func WithCatalogue(catalogue ports.Catalogue, records []domain.CatalogueRecord)
265		}	265		}
266		}	266		}
267			267
			268	+	func (d *Daemon) catalogueRecord(name string) (domain.CatalogueRecord, bool, error) {
			269	+	if d == nil \|\| d.catalogue == nil {
			270	+	return domain.CatalogueRecord{}, false, nil
			271	+	}
			272	+	return d.catalogue.Record(name)
			273	+	}
			274	+
			275	+	// markCatalogueDirty buffers metadata. The periodic catalogue timer, the next
			276	+	// identity write, or shutdown provides the durability barrier.
			277	+	func (d *Daemon) markCatalogueDirty(update domain.CatalogueMetadataUpdate) {
			278	+	if d == nil \|\| d.catalogue == nil {
			279	+	return
			280	+	}
			281	+	if err := d.catalogue.UpdateMetadata(update); err != nil {
			282	+	d.log.Warn("buffering session metadata failed", "err", err, "session", update.Name)
			283	+	}
			284	+	}
			285	+
			286	+	func (d *Daemon) flushCatalogue() error {
			287	+	if d == nil \|\| d.catalogue == nil {
			288	+	return nil
			289	+	}
			290	+	return d.catalogue.Sync()
			291	+	}
			292	+
268		// WithRecoveryCoordinator installs the durable recovery coordinator.	293		// WithRecoveryCoordinator installs the durable recovery coordinator.
269		func WithRecoveryCoordinator(coordinator *recoveryusecase.Coordinator) Option {	294		func WithRecoveryCoordinator(coordinator *recoveryusecase.Coordinator) Option {
270		return func(d *Daemon) {	295		return func(d *Daemon) {

@@ -510,7 +535,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
510		d.sessWg.Go(func() {	535		d.sessWg.Go(func() {
511		d.barScriptPoller(d.serveCtx)	536		d.barScriptPoller(d.serveCtx)
512		})	537		})
513	-	if d.persistEnabled && d.procCwd != nil {	538	+	if d.persistEnabled {
514		d.sessWg.Go(func() {	539		d.sessWg.Go(func() {
515		d.cwdSampler(d.serveCtx)	540		d.cwdSampler(d.serveCtx)
516		})	541		})

@@ -589,12 +614,13 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
589		d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)	614		d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
590		d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)	615		d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
591		d.waitNotifies()	616		d.waitNotifies()
592	-	var closeErr error	617	+	if err := d.flushCatalogue(); err != nil {
593	-	if d.catalogue != nil {	618	+	d.log.Warn("flushing session catalogue at shutdown failed", "err", err)
594	-	closeErr = d.catalogue.Close()
595		}	619		}
596	-	if closeErr != nil {	620	+	if d.catalogue != nil {
597	-	d.log.Warn("closing session catalogue failed", "err", closeErr)	621	+	if err := d.catalogue.Close(); err != nil {
			622	+	d.log.Warn("closing session catalogue failed", "err", err)
			623	+	}
598		}	624		}
599		return nil	625		return nil
600		}	626		}

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


@@ -115,6 +115,7 @@ func (c *durableRecoveryCatalogue) MetadataUpdates() []domain.CatalogueMetadataU
115	115		return append([]domain.CatalogueMetadataUpdate(nil), c.metadataUpdates...)
116	116		}
117	117
	118	+	func (c *durableRecoveryCatalogue) Sync() error { return nil }
118	119		func (c *durableRecoveryCatalogue) Close() error { return nil }
119	120
120	121		type durableRecoveryRepository struct {


@@ -115,6 +115,7 @@ func (c *durableRecoveryCatalogue) MetadataUpdates() []domain.CatalogueMetadataU
115	return append([]domain.CatalogueMetadataUpdate(nil), c.metadataUpdates...)	115		return append([]domain.CatalogueMetadataUpdate(nil), c.metadataUpdates...)
116	}	116		}
117		117
		118	+	func (c *durableRecoveryCatalogue) Sync() error { return nil }
118	func (c *durableRecoveryCatalogue) Close() error { return nil }	119		func (c *durableRecoveryCatalogue) Close() error { return nil }
119		120
120	type durableRecoveryRepository struct {	121		type durableRecoveryRepository struct {

M internal/usecase/daemon/metadata_order_test.go +52 -251

@@ -1,294 +1,95 @@
1	1		package daemon
2	2
3	3		import (
4		-	"errors"
	4	+	"bytes"
5	5		"sync"
6	6		"testing"
7		-	"time"
8	7
9	8		"github.com/stretchr/testify/require"
10	9
11	10		"github.com/bnema/vev/internal/domain"
12	11		"github.com/bnema/vev/internal/ports"
	12	+	recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
13	13		)
14	14
15		-	type blockingMetadataCatalogue struct {
16		-	ports.Catalogue
	15	+	type syncCountingCatalogue struct {
	16	+	*durableRecoveryCatalogue
17	17
18		-	mu sync.Mutex
19		-	calls int
20		-	firstStarted chan struct{}
21		-	releaseFirst chan struct{}
22		-	firstErr error
23		-	secondErr error
	18	+	muSync sync.Mutex
	19	+	syncs int
24	20		}
25	21
26		-	func (c *blockingMetadataCatalogue) UpdateMetadata(update domain.CatalogueMetadataUpdate) error {
27		-	c.mu.Lock()
28		-	c.calls++
29		-	call := c.calls
30		-	c.mu.Unlock()
31		-	if call == 1 {
32		-	close(c.firstStarted)
33		-	<-c.releaseFirst
34		-	if c.firstErr != nil {
35		-	return c.firstErr
36		-	}
37		-	}
38		-	if call == 2 && c.secondErr != nil {
39		-	return c.secondErr
40		-	}
41		-	return c.Catalogue.UpdateMetadata(update)
42		-	}
43		-
44		-	func newBlockingCatalogue(sess session, firstErr error) (durableRecoveryCatalogue, *blockingMetadataCatalogue) {
45		-	sess.mu.Lock()
46		-	record := sess.persistRecordLocked(sess.createdAt)
47		-	sess.mu.Unlock()
48		-	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
49		-	return catalogue, &blockingMetadataCatalogue{
50		-	Catalogue: catalogue,
51		-	firstStarted: make(chan struct{}),
52		-	releaseFirst: make(chan struct{}),
53		-	firstErr: firstErr,
54		-	}
55		-	}
56		-
57		-	func TestMetadataWritesPreserveLaterMutationWhenFirstWriteIsBlocked(t *testing.T) {
58		-	firstFailure := errors.New("first metadata write failed")
59		-	for _, tt := range []struct {
60		-	name string
61		-	firstErr error
62		-	}{
63		-	{name: "delayed success"},
64		-	{name: "delayed failure does not roll back newer state", firstErr: firstFailure},
65		-	} {
66		-	t.Run(tt.name, func(t *testing.T) {
67		-	sess := newSnapshotTestSession(t, "work", false, "/work")
68		-	catalogue, blocking := newBlockingCatalogue(sess, tt.firstErr)
69		-	d := newTestDaemon(t, nil, stubClock{})
70		-	d.catalogue = blocking
71		-	d.sessions[sess.id] = sess
72		-
73		-	firstDone := make(chan error, 1)
74		-	go func() { firstDone <- d.renameTab(sess, sess.tabs[0], "first") }()
75		-	select {
76		-	case <-blocking.firstStarted:
77		-	case <-time.After(testWaitTimeout):
78		-	t.Fatal("first metadata write did not block")
79		-	}
80		-
81		-	secondDone := make(chan error, 1)
82		-	go func() { secondDone <- d.renameTab(sess, sess.tabs[0], "second") }()
83		-	require.Eventually(t, func() bool {
84		-	sess.mu.Lock()
85		-	defer sess.mu.Unlock()
86		-	return sess.tabs[0].name == "second"
87		-	}, testWaitTimeout, time.Millisecond, "later mutation did not proceed while storage was blocked")
88		-
89		-	close(blocking.releaseFirst)
90		-	firstResult := <-firstDone
91		-	if tt.firstErr == nil {
92		-	require.NoError(t, firstResult)
93		-	} else {
94		-	require.ErrorIs(t, firstResult, tt.firstErr, "an in-memory revision cannot subsume a failed durable write")
95		-	}
96		-	require.NoError(t, <-secondDone)
97		-
98		-	persisted, ok, err := catalogue.Record("work")
99		-	require.NoError(t, err)
100		-	require.True(t, ok)
101		-	require.Equal(t, []string{"second"}, persisted.TabNames)
102		-	require.Equal(t, "second", sess.tabs[0].name)
103		-	})
104		-	}
	22	+	func (c *syncCountingCatalogue) Sync() error {
	23	+	c.muSync.Lock()
	24	+	defer c.muSync.Unlock()
	25	+	c.syncs++
	26	+	return nil
105	27		}
106	28
107		-	func TestTwoConsecutiveMetadataFailuresAreBothReported(t *testing.T) {
108		-	firstFailure := errors.New("first metadata write failed")
109		-	secondFailure := errors.New("second metadata write failed")
110		-	sess := newSnapshotTestSession(t, "work", false, "/work")
111		-	catalogue, blocking := newBlockingCatalogue(sess, firstFailure)
112		-	blocking.secondErr = secondFailure
113		-	close(blocking.releaseFirst)
114		-	d := newTestDaemon(t, nil, stubClock{})
115		-	d.catalogue = blocking
116		-	d.sessions[sess.id] = sess
117		-
118		-	require.ErrorIs(t, d.renameTab(sess, sess.tabs[0], "first"), firstFailure)
119		-	require.ErrorIs(t, d.renameTab(sess, sess.tabs[0], "second"), secondFailure)
120		-
121		-	persisted, ok, err := catalogue.Record("work")
122		-	require.NoError(t, err)
123		-	require.True(t, ok)
124		-	require.Empty(t, persisted.TabNames)
125		-	sess.mu.Lock()
126		-	require.Empty(t, sess.tabs[0].name)
127		-	sess.mu.Unlock()
	29	+	func (c *syncCountingCatalogue) syncCount() int {
	30	+	c.muSync.Lock()
	31	+	defer c.muSync.Unlock()
	32	+	return c.syncs
128	33		}
129	34
130		-	func TestConsecutiveMetadataFailuresRestoreDurableState(t *testing.T) {
131		-	firstFailure := errors.New("first metadata write failed")
132		-	secondFailure := errors.New("second metadata write failed")
133		-	sess := newSnapshotTestSession(t, "work", false, "/work")
134		-	catalogue, blocking := newBlockingCatalogue(sess, firstFailure)
135		-	blocking.secondErr = secondFailure
136		-	d := newTestDaemon(t, nil, stubClock{})
137		-	d.catalogue = blocking
138		-	d.sessions[sess.id] = sess
139		-
140		-	firstDone := make(chan error, 1)
141		-	go func() { firstDone <- d.renameTab(sess, sess.tabs[0], "first") }()
142		-	select {
143		-	case <-blocking.firstStarted:
144		-	case <-time.After(testWaitTimeout):
145		-	t.Fatal("first metadata write did not block")
	35	+	func (c *syncCountingCatalogue) Create(record domain.CatalogueRecord) error {
	36	+	c.mu.Lock()
	37	+	if _, exists := c.records[record.Name]; exists {
	38	+	c.mu.Unlock()
	39	+	return errSessionNameInUse
146	40		}
147		-	secondDone := make(chan error, 1)
148		-	go func() { secondDone <- d.renameTab(sess, sess.tabs[0], "second") }()
149		-	require.Eventually(t, func() bool {
150		-	sess.mu.Lock()
151		-	defer sess.mu.Unlock()
152		-	return sess.tabs[0].name == "second"
153		-	}, testWaitTimeout, time.Millisecond)
154		-
155		-	close(blocking.releaseFirst)
156		-	require.ErrorIs(t, <-firstDone, firstFailure)
157		-	require.ErrorIs(t, <-secondDone, secondFailure)
158		-
159		-	persisted, ok, err := catalogue.Record("work")
160		-	require.NoError(t, err)
161		-	require.True(t, ok)
162		-	require.Empty(t, persisted.TabNames)
163		-	sess.mu.Lock()
164		-	require.Empty(t, sess.tabs[0].name)
165		-	sess.mu.Unlock()
	41	+	c.records[record.Name] = record
	42	+	c.mu.Unlock()
	43	+	return c.Sync()
166	44		}
167	45
168		-	func TestNewerMetadataFailureDoesNotSkipOlderSnapshot(t *testing.T) {
169		-	failure := errors.New("newer metadata write failed")
	46	+	func TestMetadataWritesAreDeferred(t *testing.T) {
	47	+	t.Parallel()
170	48		sess := newSnapshotTestSession(t, "work", false, "/work")
171	49		sess.mu.Lock()
172	50		record := sess.persistRecordLocked(sess.createdAt)
173	51		sess.mu.Unlock()
174		-	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
175		-	blocking := &blockingMetadataCatalogue{
176		-	Catalogue: catalogue,
177		-	firstStarted: make(chan struct{}),
178		-	releaseFirst: make(chan struct{}),
179		-	firstErr: failure,
180		-	}
181		-	close(blocking.releaseFirst)
	52	+	cat := &syncCountingCatalogue{durableRecoveryCatalogue: newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})}
182	53		d := newTestDaemon(t, nil, stubClock{})
183		-	d.catalogue = blocking
	54	+	d.catalogue = cat
	55	+	d.persistEnabled = true
184	56		d.sessions[sess.id] = sess
185		-	tb := sess.tabs[0]
186		-
187		-	sess.mu.Lock()
188		-	tb.name = "first"
189		-	firstRecord, firstVersion := sess.nextPersistRecordLocked(sess.createdAt + 1)
190		-	tb.name = "second"
191		-	secondRecord, secondVersion := sess.nextPersistRecordLocked(sess.createdAt + 2)
192		-	sess.mu.Unlock()
193	57
194		-	secondRollback := func() bool {
195		-	sess.mu.Lock()
196		-	defer sess.mu.Unlock()
197		-	if tb.name != "second" {
198		-	return false
199		-	}
200		-	tb.name = "first"
201		-	return true
202		-	}
203		-	firstRollback := func() bool {
204		-	sess.mu.Lock()
205		-	defer sess.mu.Unlock()
206		-	if tb.name != "first" {
207		-	return false
208		-	}
209		-	tb.name = ""
210		-	return true
211		-	}
	58	+	syncsBefore := cat.syncCount()
	59	+	require.NoError(t, d.renameTab(sess, sess.tabs[0], "build"))
	60	+	d.touchMRU(sess)
	61	+	require.Equal(t, syncsBefore, cat.syncCount(), "metadata updates must not force a durable write")
212	62
213		-	_, err := d.persistSessionMetadata(sess, secondVersion, secondRecord.MetadataUpdate(), secondRollback)
214		-	require.ErrorIs(t, err, failure)
215		-	_, err = d.persistSessionMetadata(sess, firstVersion, firstRecord.MetadataUpdate(), firstRollback)
216		-	require.NoError(t, err)
	63	+	require.NoError(t, d.flushCatalogue())
	64	+	require.Greater(t, cat.syncCount(), syncsBefore)
217	65
218		-	persisted, ok, err := catalogue.Record("work")
	66	+	got, ok, err := cat.Record("work")
219	67		require.NoError(t, err)
220	68		require.True(t, ok)
221		-	require.Equal(t, []string{"first"}, persisted.TabNames)
222		-	sess.mu.Lock()
223		-	require.Equal(t, "first", tb.name)
224		-	sess.mu.Unlock()
	69	+	require.Equal(t, []string{"build"}, got.TabNames)
225	70		}
226	71
227		-	func TestLatestMetadataFailureRollsBackRename(t *testing.T) {
228		-	failure := errors.New("metadata write failed")
229		-	sess := newSnapshotTestSession(t, "work", false, "/work")
230		-	catalogue, blocking := newBlockingCatalogue(sess, failure)
	72	+	func TestIdentityWritesAreSynchronous(t *testing.T) {
	73	+	t.Parallel()
	74	+	sess := newSnapshotTestSession(t, "0", true, "/work")
	75	+	cat := &syncCountingCatalogue{durableRecoveryCatalogue: newDurableRecoveryCatalogue(nil)}
231	76		d := newTestDaemon(t, nil, stubClock{})
232		-	d.catalogue = blocking
	77	+	d.catalogue = cat
	78	+	d.persistEnabled = true
	79	+	d.recovery = recoveryusecase.NewCoordinator(cat, nil, bytes.NewReader(bytes.Repeat([]byte{1}, 16)))
233	80		d.sessions[sess.id] = sess
234	81
235		-	done := make(chan error, 1)
236		-	go func() { done <- d.renameTab(sess, sess.tabs[0], "failed") }()
237		-	select {
238		-	case <-blocking.firstStarted:
239		-	case <-time.After(testWaitTimeout):
240		-	t.Fatal("metadata write did not block")
241		-	}
242		-	close(blocking.releaseFirst)
243		-	require.ErrorIs(t, <-done, failure)
	82	+	syncsBefore := cat.syncCount()
	83	+	require.NoError(t, d.renameSession(sess, "work"))
	84	+	require.Greater(t, cat.syncCount(), syncsBefore, "creating a named session must be durable before it is exposed")
244	85
245		-	persisted, ok, err := catalogue.Record("work")
246		-	require.NoError(t, err)
247		-	require.True(t, ok)
248		-	require.Empty(t, persisted.TabNames)
	86	+	d.mu.Lock()
	87	+	published := d.findByNameLocked("work")
	88	+	d.mu.Unlock()
	89	+	require.Same(t, sess, published)
249	90		sess.mu.Lock()
250		-	require.Empty(t, sess.tabs[0].name)
	91	+	require.False(t, sess.ephemeral)
251	92		sess.mu.Unlock()
252	93		}
253	94
254		-	func TestBlockedCwdWriteCannotOverwriteLaterTabMetadata(t *testing.T) {
255		-	sess := newSnapshotTestSession(t, "work", false, "/work")
256		-	catalogue, blocking := newBlockingCatalogue(sess, nil)
257		-	d := newTestDaemon(t, nil, stubClock{})
258		-	d.catalogue = blocking
259		-	d.sessions[sess.id] = sess
260		-	d.procCwd = func(int) (string, error) { return "/later-cwd", nil }
261		-
262		-	cwdDone := make(chan struct{})
263		-	go func() {
264		-	d.refreshSessionCwd(sess)
265		-	close(cwdDone)
266		-	}()
267		-	select {
268		-	case <-blocking.firstStarted:
269		-	case <-time.After(testWaitTimeout):
270		-	t.Fatal("cwd metadata write did not block")
271		-	}
272		-
273		-	renameDone := make(chan error, 1)
274		-	go func() { renameDone <- d.renameTab(sess, sess.tabs[0], "later-tab") }()
275		-	require.Eventually(t, func() bool {
276		-	sess.mu.Lock()
277		-	defer sess.mu.Unlock()
278		-	return sess.tabs[0].name == "later-tab"
279		-	}, testWaitTimeout, time.Millisecond)
280		-
281		-	close(blocking.releaseFirst)
282		-	select {
283		-	case <-cwdDone:
284		-	case <-time.After(testWaitTimeout):
285		-	t.Fatal("cwd refresh deadlocked")
286		-	}
287		-	require.NoError(t, <-renameDone)
288		-
289		-	persisted, ok, err := catalogue.Record("work")
290		-	require.NoError(t, err)
291		-	require.True(t, ok)
292		-	require.Equal(t, "/later-cwd", persisted.Cwd)
293		-	require.Equal(t, []string{"later-tab"}, persisted.TabNames)
294		-	}
	95	+	var _ ports.Catalogue = (*syncCountingCatalogue)(nil)

@@ -1,294 +1,95 @@
1		package daemon	1		package daemon
2			2
3		import (	3		import (
4	-	"errors"	4	+	"bytes"
5		"sync"	5		"sync"
6		"testing"	6		"testing"
7	-	"time"
8			7
9		"github.com/stretchr/testify/require"	8		"github.com/stretchr/testify/require"
10			9
11		"github.com/bnema/vev/internal/domain"	10		"github.com/bnema/vev/internal/domain"
12		"github.com/bnema/vev/internal/ports"	11		"github.com/bnema/vev/internal/ports"
			12	+	recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
13		)	13		)
14			14
15	-	type blockingMetadataCatalogue struct {	15	+	type syncCountingCatalogue struct {
16	-	ports.Catalogue	16	+	*durableRecoveryCatalogue
17			17
18	-	mu sync.Mutex	18	+	muSync sync.Mutex
19	-	calls int	19	+	syncs int
20	-	firstStarted chan struct{}
21	-	releaseFirst chan struct{}
22	-	firstErr error
23	-	secondErr error
24		}	20		}
25			21
26	-	func (c *blockingMetadataCatalogue) UpdateMetadata(update domain.CatalogueMetadataUpdate) error {	22	+	func (c *syncCountingCatalogue) Sync() error {
27	-	c.mu.Lock()	23	+	c.muSync.Lock()
28	-	c.calls++	24	+	defer c.muSync.Unlock()
29	-	call := c.calls	25	+	c.syncs++
30	-	c.mu.Unlock()	26	+	return nil
31	-	if call == 1 {
32	-	close(c.firstStarted)
33	-	<-c.releaseFirst
34	-	if c.firstErr != nil {
35	-	return c.firstErr
36	-	}
37	-	}
38	-	if call == 2 && c.secondErr != nil {
39	-	return c.secondErr
40	-	}
41	-	return c.Catalogue.UpdateMetadata(update)
42	-	}
43	-
44	-	func newBlockingCatalogue(sess session, firstErr error) (durableRecoveryCatalogue, *blockingMetadataCatalogue) {
45	-	sess.mu.Lock()
46	-	record := sess.persistRecordLocked(sess.createdAt)
47	-	sess.mu.Unlock()
48	-	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
49	-	return catalogue, &blockingMetadataCatalogue{
50	-	Catalogue: catalogue,
51	-	firstStarted: make(chan struct{}),
52	-	releaseFirst: make(chan struct{}),
53	-	firstErr: firstErr,
54	-	}
55	-	}
56	-
57	-	func TestMetadataWritesPreserveLaterMutationWhenFirstWriteIsBlocked(t *testing.T) {
58	-	firstFailure := errors.New("first metadata write failed")
59	-	for _, tt := range []struct {
60	-	name string
61	-	firstErr error
62	-	}{
63	-	{name: "delayed success"},
64	-	{name: "delayed failure does not roll back newer state", firstErr: firstFailure},
65	-	} {
66	-	t.Run(tt.name, func(t *testing.T) {
67	-	sess := newSnapshotTestSession(t, "work", false, "/work")
68	-	catalogue, blocking := newBlockingCatalogue(sess, tt.firstErr)
69	-	d := newTestDaemon(t, nil, stubClock{})
70	-	d.catalogue = blocking
71	-	d.sessions[sess.id] = sess
72	-
73	-	firstDone := make(chan error, 1)
74	-	go func() { firstDone <- d.renameTab(sess, sess.tabs[0], "first") }()
75	-	select {
76	-	case <-blocking.firstStarted:
77	-	case <-time.After(testWaitTimeout):
78	-	t.Fatal("first metadata write did not block")
79	-	}
80	-
81	-	secondDone := make(chan error, 1)
82	-	go func() { secondDone <- d.renameTab(sess, sess.tabs[0], "second") }()
83	-	require.Eventually(t, func() bool {
84	-	sess.mu.Lock()
85	-	defer sess.mu.Unlock()
86	-	return sess.tabs[0].name == "second"
87	-	}, testWaitTimeout, time.Millisecond, "later mutation did not proceed while storage was blocked")
88	-
89	-	close(blocking.releaseFirst)
90	-	firstResult := <-firstDone
91	-	if tt.firstErr == nil {
92	-	require.NoError(t, firstResult)
93	-	} else {
94	-	require.ErrorIs(t, firstResult, tt.firstErr, "an in-memory revision cannot subsume a failed durable write")
95	-	}
96	-	require.NoError(t, <-secondDone)
97	-
98	-	persisted, ok, err := catalogue.Record("work")
99	-	require.NoError(t, err)
100	-	require.True(t, ok)
101	-	require.Equal(t, []string{"second"}, persisted.TabNames)
102	-	require.Equal(t, "second", sess.tabs[0].name)
103	-	})
104	-	}
105		}	27		}
106			28
107	-	func TestTwoConsecutiveMetadataFailuresAreBothReported(t *testing.T) {	29	+	func (c *syncCountingCatalogue) syncCount() int {
108	-	firstFailure := errors.New("first metadata write failed")	30	+	c.muSync.Lock()
109	-	secondFailure := errors.New("second metadata write failed")	31	+	defer c.muSync.Unlock()
110	-	sess := newSnapshotTestSession(t, "work", false, "/work")	32	+	return c.syncs
111	-	catalogue, blocking := newBlockingCatalogue(sess, firstFailure)
112	-	blocking.secondErr = secondFailure
113	-	close(blocking.releaseFirst)
114	-	d := newTestDaemon(t, nil, stubClock{})
115	-	d.catalogue = blocking
116	-	d.sessions[sess.id] = sess
117	-
118	-	require.ErrorIs(t, d.renameTab(sess, sess.tabs[0], "first"), firstFailure)
119	-	require.ErrorIs(t, d.renameTab(sess, sess.tabs[0], "second"), secondFailure)
120	-
121	-	persisted, ok, err := catalogue.Record("work")
122	-	require.NoError(t, err)
123	-	require.True(t, ok)
124	-	require.Empty(t, persisted.TabNames)
125	-	sess.mu.Lock()
126	-	require.Empty(t, sess.tabs[0].name)
127	-	sess.mu.Unlock()
128		}	33		}
129			34
130	-	func TestConsecutiveMetadataFailuresRestoreDurableState(t *testing.T) {	35	+	func (c *syncCountingCatalogue) Create(record domain.CatalogueRecord) error {
131	-	firstFailure := errors.New("first metadata write failed")	36	+	c.mu.Lock()
132	-	secondFailure := errors.New("second metadata write failed")	37	+	if _, exists := c.records[record.Name]; exists {
133	-	sess := newSnapshotTestSession(t, "work", false, "/work")	38	+	c.mu.Unlock()
134	-	catalogue, blocking := newBlockingCatalogue(sess, firstFailure)	39	+	return errSessionNameInUse
135	-	blocking.secondErr = secondFailure
136	-	d := newTestDaemon(t, nil, stubClock{})
137	-	d.catalogue = blocking
138	-	d.sessions[sess.id] = sess
139	-
140	-	firstDone := make(chan error, 1)
141	-	go func() { firstDone <- d.renameTab(sess, sess.tabs[0], "first") }()
142	-	select {
143	-	case <-blocking.firstStarted:
144	-	case <-time.After(testWaitTimeout):
145	-	t.Fatal("first metadata write did not block")
146		}	40		}
147	-	secondDone := make(chan error, 1)	41	+	c.records[record.Name] = record
148	-	go func() { secondDone <- d.renameTab(sess, sess.tabs[0], "second") }()	42	+	c.mu.Unlock()
149	-	require.Eventually(t, func() bool {	43	+	return c.Sync()
150	-	sess.mu.Lock()
151	-	defer sess.mu.Unlock()
152	-	return sess.tabs[0].name == "second"
153	-	}, testWaitTimeout, time.Millisecond)
154	-
155	-	close(blocking.releaseFirst)
156	-	require.ErrorIs(t, <-firstDone, firstFailure)
157	-	require.ErrorIs(t, <-secondDone, secondFailure)
158	-
159	-	persisted, ok, err := catalogue.Record("work")
160	-	require.NoError(t, err)
161	-	require.True(t, ok)
162	-	require.Empty(t, persisted.TabNames)
163	-	sess.mu.Lock()
164	-	require.Empty(t, sess.tabs[0].name)
165	-	sess.mu.Unlock()
166		}	44		}
167			45
168	-	func TestNewerMetadataFailureDoesNotSkipOlderSnapshot(t *testing.T) {	46	+	func TestMetadataWritesAreDeferred(t *testing.T) {
169	-	failure := errors.New("newer metadata write failed")	47	+	t.Parallel()
170		sess := newSnapshotTestSession(t, "work", false, "/work")	48		sess := newSnapshotTestSession(t, "work", false, "/work")
171		sess.mu.Lock()	49		sess.mu.Lock()
172		record := sess.persistRecordLocked(sess.createdAt)	50		record := sess.persistRecordLocked(sess.createdAt)
173		sess.mu.Unlock()	51		sess.mu.Unlock()
174	-	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})	52	+	cat := &syncCountingCatalogue{durableRecoveryCatalogue: newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})}
175	-	blocking := &blockingMetadataCatalogue{
176	-	Catalogue: catalogue,
177	-	firstStarted: make(chan struct{}),
178	-	releaseFirst: make(chan struct{}),
179	-	firstErr: failure,
180	-	}
181	-	close(blocking.releaseFirst)
182		d := newTestDaemon(t, nil, stubClock{})	53		d := newTestDaemon(t, nil, stubClock{})
183	-	d.catalogue = blocking	54	+	d.catalogue = cat
			55	+	d.persistEnabled = true
184		d.sessions[sess.id] = sess	56		d.sessions[sess.id] = sess
185	-	tb := sess.tabs[0]
186	-
187	-	sess.mu.Lock()
188	-	tb.name = "first"
189	-	firstRecord, firstVersion := sess.nextPersistRecordLocked(sess.createdAt + 1)
190	-	tb.name = "second"
191	-	secondRecord, secondVersion := sess.nextPersistRecordLocked(sess.createdAt + 2)
192	-	sess.mu.Unlock()
193			57
194	-	secondRollback := func() bool {	58	+	syncsBefore := cat.syncCount()
195	-	sess.mu.Lock()	59	+	require.NoError(t, d.renameTab(sess, sess.tabs[0], "build"))
196	-	defer sess.mu.Unlock()	60	+	d.touchMRU(sess)
197	-	if tb.name != "second" {	61	+	require.Equal(t, syncsBefore, cat.syncCount(), "metadata updates must not force a durable write")
198	-	return false
199	-	}
200	-	tb.name = "first"
201	-	return true
202	-	}
203	-	firstRollback := func() bool {
204	-	sess.mu.Lock()
205	-	defer sess.mu.Unlock()
206	-	if tb.name != "first" {
207	-	return false
208	-	}
209	-	tb.name = ""
210	-	return true
211	-	}
212			62
213	-	_, err := d.persistSessionMetadata(sess, secondVersion, secondRecord.MetadataUpdate(), secondRollback)	63	+	require.NoError(t, d.flushCatalogue())
214	-	require.ErrorIs(t, err, failure)	64	+	require.Greater(t, cat.syncCount(), syncsBefore)
215	-	_, err = d.persistSessionMetadata(sess, firstVersion, firstRecord.MetadataUpdate(), firstRollback)
216	-	require.NoError(t, err)
217			65
218	-	persisted, ok, err := catalogue.Record("work")	66	+	got, ok, err := cat.Record("work")
219		require.NoError(t, err)	67		require.NoError(t, err)
220		require.True(t, ok)	68		require.True(t, ok)
221	-	require.Equal(t, []string{"first"}, persisted.TabNames)	69	+	require.Equal(t, []string{"build"}, got.TabNames)
222	-	sess.mu.Lock()
223	-	require.Equal(t, "first", tb.name)
224	-	sess.mu.Unlock()
225		}	70		}
226			71
227	-	func TestLatestMetadataFailureRollsBackRename(t *testing.T) {	72	+	func TestIdentityWritesAreSynchronous(t *testing.T) {
228	-	failure := errors.New("metadata write failed")	73	+	t.Parallel()
229	-	sess := newSnapshotTestSession(t, "work", false, "/work")	74	+	sess := newSnapshotTestSession(t, "0", true, "/work")
230	-	catalogue, blocking := newBlockingCatalogue(sess, failure)	75	+	cat := &syncCountingCatalogue{durableRecoveryCatalogue: newDurableRecoveryCatalogue(nil)}
231		d := newTestDaemon(t, nil, stubClock{})	76		d := newTestDaemon(t, nil, stubClock{})
232	-	d.catalogue = blocking	77	+	d.catalogue = cat
			78	+	d.persistEnabled = true
			79	+	d.recovery = recoveryusecase.NewCoordinator(cat, nil, bytes.NewReader(bytes.Repeat([]byte{1}, 16)))
233		d.sessions[sess.id] = sess	80		d.sessions[sess.id] = sess
234			81
235	-	done := make(chan error, 1)	82	+	syncsBefore := cat.syncCount()
236	-	go func() { done <- d.renameTab(sess, sess.tabs[0], "failed") }()	83	+	require.NoError(t, d.renameSession(sess, "work"))
237	-	select {	84	+	require.Greater(t, cat.syncCount(), syncsBefore, "creating a named session must be durable before it is exposed")
238	-	case <-blocking.firstStarted:
239	-	case <-time.After(testWaitTimeout):
240	-	t.Fatal("metadata write did not block")
241	-	}
242	-	close(blocking.releaseFirst)
243	-	require.ErrorIs(t, <-done, failure)
244			85
245	-	persisted, ok, err := catalogue.Record("work")	86	+	d.mu.Lock()
246	-	require.NoError(t, err)	87	+	published := d.findByNameLocked("work")
247	-	require.True(t, ok)	88	+	d.mu.Unlock()
248	-	require.Empty(t, persisted.TabNames)	89	+	require.Same(t, sess, published)
249		sess.mu.Lock()	90		sess.mu.Lock()
250	-	require.Empty(t, sess.tabs[0].name)	91	+	require.False(t, sess.ephemeral)
251		sess.mu.Unlock()	92		sess.mu.Unlock()
252		}	93		}
253			94
254	-	func TestBlockedCwdWriteCannotOverwriteLaterTabMetadata(t *testing.T) {	95	+	var _ ports.Catalogue = (*syncCountingCatalogue)(nil)
255	-	sess := newSnapshotTestSession(t, "work", false, "/work")
256	-	catalogue, blocking := newBlockingCatalogue(sess, nil)
257	-	d := newTestDaemon(t, nil, stubClock{})
258	-	d.catalogue = blocking
259	-	d.sessions[sess.id] = sess
260	-	d.procCwd = func(int) (string, error) { return "/later-cwd", nil }
261	-
262	-	cwdDone := make(chan struct{})
263	-	go func() {
264	-	d.refreshSessionCwd(sess)
265	-	close(cwdDone)
266	-	}()
267	-	select {
268	-	case <-blocking.firstStarted:
269	-	case <-time.After(testWaitTimeout):
270	-	t.Fatal("cwd metadata write did not block")
271	-	}
272	-
273	-	renameDone := make(chan error, 1)
274	-	go func() { renameDone <- d.renameTab(sess, sess.tabs[0], "later-tab") }()
275	-	require.Eventually(t, func() bool {
276	-	sess.mu.Lock()
277	-	defer sess.mu.Unlock()
278	-	return sess.tabs[0].name == "later-tab"
279	-	}, testWaitTimeout, time.Millisecond)
280	-
281	-	close(blocking.releaseFirst)
282	-	select {
283	-	case <-cwdDone:
284	-	case <-time.After(testWaitTimeout):
285	-	t.Fatal("cwd refresh deadlocked")
286	-	}
287	-	require.NoError(t, <-renameDone)
288	-
289	-	persisted, ok, err := catalogue.Record("work")
290	-	require.NoError(t, err)
291	-	require.True(t, ok)
292	-	require.Equal(t, "/later-cwd", persisted.Cwd)
293	-	require.Equal(t, []string{"later-tab"}, persisted.TabNames)
294	-	}

M internal/usecase/daemon/session.go +46 -258


@@ -55,25 +55,15 @@ type session struct {
55	55		mu sync.Mutex // guards tabs, active, client, restoreDone, clipFiles, and clipboard queue state
56	56		// restoreDone remains attached to a restored session after it is published in
57	57		// the live registry, so racing attaches cannot bypass restoration completion.
58		-	restoreDone chan struct{}
59		-	// metadataPersistMu serializes authority writes and post-I/O rollback. State
60		-	// mutation paths must release d.mu and mu before acquiring it. After durable
61		-	// I/O completes it may acquire d.mu then mu to reconcile failed revisions.
62		-	metadataPersistMu sync.Mutex
63		-	metadataVersion uint64 // guarded by mu; every durable metadata snapshot receives a revision
64		-	metadataDurableVersion uint64 // guarded by metadataPersistMu
65		-	metadataLiveVersion uint64 // rollback cursor guarded by metadataPersistMu
66		-	metadataFailedRollbacks map[uint64]func() bool // guarded by metadataPersistMu
67		-	themeMu sync.Mutex
68		-	tabs []*tab
69		-	active int
70		-	client *attachedClient
71		-	clipboardQueue []clipboardForward
72		-	clipboardWorkerRunning bool
73		-	renameInProgress bool
74		-	renameDone chan struct{}
75		-	cwd string
76		-	terminal terminalEnv
	58	+	restoreDone chan struct{}
	59	+	themeMu sync.Mutex
	60	+	tabs []*tab
	61	+	active int
	62	+	client *attachedClient
	63	+	clipboardQueue []clipboardForward
	64	+	clipboardWorkerRunning bool
	65	+	cwd string
	66	+	terminal terminalEnv
77	67		// env is the authoritative immutable environment snapshot for future PTY children.
78	68		// It is guarded by mu and is always copied on ingress and egress.
79	69		env []string

@@ -177,39 +167,20 @@ func (d Daemon) touchMRU(sess session) {
177	167		return
178	168		}
179	169		seq := d.mruSeq.Add(1)
180		-	updated := false
181		-	var previous uint64
182	170		for {
183	171		old := sess.mruAt.Load()
184	172		if old >= seq {
185	173		return
186	174		}
187	175		if sess.mruAt.CompareAndSwap(old, seq) {
188		-	previous = old
189		-	updated = true
190	176		break
191	177		}
192	178		}
193		-	if !updated {
194		-	return
195		-	}
196	179		sess.mu.Lock()
197		-	name := sess.name
198		-	ephemeral := sess.ephemeral
199		-	var record domain.CatalogueRecord
200		-	var version uint64
201		-	if !ephemeral {
202		-	record, version = sess.nextPersistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))
	180	+	if !sess.ephemeral {
	181	+	d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
203	182		}
204	183		sess.mu.Unlock()
205		-	if !ephemeral {
206		-	rollback := func() bool {
207		-	return sess.mruAt.CompareAndSwap(seq, previous)
208		-	}
209		-	if _, err := d.persistSessionMetadata(sess, version, record.MetadataUpdate(), rollback); err != nil {
210		-	d.log.Warn("touching persisted session recency failed", "err", err, "session", name)
211		-	}
212		-	}
213	184		}
214	185
215	186		func (d Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz domain.Size, term terminalEnv, env []string, restoredTabNames ...[]string) (session, error) {

@@ -329,39 +300,20 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz
329	300		cancel()
330	301		return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", errors.New("durable session authority is not configured"))
331	302		}
332		-	// The name reservation remains visible while durable authority I/O runs,
333		-	// but daemon/session mutexes are released so storage cannot block routing.
334		-	d.mu.Unlock()
335	303		var err error
336	304		if resuming && authoritativeExists {
337		-	err = d.updateCatalogueMetadata(record.MetadataUpdate())
	305	+	err = d.catalogue.UpdateMetadata(record.MetadataUpdate())
	306	+	if err == nil {
	307	+	err = d.catalogue.Sync()
	308	+	}
338	309		} else {
339	310		record, err = d.recovery.Create(d.serveCtx, record)
340	311		}
341		-	d.mu.Lock()
342	312		if err != nil {
343	313		closeTabs(tabs)
344	314		cancel()
345	315		return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)
346	316		}
347		-	if d.closing {
348		-	// Creation lost its race with shutdown. Roll back a newly committed
349		-	// authority record without holding mu; resumed metadata remains owned
350		-	// by the existing stopped lifecycle.
351		-	if !resuming {
352		-	d.mu.Unlock()
353		-	rollbackCtx, rollbackCancel := context.WithTimeout(context.WithoutCancel(d.serveCtx), snapshotFinalFlushTimeout)
354		-	rollbackErr := d.recovery.Delete(rollbackCtx, name)
355		-	rollbackCancel()
356		-	d.mu.Lock()
357		-	if rollbackErr != nil {
358		-	d.log.Warn("rolling back session creation during shutdown failed", "session", name, "err", rollbackErr)
359		-	}
360		-	}
361		-	closeTabs(tabs)
362		-	cancel()
363		-	return nil, errors.New("daemon is shutting down")
364		-	}
365	317		sess.incarnation = record.IncarnationID
366	318		}
367	319		delete(d.stopped, name)

@@ -494,58 +446,14 @@ func (d Daemon) createTab(sess session, sz domain.Size) error {
494	446		p.ctx, p.cancel = context.WithCancel(tb.ctx)
495	447		}
496	448		sess.mu.Lock()
497		-	oldActive := sess.active
498	449		sess.tabs = append(sess.tabs, tb)
499	450		sess.active = len(sess.tabs) - 1
500	451		tabIndex := sess.active
501		-	record, metadataVersion := sess.nextPersistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))
502		-	ephemeral := sess.ephemeral
	452	+	if !sess.ephemeral {
	453	+	d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
	454	+	}
503	455		sess.mu.Unlock()
504	456		d.mu.Unlock()
505		-	if !ephemeral {
506		-	rollback := func() bool {
507		-	d.mu.Lock()
508		-	defer d.mu.Unlock()
509		-	if d.sessions[sess.id] != sess {
510		-	return true
511		-	}
512		-	sess.mu.Lock()
513		-	if !slices.Contains(sess.tabs, tb) {
514		-	sess.mu.Unlock()
515		-	return true
516		-	}
517		-	if sess.tabs[len(sess.tabs)-1] != tb {
518		-	sess.mu.Unlock()
519		-	return false
520		-	}
521		-	sess.tabs = slices.DeleteFunc(sess.tabs, func(candidate *tab) bool { return candidate == tb })
522		-	sess.active = min(oldActive, len(sess.tabs)-1)
523		-	sess.mu.Unlock()
524		-	if tb.cancel != nil {
525		-	tb.cancel()
526		-	}
527		-	_ = pty.Close()
528		-	return true
529		-	}
530		-	rollbackRejected, err := d.persistSessionMetadata(sess, metadataVersion, record.MetadataUpdate(), rollback)
531		-	if err != nil {
532		-	if rollbackRejected {
533		-	d.mu.Lock()
534		-	if d.sessions[sess.id] == sess {
535		-	sess.mu.Lock()
536		-	sess.tabs = slices.DeleteFunc(sess.tabs, func(candidate *tab) bool { return candidate == tb })
537		-	sess.active = min(sess.active, len(sess.tabs)-1)
538		-	sess.mu.Unlock()
539		-	}
540		-	d.mu.Unlock()
541		-	if tb.cancel != nil {
542		-	tb.cancel()
543		-	}
544		-	tb.closeAllPanes()
545		-	}
546		-	return err
547		-	}
548		-	}
549	457		d.log.Info("tab created", "session", name, "tab", tabIndex)
550	458		d.startTabGoroutines(sess, tb)
551	459		d.activateTab(sess, tb)

@@ -731,19 +639,14 @@ func (d Daemon) renameSession(sess session, name string) error {
731	639		return err
732	640		}
733	641
734		-	// Reserve the rename while state is inspected, then persist the new name
735		-	// without changing the incarnation-keyed snapshot namespace.
	642	+	// Hold daemon then session state across the durable identity write. The
	643	+	// incarnation-keyed snapshot namespace is never relocated.
736	644		d.mu.Lock()
737	645		if taken := d.findByNameLocked(name); taken != nil && taken != sess {
738	646		d.mu.Unlock()
739	647		return errSessionNameInUse
740	648		}
741	649		sess.mu.Lock()
742		-	if sess.renameInProgress {
743		-	sess.mu.Unlock()
744		-	d.mu.Unlock()
745		-	return errors.New("session rename already in progress")
746		-	}
747	650		if stopped, ok := d.stopped[name]; ok && stopped.name != sess.name {
748	651		sess.mu.Unlock()
749	652		d.mu.Unlock()

@@ -773,20 +676,10 @@ func (d Daemon) renameSession(sess session, name string) error {
773	676		record.Name = name
774	677		record.CreatedAt = createdAt
775	678		record.LastUsedSeq = lastUsedSeq
776		-	sess.renameInProgress = true
777		-	sess.renameDone = make(chan struct{})
778		-	sess.mu.Unlock()
779		-	d.mu.Unlock()
780		-
781	679		rollback := func(err error) error {
782		-	sess.mu.Lock()
783	680		sess.incarnation = priorIncarnation
784		-	sess.renameInProgress = false
785		-	if sess.renameDone != nil {
786		-	close(sess.renameDone)
787		-	sess.renameDone = nil
788		-	}
789	681		sess.mu.Unlock()
	682	+	d.mu.Unlock()
790	683		return err
791	684		}
792	685		// Durable named renames are one atomic catalogue batch. The incarnation-keyed

@@ -810,21 +703,12 @@ func (d Daemon) renameSession(sess session, name string) error {
810	703		}
811	704		}
812	705
813		-	d.mu.Lock()
814		-	sess.mu.Lock()
815		-	// The reservation prevents another rename. A closing session is allowed to
816		-	// finish this short commit; killSession subsequently owns its teardown.
817	706		delete(d.stopped, oldName)
818	707		delete(d.stopped, name)
819	708		sess.name = name
820	709		sess.createdAt = createdAt
821	710		sess.incarnation = record.IncarnationID
822	711		sess.ephemeral = false
823		-	sess.renameInProgress = false
824		-	if sess.renameDone != nil {
825		-	close(sess.renameDone)
826		-	sess.renameDone = nil
827		-	}
828	712		sess.snapEligible.Store(name != "")
829	713		sess.mu.Unlock()
830	714		d.mu.Unlock()

@@ -852,43 +736,15 @@ func (d Daemon) renameTab(sess session, tb *tab, name string) error {
852	736		d.mu.Unlock()
853	737		return errors.New("tab not found")
854	738		}
855		-	oldName := tb.name
856	739		tb.name = name
857		-	record, metadataVersion := sess.nextPersistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))
858		-	ephemeral := sess.ephemeral
	740	+	if !sess.ephemeral {
	741	+	d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
	742	+	}
859	743		sess.mu.Unlock()
860	744		d.mu.Unlock()
861		-	if ephemeral {
862		-	return nil
863		-	}
864		-	rollback := func() bool {
865		-	d.mu.Lock()
866		-	defer d.mu.Unlock()
867		-	if d.sessions[sess.id] != sess {
868		-	return true
869		-	}
870		-	sess.mu.Lock()
871		-	defer sess.mu.Unlock()
872		-	if !slices.Contains(sess.tabs, tb) {
873		-	return true
874		-	}
875		-	if tb.name != name {
876		-	return false
877		-	}
878		-	tb.name = oldName
879		-	return true
880		-	}
881		-	if _, err := d.persistSessionMetadata(sess, metadataVersion, record.MetadataUpdate(), rollback); err != nil {
882		-	return err
883		-	}
884	745		return nil
885	746		}
886	747
887		-	func (s *session) nextPersistRecordLocked(updatedAt int64) (domain.CatalogueRecord, uint64) {
888		-	s.metadataVersion++
889		-	return s.persistRecordLocked(updatedAt), s.metadataVersion
890		-	}
891		-
892	748		func (s *session) persistRecordLocked(updatedAt int64) domain.CatalogueRecord {
893	749		createdAt := s.createdAt
894	750		tabNames := make([]string, len(s.tabs))

@@ -943,10 +799,8 @@ func (d Daemon) closeTab(sess session, tb *tab, repaint bool) error {
943	799		return nil
944	800		}
945	801		ringing := tb.attention
946		-	oldActive := sess.active
947	802		wasActive := idx == sess.active
948	803		sess.tabs = append(sess.tabs[:idx], sess.tabs[idx+1:]...)
949		-	tabsAfterClose := len(sess.tabs)
950	804		if sess.active >= len(sess.tabs) {
951	805		sess.active = len(sess.tabs) - 1
952	806		} else if idx < sess.active {

@@ -955,41 +809,11 @@ func (d Daemon) closeTab(sess session, tb *tab, repaint bool) error {
955	809		destination := sess.tabs[sess.active]
956	810		ac := sess.client
957	811		name := sess.name
958		-	record, metadataVersion := sess.nextPersistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))
959		-	ephemeral := sess.ephemeral
	812	+	if !sess.ephemeral {
	813	+	d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
	814	+	}
960	815		sess.mu.Unlock()
961	816		d.mu.Unlock()
962		-	if !ephemeral {
963		-	rollback := func() bool {
964		-	d.mu.Lock()
965		-	defer d.mu.Unlock()
966		-	if d.sessions[sess.id] != sess {
967		-	return true
968		-	}
969		-	sess.mu.Lock()
970		-	defer sess.mu.Unlock()
971		-	if slices.Contains(sess.tabs, tb) {
972		-	return true
973		-	}
974		-	if len(sess.tabs) != tabsAfterClose {
975		-	return false
976		-	}
977		-	sess.tabs = slices.Insert(sess.tabs, min(idx, len(sess.tabs)), tb)
978		-	sess.active = min(oldActive, len(sess.tabs)-1)
979		-	return true
980		-	}
981		-	rollbackRejected, err := d.persistSessionMetadata(sess, metadataVersion, record.MetadataUpdate(), rollback)
982		-	if err != nil {
983		-	d.log.Warn("persisting closed tab failed", "err", err, "session", name)
984		-	if rollbackRejected {
985		-	if tb.cancel != nil {
986		-	tb.cancel()
987		-	}
988		-	tb.closeAllPanes()
989		-	}
990		-	return err
991		-	}
992		-	}
993	817		d.log.Info("tab closed", "session", name)
994	818		markSnapshotDirty(sess)
995	819

@@ -1071,15 +895,15 @@ func (d *Daemon) retryStoppedPurgeContext(ctx context.Context, name string) erro
1071	895		}
1072	896		stopped.purging = true
1073	897		d.stopped[name] = stopped
1074		-	d.mu.Unlock()
1075	898
1076	899		if err := d.beginSnapshotPurge(name, stopped.incarnation); err != nil {
	900	+	d.mu.Unlock()
1077	901		return err
1078	902		}
1079	903		if err := d.finishSnapshotPurge(ctx, name, stopped.incarnation); err != nil {
	904	+	d.mu.Unlock()
1080	905		return err
1081	906		}
1082		-	d.mu.Lock()
1083	907		if current, ok := d.stopped[name]; ok && current.purging {
1084	908		delete(d.stopped, name)
1085	909		}

@@ -1204,23 +1028,6 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
1204	1028		}
1205	1029		defer sess.finishTeardown()
1206	1030
1207		-	// A rename reserves durable identity while coordinator I/O runs unlocked.
1208		-	// Teardown waits for that reservation before selecting the purge name.
1209		-	sess.mu.Lock()
1210		-	renameDone := sess.renameDone
1211		-	sess.mu.Unlock()
1212		-	if renameDone != nil {
1213		-	if deadline == nil {
1214		-	<-renameDone
1215		-	} else {
1216		-	select {
1217		-	case <-renameDone:
1218		-	case <-deadline.Done():
1219		-	return context.DeadlineExceeded
1220		-	}
1221		-	}
1222		-	}
1223		-
1224	1031		d.mu.Lock()
1225	1032		current := d.sessions[sess.id]
1226	1033		d.mu.Unlock()

@@ -1378,16 +1185,17 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
1378	1185		// another generation after this destructive source deletion. Keep the hidden
1379	1186		// stopped record when deletion fails so a repeated live kill can retry.
1380	1187		if !ephemeral && purge {
1381		-	if err := d.finishSnapshotPurge(d.serveCtx, stoppedName, incarnation); err != nil {
	1188	+	d.mu.Lock()
	1189	+	sess.mu.Lock()
	1190	+	err := d.finishSnapshotPurge(d.serveCtx, stoppedName, incarnation)
	1191	+	sess.mu.Unlock()
	1192	+	if err != nil {
1382	1193		purgeErr = errors.Join(purgeErr, err)
1383	1194		d.log.Warn("finishing session snapshot purge failed", "err", err, "session", stoppedName)
1384		-	} else {
1385		-	d.mu.Lock()
1386		-	if stopped, ok := d.stopped[stoppedName]; ok && stopped.purging {
1387		-	delete(d.stopped, stoppedName)
1388		-	}
1389		-	d.mu.Unlock()
	1195	+	} else if stopped, ok := d.stopped[stoppedName]; ok && stopped.purging {
	1196	+	delete(d.stopped, stoppedName)
1390	1197		}
	1198	+	d.mu.Unlock()
1391	1199		}
1392	1200		if empty {
1393	1201		d.doneOnce.Do(func() { close(d.done) })

@@ -1448,7 +1256,12 @@ func (d *Daemon) cwdSampler(ctx context.Context) {
1448	1256		return
1449	1257		case <-t.C():
1450	1258		}
1451		-	d.refreshNamedSessionCwds()
	1259	+	if d.procCwd != nil {
	1260	+	d.refreshNamedSessionCwds()
	1261	+	}
	1262	+	if err := d.flushCatalogue(); err != nil {
	1263	+	d.log.Warn("flushing session catalogue failed", "err", err)
	1264	+	}
1452	1265		t.Reset(5 * time.Second)
1453	1266		}
1454	1267		}

@@ -1496,38 +1309,13 @@ func (d Daemon) refreshSessionCwd(sess session) {
1496	1309		d.mu.Unlock()
1497	1310		return
1498	1311		}
1499		-	oldCwd := sess.cwd
1500	1312		sess.cwd = cwd
1501		-	name := sess.name
1502		-	ephemeral := sess.ephemeral
1503		-	var record domain.CatalogueRecord
1504		-	var metadataVersion uint64
1505		-	if !ephemeral {
1506		-	record, metadataVersion = sess.nextPersistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))
	1313	+	if !sess.ephemeral {
	1314	+	d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
1507	1315		}
1508	1316		sess.mu.Unlock()
1509	1317		d.mu.Unlock()
1510		-	if !ephemeral {
1511		-	rollback := func() bool {
1512		-	d.mu.Lock()
1513		-	defer d.mu.Unlock()
1514		-	if d.sessions[sess.id] != sess {
1515		-	return true
1516		-	}
1517		-	sess.mu.Lock()
1518		-	defer sess.mu.Unlock()
1519		-	if sess.cwd != cwd {
1520		-	return false
1521		-	}
1522		-	sess.cwd = oldCwd
1523		-	return true
1524		-	}
1525		-	if _, err := d.persistSessionMetadata(sess, metadataVersion, record.MetadataUpdate(), rollback); err != nil {
1526		-	d.log.Warn("touching persisted session cwd failed", "err", err, "session", name)
1527		-	return
1528		-	}
1529		-	markSnapshotDirty(sess)
1530		-	}
	1318	+	markSnapshotDirty(sess)
1531	1319		}
1532	1320
1533	1321		// childEnv retains the daemon-environment helper for daemon-local legacy callers.


@@ -55,25 +55,15 @@ type session struct {
55		mu sync.Mutex // guards tabs, active, client, restoreDone, clipFiles, and clipboard queue state	55		mu sync.Mutex // guards tabs, active, client, restoreDone, clipFiles, and clipboard queue state
56		// restoreDone remains attached to a restored session after it is published in	56		// restoreDone remains attached to a restored session after it is published in
57		// the live registry, so racing attaches cannot bypass restoration completion.	57		// the live registry, so racing attaches cannot bypass restoration completion.
58	-	restoreDone chan struct{}	58	+	restoreDone chan struct{}
59	-	// metadataPersistMu serializes authority writes and post-I/O rollback. State	59	+	themeMu sync.Mutex
60	-	// mutation paths must release d.mu and mu before acquiring it. After durable	60	+	tabs []*tab
61	-	// I/O completes it may acquire d.mu then mu to reconcile failed revisions.	61	+	active int
62	-	metadataPersistMu sync.Mutex	62	+	client *attachedClient
63	-	metadataVersion uint64 // guarded by mu; every durable metadata snapshot receives a revision	63	+	clipboardQueue []clipboardForward
64	-	metadataDurableVersion uint64 // guarded by metadataPersistMu	64	+	clipboardWorkerRunning bool
65	-	metadataLiveVersion uint64 // rollback cursor guarded by metadataPersistMu	65	+	cwd string
66	-	metadataFailedRollbacks map[uint64]func() bool // guarded by metadataPersistMu	66	+	terminal terminalEnv
67	-	themeMu sync.Mutex
68	-	tabs []*tab
69	-	active int
70	-	client *attachedClient
71	-	clipboardQueue []clipboardForward
72	-	clipboardWorkerRunning bool
73	-	renameInProgress bool
74	-	renameDone chan struct{}
75	-	cwd string
76	-	terminal terminalEnv
77		// env is the authoritative immutable environment snapshot for future PTY children.	67		// env is the authoritative immutable environment snapshot for future PTY children.
78		// It is guarded by mu and is always copied on ingress and egress.	68		// It is guarded by mu and is always copied on ingress and egress.
79		env []string	69		env []string

@@ -177,39 +167,20 @@ func (d Daemon) touchMRU(sess session) {
177		return	167		return
178		}	168		}
179		seq := d.mruSeq.Add(1)	169		seq := d.mruSeq.Add(1)
180	-	updated := false
181	-	var previous uint64
182		for {	170		for {
183		old := sess.mruAt.Load()	171		old := sess.mruAt.Load()
184		if old >= seq {	172		if old >= seq {
185		return	173		return
186		}	174		}
187		if sess.mruAt.CompareAndSwap(old, seq) {	175		if sess.mruAt.CompareAndSwap(old, seq) {
188	-	previous = old
189	-	updated = true
190		break	176		break
191		}	177		}
192		}	178		}
193	-	if !updated {
194	-	return
195	-	}
196		sess.mu.Lock()	179		sess.mu.Lock()
197	-	name := sess.name	180	+	if !sess.ephemeral {
198	-	ephemeral := sess.ephemeral	181	+	d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
199	-	var record domain.CatalogueRecord
200	-	var version uint64
201	-	if !ephemeral {
202	-	record, version = sess.nextPersistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))
203		}	182		}
204		sess.mu.Unlock()	183		sess.mu.Unlock()
205	-	if !ephemeral {
206	-	rollback := func() bool {
207	-	return sess.mruAt.CompareAndSwap(seq, previous)
208	-	}
209	-	if _, err := d.persistSessionMetadata(sess, version, record.MetadataUpdate(), rollback); err != nil {
210	-	d.log.Warn("touching persisted session recency failed", "err", err, "session", name)
211	-	}
212	-	}
213		}	184		}
214			185
215		func (d Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz domain.Size, term terminalEnv, env []string, restoredTabNames ...[]string) (session, error) {	186		func (d Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz domain.Size, term terminalEnv, env []string, restoredTabNames ...[]string) (session, error) {

@@ -329,39 +300,20 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz
329		cancel()	300		cancel()
330		return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", errors.New("durable session authority is not configured"))	301		return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", errors.New("durable session authority is not configured"))
331		}	302		}
332	-	// The name reservation remains visible while durable authority I/O runs,
333	-	// but daemon/session mutexes are released so storage cannot block routing.
334	-	d.mu.Unlock()
335		var err error	303		var err error
336		if resuming && authoritativeExists {	304		if resuming && authoritativeExists {
337	-	err = d.updateCatalogueMetadata(record.MetadataUpdate())	305	+	err = d.catalogue.UpdateMetadata(record.MetadataUpdate())
			306	+	if err == nil {
			307	+	err = d.catalogue.Sync()
			308	+	}
338		} else {	309		} else {
339		record, err = d.recovery.Create(d.serveCtx, record)	310		record, err = d.recovery.Create(d.serveCtx, record)
340		}	311		}
341	-	d.mu.Lock()
342		if err != nil {	312		if err != nil {
343		closeTabs(tabs)	313		closeTabs(tabs)
344		cancel()	314		cancel()
345		return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)	315		return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)
346		}	316		}
347	-	if d.closing {
348	-	// Creation lost its race with shutdown. Roll back a newly committed
349	-	// authority record without holding mu; resumed metadata remains owned
350	-	// by the existing stopped lifecycle.
351	-	if !resuming {
352	-	d.mu.Unlock()
353	-	rollbackCtx, rollbackCancel := context.WithTimeout(context.WithoutCancel(d.serveCtx), snapshotFinalFlushTimeout)
354	-	rollbackErr := d.recovery.Delete(rollbackCtx, name)
355	-	rollbackCancel()
356	-	d.mu.Lock()
357	-	if rollbackErr != nil {
358	-	d.log.Warn("rolling back session creation during shutdown failed", "session", name, "err", rollbackErr)
359	-	}
360	-	}
361	-	closeTabs(tabs)
362	-	cancel()
363	-	return nil, errors.New("daemon is shutting down")
364	-	}
365		sess.incarnation = record.IncarnationID	317		sess.incarnation = record.IncarnationID
366		}	318		}
367		delete(d.stopped, name)	319		delete(d.stopped, name)

@@ -494,58 +446,14 @@ func (d Daemon) createTab(sess session, sz domain.Size) error {
494		p.ctx, p.cancel = context.WithCancel(tb.ctx)	446		p.ctx, p.cancel = context.WithCancel(tb.ctx)
495		}	447		}
496		sess.mu.Lock()	448		sess.mu.Lock()
497	-	oldActive := sess.active
498		sess.tabs = append(sess.tabs, tb)	449		sess.tabs = append(sess.tabs, tb)
499		sess.active = len(sess.tabs) - 1	450		sess.active = len(sess.tabs) - 1
500		tabIndex := sess.active	451		tabIndex := sess.active
501	-	record, metadataVersion := sess.nextPersistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))	452	+	if !sess.ephemeral {
502	-	ephemeral := sess.ephemeral	453	+	d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
			454	+	}
503		sess.mu.Unlock()	455		sess.mu.Unlock()
504		d.mu.Unlock()	456		d.mu.Unlock()
505	-	if !ephemeral {
506	-	rollback := func() bool {
507	-	d.mu.Lock()
508	-	defer d.mu.Unlock()
509	-	if d.sessions[sess.id] != sess {
510	-	return true
511	-	}
512	-	sess.mu.Lock()
513	-	if !slices.Contains(sess.tabs, tb) {
514	-	sess.mu.Unlock()
515	-	return true
516	-	}
517	-	if sess.tabs[len(sess.tabs)-1] != tb {
518	-	sess.mu.Unlock()
519	-	return false
520	-	}
521	-	sess.tabs = slices.DeleteFunc(sess.tabs, func(candidate *tab) bool { return candidate == tb })
522	-	sess.active = min(oldActive, len(sess.tabs)-1)
523	-	sess.mu.Unlock()
524	-	if tb.cancel != nil {
525	-	tb.cancel()
526	-	}
527	-	_ = pty.Close()
528	-	return true
529	-	}
530	-	rollbackRejected, err := d.persistSessionMetadata(sess, metadataVersion, record.MetadataUpdate(), rollback)
531	-	if err != nil {
532	-	if rollbackRejected {
533	-	d.mu.Lock()
534	-	if d.sessions[sess.id] == sess {
535	-	sess.mu.Lock()
536	-	sess.tabs = slices.DeleteFunc(sess.tabs, func(candidate *tab) bool { return candidate == tb })
537	-	sess.active = min(sess.active, len(sess.tabs)-1)
538	-	sess.mu.Unlock()
539	-	}
540	-	d.mu.Unlock()
541	-	if tb.cancel != nil {
542	-	tb.cancel()
543	-	}
544	-	tb.closeAllPanes()
545	-	}
546	-	return err
547	-	}
548	-	}
549		d.log.Info("tab created", "session", name, "tab", tabIndex)	457		d.log.Info("tab created", "session", name, "tab", tabIndex)
550		d.startTabGoroutines(sess, tb)	458		d.startTabGoroutines(sess, tb)
551		d.activateTab(sess, tb)	459		d.activateTab(sess, tb)

@@ -731,19 +639,14 @@ func (d Daemon) renameSession(sess session, name string) error {
731		return err	639		return err
732		}	640		}
733			641
734	-	// Reserve the rename while state is inspected, then persist the new name	642	+	// Hold daemon then session state across the durable identity write. The
735	-	// without changing the incarnation-keyed snapshot namespace.	643	+	// incarnation-keyed snapshot namespace is never relocated.
736		d.mu.Lock()	644		d.mu.Lock()
737		if taken := d.findByNameLocked(name); taken != nil && taken != sess {	645		if taken := d.findByNameLocked(name); taken != nil && taken != sess {
738		d.mu.Unlock()	646		d.mu.Unlock()
739		return errSessionNameInUse	647		return errSessionNameInUse
740		}	648		}
741		sess.mu.Lock()	649		sess.mu.Lock()
742	-	if sess.renameInProgress {
743	-	sess.mu.Unlock()
744	-	d.mu.Unlock()
745	-	return errors.New("session rename already in progress")
746	-	}
747		if stopped, ok := d.stopped[name]; ok && stopped.name != sess.name {	650		if stopped, ok := d.stopped[name]; ok && stopped.name != sess.name {
748		sess.mu.Unlock()	651		sess.mu.Unlock()
749		d.mu.Unlock()	652		d.mu.Unlock()

@@ -773,20 +676,10 @@ func (d Daemon) renameSession(sess session, name string) error {
773		record.Name = name	676		record.Name = name
774		record.CreatedAt = createdAt	677		record.CreatedAt = createdAt
775		record.LastUsedSeq = lastUsedSeq	678		record.LastUsedSeq = lastUsedSeq
776	-	sess.renameInProgress = true
777	-	sess.renameDone = make(chan struct{})
778	-	sess.mu.Unlock()
779	-	d.mu.Unlock()
780	-
781		rollback := func(err error) error {	679		rollback := func(err error) error {
782	-	sess.mu.Lock()
783		sess.incarnation = priorIncarnation	680		sess.incarnation = priorIncarnation
784	-	sess.renameInProgress = false
785	-	if sess.renameDone != nil {
786	-	close(sess.renameDone)
787	-	sess.renameDone = nil
788	-	}
789		sess.mu.Unlock()	681		sess.mu.Unlock()
			682	+	d.mu.Unlock()
790		return err	683		return err
791		}	684		}
792		// Durable named renames are one atomic catalogue batch. The incarnation-keyed	685		// Durable named renames are one atomic catalogue batch. The incarnation-keyed

@@ -810,21 +703,12 @@ func (d Daemon) renameSession(sess session, name string) error {
810		}	703		}
811		}	704		}
812			705
813	-	d.mu.Lock()
814	-	sess.mu.Lock()
815	-	// The reservation prevents another rename. A closing session is allowed to
816	-	// finish this short commit; killSession subsequently owns its teardown.
817		delete(d.stopped, oldName)	706		delete(d.stopped, oldName)
818		delete(d.stopped, name)	707		delete(d.stopped, name)
819		sess.name = name	708		sess.name = name
820		sess.createdAt = createdAt	709		sess.createdAt = createdAt
821		sess.incarnation = record.IncarnationID	710		sess.incarnation = record.IncarnationID
822		sess.ephemeral = false	711		sess.ephemeral = false
823	-	sess.renameInProgress = false
824	-	if sess.renameDone != nil {
825	-	close(sess.renameDone)
826	-	sess.renameDone = nil
827	-	}
828		sess.snapEligible.Store(name != "")	712		sess.snapEligible.Store(name != "")
829		sess.mu.Unlock()	713		sess.mu.Unlock()
830		d.mu.Unlock()	714		d.mu.Unlock()

@@ -852,43 +736,15 @@ func (d Daemon) renameTab(sess session, tb *tab, name string) error {
852		d.mu.Unlock()	736		d.mu.Unlock()
853		return errors.New("tab not found")	737		return errors.New("tab not found")
854		}	738		}
855	-	oldName := tb.name
856		tb.name = name	739		tb.name = name
857	-	record, metadataVersion := sess.nextPersistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))	740	+	if !sess.ephemeral {
858	-	ephemeral := sess.ephemeral	741	+	d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
			742	+	}
859		sess.mu.Unlock()	743		sess.mu.Unlock()
860		d.mu.Unlock()	744		d.mu.Unlock()
861	-	if ephemeral {
862	-	return nil
863	-	}
864	-	rollback := func() bool {
865	-	d.mu.Lock()
866	-	defer d.mu.Unlock()
867	-	if d.sessions[sess.id] != sess {
868	-	return true
869	-	}
870	-	sess.mu.Lock()
871	-	defer sess.mu.Unlock()
872	-	if !slices.Contains(sess.tabs, tb) {
873	-	return true
874	-	}
875	-	if tb.name != name {
876	-	return false
877	-	}
878	-	tb.name = oldName
879	-	return true
880	-	}
881	-	if _, err := d.persistSessionMetadata(sess, metadataVersion, record.MetadataUpdate(), rollback); err != nil {
882	-	return err
883	-	}
884		return nil	745		return nil
885		}	746		}
886			747
887	-	func (s *session) nextPersistRecordLocked(updatedAt int64) (domain.CatalogueRecord, uint64) {
888	-	s.metadataVersion++
889	-	return s.persistRecordLocked(updatedAt), s.metadataVersion
890	-	}
891	-
892		func (s *session) persistRecordLocked(updatedAt int64) domain.CatalogueRecord {	748		func (s *session) persistRecordLocked(updatedAt int64) domain.CatalogueRecord {
893		createdAt := s.createdAt	749		createdAt := s.createdAt
894		tabNames := make([]string, len(s.tabs))	750		tabNames := make([]string, len(s.tabs))

@@ -943,10 +799,8 @@ func (d Daemon) closeTab(sess session, tb *tab, repaint bool) error {
943		return nil	799		return nil
944		}	800		}
945		ringing := tb.attention	801		ringing := tb.attention
946	-	oldActive := sess.active
947		wasActive := idx == sess.active	802		wasActive := idx == sess.active
948		sess.tabs = append(sess.tabs[:idx], sess.tabs[idx+1:]...)	803		sess.tabs = append(sess.tabs[:idx], sess.tabs[idx+1:]...)
949	-	tabsAfterClose := len(sess.tabs)
950		if sess.active >= len(sess.tabs) {	804		if sess.active >= len(sess.tabs) {
951		sess.active = len(sess.tabs) - 1	805		sess.active = len(sess.tabs) - 1
952		} else if idx < sess.active {	806		} else if idx < sess.active {

@@ -955,41 +809,11 @@ func (d Daemon) closeTab(sess session, tb *tab, repaint bool) error {
955		destination := sess.tabs[sess.active]	809		destination := sess.tabs[sess.active]
956		ac := sess.client	810		ac := sess.client
957		name := sess.name	811		name := sess.name
958	-	record, metadataVersion := sess.nextPersistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))	812	+	if !sess.ephemeral {
959	-	ephemeral := sess.ephemeral	813	+	d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
			814	+	}
960		sess.mu.Unlock()	815		sess.mu.Unlock()
961		d.mu.Unlock()	816		d.mu.Unlock()
962	-	if !ephemeral {
963	-	rollback := func() bool {
964	-	d.mu.Lock()
965	-	defer d.mu.Unlock()
966	-	if d.sessions[sess.id] != sess {
967	-	return true
968	-	}
969	-	sess.mu.Lock()
970	-	defer sess.mu.Unlock()
971	-	if slices.Contains(sess.tabs, tb) {
972	-	return true
973	-	}
974	-	if len(sess.tabs) != tabsAfterClose {
975	-	return false
976	-	}
977	-	sess.tabs = slices.Insert(sess.tabs, min(idx, len(sess.tabs)), tb)
978	-	sess.active = min(oldActive, len(sess.tabs)-1)
979	-	return true
980	-	}
981	-	rollbackRejected, err := d.persistSessionMetadata(sess, metadataVersion, record.MetadataUpdate(), rollback)
982	-	if err != nil {
983	-	d.log.Warn("persisting closed tab failed", "err", err, "session", name)
984	-	if rollbackRejected {
985	-	if tb.cancel != nil {
986	-	tb.cancel()
987	-	}
988	-	tb.closeAllPanes()
989	-	}
990	-	return err
991	-	}
992	-	}
993		d.log.Info("tab closed", "session", name)	817		d.log.Info("tab closed", "session", name)
994		markSnapshotDirty(sess)	818		markSnapshotDirty(sess)
995			819

@@ -1071,15 +895,15 @@ func (d *Daemon) retryStoppedPurgeContext(ctx context.Context, name string) erro
1071		}	895		}
1072		stopped.purging = true	896		stopped.purging = true
1073		d.stopped[name] = stopped	897		d.stopped[name] = stopped
1074	-	d.mu.Unlock()
1075			898
1076		if err := d.beginSnapshotPurge(name, stopped.incarnation); err != nil {	899		if err := d.beginSnapshotPurge(name, stopped.incarnation); err != nil {
			900	+	d.mu.Unlock()
1077		return err	901		return err
1078		}	902		}
1079		if err := d.finishSnapshotPurge(ctx, name, stopped.incarnation); err != nil {	903		if err := d.finishSnapshotPurge(ctx, name, stopped.incarnation); err != nil {
			904	+	d.mu.Unlock()
1080		return err	905		return err
1081		}	906		}
1082	-	d.mu.Lock()
1083		if current, ok := d.stopped[name]; ok && current.purging {	907		if current, ok := d.stopped[name]; ok && current.purging {
1084		delete(d.stopped, name)	908		delete(d.stopped, name)
1085		}	909		}

@@ -1204,23 +1028,6 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
1204		}	1028		}
1205		defer sess.finishTeardown()	1029		defer sess.finishTeardown()
1206			1030
1207	-	// A rename reserves durable identity while coordinator I/O runs unlocked.
1208	-	// Teardown waits for that reservation before selecting the purge name.
1209	-	sess.mu.Lock()
1210	-	renameDone := sess.renameDone
1211	-	sess.mu.Unlock()
1212	-	if renameDone != nil {
1213	-	if deadline == nil {
1214	-	<-renameDone
1215	-	} else {
1216	-	select {
1217	-	case <-renameDone:
1218	-	case <-deadline.Done():
1219	-	return context.DeadlineExceeded
1220	-	}
1221	-	}
1222	-	}
1223	-
1224		d.mu.Lock()	1031		d.mu.Lock()
1225		current := d.sessions[sess.id]	1032		current := d.sessions[sess.id]
1226		d.mu.Unlock()	1033		d.mu.Unlock()

@@ -1378,16 +1185,17 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
1378		// another generation after this destructive source deletion. Keep the hidden	1185		// another generation after this destructive source deletion. Keep the hidden
1379		// stopped record when deletion fails so a repeated live kill can retry.	1186		// stopped record when deletion fails so a repeated live kill can retry.
1380		if !ephemeral && purge {	1187		if !ephemeral && purge {
1381	-	if err := d.finishSnapshotPurge(d.serveCtx, stoppedName, incarnation); err != nil {	1188	+	d.mu.Lock()
			1189	+	sess.mu.Lock()
			1190	+	err := d.finishSnapshotPurge(d.serveCtx, stoppedName, incarnation)
			1191	+	sess.mu.Unlock()
			1192	+	if err != nil {
1382		purgeErr = errors.Join(purgeErr, err)	1193		purgeErr = errors.Join(purgeErr, err)
1383		d.log.Warn("finishing session snapshot purge failed", "err", err, "session", stoppedName)	1194		d.log.Warn("finishing session snapshot purge failed", "err", err, "session", stoppedName)
1384	-	} else {	1195	+	} else if stopped, ok := d.stopped[stoppedName]; ok && stopped.purging {
1385	-	d.mu.Lock()	1196	+	delete(d.stopped, stoppedName)
1386	-	if stopped, ok := d.stopped[stoppedName]; ok && stopped.purging {
1387	-	delete(d.stopped, stoppedName)
1388	-	}
1389	-	d.mu.Unlock()
1390		}	1197		}
			1198	+	d.mu.Unlock()
1391		}	1199		}
1392		if empty {	1200		if empty {
1393		d.doneOnce.Do(func() { close(d.done) })	1201		d.doneOnce.Do(func() { close(d.done) })

@@ -1448,7 +1256,12 @@ func (d *Daemon) cwdSampler(ctx context.Context) {
1448		return	1256		return
1449		case <-t.C():	1257		case <-t.C():
1450		}	1258		}
1451	-	d.refreshNamedSessionCwds()	1259	+	if d.procCwd != nil {
			1260	+	d.refreshNamedSessionCwds()
			1261	+	}
			1262	+	if err := d.flushCatalogue(); err != nil {
			1263	+	d.log.Warn("flushing session catalogue failed", "err", err)
			1264	+	}
1452		t.Reset(5 * time.Second)	1265		t.Reset(5 * time.Second)
1453		}	1266		}
1454		}	1267		}

@@ -1496,38 +1309,13 @@ func (d Daemon) refreshSessionCwd(sess session) {
1496		d.mu.Unlock()	1309		d.mu.Unlock()
1497		return	1310		return
1498		}	1311		}
1499	-	oldCwd := sess.cwd
1500		sess.cwd = cwd	1312		sess.cwd = cwd
1501	-	name := sess.name	1313	+	if !sess.ephemeral {
1502	-	ephemeral := sess.ephemeral	1314	+	d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
1503	-	var record domain.CatalogueRecord
1504	-	var metadataVersion uint64
1505	-	if !ephemeral {
1506	-	record, metadataVersion = sess.nextPersistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))
1507		}	1315		}
1508		sess.mu.Unlock()	1316		sess.mu.Unlock()
1509		d.mu.Unlock()	1317		d.mu.Unlock()
1510	-	if !ephemeral {	1318	+	markSnapshotDirty(sess)
1511	-	rollback := func() bool {
1512	-	d.mu.Lock()
1513	-	defer d.mu.Unlock()
1514	-	if d.sessions[sess.id] != sess {
1515	-	return true
1516	-	}
1517	-	sess.mu.Lock()
1518	-	defer sess.mu.Unlock()
1519	-	if sess.cwd != cwd {
1520	-	return false
1521	-	}
1522	-	sess.cwd = oldCwd
1523	-	return true
1524	-	}
1525	-	if _, err := d.persistSessionMetadata(sess, metadataVersion, record.MetadataUpdate(), rollback); err != nil {
1526	-	d.log.Warn("touching persisted session cwd failed", "err", err, "session", name)
1527	-	return
1528	-	}
1529	-	markSnapshotDirty(sess)
1530	-	}
1531		}	1319		}
1532			1320
1533		// childEnv retains the daemon-environment helper for daemon-local legacy callers.	1321		// childEnv retains the daemon-environment helper for daemon-local legacy callers.

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


@@ -969,17 +969,15 @@ func TestRenameTabPersistsForNamedSession(t *testing.T) {
969	969		require.Equal(t, []string{"shell"}, records[0].TabNames)
970	970		}
971	971
972		-	func TestRenameTabDoesNotHoldDaemonOrSessionMutexDuringCatalogueIO(t *testing.T) {
	972	+	func TestRenameTabBuffersMetadataUnderStateLocks(t *testing.T) {
973	973		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
974	974		sess := newSnapshotTestSession(t, "work", false, "/work")
975	975		d.sessions[sess.id] = sess
976	976		catalogue := portsmocks.NewMockCatalogue(t)
977	977		WithCatalogue(catalogue, nil)(d)
978	978		catalogue.EXPECT().UpdateMetadata(mock.Anything).RunAndReturn(func(domain.CatalogueMetadataUpdate) error {
979		-	require.True(t, d.mu.TryLock(), "daemon mutex held during catalogue I/O")
980		-	d.mu.Unlock()
981		-	require.True(t, sess.mu.TryLock(), "session mutex held during catalogue I/O")
982		-	sess.mu.Unlock()
	979	+	require.False(t, d.mu.TryLock(), "daemon mutex must guard the metadata snapshot")
	980	+	require.False(t, sess.mu.TryLock(), "session mutex must guard the metadata snapshot")
983	981		return nil
984	982		}).Once()
985	983

 	require.Equal(t, []string{"shell"}, records[0].TabNames)
 }
-func TestRenameTabDoesNotHoldDaemonOrSessionMutexDuringCatalogueIO(t *testing.T) {
+func TestRenameTabBuffersMetadataUnderStateLocks(t *testing.T) {
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
 	sess := newSnapshotTestSession(t, "work", false, "/work")
 	d.sessions[sess.id] = sess
 	catalogue := portsmocks.NewMockCatalogue(t)
 	WithCatalogue(catalogue, nil)(d)
 	catalogue.EXPECT().UpdateMetadata(mock.Anything).RunAndReturn(func(domain.CatalogueMetadataUpdate) error {
-		require.True(t, d.mu.TryLock(), "daemon mutex held during catalogue I/O")
+		require.False(t, d.mu.TryLock(), "daemon mutex must guard the metadata snapshot")
-		d.mu.Unlock()
+		require.False(t, sess.mu.TryLock(), "session mutex must guard the metadata snapshot")
-		require.True(t, sess.mu.TryLock(), "session mutex held during catalogue I/O")
-		sess.mu.Unlock()
 		return nil
 	}).Once()

M internal/usecase/recovery/checkpoint_test.go +1 -0


@@ -33,6 +33,7 @@ func (c *checkpointCatalogue) Create(domain.CatalogueRecord) error { return nil
33	33		func (c *checkpointCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error {
34	34		return nil
35	35		}
	36	+	func (c *checkpointCatalogue) Sync() error { return nil }
36	37		func (c *checkpointCatalogue) Replace(name string, record domain.CatalogueRecord) error {
37	38		c.mu.Lock()
38	39		defer c.mu.Unlock()


@@ -33,6 +33,7 @@ func (c *checkpointCatalogue) Create(domain.CatalogueRecord) error { return nil
33	func (c *checkpointCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error {	33		func (c *checkpointCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error {
34	return nil	34		return nil
35	}	35		}
		36	+	func (c *checkpointCatalogue) Sync() error { return nil }
36	func (c *checkpointCatalogue) Replace(name string, record domain.CatalogueRecord) error {	37		func (c *checkpointCatalogue) Replace(name string, record domain.CatalogueRecord) error {
37	c.mu.Lock()	38		c.mu.Lock()
38	defer c.mu.Unlock()	39		defer c.mu.Unlock()

M internal/usecase/recovery/transactions_test.go +1 -0


@@ -60,6 +60,7 @@ func (c *transactionCatalogue) Create(record domain.CatalogueRecord) error {
60	60		}
61	61
62	62		func (c *transactionCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error { return nil }
	63	+	func (c *transactionCatalogue) Sync() error { return nil }
63	64
64	65		func (c *transactionCatalogue) Replace(name string, record domain.CatalogueRecord) error {
65	66		c.mu.Lock()


@@ -60,6 +60,7 @@ func (c *transactionCatalogue) Create(record domain.CatalogueRecord) error {
60	}	60		}
61		61
62	func (c *transactionCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error { return nil }	62		func (c *transactionCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error { return nil }
		63	+	func (c *transactionCatalogue) Sync() error { return nil }
63		64
64	func (c *transactionCatalogue) Replace(name string, record domain.CatalogueRecord) error {	65		func (c *transactionCatalogue) Replace(name string, record domain.CatalogueRecord) error {
65	c.mu.Lock()	66		c.mu.Lock()