feat(recovery): make session lifecycle transactions durable

commit: 66c3cd3725507e6e384ddd4b0793bd7bb57edde8
author: brice · 2mo ago · 2026-07-26 00:49 UTC
parent: ff3aadc
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

11 files · +866 -215

M internal/adapters/snapshot/maintenance_delete_test.go +36 -0


@@ -5,6 +5,7 @@ import (
5	5		"encoding/base64"
6	6		"encoding/hex"
7	7		"errors"
	8	+	"os"
8	9		"path/filepath"
9	10		"slices"
10	11		"testing"

@@ -37,6 +38,22 @@ func TestDeletionTombstoneCodec(t *testing.T) {
37	38		if _, err := decodeDeletionTombstone(append(encoded, 0)); err == nil {
38	39		t.Fatal("trailing byte accepted")
39	40		}
	41	+	for _, test := range []struct {
	42	+	name string
	43	+	mutate func([]byte) []byte
	44	+	}{
	45	+	{name: "wrong magic", mutate: func(data []byte) []byte { data[0] ^= 0xff; return data }},
	46	+	{name: "wrong version", mutate: func(data []byte) []byte { data[5]++; return data }},
	47	+	{name: "invalid name", mutate: func(data []byte) []byte { copy(data[8:12], "../x"); return data }},
	48	+	{name: "zero ID", mutate: func(data []byte) []byte { clear(data[len(data)-16:]); return data }},
	49	+	} {
	50	+	t.Run(test.name, func(t *testing.T) {
	51	+	malformed := test.mutate(append([]byte(nil), encoded...))
	52	+	if _, err := decodeDeletionTombstone(malformed); err == nil {
	53	+	t.Fatal("malformed tombstone accepted")
	54	+	}
	55	+	})
	56	+	}
40	57
41	58		for _, boundary := range []string{"file-sync", "install", "directory-sync"} {
42	59		t.Run(boundary, func(t *testing.T) {

@@ -117,6 +134,25 @@ func TestDeletionTombstoneListingResumesAcrossRepositoryRecreation(t *testing.T)
117	134		}
118	135		}
119	136
	137	+	func TestMalformedDeletionTombstone(t *testing.T) {
	138	+	dir := privateDir(t)
	139	+	repo := NewRepository(dir)
	140	+	tombstone := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{7}}
	141	+	if err := repo.WriteDeletionTombstone(context.Background(), tombstone); err != nil {
	142	+	t.Fatal(err)
	143	+	}
	144	+	path := repo.deletionTombstonePath(tombstone.IncarnationID)
	145	+	if err := os.WriteFile(path, []byte("malformed"), 0o600); err != nil {
	146	+	t.Fatal(err)
	147	+	}
	148	+	if _, err := repo.ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 64, Bytes: 1024}); err == nil {
	149	+	t.Fatal("malformed tombstone was skipped")
	150	+	}
	151	+	if _, err := os.Stat(path); err != nil {
	152	+	t.Fatalf("malformed tombstone was removed: %v", err)
	153	+	}
	154	+	}
	155	+
120	156		func TestDeletionTombstoneListingRejectsMalformedContinuation(t *testing.T) {
121	157		repo := NewRepository(privateDir(t))
122	158		ctx := context.Background()


@@ -5,6 +5,7 @@ import (
5	"encoding/base64"	5		"encoding/base64"
6	"encoding/hex"	6		"encoding/hex"
7	"errors"	7		"errors"
		8	+	"os"
8	"path/filepath"	9		"path/filepath"
9	"slices"	10		"slices"
10	"testing"	11		"testing"

@@ -37,6 +38,22 @@ func TestDeletionTombstoneCodec(t *testing.T) {
37	if _, err := decodeDeletionTombstone(append(encoded, 0)); err == nil {	38		if _, err := decodeDeletionTombstone(append(encoded, 0)); err == nil {
38	t.Fatal("trailing byte accepted")	39		t.Fatal("trailing byte accepted")
39	}	40		}
		41	+	for _, test := range []struct {
		42	+	name string
		43	+	mutate func([]byte) []byte
		44	+	}{
		45	+	{name: "wrong magic", mutate: func(data []byte) []byte { data[0] ^= 0xff; return data }},
		46	+	{name: "wrong version", mutate: func(data []byte) []byte { data[5]++; return data }},
		47	+	{name: "invalid name", mutate: func(data []byte) []byte { copy(data[8:12], "../x"); return data }},
		48	+	{name: "zero ID", mutate: func(data []byte) []byte { clear(data[len(data)-16:]); return data }},
		49	+	} {
		50	+	t.Run(test.name, func(t *testing.T) {
		51	+	malformed := test.mutate(append([]byte(nil), encoded...))
		52	+	if _, err := decodeDeletionTombstone(malformed); err == nil {
		53	+	t.Fatal("malformed tombstone accepted")
		54	+	}
		55	+	})
		56	+	}
40		57
41	for _, boundary := range []string{"file-sync", "install", "directory-sync"} {	58		for _, boundary := range []string{"file-sync", "install", "directory-sync"} {
42	t.Run(boundary, func(t *testing.T) {	59		t.Run(boundary, func(t *testing.T) {

@@ -117,6 +134,25 @@ func TestDeletionTombstoneListingResumesAcrossRepositoryRecreation(t *testing.T)
117	}	134		}
118	}	135		}
119		136
		137	+	func TestMalformedDeletionTombstone(t *testing.T) {
		138	+	dir := privateDir(t)
		139	+	repo := NewRepository(dir)
		140	+	tombstone := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{7}}
		141	+	if err := repo.WriteDeletionTombstone(context.Background(), tombstone); err != nil {
		142	+	t.Fatal(err)
		143	+	}
		144	+	path := repo.deletionTombstonePath(tombstone.IncarnationID)
		145	+	if err := os.WriteFile(path, []byte("malformed"), 0o600); err != nil {
		146	+	t.Fatal(err)
		147	+	}
		148	+	if _, err := repo.ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, ports.MaintenanceBudget{Entries: 64, Bytes: 1024}); err == nil {
		149	+	t.Fatal("malformed tombstone was skipped")
		150	+	}
		151	+	if _, err := os.Stat(path); err != nil {
		152	+	t.Fatalf("malformed tombstone was removed: %v", err)
		153	+	}
		154	+	}
		155	+
120	func TestDeletionTombstoneListingRejectsMalformedContinuation(t *testing.T) {	156		func TestDeletionTombstoneListingRejectsMalformedContinuation(t *testing.T) {
121	repo := NewRepository(privateDir(t))	157		repo := NewRepository(privateDir(t))
122	ctx := context.Background()	158		ctx := context.Background()

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


@@ -530,7 +530,7 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
530	530		if err := coordinator.Recover(ctx); err != nil {
531	531		return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
532	532		}
533		-	daemonOpts = append(daemonOpts, daemon.WithCheckpointCoordinator(coordinator))
	533	+	daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
534	534		log.Info("session persistence enabled", "path", storePath)
535	535		daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
536	536


@@ -530,7 +530,7 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
530		if err := coordinator.Recover(ctx); err != nil {	530		if err := coordinator.Recover(ctx); err != nil {
531		return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())	531		return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
532		}	532		}
533	-	daemonOpts = append(daemonOpts, daemon.WithCheckpointCoordinator(coordinator))	533	+	daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
534		log.Info("session persistence enabled", "path", storePath)	534		log.Info("session persistence enabled", "path", storePath)
535		daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))	535		daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
536			536

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


@@ -605,6 +605,372 @@ func (_c *MockCheckpointCoordinator_PublishCheckpoint_Call) RunAndReturn(run fun
605	605		return _c
606	606		}
607	607
	608	+	// NewMockSessionLifecycleCoordinator creates a new instance of MockSessionLifecycleCoordinator. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
	609	+	// The first argument is typically a *testing.T value.
	610	+	func NewMockSessionLifecycleCoordinator(t interface {
	611	+	mock.TestingT
	612	+	Cleanup(func())
	613	+	}) *MockSessionLifecycleCoordinator {
	614	+	mock := &MockSessionLifecycleCoordinator{}
	615	+	mock.Mock.Test(t)
	616	+
	617	+	t.Cleanup(func() { mock.AssertExpectations(t) })
	618	+
	619	+	return mock
	620	+	}
	621	+
	622	+	// MockSessionLifecycleCoordinator is an autogenerated mock type for the SessionLifecycleCoordinator type
	623	+	type MockSessionLifecycleCoordinator struct {
	624	+	mock.Mock
	625	+	}
	626	+
	627	+	type MockSessionLifecycleCoordinator_Expecter struct {
	628	+	mock *mock.Mock
	629	+	}
	630	+
	631	+	func (_m MockSessionLifecycleCoordinator) EXPECT() MockSessionLifecycleCoordinator_Expecter {
	632	+	return &MockSessionLifecycleCoordinator_Expecter{mock: &_m.Mock}
	633	+	}
	634	+
	635	+	// Create provides a mock function for the type MockSessionLifecycleCoordinator
	636	+	func (_mock *MockSessionLifecycleCoordinator) Create(context1 context.Context, catalogueRecord domain.CatalogueRecord) (domain.CatalogueRecord, error) {
	637	+	ret := _mock.Called(context1, catalogueRecord)
	638	+
	639	+	if len(ret) == 0 {
	640	+	panic("no return value specified for Create")
	641	+	}
	642	+
	643	+	var r0 domain.CatalogueRecord
	644	+	var r1 error
	645	+	if returnFunc, ok := ret.Get(0).(func(context.Context, domain.CatalogueRecord) (domain.CatalogueRecord, error)); ok {
	646	+	return returnFunc(context1, catalogueRecord)
	647	+	}
	648	+	if returnFunc, ok := ret.Get(0).(func(context.Context, domain.CatalogueRecord) domain.CatalogueRecord); ok {
	649	+	r0 = returnFunc(context1, catalogueRecord)
	650	+	} else {
	651	+	r0 = ret.Get(0).(domain.CatalogueRecord)
	652	+	}
	653	+	if returnFunc, ok := ret.Get(1).(func(context.Context, domain.CatalogueRecord) error); ok {
	654	+	r1 = returnFunc(context1, catalogueRecord)
	655	+	} else {
	656	+	r1 = ret.Error(1)
	657	+	}
	658	+	return r0, r1
	659	+	}
	660	+
	661	+	// MockSessionLifecycleCoordinator_Create_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Create'
	662	+	type MockSessionLifecycleCoordinator_Create_Call struct {
	663	+	*mock.Call
	664	+	}
	665	+
	666	+	// Create is a helper method to define mock.On call
	667	+	// - context1 context.Context
	668	+	// - catalogueRecord domain.CatalogueRecord
	669	+	func (_e MockSessionLifecycleCoordinator_Expecter) Create(context1 any, catalogueRecord any) MockSessionLifecycleCoordinator_Create_Call {
	670	+	return &MockSessionLifecycleCoordinator_Create_Call{Call: _e.mock.On("Create", context1, catalogueRecord)}
	671	+	}
	672	+
	673	+	func (_c MockSessionLifecycleCoordinator_Create_Call) Run(run func(context1 context.Context, catalogueRecord domain.CatalogueRecord)) MockSessionLifecycleCoordinator_Create_Call {
	674	+	_c.Call.Run(func(args mock.Arguments) {
	675	+	var arg0 context.Context
	676	+	if args[0] != nil {
	677	+	arg0 = args[0].(context.Context)
	678	+	}
	679	+	var arg1 domain.CatalogueRecord
	680	+	if args[1] != nil {
	681	+	arg1 = args[1].(domain.CatalogueRecord)
	682	+	}
	683	+	run(
	684	+	arg0,
	685	+	arg1,
	686	+	)
	687	+	})
	688	+	return _c
	689	+	}
	690	+
	691	+	func (_c MockSessionLifecycleCoordinator_Create_Call) Return(catalogueRecord1 domain.CatalogueRecord, err error) MockSessionLifecycleCoordinator_Create_Call {
	692	+	_c.Call.Return(catalogueRecord1, err)
	693	+	return _c
	694	+	}
	695	+
	696	+	func (_c MockSessionLifecycleCoordinator_Create_Call) RunAndReturn(run func(context1 context.Context, catalogueRecord domain.CatalogueRecord) (domain.CatalogueRecord, error)) MockSessionLifecycleCoordinator_Create_Call {
	697	+	_c.Call.Return(run)
	698	+	return _c
	699	+	}
	700	+
	701	+	// Delete provides a mock function for the type MockSessionLifecycleCoordinator
	702	+	func (_mock *MockSessionLifecycleCoordinator) Delete(context1 context.Context, s string) error {
	703	+	ret := _mock.Called(context1, s)
	704	+
	705	+	if len(ret) == 0 {
	706	+	panic("no return value specified for Delete")
	707	+	}
	708	+
	709	+	var r0 error
	710	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string) error); ok {
	711	+	r0 = returnFunc(context1, s)
	712	+	} else {
	713	+	r0 = ret.Error(0)
	714	+	}
	715	+	return r0
	716	+	}
	717	+
	718	+	// MockSessionLifecycleCoordinator_Delete_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Delete'
	719	+	type MockSessionLifecycleCoordinator_Delete_Call struct {
	720	+	*mock.Call
	721	+	}
	722	+
	723	+	// Delete is a helper method to define mock.On call
	724	+	// - context1 context.Context
	725	+	// - s string
	726	+	func (_e MockSessionLifecycleCoordinator_Expecter) Delete(context1 any, s any) MockSessionLifecycleCoordinator_Delete_Call {
	727	+	return &MockSessionLifecycleCoordinator_Delete_Call{Call: _e.mock.On("Delete", context1, s)}
	728	+	}
	729	+
	730	+	func (_c MockSessionLifecycleCoordinator_Delete_Call) Run(run func(context1 context.Context, s string)) MockSessionLifecycleCoordinator_Delete_Call {
	731	+	_c.Call.Run(func(args mock.Arguments) {
	732	+	var arg0 context.Context
	733	+	if args[0] != nil {
	734	+	arg0 = args[0].(context.Context)
	735	+	}
	736	+	var arg1 string
	737	+	if args[1] != nil {
	738	+	arg1 = args[1].(string)
	739	+	}
	740	+	run(
	741	+	arg0,
	742	+	arg1,
	743	+	)
	744	+	})
	745	+	return _c
	746	+	}
	747	+
	748	+	func (_c MockSessionLifecycleCoordinator_Delete_Call) Return(err error) MockSessionLifecycleCoordinator_Delete_Call {
	749	+	_c.Call.Return(err)
	750	+	return _c
	751	+	}
	752	+
	753	+	func (_c MockSessionLifecycleCoordinator_Delete_Call) RunAndReturn(run func(context1 context.Context, s string) error) MockSessionLifecycleCoordinator_Delete_Call {
	754	+	_c.Call.Return(run)
	755	+	return _c
	756	+	}
	757	+
	758	+	// PromoteFallback provides a mock function for the type MockSessionLifecycleCoordinator
	759	+	func (_mock *MockSessionLifecycleCoordinator) PromoteFallback(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (ports.FallbackPromotionOutcome, error) {
	760	+	ret := _mock.Called(context1, s, checkpointRef)
	761	+
	762	+	if len(ret) == 0 {
	763	+	panic("no return value specified for PromoteFallback")
	764	+	}
	765	+
	766	+	var r0 ports.FallbackPromotionOutcome
	767	+	var r1 error
	768	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) (ports.FallbackPromotionOutcome, error)); ok {
	769	+	return returnFunc(context1, s, checkpointRef)
	770	+	}
	771	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) ports.FallbackPromotionOutcome); ok {
	772	+	r0 = returnFunc(context1, s, checkpointRef)
	773	+	} else {
	774	+	r0 = ret.Get(0).(ports.FallbackPromotionOutcome)
	775	+	}
	776	+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, domain.CheckpointRef) error); ok {
	777	+	r1 = returnFunc(context1, s, checkpointRef)
	778	+	} else {
	779	+	r1 = ret.Error(1)
	780	+	}
	781	+	return r0, r1
	782	+	}
	783	+
	784	+	// MockSessionLifecycleCoordinator_PromoteFallback_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'PromoteFallback'
	785	+	type MockSessionLifecycleCoordinator_PromoteFallback_Call struct {
	786	+	*mock.Call
	787	+	}
	788	+
	789	+	// PromoteFallback is a helper method to define mock.On call
	790	+	// - context1 context.Context
	791	+	// - s string
	792	+	// - checkpointRef domain.CheckpointRef
	793	+	func (_e MockSessionLifecycleCoordinator_Expecter) PromoteFallback(context1 any, s any, checkpointRef any) MockSessionLifecycleCoordinator_PromoteFallback_Call {
	794	+	return &MockSessionLifecycleCoordinator_PromoteFallback_Call{Call: _e.mock.On("PromoteFallback", context1, s, checkpointRef)}
	795	+	}
	796	+
	797	+	func (_c MockSessionLifecycleCoordinator_PromoteFallback_Call) Run(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef)) MockSessionLifecycleCoordinator_PromoteFallback_Call {
	798	+	_c.Call.Run(func(args mock.Arguments) {
	799	+	var arg0 context.Context
	800	+	if args[0] != nil {
	801	+	arg0 = args[0].(context.Context)
	802	+	}
	803	+	var arg1 string
	804	+	if args[1] != nil {
	805	+	arg1 = args[1].(string)
	806	+	}
	807	+	var arg2 domain.CheckpointRef
	808	+	if args[2] != nil {
	809	+	arg2 = args[2].(domain.CheckpointRef)
	810	+	}
	811	+	run(
	812	+	arg0,
	813	+	arg1,
	814	+	arg2,
	815	+	)
	816	+	})
	817	+	return _c
	818	+	}
	819	+
	820	+	func (_c MockSessionLifecycleCoordinator_PromoteFallback_Call) Return(fallbackPromotionOutcome ports.FallbackPromotionOutcome, err error) MockSessionLifecycleCoordinator_PromoteFallback_Call {
	821	+	_c.Call.Return(fallbackPromotionOutcome, err)
	822	+	return _c
	823	+	}
	824	+
	825	+	func (_c MockSessionLifecycleCoordinator_PromoteFallback_Call) RunAndReturn(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (ports.FallbackPromotionOutcome, error)) MockSessionLifecycleCoordinator_PromoteFallback_Call {
	826	+	_c.Call.Return(run)
	827	+	return _c
	828	+	}
	829	+
	830	+	// PublishCheckpoint provides a mock function for the type MockSessionLifecycleCoordinator
	831	+	func (_mock *MockSessionLifecycleCoordinator) PublishCheckpoint(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication) (domain.CatalogueRecord, error) {
	832	+	ret := _mock.Called(context1, s, snapshotPublication)
	833	+
	834	+	if len(ret) == 0 {
	835	+	panic("no return value specified for PublishCheckpoint")
	836	+	}
	837	+
	838	+	var r0 domain.CatalogueRecord
	839	+	var r1 error
	840	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, ports.SnapshotPublication) (domain.CatalogueRecord, error)); ok {
	841	+	return returnFunc(context1, s, snapshotPublication)
	842	+	}
	843	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, ports.SnapshotPublication) domain.CatalogueRecord); ok {
	844	+	r0 = returnFunc(context1, s, snapshotPublication)
	845	+	} else {
	846	+	r0 = ret.Get(0).(domain.CatalogueRecord)
	847	+	}
	848	+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, ports.SnapshotPublication) error); ok {
	849	+	r1 = returnFunc(context1, s, snapshotPublication)
	850	+	} else {
	851	+	r1 = ret.Error(1)
	852	+	}
	853	+	return r0, r1
	854	+	}
	855	+
	856	+	// MockSessionLifecycleCoordinator_PublishCheckpoint_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'PublishCheckpoint'
	857	+	type MockSessionLifecycleCoordinator_PublishCheckpoint_Call struct {
	858	+	*mock.Call
	859	+	}
	860	+
	861	+	// PublishCheckpoint is a helper method to define mock.On call
	862	+	// - context1 context.Context
	863	+	// - s string
	864	+	// - snapshotPublication ports.SnapshotPublication
	865	+	func (_e MockSessionLifecycleCoordinator_Expecter) PublishCheckpoint(context1 any, s any, snapshotPublication any) MockSessionLifecycleCoordinator_PublishCheckpoint_Call {
	866	+	return &MockSessionLifecycleCoordinator_PublishCheckpoint_Call{Call: _e.mock.On("PublishCheckpoint", context1, s, snapshotPublication)}
	867	+	}
	868	+
	869	+	func (_c MockSessionLifecycleCoordinator_PublishCheckpoint_Call) Run(run func(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication)) MockSessionLifecycleCoordinator_PublishCheckpoint_Call {
	870	+	_c.Call.Run(func(args mock.Arguments) {
	871	+	var arg0 context.Context
	872	+	if args[0] != nil {
	873	+	arg0 = args[0].(context.Context)
	874	+	}
	875	+	var arg1 string
	876	+	if args[1] != nil {
	877	+	arg1 = args[1].(string)
	878	+	}
	879	+	var arg2 ports.SnapshotPublication
	880	+	if args[2] != nil {
	881	+	arg2 = args[2].(ports.SnapshotPublication)
	882	+	}
	883	+	run(
	884	+	arg0,
	885	+	arg1,
	886	+	arg2,
	887	+	)
	888	+	})
	889	+	return _c
	890	+	}
	891	+
	892	+	func (_c MockSessionLifecycleCoordinator_PublishCheckpoint_Call) Return(catalogueRecord domain.CatalogueRecord, err error) MockSessionLifecycleCoordinator_PublishCheckpoint_Call {
	893	+	_c.Call.Return(catalogueRecord, err)
	894	+	return _c
	895	+	}
	896	+
	897	+	func (_c MockSessionLifecycleCoordinator_PublishCheckpoint_Call) RunAndReturn(run func(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication) (domain.CatalogueRecord, error)) MockSessionLifecycleCoordinator_PublishCheckpoint_Call {
	898	+	_c.Call.Return(run)
	899	+	return _c
	900	+	}
	901	+
	902	+	// Rename provides a mock function for the type MockSessionLifecycleCoordinator
	903	+	func (_mock *MockSessionLifecycleCoordinator) Rename(context1 context.Context, s string, s1 string) (domain.CatalogueRecord, error) {
	904	+	ret := _mock.Called(context1, s, s1)
	905	+
	906	+	if len(ret) == 0 {
	907	+	panic("no return value specified for Rename")
	908	+	}
	909	+
	910	+	var r0 domain.CatalogueRecord
	911	+	var r1 error
	912	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) (domain.CatalogueRecord, error)); ok {
	913	+	return returnFunc(context1, s, s1)
	914	+	}
	915	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) domain.CatalogueRecord); ok {
	916	+	r0 = returnFunc(context1, s, s1)
	917	+	} else {
	918	+	r0 = ret.Get(0).(domain.CatalogueRecord)
	919	+	}
	920	+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, string) error); ok {
	921	+	r1 = returnFunc(context1, s, s1)
	922	+	} else {
	923	+	r1 = ret.Error(1)
	924	+	}
	925	+	return r0, r1
	926	+	}
	927	+
	928	+	// MockSessionLifecycleCoordinator_Rename_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Rename'
	929	+	type MockSessionLifecycleCoordinator_Rename_Call struct {
	930	+	*mock.Call
	931	+	}
	932	+
	933	+	// Rename is a helper method to define mock.On call
	934	+	// - context1 context.Context
	935	+	// - s string
	936	+	// - s1 string
	937	+	func (_e MockSessionLifecycleCoordinator_Expecter) Rename(context1 any, s any, s1 any) MockSessionLifecycleCoordinator_Rename_Call {
	938	+	return &MockSessionLifecycleCoordinator_Rename_Call{Call: _e.mock.On("Rename", context1, s, s1)}
	939	+	}
	940	+
	941	+	func (_c MockSessionLifecycleCoordinator_Rename_Call) Run(run func(context1 context.Context, s string, s1 string)) MockSessionLifecycleCoordinator_Rename_Call {
	942	+	_c.Call.Run(func(args mock.Arguments) {
	943	+	var arg0 context.Context
	944	+	if args[0] != nil {
	945	+	arg0 = args[0].(context.Context)
	946	+	}
	947	+	var arg1 string
	948	+	if args[1] != nil {
	949	+	arg1 = args[1].(string)
	950	+	}
	951	+	var arg2 string
	952	+	if args[2] != nil {
	953	+	arg2 = args[2].(string)
	954	+	}
	955	+	run(
	956	+	arg0,
	957	+	arg1,
	958	+	arg2,
	959	+	)
	960	+	})
	961	+	return _c
	962	+	}
	963	+
	964	+	func (_c MockSessionLifecycleCoordinator_Rename_Call) Return(catalogueRecord domain.CatalogueRecord, err error) MockSessionLifecycleCoordinator_Rename_Call {
	965	+	_c.Call.Return(catalogueRecord, err)
	966	+	return _c
	967	+	}
	968	+
	969	+	func (_c MockSessionLifecycleCoordinator_Rename_Call) RunAndReturn(run func(context1 context.Context, s string, s1 string) (domain.CatalogueRecord, error)) MockSessionLifecycleCoordinator_Rename_Call {
	970	+	_c.Call.Return(run)
	971	+	return _c
	972	+	}
	973	+
608	974		// NewMockCatalogue creates a new instance of MockCatalogue. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
609	975		// The first argument is typically a *testing.T value.
610	976		func NewMockCatalogue(t interface {


@@ -605,6 +605,372 @@ func (_c *MockCheckpointCoordinator_PublishCheckpoint_Call) RunAndReturn(run fun
605	return _c	605		return _c
606	}	606		}
607		607
		608	+	// NewMockSessionLifecycleCoordinator creates a new instance of MockSessionLifecycleCoordinator. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
		609	+	// The first argument is typically a *testing.T value.
		610	+	func NewMockSessionLifecycleCoordinator(t interface {
		611	+	mock.TestingT
		612	+	Cleanup(func())
		613	+	}) *MockSessionLifecycleCoordinator {
		614	+	mock := &MockSessionLifecycleCoordinator{}
		615	+	mock.Mock.Test(t)
		616	+
		617	+	t.Cleanup(func() { mock.AssertExpectations(t) })
		618	+
		619	+	return mock
		620	+	}
		621	+
		622	+	// MockSessionLifecycleCoordinator is an autogenerated mock type for the SessionLifecycleCoordinator type
		623	+	type MockSessionLifecycleCoordinator struct {
		624	+	mock.Mock
		625	+	}
		626	+
		627	+	type MockSessionLifecycleCoordinator_Expecter struct {
		628	+	mock *mock.Mock
		629	+	}
		630	+
		631	+	func (_m MockSessionLifecycleCoordinator) EXPECT() MockSessionLifecycleCoordinator_Expecter {
		632	+	return &MockSessionLifecycleCoordinator_Expecter{mock: &_m.Mock}
		633	+	}
		634	+
		635	+	// Create provides a mock function for the type MockSessionLifecycleCoordinator
		636	+	func (_mock *MockSessionLifecycleCoordinator) Create(context1 context.Context, catalogueRecord domain.CatalogueRecord) (domain.CatalogueRecord, error) {
		637	+	ret := _mock.Called(context1, catalogueRecord)
		638	+
		639	+	if len(ret) == 0 {
		640	+	panic("no return value specified for Create")
		641	+	}
		642	+
		643	+	var r0 domain.CatalogueRecord
		644	+	var r1 error
		645	+	if returnFunc, ok := ret.Get(0).(func(context.Context, domain.CatalogueRecord) (domain.CatalogueRecord, error)); ok {
		646	+	return returnFunc(context1, catalogueRecord)
		647	+	}
		648	+	if returnFunc, ok := ret.Get(0).(func(context.Context, domain.CatalogueRecord) domain.CatalogueRecord); ok {
		649	+	r0 = returnFunc(context1, catalogueRecord)
		650	+	} else {
		651	+	r0 = ret.Get(0).(domain.CatalogueRecord)
		652	+	}
		653	+	if returnFunc, ok := ret.Get(1).(func(context.Context, domain.CatalogueRecord) error); ok {
		654	+	r1 = returnFunc(context1, catalogueRecord)
		655	+	} else {
		656	+	r1 = ret.Error(1)
		657	+	}
		658	+	return r0, r1
		659	+	}
		660	+
		661	+	// MockSessionLifecycleCoordinator_Create_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Create'
		662	+	type MockSessionLifecycleCoordinator_Create_Call struct {
		663	+	*mock.Call
		664	+	}
		665	+
		666	+	// Create is a helper method to define mock.On call
		667	+	// - context1 context.Context
		668	+	// - catalogueRecord domain.CatalogueRecord
		669	+	func (_e MockSessionLifecycleCoordinator_Expecter) Create(context1 any, catalogueRecord any) MockSessionLifecycleCoordinator_Create_Call {
		670	+	return &MockSessionLifecycleCoordinator_Create_Call{Call: _e.mock.On("Create", context1, catalogueRecord)}
		671	+	}
		672	+
		673	+	func (_c MockSessionLifecycleCoordinator_Create_Call) Run(run func(context1 context.Context, catalogueRecord domain.CatalogueRecord)) MockSessionLifecycleCoordinator_Create_Call {
		674	+	_c.Call.Run(func(args mock.Arguments) {
		675	+	var arg0 context.Context
		676	+	if args[0] != nil {
		677	+	arg0 = args[0].(context.Context)
		678	+	}
		679	+	var arg1 domain.CatalogueRecord
		680	+	if args[1] != nil {
		681	+	arg1 = args[1].(domain.CatalogueRecord)
		682	+	}
		683	+	run(
		684	+	arg0,
		685	+	arg1,
		686	+	)
		687	+	})
		688	+	return _c
		689	+	}
		690	+
		691	+	func (_c MockSessionLifecycleCoordinator_Create_Call) Return(catalogueRecord1 domain.CatalogueRecord, err error) MockSessionLifecycleCoordinator_Create_Call {
		692	+	_c.Call.Return(catalogueRecord1, err)
		693	+	return _c
		694	+	}
		695	+
		696	+	func (_c MockSessionLifecycleCoordinator_Create_Call) RunAndReturn(run func(context1 context.Context, catalogueRecord domain.CatalogueRecord) (domain.CatalogueRecord, error)) MockSessionLifecycleCoordinator_Create_Call {
		697	+	_c.Call.Return(run)
		698	+	return _c
		699	+	}
		700	+
		701	+	// Delete provides a mock function for the type MockSessionLifecycleCoordinator
		702	+	func (_mock *MockSessionLifecycleCoordinator) Delete(context1 context.Context, s string) error {
		703	+	ret := _mock.Called(context1, s)
		704	+
		705	+	if len(ret) == 0 {
		706	+	panic("no return value specified for Delete")
		707	+	}
		708	+
		709	+	var r0 error
		710	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string) error); ok {
		711	+	r0 = returnFunc(context1, s)
		712	+	} else {
		713	+	r0 = ret.Error(0)
		714	+	}
		715	+	return r0
		716	+	}
		717	+
		718	+	// MockSessionLifecycleCoordinator_Delete_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Delete'
		719	+	type MockSessionLifecycleCoordinator_Delete_Call struct {
		720	+	*mock.Call
		721	+	}
		722	+
		723	+	// Delete is a helper method to define mock.On call
		724	+	// - context1 context.Context
		725	+	// - s string
		726	+	func (_e MockSessionLifecycleCoordinator_Expecter) Delete(context1 any, s any) MockSessionLifecycleCoordinator_Delete_Call {
		727	+	return &MockSessionLifecycleCoordinator_Delete_Call{Call: _e.mock.On("Delete", context1, s)}
		728	+	}
		729	+
		730	+	func (_c MockSessionLifecycleCoordinator_Delete_Call) Run(run func(context1 context.Context, s string)) MockSessionLifecycleCoordinator_Delete_Call {
		731	+	_c.Call.Run(func(args mock.Arguments) {
		732	+	var arg0 context.Context
		733	+	if args[0] != nil {
		734	+	arg0 = args[0].(context.Context)
		735	+	}
		736	+	var arg1 string
		737	+	if args[1] != nil {
		738	+	arg1 = args[1].(string)
		739	+	}
		740	+	run(
		741	+	arg0,
		742	+	arg1,
		743	+	)
		744	+	})
		745	+	return _c
		746	+	}
		747	+
		748	+	func (_c MockSessionLifecycleCoordinator_Delete_Call) Return(err error) MockSessionLifecycleCoordinator_Delete_Call {
		749	+	_c.Call.Return(err)
		750	+	return _c
		751	+	}
		752	+
		753	+	func (_c MockSessionLifecycleCoordinator_Delete_Call) RunAndReturn(run func(context1 context.Context, s string) error) MockSessionLifecycleCoordinator_Delete_Call {
		754	+	_c.Call.Return(run)
		755	+	return _c
		756	+	}
		757	+
		758	+	// PromoteFallback provides a mock function for the type MockSessionLifecycleCoordinator
		759	+	func (_mock *MockSessionLifecycleCoordinator) PromoteFallback(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (ports.FallbackPromotionOutcome, error) {
		760	+	ret := _mock.Called(context1, s, checkpointRef)
		761	+
		762	+	if len(ret) == 0 {
		763	+	panic("no return value specified for PromoteFallback")
		764	+	}
		765	+
		766	+	var r0 ports.FallbackPromotionOutcome
		767	+	var r1 error
		768	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) (ports.FallbackPromotionOutcome, error)); ok {
		769	+	return returnFunc(context1, s, checkpointRef)
		770	+	}
		771	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) ports.FallbackPromotionOutcome); ok {
		772	+	r0 = returnFunc(context1, s, checkpointRef)
		773	+	} else {
		774	+	r0 = ret.Get(0).(ports.FallbackPromotionOutcome)
		775	+	}
		776	+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, domain.CheckpointRef) error); ok {
		777	+	r1 = returnFunc(context1, s, checkpointRef)
		778	+	} else {
		779	+	r1 = ret.Error(1)
		780	+	}
		781	+	return r0, r1
		782	+	}
		783	+
		784	+	// MockSessionLifecycleCoordinator_PromoteFallback_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'PromoteFallback'
		785	+	type MockSessionLifecycleCoordinator_PromoteFallback_Call struct {
		786	+	*mock.Call
		787	+	}
		788	+
		789	+	// PromoteFallback is a helper method to define mock.On call
		790	+	// - context1 context.Context
		791	+	// - s string
		792	+	// - checkpointRef domain.CheckpointRef
		793	+	func (_e MockSessionLifecycleCoordinator_Expecter) PromoteFallback(context1 any, s any, checkpointRef any) MockSessionLifecycleCoordinator_PromoteFallback_Call {
		794	+	return &MockSessionLifecycleCoordinator_PromoteFallback_Call{Call: _e.mock.On("PromoteFallback", context1, s, checkpointRef)}
		795	+	}
		796	+
		797	+	func (_c MockSessionLifecycleCoordinator_PromoteFallback_Call) Run(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef)) MockSessionLifecycleCoordinator_PromoteFallback_Call {
		798	+	_c.Call.Run(func(args mock.Arguments) {
		799	+	var arg0 context.Context
		800	+	if args[0] != nil {
		801	+	arg0 = args[0].(context.Context)
		802	+	}
		803	+	var arg1 string
		804	+	if args[1] != nil {
		805	+	arg1 = args[1].(string)
		806	+	}
		807	+	var arg2 domain.CheckpointRef
		808	+	if args[2] != nil {
		809	+	arg2 = args[2].(domain.CheckpointRef)
		810	+	}
		811	+	run(
		812	+	arg0,
		813	+	arg1,
		814	+	arg2,
		815	+	)
		816	+	})
		817	+	return _c
		818	+	}
		819	+
		820	+	func (_c MockSessionLifecycleCoordinator_PromoteFallback_Call) Return(fallbackPromotionOutcome ports.FallbackPromotionOutcome, err error) MockSessionLifecycleCoordinator_PromoteFallback_Call {
		821	+	_c.Call.Return(fallbackPromotionOutcome, err)
		822	+	return _c
		823	+	}
		824	+
		825	+	func (_c MockSessionLifecycleCoordinator_PromoteFallback_Call) RunAndReturn(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (ports.FallbackPromotionOutcome, error)) MockSessionLifecycleCoordinator_PromoteFallback_Call {
		826	+	_c.Call.Return(run)
		827	+	return _c
		828	+	}
		829	+
		830	+	// PublishCheckpoint provides a mock function for the type MockSessionLifecycleCoordinator
		831	+	func (_mock *MockSessionLifecycleCoordinator) PublishCheckpoint(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication) (domain.CatalogueRecord, error) {
		832	+	ret := _mock.Called(context1, s, snapshotPublication)
		833	+
		834	+	if len(ret) == 0 {
		835	+	panic("no return value specified for PublishCheckpoint")
		836	+	}
		837	+
		838	+	var r0 domain.CatalogueRecord
		839	+	var r1 error
		840	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, ports.SnapshotPublication) (domain.CatalogueRecord, error)); ok {
		841	+	return returnFunc(context1, s, snapshotPublication)
		842	+	}
		843	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, ports.SnapshotPublication) domain.CatalogueRecord); ok {
		844	+	r0 = returnFunc(context1, s, snapshotPublication)
		845	+	} else {
		846	+	r0 = ret.Get(0).(domain.CatalogueRecord)
		847	+	}
		848	+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, ports.SnapshotPublication) error); ok {
		849	+	r1 = returnFunc(context1, s, snapshotPublication)
		850	+	} else {
		851	+	r1 = ret.Error(1)
		852	+	}
		853	+	return r0, r1
		854	+	}
		855	+
		856	+	// MockSessionLifecycleCoordinator_PublishCheckpoint_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'PublishCheckpoint'
		857	+	type MockSessionLifecycleCoordinator_PublishCheckpoint_Call struct {
		858	+	*mock.Call
		859	+	}
		860	+
		861	+	// PublishCheckpoint is a helper method to define mock.On call
		862	+	// - context1 context.Context
		863	+	// - s string
		864	+	// - snapshotPublication ports.SnapshotPublication
		865	+	func (_e MockSessionLifecycleCoordinator_Expecter) PublishCheckpoint(context1 any, s any, snapshotPublication any) MockSessionLifecycleCoordinator_PublishCheckpoint_Call {
		866	+	return &MockSessionLifecycleCoordinator_PublishCheckpoint_Call{Call: _e.mock.On("PublishCheckpoint", context1, s, snapshotPublication)}
		867	+	}
		868	+
		869	+	func (_c MockSessionLifecycleCoordinator_PublishCheckpoint_Call) Run(run func(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication)) MockSessionLifecycleCoordinator_PublishCheckpoint_Call {
		870	+	_c.Call.Run(func(args mock.Arguments) {
		871	+	var arg0 context.Context
		872	+	if args[0] != nil {
		873	+	arg0 = args[0].(context.Context)
		874	+	}
		875	+	var arg1 string
		876	+	if args[1] != nil {
		877	+	arg1 = args[1].(string)
		878	+	}
		879	+	var arg2 ports.SnapshotPublication
		880	+	if args[2] != nil {
		881	+	arg2 = args[2].(ports.SnapshotPublication)
		882	+	}
		883	+	run(
		884	+	arg0,
		885	+	arg1,
		886	+	arg2,
		887	+	)
		888	+	})
		889	+	return _c
		890	+	}
		891	+
		892	+	func (_c MockSessionLifecycleCoordinator_PublishCheckpoint_Call) Return(catalogueRecord domain.CatalogueRecord, err error) MockSessionLifecycleCoordinator_PublishCheckpoint_Call {
		893	+	_c.Call.Return(catalogueRecord, err)
		894	+	return _c
		895	+	}
		896	+
		897	+	func (_c MockSessionLifecycleCoordinator_PublishCheckpoint_Call) RunAndReturn(run func(context1 context.Context, s string, snapshotPublication ports.SnapshotPublication) (domain.CatalogueRecord, error)) MockSessionLifecycleCoordinator_PublishCheckpoint_Call {
		898	+	_c.Call.Return(run)
		899	+	return _c
		900	+	}
		901	+
		902	+	// Rename provides a mock function for the type MockSessionLifecycleCoordinator
		903	+	func (_mock *MockSessionLifecycleCoordinator) Rename(context1 context.Context, s string, s1 string) (domain.CatalogueRecord, error) {
		904	+	ret := _mock.Called(context1, s, s1)
		905	+
		906	+	if len(ret) == 0 {
		907	+	panic("no return value specified for Rename")
		908	+	}
		909	+
		910	+	var r0 domain.CatalogueRecord
		911	+	var r1 error
		912	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) (domain.CatalogueRecord, error)); ok {
		913	+	return returnFunc(context1, s, s1)
		914	+	}
		915	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) domain.CatalogueRecord); ok {
		916	+	r0 = returnFunc(context1, s, s1)
		917	+	} else {
		918	+	r0 = ret.Get(0).(domain.CatalogueRecord)
		919	+	}
		920	+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, string) error); ok {
		921	+	r1 = returnFunc(context1, s, s1)
		922	+	} else {
		923	+	r1 = ret.Error(1)
		924	+	}
		925	+	return r0, r1
		926	+	}
		927	+
		928	+	// MockSessionLifecycleCoordinator_Rename_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Rename'
		929	+	type MockSessionLifecycleCoordinator_Rename_Call struct {
		930	+	*mock.Call
		931	+	}
		932	+
		933	+	// Rename is a helper method to define mock.On call
		934	+	// - context1 context.Context
		935	+	// - s string
		936	+	// - s1 string
		937	+	func (_e MockSessionLifecycleCoordinator_Expecter) Rename(context1 any, s any, s1 any) MockSessionLifecycleCoordinator_Rename_Call {
		938	+	return &MockSessionLifecycleCoordinator_Rename_Call{Call: _e.mock.On("Rename", context1, s, s1)}
		939	+	}
		940	+
		941	+	func (_c MockSessionLifecycleCoordinator_Rename_Call) Run(run func(context1 context.Context, s string, s1 string)) MockSessionLifecycleCoordinator_Rename_Call {
		942	+	_c.Call.Run(func(args mock.Arguments) {
		943	+	var arg0 context.Context
		944	+	if args[0] != nil {
		945	+	arg0 = args[0].(context.Context)
		946	+	}
		947	+	var arg1 string
		948	+	if args[1] != nil {
		949	+	arg1 = args[1].(string)
		950	+	}
		951	+	var arg2 string
		952	+	if args[2] != nil {
		953	+	arg2 = args[2].(string)
		954	+	}
		955	+	run(
		956	+	arg0,
		957	+	arg1,
		958	+	arg2,
		959	+	)
		960	+	})
		961	+	return _c
		962	+	}
		963	+
		964	+	func (_c MockSessionLifecycleCoordinator_Rename_Call) Return(catalogueRecord domain.CatalogueRecord, err error) MockSessionLifecycleCoordinator_Rename_Call {
		965	+	_c.Call.Return(catalogueRecord, err)
		966	+	return _c
		967	+	}
		968	+
		969	+	func (_c MockSessionLifecycleCoordinator_Rename_Call) RunAndReturn(run func(context1 context.Context, s string, s1 string) (domain.CatalogueRecord, error)) MockSessionLifecycleCoordinator_Rename_Call {
		970	+	_c.Call.Return(run)
		971	+	return _c
		972	+	}
		973	+
608	// NewMockCatalogue creates a new instance of MockCatalogue. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.	974		// NewMockCatalogue creates a new instance of MockCatalogue. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
609	// The first argument is typically a *testing.T value.	975		// The first argument is typically a *testing.T value.
610	func NewMockCatalogue(t interface {	976		func NewMockCatalogue(t interface {

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


@@ -134,6 +134,16 @@ type CheckpointCoordinator interface {
134	134		PromoteFallback(context.Context, string, domain.CheckpointRef) (FallbackPromotionOutcome, error)
135	135		}
136	136
	137	+	// SessionLifecycleCoordinator owns durable named-session create, rename, and
	138	+	// delete commit protocols. Runtime registries update only after these calls
	139	+	// complete successfully.
	140	+	type SessionLifecycleCoordinator interface {
	141	+	CheckpointCoordinator
	142	+	Create(context.Context, domain.CatalogueRecord) (domain.CatalogueRecord, error)
	143	+	Rename(context.Context, string, string) (domain.CatalogueRecord, error)
	144	+	Delete(context.Context, string) error
	145	+	}
	146	+
137	147		type Catalogue interface {
138	148		Records() []domain.CatalogueRecord
139	149		Record(string) (domain.CatalogueRecord, bool)


@@ -134,6 +134,16 @@ type CheckpointCoordinator interface {
134	PromoteFallback(context.Context, string, domain.CheckpointRef) (FallbackPromotionOutcome, error)	134		PromoteFallback(context.Context, string, domain.CheckpointRef) (FallbackPromotionOutcome, error)
135	}	135		}
136		136
		137	+	// SessionLifecycleCoordinator owns durable named-session create, rename, and
		138	+	// delete commit protocols. Runtime registries update only after these calls
		139	+	// complete successfully.
		140	+	type SessionLifecycleCoordinator interface {
		141	+	CheckpointCoordinator
		142	+	Create(context.Context, domain.CatalogueRecord) (domain.CatalogueRecord, error)
		143	+	Rename(context.Context, string, string) (domain.CatalogueRecord, error)
		144	+	Delete(context.Context, string) error
		145	+	}
		146	+
137	type Catalogue interface {	147		type Catalogue interface {
138	Records() []domain.CatalogueRecord	148		Records() []domain.CatalogueRecord
139	Record(string) (domain.CatalogueRecord, bool)	149		Record(string) (domain.CatalogueRecord, bool)

M internal/usecase/daemon/daemon.go +10 -0


@@ -126,6 +126,7 @@ type Daemon struct {
126	126		// read-only migration input and is never used for new writes.
127	127		snapshotRepository ports.SnapshotRepository
128	128		checkpointRecovery ports.CheckpointCoordinator
	129	+	lifecycleRecovery ports.SessionLifecycleCoordinator
129	130		legacySnapshots ports.LegacySnapshotSource
130	131		snapsEnabled bool
131	132		noticeStore ports.NoticeStore

@@ -279,6 +280,15 @@ func WithCheckpointCoordinator(coordinator ports.CheckpointCoordinator) Option {
279	280		}
280	281		}
281	282
	283	+	// WithRecoveryCoordinator installs all durable checkpoint and named-session
	284	+	// lifecycle transaction seams.
	285	+	func WithRecoveryCoordinator(coordinator ports.SessionLifecycleCoordinator) Option {
	286	+	return func(d *Daemon) {
	287	+	d.checkpointRecovery = coordinator
	288	+	d.lifecycleRecovery = coordinator
	289	+	}
	290	+	}
	291	+
282	292		// WithSnapshotRepository enables content-addressed incremental snapshots. The
283	293		// legacy source is read-only migration input and is never used for new writes.
284	294		func WithSnapshotRepository(repository any, legacy ports.LegacySnapshotSource) Option {


@@ -126,6 +126,7 @@ type Daemon struct {
126	// read-only migration input and is never used for new writes.	126		// read-only migration input and is never used for new writes.
127	snapshotRepository ports.SnapshotRepository	127		snapshotRepository ports.SnapshotRepository
128	checkpointRecovery ports.CheckpointCoordinator	128		checkpointRecovery ports.CheckpointCoordinator
		129	+	lifecycleRecovery ports.SessionLifecycleCoordinator
129	legacySnapshots ports.LegacySnapshotSource	130		legacySnapshots ports.LegacySnapshotSource
130	snapsEnabled bool	131		snapsEnabled bool
131	noticeStore ports.NoticeStore	132		noticeStore ports.NoticeStore

@@ -279,6 +280,15 @@ func WithCheckpointCoordinator(coordinator ports.CheckpointCoordinator) Option {
279	}	280		}
280	}	281		}
281		282
		283	+	// WithRecoveryCoordinator installs all durable checkpoint and named-session
		284	+	// lifecycle transaction seams.
		285	+	func WithRecoveryCoordinator(coordinator ports.SessionLifecycleCoordinator) Option {
		286	+	return func(d *Daemon) {
		287	+	d.checkpointRecovery = coordinator
		288	+	d.lifecycleRecovery = coordinator
		289	+	}
		290	+	}
		291	+
282	// WithSnapshotRepository enables content-addressed incremental snapshots. The	292		// WithSnapshotRepository enables content-addressed incremental snapshots. The
283	// legacy source is read-only migration input and is never used for new writes.	293		// legacy source is read-only migration input and is never used for new writes.
284	func WithSnapshotRepository(repository any, legacy ports.LegacySnapshotSource) Option {	294		func WithSnapshotRepository(repository any, legacy ports.LegacySnapshotSource) Option {

M internal/usecase/daemon/session.go +45 -33


@@ -292,9 +292,13 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz
292	292		if resuming {
293	293		if _, ok := d.catalogueRecord(name); ok {
294	294		err = d.updateCatalogueMetadata(record.MetadataUpdate())
	295	+	} else if d.lifecycleRecovery != nil {
	296	+	record, err = d.lifecycleRecovery.Create(context.Background(), record)
295	297		} else {
296	298		err = d.createCatalogueRecord(record)
297	299		}
	300	+	} else if d.lifecycleRecovery != nil {
	301	+	record, err = d.lifecycleRecovery.Create(context.Background(), record)
298	302		} else {
299	303		err = d.createCatalogueRecord(record)
300	304		}

@@ -303,6 +307,7 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz
303	307		cancel()
304	308		return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)
305	309		}
	310	+	sess.incarnation = record.IncarnationID
306	311		}
307	312		delete(d.stopped, name)
308	313		}

@@ -696,26 +701,37 @@ func (d Daemon) renameSession(sess session, name string) error {
696	701		sess.mu.Unlock()
697	702		return err
698	703		}
699		-	var quarantine snapshotCoordinatorQuarantine
700		-	if !wasEphemeral && oldName != name {
701		-	quarantine = quarantineSnapshotCoordinatorWithEpoch(sess)
702		-	<-quarantine.done
703		-	if err := d.beginSnapshotPurge(oldName, sess.incarnation); err != nil {
704		-	return rollback(err)
705		-	}
706		-	// finishSnapshotPurge preserves the tombstone through both source
707		-	// deletions and the old metadata delete, including a retained legacy
708		-	// import whose verified source delete is still pending.
709		-	if err := d.finishSnapshotPurge(oldName, sess.incarnation); err != nil {
710		-	return rollback(err)
711		-	}
712		-	}
713		-	// For a named rename this is the durable commit point, deliberately after
714		-	// the old tombstone transaction. Ephemeral promotion has no old durable
715		-	// identity and can create its first record directly.
	704	+	// Durable named renames are one atomic catalogue batch. The incarnation-keyed
	705	+	// snapshot namespace is intentionally untouched.
716	706		if (wasEphemeral \|\| oldName != name) && d.persistEnabled {
717		-	if err := d.createCatalogueRecord(record); err != nil {
718		-	return rollback(err)
	707	+	if d.lifecycleRecovery != nil {
	708	+	var committed domain.CatalogueRecord
	709	+	var err error
	710	+	if wasEphemeral {
	711	+	committed, err = d.lifecycleRecovery.Create(context.Background(), record)
	712	+	} else {
	713	+	committed, err = d.lifecycleRecovery.Rename(context.Background(), oldName, name)
	714	+	}
	715	+	if err != nil {
	716	+	return rollback(err)
	717	+	}
	718	+	record = committed
	719	+	} else if wasEphemeral {
	720	+	if err := d.createCatalogueRecord(record); err != nil {
	721	+	return rollback(err)
	722	+	}
	723	+	} else {
	724	+	if current, ok := d.catalogueRecord(oldName); ok {
	725	+	current.Name = name
	726	+	current.Cwd = record.Cwd
	727	+	current.UpdatedAt = record.UpdatedAt
	728	+	current.LastUsedSeq = record.LastUsedSeq
	729	+	current.TabNames = append([]string(nil), record.TabNames...)
	730	+	record = current
	731	+	}
	732	+	if err := d.catalogue.Rename(oldName, record); err != nil {
	733	+	return rollback(err)
	734	+	}
719	735		}
720	736		}
721	737

@@ -727,24 +743,12 @@ func (d Daemon) renameSession(sess session, name string) error {
727	743		delete(d.stopped, name)
728	744		sess.name = name
729	745		sess.createdAt = createdAt
	746	+	sess.incarnation = record.IncarnationID
730	747		sess.ephemeral = false
731	748		sess.renameInProgress = false
732		-	// Keep the coordinator in quarantine until the new in-memory identity is
733		-	// fully committed. The fresh coordinator below makes the new repository
734		-	// start at generation one.
735		-	if !wasEphemeral && oldName != name {
736		-	sess.snapEligible.Store(false)
737		-	} else {
738		-	sess.snapEligible.Store(name != "")
739		-	}
	749	+	sess.snapEligible.Store(name != "")
740	750		sess.mu.Unlock()
741	751		d.mu.Unlock()
742		-	if !wasEphemeral && oldName != name {
743		-	// A concurrent teardown may have quarantined after this rename did. In
744		-	// that case its newer epoch owns the stopped coordinator.
745		-	resumeSnapshotCoordinatorForNewIdentity(sess, quarantine)
746		-	return nil
747		-	}
748	752		if wasEphemeral {
749	753		sess.snapshotMu.Lock()
750	754		sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)

@@ -898,6 +902,11 @@ func (d Daemon) closeTab(sess session, tb *tab, repaint bool) error {
898	902		// identity or persisted metadata is removed. It never runs under daemon or
899	903		// session locks.
900	904		func (d *Daemon) beginSnapshotPurge(name string, incarnation domain.IncarnationID) error {
	905	+	if d.lifecycleRecovery != nil {
	906	+	// The lifecycle coordinator performs the complete ordered protocol after
	907	+	// snapshot writers have joined.
	908	+	return nil
	909	+	}
901	910		if d.snapshotRepository == nil \|\| name == "" {
902	911		return nil
903	912		}

@@ -909,6 +918,9 @@ func (d *Daemon) beginSnapshotPurge(name string, incarnation domain.IncarnationI
909	918		// leaves the marker in place so startup cannot restore or import the name and
910	919		// a later live/offline kill can retry the idempotent source deletes.
911	920		func (d *Daemon) finishSnapshotPurge(name string, incarnation domain.IncarnationID) error {
	921	+	if d.lifecycleRecovery != nil {
	922	+	return d.lifecycleRecovery.Delete(context.Background(), name)
	923	+	}
912	924		if d.snapshotRepository == nil {
913	925		return d.deleteCatalogueRecord(name)
914	926		}

 			if resuming {
 				if _, ok := d.catalogueRecord(name); ok {
 					err = d.updateCatalogueMetadata(record.MetadataUpdate())
+				} else if d.lifecycleRecovery != nil {
+					record, err = d.lifecycleRecovery.Create(context.Background(), record)
 				} else {
 					err = d.createCatalogueRecord(record)
 				}
+			} else if d.lifecycleRecovery != nil {
+				record, err = d.lifecycleRecovery.Create(context.Background(), record)
 			} else {
 				err = d.createCatalogueRecord(record)
 			}
 				cancel()
 				return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)
 			}
+			sess.incarnation = record.IncarnationID
 		}
 		delete(d.stopped, name)
 	}
 		sess.mu.Unlock()
 		return err
 	}
-	var quarantine snapshotCoordinatorQuarantine
+	// Durable named renames are one atomic catalogue batch. The incarnation-keyed
-	if !wasEphemeral && oldName != name {
+	// snapshot namespace is intentionally untouched.
-		quarantine = quarantineSnapshotCoordinatorWithEpoch(sess)
-		<-quarantine.done
-		if err := d.beginSnapshotPurge(oldName, sess.incarnation); err != nil {
-			return rollback(err)
-		}
-		// finishSnapshotPurge preserves the tombstone through both source
-		// deletions and the old metadata delete, including a retained legacy
-		// import whose verified source delete is still pending.
-		if err := d.finishSnapshotPurge(oldName, sess.incarnation); err != nil {
-			return rollback(err)
-		}
-	}
-	// For a named rename this is the durable commit point, deliberately after
-	// the old tombstone transaction. Ephemeral promotion has no old durable
-	// identity and can create its first record directly.
 	if (wasEphemeral || oldName != name) && d.persistEnabled {
-		if err := d.createCatalogueRecord(record); err != nil {
+		if d.lifecycleRecovery != nil {
-			return rollback(err)
+			var committed domain.CatalogueRecord
+			var err error
+			if wasEphemeral {
+				committed, err = d.lifecycleRecovery.Create(context.Background(), record)
+			} else {
+				committed, err = d.lifecycleRecovery.Rename(context.Background(), oldName, name)
+			}
+			if err != nil {
+				return rollback(err)
+			}
+			record = committed
+		} else if wasEphemeral {
+			if err := d.createCatalogueRecord(record); err != nil {
+				return rollback(err)
+			}
+		} else {
+			if current, ok := d.catalogueRecord(oldName); ok {
+				current.Name = name
+				current.Cwd = record.Cwd
+				current.UpdatedAt = record.UpdatedAt
+				current.LastUsedSeq = record.LastUsedSeq
+				current.TabNames = append([]string(nil), record.TabNames...)
+				record = current
+			}
+			if err := d.catalogue.Rename(oldName, record); err != nil {
+				return rollback(err)
+			}
 		}
 	}
 	delete(d.stopped, name)
 	sess.name = name
 	sess.createdAt = createdAt
+	sess.incarnation = record.IncarnationID
 	sess.ephemeral = false
 	sess.renameInProgress = false
-	// Keep the coordinator in quarantine until the new in-memory identity is
+	sess.snapEligible.Store(name != "")
-	// fully committed. The fresh coordinator below makes the new repository
-	// start at generation one.
-	if !wasEphemeral && oldName != name {
-		sess.snapEligible.Store(false)
-	} else {
-		sess.snapEligible.Store(name != "")
-	}
 	sess.mu.Unlock()
 	d.mu.Unlock()
-	if !wasEphemeral && oldName != name {
-		// A concurrent teardown may have quarantined after this rename did. In
-		// that case its newer epoch owns the stopped coordinator.
-		resumeSnapshotCoordinatorForNewIdentity(sess, quarantine)
-		return nil
-	}
 	if wasEphemeral {
 		sess.snapshotMu.Lock()
 		sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)
 // identity or persisted metadata is removed. It never runs under daemon or
 // session locks.
 func (d *Daemon) beginSnapshotPurge(name string, incarnation domain.IncarnationID) error {
+	if d.lifecycleRecovery != nil {
+		// The lifecycle coordinator performs the complete ordered protocol after
+		// snapshot writers have joined.
+		return nil
+	}
 	if d.snapshotRepository == nil || name == "" {
 		return nil
 	}
 // leaves the marker in place so startup cannot restore or import the name and
 // a later live/offline kill can retry the idempotent source deletes.
 func (d *Daemon) finishSnapshotPurge(name string, incarnation domain.IncarnationID) error {
+	if d.lifecycleRecovery != nil {
+		return d.lifecycleRecovery.Delete(context.Background(), name)
+	}
 	if d.snapshotRepository == nil {
 		return d.deleteCatalogueRecord(name)
 	}

M internal/usecase/daemon/session_rename_purge_test.go +9 -11

@@ -1,7 +1,6 @@
1	1		package daemon
2	2
3	3		import (
4		-	"errors"
5	4		"testing"
6	5
7	6		"github.com/stretchr/testify/require"

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

@@ -21,15 +20,15 @@ func TestRenamePurgesOldIncrementalAndLegacyBeforeCommittingNewIdentity(t *testi
21	20		require.NoError(t, d.persist.Save(sess.persistRecordLocked(1)))
22	21
23	22		require.NoError(t, d.renameSession(sess, "new"))
24		-	require.Equal(t, []string{"tombstone", "incremental", "legacy", "clear tombstone"}, repository.calls)
	23	+	require.Empty(t, repository.calls, "rename must not touch incarnation-keyed snapshot storage")
25	24		require.False(t, state.has("old"))
26	25		require.True(t, state.has("new"))
27		-	require.False(t, repository.tombstoned["old"])
	26	+	require.Equal(t, sess.incarnation, state.record(t, "new").IncarnationID)
28	27		}
29	28
30		-	func TestRenameLegacyFailureLeavesOldIdentityFenced(t *testing.T) {
	29	+	func TestRenameIgnoresUnrelatedLegacySourceFailure(t *testing.T) {
31	30		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
32		-	repository := &retryablePurgeRepository{legacyErr: errors.New("legacy delete failed")}
	31	+	repository := &retryablePurgeRepository{}
33	32		WithSnapshotRepository(repository, repository)(d)
34	33		store, state := newMockStore(t)
35	34		WithStore(store)(d)

@@ -37,9 +36,8 @@ func TestRenameLegacyFailureLeavesOldIdentityFenced(t *testing.T) {
37	36		d.sessions = map[domain.SessionID]*session{sess.id: sess}
38	37		require.NoError(t, d.persist.Save(sess.persistRecordLocked(1)))
39	38
40		-	require.Error(t, d.renameSession(sess, "new"))
41		-	require.Equal(t, []string{"tombstone", "incremental", "legacy"}, repository.calls)
42		-	require.True(t, repository.tombstoned["old"])
43		-	require.True(t, state.has("old"))
44		-	require.False(t, state.has("new"))
	39	+	require.NoError(t, d.renameSession(sess, "new"))
	40	+	require.Empty(t, repository.calls)
	41	+	require.False(t, state.has("old"))
	42	+	require.True(t, state.has("new"))
45	43		}

 package daemon
 import (
-	"errors"
 	"testing"
 	"github.com/stretchr/testify/require"
 	portsmocks "github.com/bnema/vev/internal/ports/mocks"
 )
-func TestRenamePurgesOldIncrementalAndLegacyBeforeCommittingNewIdentity(t *testing.T) {
+func TestRenamePreservesIncarnationSnapshotSources(t *testing.T) {
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
 	repository := &retryablePurgeRepository{}
 	WithSnapshotRepository(repository, repository)(d)
 	require.NoError(t, d.persist.Save(sess.persistRecordLocked(1)))
 	require.NoError(t, d.renameSession(sess, "new"))
-	require.Equal(t, []string{"tombstone", "incremental", "legacy", "clear tombstone"}, repository.calls)
+	require.Empty(t, repository.calls, "rename must not touch incarnation-keyed snapshot storage")
 	require.False(t, state.has("old"))
 	require.True(t, state.has("new"))
-	require.False(t, repository.tombstoned["old"])
+	require.Equal(t, sess.incarnation, state.record(t, "new").IncarnationID)
 }
-func TestRenameLegacyFailureLeavesOldIdentityFenced(t *testing.T) {
+func TestRenameIgnoresUnrelatedLegacySourceFailure(t *testing.T) {
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
-	repository := &retryablePurgeRepository{legacyErr: errors.New("legacy delete failed")}
+	repository := &retryablePurgeRepository{}
 	WithSnapshotRepository(repository, repository)(d)
 	store, state := newMockStore(t)
 	WithStore(store)(d)
 	d.sessions = map[domain.SessionID]*session{sess.id: sess}
 	require.NoError(t, d.persist.Save(sess.persistRecordLocked(1)))
-	require.Error(t, d.renameSession(sess, "new"))
+	require.NoError(t, d.renameSession(sess, "new"))
-	require.Equal(t, []string{"tombstone", "incremental", "legacy"}, repository.calls)
+	require.Empty(t, repository.calls)
-	require.True(t, repository.tombstoned["old"])
+	require.False(t, state.has("old"))
-	require.True(t, state.has("old"))
+	require.True(t, state.has("new"))
-	require.False(t, state.has("new"))
 }

M internal/usecase/daemon/snapshot_scheduler_test.go +0 -157


@@ -2,7 +2,6 @@ package daemon
2	2
3	3		import (
4	4		"context"
5		-	"errors"
6	5		"io"
7	6		"sync"
8	7		"sync/atomic"

@@ -82,162 +81,6 @@ func TestSnapshotCoordinatorQuarantineJoinsInFlightPublication(t *testing.T) {
82	81		}
83	82		}
84	83
85		-	func TestRenameKeepsSnapshotCoordinatorQuarantinedUntilNewIdentityCommits(t *testing.T) {
86		-	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
87		-	store, state := newMockStore(t)
88		-	WithStore(store)(d)
89		-	repository := portsmocks.NewMockSnapshotRepository(t)
90		-	WithSnapshotRepository(repository, nil)(d)
91		-	startSnapshotEncodeWorker(t, d)
92		-
93		-	deleteEntered := make(chan struct{})
94		-	allowDelete := make(chan struct{})
95		-	deleteCompleted := make(chan struct{})
96		-	state.deleteDone = deleteCompleted
97		-	published := make(chan ports.SnapshotPublication, 2)
98		-	repository.EXPECT().WriteDeletionTombstone(mock.Anything, mock.Anything).Return(nil).Once()
99		-	repository.EXPECT().QuarantineDeletionSources(mock.Anything, mock.Anything, false).RunAndReturn(func(context.Context, domain.DeletionTombstone, bool) error {
100		-	close(deleteEntered)
101		-	<-allowDelete
102		-	return nil
103		-	}).Once()
104		-	repository.EXPECT().DeleteDeletionTombstone(mock.Anything, mock.Anything).Return(nil).Once()
105		-	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
106		-	published <- publication
107		-	return nil
108		-	}).Once()
109		-
110		-	pty, release := newBlockingPTY(t)
111		-	defer release()
112		-	d.ptys = newFactory(t, pty)
113		-	sess, err := d.createSessionLocked("work", false, "/work", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
114		-	require.NoError(t, err)
115		-	markSnapshotDirty(sess)
116		-
117		-	renameDone := make(chan error, 1)
118		-	go func() { renameDone <- d.renameSession(sess, "renamed") }()
119		-	select {
120		-	case <-deleteEntered:
121		-	case <-time.After(testWaitTimeout):
122		-	t.Fatal("rename did not begin deleting the old snapshot identity")
123		-	}
124		-
125		-	// Hold the daemon identity commit after Delete has returned. Scheduler work
126		-	// during this window must not capture or publish the old name.
127		-	d.mu.Lock()
128		-	close(allowDelete)
129		-	select {
130		-	case <-deleteCompleted:
131		-	case <-time.After(testWaitTimeout):
132		-	t.Fatal("old snapshot identity was not deleted")
133		-	}
134		-	require.False(t, state.has("work"))
135		-	sess.snapshotMu.Lock()
136		-	quarantined := sess.snapshotQuarantined
137		-	sess.snapshotMu.Unlock()
138		-	require.True(t, quarantined, "old Delete returned before the coordinator resumed")
139		-
140		-	schedulerCtx, cancelScheduler := context.WithCancel(context.Background())
141		-	defer cancelScheduler()
142		-	go d.snapshotRepositorySaver(schedulerCtx)
143		-	select {
144		-	case d.snapshotWake <- struct{}{}:
145		-	default:
146		-	}
147		-	require.True(t, d.scheduleSnapshot(sess), "old work may be coalesced while the coordinator is quarantined")
148		-	d.mu.Unlock()
149		-
150		-	require.NoError(t, <-renameDone)
151		-	select {
152		-	case publication := <-published:
153		-	require.Equal(t, "renamed", publication.Name)
154		-	require.Equal(t, uint64(1), publication.Generation)
155		-	case <-time.After(testWaitTimeout):
156		-	t.Fatal("new identity was not published")
157		-	}
158		-	}
159		-
160		-	func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) {
161		-	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
162		-	store, state := newMockStore(t)
163		-	WithStore(store)(d)
164		-	repository := portsmocks.NewMockSnapshotRepository(t)
165		-	WithSnapshotRepository(repository, nil)(d)
166		-	startSnapshotEncodeWorker(t, d)
167		-
168		-	deleteOldRecordErr := errors.New("old record delete failed")
169		-	state.mu.Lock()
170		-	state.deleteErr = func(key string) error {
171		-	if key == "work" {
172		-	return deleteOldRecordErr
173		-	}
174		-	return nil
175		-	}
176		-	state.mu.Unlock()
177		-	repository.EXPECT().WriteDeletionTombstone(mock.Anything, mock.Anything).Return(nil).Once()
178		-	repository.EXPECT().QuarantineDeletionSources(mock.Anything, mock.Anything, false).Return(nil).Once()
179		-	// No Publish expectation is installed: any publication is an unexpected mock
180		-	// call. The saver handshake below joins the scheduler before this test ends.
181		-
182		-	pty, release := newBlockingPTY(t)
183		-	defer release()
184		-	d.ptys = newFactory(t, pty)
185		-	sess, err := d.createSessionLocked("work", false, "/work", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
186		-	require.NoError(t, err)
187		-	markSnapshotDirty(sess)
188		-
189		-	require.ErrorIs(t, d.renameSession(sess, "renamed"), deleteOldRecordErr)
190		-	sess.mu.Lock()
191		-	require.Equal(t, "work", sess.name)
192		-	require.False(t, sess.ephemeral)
193		-	require.False(t, sess.renameInProgress)
194		-	sess.mu.Unlock()
195		-	require.True(t, state.has("work"))
196		-	require.False(t, state.has("renamed"))
197		-	sess.snapshotMu.Lock()
198		-	quarantined := sess.snapshotQuarantined
199		-	sess.snapshotMu.Unlock()
200		-	require.True(t, quarantined)
201		-
202		-	// Drain stale producer notifications, then use the timer-reset handshake to
203		-	// prove the saver consumed this wake and completed its scheduling pass. Stop
204		-	// and join it before asserting: no publisher can start after this point.
205		-	for {
206		-	select {
207		-	case <-d.snapshotWake:
208		-	default:
209		-	goto wakeDrained
210		-	}
211		-	}
212		-
213		-	wakeDrained:
214		-	clock := newDeterministicSnapshotClock(time.Unix(100, 0))
215		-	d.clock = clock
216		-	schedulerCtx, cancelScheduler := context.WithCancel(context.Background())
217		-	schedulerDone := make(chan struct{})
218		-	go func() {
219		-	d.snapshotRepositorySaver(schedulerCtx)
220		-	close(schedulerDone)
221		-	}()
222		-	timer := clock.timer()
223		-	require.Equal(t, 24*time.Hour, <-timer.resets)
224		-	select {
225		-	case d.snapshotWake <- struct{}{}:
226		-	default:
227		-	t.Fatal("snapshot wake unexpectedly remained full")
228		-	}
229		-	require.Equal(t, 24*time.Hour, <-timer.resets, "scheduler did not process the rollback wake")
230		-	cancelScheduler()
231		-	select {
232		-	case <-schedulerDone:
233		-	case <-time.After(testWaitTimeout):
234		-	t.Fatal("snapshot scheduler did not stop")
235		-	}
236		-	// The deterministic scheduler clock has no independently firing deadline;
237		-	// restore the daemon's normal test clock before its session cleanup runs.
238		-	d.clock = stubClock{}
239		-	}
240		-
241	84		func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) {
242	85		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
243	86		repository := portsmocks.NewMockSnapshotRepository(t)


@@ -2,7 +2,6 @@ package daemon
2			2
3		import (	3	import (
4		"context"	4	"context"
5	-	"errors"
6		"io"	5	"io"
7		"sync"	6	"sync"
8		"sync/atomic"	7	"sync/atomic"

@@ -82,162 +81,6 @@ func TestSnapshotCoordinatorQuarantineJoinsInFlightPublication(t *testing.T) {
82		}	81	}
83		}	82	}
84			83
85	-	func TestRenameKeepsSnapshotCoordinatorQuarantinedUntilNewIdentityCommits(t *testing.T) {
86	-	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
87	-	store, state := newMockStore(t)
88	-	WithStore(store)(d)
89	-	repository := portsmocks.NewMockSnapshotRepository(t)
90	-	WithSnapshotRepository(repository, nil)(d)
91	-	startSnapshotEncodeWorker(t, d)
92	-
93	-	deleteEntered := make(chan struct{})
94	-	allowDelete := make(chan struct{})
95	-	deleteCompleted := make(chan struct{})
96	-	state.deleteDone = deleteCompleted
97	-	published := make(chan ports.SnapshotPublication, 2)
98	-	repository.EXPECT().WriteDeletionTombstone(mock.Anything, mock.Anything).Return(nil).Once()
99	-	repository.EXPECT().QuarantineDeletionSources(mock.Anything, mock.Anything, false).RunAndReturn(func(context.Context, domain.DeletionTombstone, bool) error {
100	-	close(deleteEntered)
101	-	<-allowDelete
102	-	return nil
103	-	}).Once()
104	-	repository.EXPECT().DeleteDeletionTombstone(mock.Anything, mock.Anything).Return(nil).Once()
105	-	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
106	-	published <- publication
107	-	return nil
108	-	}).Once()
109	-
110	-	pty, release := newBlockingPTY(t)
111	-	defer release()
112	-	d.ptys = newFactory(t, pty)
113	-	sess, err := d.createSessionLocked("work", false, "/work", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
114	-	require.NoError(t, err)
115	-	markSnapshotDirty(sess)
116	-
117	-	renameDone := make(chan error, 1)
118	-	go func() { renameDone <- d.renameSession(sess, "renamed") }()
119	-	select {
120	-	case <-deleteEntered:
121	-	case <-time.After(testWaitTimeout):
122	-	t.Fatal("rename did not begin deleting the old snapshot identity")
123	-	}
124	-
125	-	// Hold the daemon identity commit after Delete has returned. Scheduler work
126	-	// during this window must not capture or publish the old name.
127	-	d.mu.Lock()
128	-	close(allowDelete)
129	-	select {
130	-	case <-deleteCompleted:
131	-	case <-time.After(testWaitTimeout):
132	-	t.Fatal("old snapshot identity was not deleted")
133	-	}
134	-	require.False(t, state.has("work"))
135	-	sess.snapshotMu.Lock()
136	-	quarantined := sess.snapshotQuarantined
137	-	sess.snapshotMu.Unlock()
138	-	require.True(t, quarantined, "old Delete returned before the coordinator resumed")
139	-
140	-	schedulerCtx, cancelScheduler := context.WithCancel(context.Background())
141	-	defer cancelScheduler()
142	-	go d.snapshotRepositorySaver(schedulerCtx)
143	-	select {
144	-	case d.snapshotWake <- struct{}{}:
145	-	default:
146	-	}
147	-	require.True(t, d.scheduleSnapshot(sess), "old work may be coalesced while the coordinator is quarantined")
148	-	d.mu.Unlock()
149	-
150	-	require.NoError(t, <-renameDone)
151	-	select {
152	-	case publication := <-published:
153	-	require.Equal(t, "renamed", publication.Name)
154	-	require.Equal(t, uint64(1), publication.Generation)
155	-	case <-time.After(testWaitTimeout):
156	-	t.Fatal("new identity was not published")
157	-	}
158	-	}
159	-
160	-	func TestRenameRollbackAfterOldDeleteLeavesOldCoordinatorStopped(t *testing.T) {
161	-	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
162	-	store, state := newMockStore(t)
163	-	WithStore(store)(d)
164	-	repository := portsmocks.NewMockSnapshotRepository(t)
165	-	WithSnapshotRepository(repository, nil)(d)
166	-	startSnapshotEncodeWorker(t, d)
167	-
168	-	deleteOldRecordErr := errors.New("old record delete failed")
169	-	state.mu.Lock()
170	-	state.deleteErr = func(key string) error {
171	-	if key == "work" {
172	-	return deleteOldRecordErr
173	-	}
174	-	return nil
175	-	}
176	-	state.mu.Unlock()
177	-	repository.EXPECT().WriteDeletionTombstone(mock.Anything, mock.Anything).Return(nil).Once()
178	-	repository.EXPECT().QuarantineDeletionSources(mock.Anything, mock.Anything, false).Return(nil).Once()
179	-	// No Publish expectation is installed: any publication is an unexpected mock
180	-	// call. The saver handshake below joins the scheduler before this test ends.
181	-
182	-	pty, release := newBlockingPTY(t)
183	-	defer release()
184	-	d.ptys = newFactory(t, pty)
185	-	sess, err := d.createSessionLocked("work", false, "/work", domain.Size{Cols: 80, Rows: 24}, terminalEnv{}, d.baseEnv)
186	-	require.NoError(t, err)
187	-	markSnapshotDirty(sess)
188	-
189	-	require.ErrorIs(t, d.renameSession(sess, "renamed"), deleteOldRecordErr)
190	-	sess.mu.Lock()
191	-	require.Equal(t, "work", sess.name)
192	-	require.False(t, sess.ephemeral)
193	-	require.False(t, sess.renameInProgress)
194	-	sess.mu.Unlock()
195	-	require.True(t, state.has("work"))
196	-	require.False(t, state.has("renamed"))
197	-	sess.snapshotMu.Lock()
198	-	quarantined := sess.snapshotQuarantined
199	-	sess.snapshotMu.Unlock()
200	-	require.True(t, quarantined)
201	-
202	-	// Drain stale producer notifications, then use the timer-reset handshake to
203	-	// prove the saver consumed this wake and completed its scheduling pass. Stop
204	-	// and join it before asserting: no publisher can start after this point.
205	-	for {
206	-	select {
207	-	case <-d.snapshotWake:
208	-	default:
209	-	goto wakeDrained
210	-	}
211	-	}
212	-
213	-	wakeDrained:
214	-	clock := newDeterministicSnapshotClock(time.Unix(100, 0))
215	-	d.clock = clock
216	-	schedulerCtx, cancelScheduler := context.WithCancel(context.Background())
217	-	schedulerDone := make(chan struct{})
218	-	go func() {
219	-	d.snapshotRepositorySaver(schedulerCtx)
220	-	close(schedulerDone)
221	-	}()
222	-	timer := clock.timer()
223	-	require.Equal(t, 24*time.Hour, <-timer.resets)
224	-	select {
225	-	case d.snapshotWake <- struct{}{}:
226	-	default:
227	-	t.Fatal("snapshot wake unexpectedly remained full")
228	-	}
229	-	require.Equal(t, 24*time.Hour, <-timer.resets, "scheduler did not process the rollback wake")
230	-	cancelScheduler()
231	-	select {
232	-	case <-schedulerDone:
233	-	case <-time.After(testWaitTimeout):
234	-	t.Fatal("snapshot scheduler did not stop")
235	-	}
236	-	// The deterministic scheduler clock has no independently firing deadline;
237	-	// restore the daemon's normal test clock before its session cleanup runs.
238	-	d.clock = stubClock{}
239	-	}
240	-
241		func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) {	84	func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) {
242		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})	85	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
243		repository := portsmocks.NewMockSnapshotRepository(t)	86	repository := portsmocks.NewMockSnapshotRepository(t)

M internal/usecase/recovery/coordinator.go +159 -11


@@ -5,10 +5,12 @@ package recovery
5	5		import (
6	6		"context"
7	7		"errors"
	8	+	"fmt"
8	9		"io"
9	10		"sort"
10	11		"sync"
11	12
	13	+	"github.com/bnema/vev/internal/domain"
12	14		"github.com/bnema/vev/internal/ports"
13	15		)
14	16

@@ -95,37 +97,183 @@ func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotReposito
95	97		return &Coordinator{catalogue: catalogue, repository: repository, journal: journal, locks: NewKeyLocks(), random: random}
96	98		}
97	99
98		-	// Recover is deliberately fail-closed until the transaction roll-forward
99		-	// handlers are introduced. Startup may proceed only when both intent sources
100		-	// have been completely and strictly enumerated.
	100	+	// Create commits fresh catalogue metadata before the caller creates or exposes
	101	+	// any runtime session.
	102	+	func (c *Coordinator) Create(ctx context.Context, record domain.CatalogueRecord) (domain.CatalogueRecord, error) {
	103	+	if c == nil \|\| c.catalogue == nil \|\| c.locks == nil \|\| c.random == nil {
	104	+	return domain.CatalogueRecord{}, errors.New("recovery: incomplete create dependencies")
	105	+	}
	106	+	if err := ctx.Err(); err != nil {
	107	+	return domain.CatalogueRecord{}, err
	108	+	}
	109	+	unlock := c.locks.Lock([]string{record.Name})
	110	+	defer unlock()
	111	+	if _, exists := c.catalogue.Record(record.Name); exists {
	112	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q already exists", record.Name)
	113	+	}
	114	+	id, err := domain.NewIncarnationID(c.random)
	115	+	if err != nil {
	116	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: generate incarnation: %w", err)
	117	+	}
	118	+	record.IncarnationID = id
	119	+	record.RecoveryState = domain.RecoveryFresh
	120	+	record.Committed = nil
	121	+	record.Fallbacks = [2]*domain.CheckpointRef{}
	122	+	record.DegradedReason = ""
	123	+	if err := record.Validate(); err != nil {
	124	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid fresh record: %w", err)
	125	+	}
	126	+	if err := c.catalogue.Create(record); err != nil {
	127	+	return domain.CatalogueRecord{}, err
	128	+	}
	129	+	return record, nil
	130	+	}
	131	+
	132	+	// Rename atomically moves one catalogue key while retaining the immutable
	133	+	// incarnation and all checkpoint references.
	134	+	func (c *Coordinator) Rename(ctx context.Context, oldName, newName string) (domain.CatalogueRecord, error) {
	135	+	if c == nil \|\| c.catalogue == nil \|\| c.locks == nil {
	136	+	return domain.CatalogueRecord{}, errors.New("recovery: incomplete rename dependencies")
	137	+	}
	138	+	if err := ctx.Err(); err != nil {
	139	+	return domain.CatalogueRecord{}, err
	140	+	}
	141	+	unlock := c.locks.Lock([]string{oldName, newName})
	142	+	defer unlock()
	143	+	record, ok := c.catalogue.Record(oldName)
	144	+	if !ok {
	145	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q not found", oldName)
	146	+	}
	147	+	if existing, exists := c.catalogue.Record(newName); exists && (newName != oldName \|\| existing.IncarnationID != record.IncarnationID) {
	148	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q already exists", newName)
	149	+	}
	150	+	record.Name = newName
	151	+	if err := record.Validate(); err != nil {
	152	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid rename: %w", err)
	153	+	}
	154	+	if oldName != newName {
	155	+	if err := c.catalogue.Rename(oldName, record); err != nil {
	156	+	return domain.CatalogueRecord{}, err
	157	+	}
	158	+	}
	159	+	return record, nil
	160	+	}
	161	+
	162	+	// Delete executes the durable deletion protocol. Every completed boundary is
	163	+	// independently sufficient for Recover to roll the operation forward.
	164	+	func (c *Coordinator) Delete(ctx context.Context, name string) error {
	165	+	if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil {
	166	+	return errors.New("recovery: incomplete delete dependencies")
	167	+	}
	168	+	unlock := c.locks.Lock([]string{name})
	169	+	defer unlock()
	170	+	record, ok := c.catalogue.Record(name)
	171	+	if !ok {
	172	+	return nil
	173	+	}
	174	+	return c.deleteLocked(ctx, record)
	175	+	}
	176	+
	177	+	func (c *Coordinator) deleteLocked(ctx context.Context, record domain.CatalogueRecord) error {
	178	+	if err := ctx.Err(); err != nil {
	179	+	return err
	180	+	}
	181	+	if record.RecoveryState != domain.RecoveryDeleting {
	182	+	record.RecoveryState = domain.RecoveryDeleting
	183	+	record.DegradedReason = ""
	184	+	if err := record.Validate(); err != nil {
	185	+	return fmt.Errorf("recovery: invalid deleting record: %w", err)
	186	+	}
	187	+	if err := c.catalogue.Replace(record.Name, record); err != nil {
	188	+	return err
	189	+	}
	190	+	}
	191	+	tombstone := domain.DeletionTombstone{Name: record.Name, IncarnationID: record.IncarnationID}
	192	+	if err := c.repository.WriteDeletionTombstone(ctx, tombstone); err != nil {
	193	+	return err
	194	+	}
	195	+	if err := c.repository.QuarantineDeletionSources(ctx, tombstone, true); err != nil {
	196	+	return err
	197	+	}
	198	+	if err := c.catalogue.Delete(record.Name); err != nil {
	199	+	return err
	200	+	}
	201	+	return c.repository.DeleteDeletionTombstone(ctx, record.IncarnationID)
	202	+	}
	203	+
	204	+	// Recover rolls deleting records and strict deletion tombstones forward. The
	205	+	// two sources are enumerated independently because catalogue removal precedes
	206	+	// tombstone removal in the commit protocol.
101	207		func (c *Coordinator) Recover(ctx context.Context) error {
102		-	if c == nil \|\| c.repository == nil \|\| c.journal == nil {
	208	+	if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.journal == nil \|\| c.locks == nil {
103	209		return errors.New("recovery: incomplete coordinator dependencies")
104	210		}
105	211		intents, err := c.journal.ListDiscards(ctx)
106	212		if err != nil {
107	213		return err
108	214		}
109		-	pending := len(intents) != 0
	215	+	for _, candidate := range c.catalogue.Records() {
	216	+	if candidate.RecoveryState != domain.RecoveryDeleting {
	217	+	continue
	218	+	}
	219	+	unlock := c.locks.Lock([]string{candidate.Name})
	220	+	current, ok := c.catalogue.Record(candidate.Name)
	221	+	if ok && current.IncarnationID == candidate.IncarnationID && current.RecoveryState == domain.RecoveryDeleting {
	222	+	err = c.deleteLocked(ctx, current)
	223	+	}
	224	+	unlock()
	225	+	if err != nil {
	226	+	return err
	227	+	}
	228	+	}
	229	+
110	230		cursor := ports.DeletionTombstoneCursor{}
111	231		for {
112	232		page, err := c.repository.ListDeletionTombstones(ctx, cursor, recoveryListingBudget)
113	233		if err != nil {
114	234		return err
115	235		}
116		-	if len(page.Tombstones) != 0 {
117		-	pending = true
	236	+	for _, tombstone := range page.Tombstones {
	237	+	if err := c.recoverDeletionTombstone(ctx, tombstone); err != nil {
	238	+	return err
	239	+	}
118	240		}
119	241		if page.Done {
120		-	break
	242	+	if len(intents) != 0 {
	243	+	return ErrPendingRecoveryUnsupported
	244	+	}
	245	+	return nil
121	246		}
122	247		if page.Next.After == cursor.After {
123	248		return errors.New("recovery: tombstone listing did not advance")
124	249		}
125	250		cursor = page.Next
126	251		}
127		-	if pending {
128		-	return ErrPendingRecoveryUnsupported
	252	+	}
	253	+
	254	+	func (c *Coordinator) recoverDeletionTombstone(ctx context.Context, tombstone domain.DeletionTombstone) error {
	255	+	if err := tombstone.Validate(); err != nil {
	256	+	return err
	257	+	}
	258	+	unlock := c.locks.Lock([]string{tombstone.Name})
	259	+	defer unlock()
	260	+	record, exists := c.catalogue.Record(tombstone.Name)
	261	+	includeLegacyName := true
	262	+	if exists {
	263	+	switch {
	264	+	case record.IncarnationID != tombstone.IncarnationID:
	265	+	includeLegacyName = false
	266	+	case record.RecoveryState != domain.RecoveryDeleting:
	267	+	return fmt.Errorf("recovery: deletion tombstone conflicts with non-deleting session %q", tombstone.Name)
	268	+	}
	269	+	}
	270	+	if err := c.repository.QuarantineDeletionSources(ctx, tombstone, includeLegacyName); err != nil {
	271	+	return err
	272	+	}
	273	+	if exists && record.IncarnationID == tombstone.IncarnationID {
	274	+	if err := c.catalogue.Delete(tombstone.Name); err != nil {
	275	+	return err
	276	+	}
129	277		}
130		-	return nil
	278	+	return c.repository.DeleteDeletionTombstone(ctx, tombstone.IncarnationID)
131	279		}


@@ -5,10 +5,12 @@ package recovery
5		import (	5		import (
6		"context"	6		"context"
7		"errors"	7		"errors"
			8	+	"fmt"
8		"io"	9		"io"
9		"sort"	10		"sort"
10		"sync"	11		"sync"
11			12
			13	+	"github.com/bnema/vev/internal/domain"
12		"github.com/bnema/vev/internal/ports"	14		"github.com/bnema/vev/internal/ports"
13		)	15		)
14			16

@@ -95,37 +97,183 @@ func NewCoordinator(catalogue ports.Catalogue, repository ports.SnapshotReposito
95		return &Coordinator{catalogue: catalogue, repository: repository, journal: journal, locks: NewKeyLocks(), random: random}	97		return &Coordinator{catalogue: catalogue, repository: repository, journal: journal, locks: NewKeyLocks(), random: random}
96		}	98		}
97			99
98	-	// Recover is deliberately fail-closed until the transaction roll-forward	100	+	// Create commits fresh catalogue metadata before the caller creates or exposes
99	-	// handlers are introduced. Startup may proceed only when both intent sources	101	+	// any runtime session.
100	-	// have been completely and strictly enumerated.	102	+	func (c *Coordinator) Create(ctx context.Context, record domain.CatalogueRecord) (domain.CatalogueRecord, error) {
			103	+	if c == nil \|\| c.catalogue == nil \|\| c.locks == nil \|\| c.random == nil {
			104	+	return domain.CatalogueRecord{}, errors.New("recovery: incomplete create dependencies")
			105	+	}
			106	+	if err := ctx.Err(); err != nil {
			107	+	return domain.CatalogueRecord{}, err
			108	+	}
			109	+	unlock := c.locks.Lock([]string{record.Name})
			110	+	defer unlock()
			111	+	if _, exists := c.catalogue.Record(record.Name); exists {
			112	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q already exists", record.Name)
			113	+	}
			114	+	id, err := domain.NewIncarnationID(c.random)
			115	+	if err != nil {
			116	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: generate incarnation: %w", err)
			117	+	}
			118	+	record.IncarnationID = id
			119	+	record.RecoveryState = domain.RecoveryFresh
			120	+	record.Committed = nil
			121	+	record.Fallbacks = [2]*domain.CheckpointRef{}
			122	+	record.DegradedReason = ""
			123	+	if err := record.Validate(); err != nil {
			124	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid fresh record: %w", err)
			125	+	}
			126	+	if err := c.catalogue.Create(record); err != nil {
			127	+	return domain.CatalogueRecord{}, err
			128	+	}
			129	+	return record, nil
			130	+	}
			131	+
			132	+	// Rename atomically moves one catalogue key while retaining the immutable
			133	+	// incarnation and all checkpoint references.
			134	+	func (c *Coordinator) Rename(ctx context.Context, oldName, newName string) (domain.CatalogueRecord, error) {
			135	+	if c == nil \|\| c.catalogue == nil \|\| c.locks == nil {
			136	+	return domain.CatalogueRecord{}, errors.New("recovery: incomplete rename dependencies")
			137	+	}
			138	+	if err := ctx.Err(); err != nil {
			139	+	return domain.CatalogueRecord{}, err
			140	+	}
			141	+	unlock := c.locks.Lock([]string{oldName, newName})
			142	+	defer unlock()
			143	+	record, ok := c.catalogue.Record(oldName)
			144	+	if !ok {
			145	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q not found", oldName)
			146	+	}
			147	+	if existing, exists := c.catalogue.Record(newName); exists && (newName != oldName \|\| existing.IncarnationID != record.IncarnationID) {
			148	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: session %q already exists", newName)
			149	+	}
			150	+	record.Name = newName
			151	+	if err := record.Validate(); err != nil {
			152	+	return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid rename: %w", err)
			153	+	}
			154	+	if oldName != newName {
			155	+	if err := c.catalogue.Rename(oldName, record); err != nil {
			156	+	return domain.CatalogueRecord{}, err
			157	+	}
			158	+	}
			159	+	return record, nil
			160	+	}
			161	+
			162	+	// Delete executes the durable deletion protocol. Every completed boundary is
			163	+	// independently sufficient for Recover to roll the operation forward.
			164	+	func (c *Coordinator) Delete(ctx context.Context, name string) error {
			165	+	if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil {
			166	+	return errors.New("recovery: incomplete delete dependencies")
			167	+	}
			168	+	unlock := c.locks.Lock([]string{name})
			169	+	defer unlock()
			170	+	record, ok := c.catalogue.Record(name)
			171	+	if !ok {
			172	+	return nil
			173	+	}
			174	+	return c.deleteLocked(ctx, record)
			175	+	}
			176	+
			177	+	func (c *Coordinator) deleteLocked(ctx context.Context, record domain.CatalogueRecord) error {
			178	+	if err := ctx.Err(); err != nil {
			179	+	return err
			180	+	}
			181	+	if record.RecoveryState != domain.RecoveryDeleting {
			182	+	record.RecoveryState = domain.RecoveryDeleting
			183	+	record.DegradedReason = ""
			184	+	if err := record.Validate(); err != nil {
			185	+	return fmt.Errorf("recovery: invalid deleting record: %w", err)
			186	+	}
			187	+	if err := c.catalogue.Replace(record.Name, record); err != nil {
			188	+	return err
			189	+	}
			190	+	}
			191	+	tombstone := domain.DeletionTombstone{Name: record.Name, IncarnationID: record.IncarnationID}
			192	+	if err := c.repository.WriteDeletionTombstone(ctx, tombstone); err != nil {
			193	+	return err
			194	+	}
			195	+	if err := c.repository.QuarantineDeletionSources(ctx, tombstone, true); err != nil {
			196	+	return err
			197	+	}
			198	+	if err := c.catalogue.Delete(record.Name); err != nil {
			199	+	return err
			200	+	}
			201	+	return c.repository.DeleteDeletionTombstone(ctx, record.IncarnationID)
			202	+	}
			203	+
			204	+	// Recover rolls deleting records and strict deletion tombstones forward. The
			205	+	// two sources are enumerated independently because catalogue removal precedes
			206	+	// tombstone removal in the commit protocol.
101		func (c *Coordinator) Recover(ctx context.Context) error {	207		func (c *Coordinator) Recover(ctx context.Context) error {
102	-	if c == nil \|\| c.repository == nil \|\| c.journal == nil {	208	+	if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.journal == nil \|\| c.locks == nil {
103		return errors.New("recovery: incomplete coordinator dependencies")	209		return errors.New("recovery: incomplete coordinator dependencies")
104		}	210		}
105		intents, err := c.journal.ListDiscards(ctx)	211		intents, err := c.journal.ListDiscards(ctx)
106		if err != nil {	212		if err != nil {
107		return err	213		return err
108		}	214		}
109	-	pending := len(intents) != 0	215	+	for _, candidate := range c.catalogue.Records() {
			216	+	if candidate.RecoveryState != domain.RecoveryDeleting {
			217	+	continue
			218	+	}
			219	+	unlock := c.locks.Lock([]string{candidate.Name})
			220	+	current, ok := c.catalogue.Record(candidate.Name)
			221	+	if ok && current.IncarnationID == candidate.IncarnationID && current.RecoveryState == domain.RecoveryDeleting {
			222	+	err = c.deleteLocked(ctx, current)
			223	+	}
			224	+	unlock()
			225	+	if err != nil {
			226	+	return err
			227	+	}
			228	+	}
			229	+
110		cursor := ports.DeletionTombstoneCursor{}	230		cursor := ports.DeletionTombstoneCursor{}
111		for {	231		for {
112		page, err := c.repository.ListDeletionTombstones(ctx, cursor, recoveryListingBudget)	232		page, err := c.repository.ListDeletionTombstones(ctx, cursor, recoveryListingBudget)
113		if err != nil {	233		if err != nil {
114		return err	234		return err
115		}	235		}
116	-	if len(page.Tombstones) != 0 {	236	+	for _, tombstone := range page.Tombstones {
117	-	pending = true	237	+	if err := c.recoverDeletionTombstone(ctx, tombstone); err != nil {
			238	+	return err
			239	+	}
118		}	240		}
119		if page.Done {	241		if page.Done {
120	-	break	242	+	if len(intents) != 0 {
			243	+	return ErrPendingRecoveryUnsupported
			244	+	}
			245	+	return nil
121		}	246		}
122		if page.Next.After == cursor.After {	247		if page.Next.After == cursor.After {
123		return errors.New("recovery: tombstone listing did not advance")	248		return errors.New("recovery: tombstone listing did not advance")
124		}	249		}
125		cursor = page.Next	250		cursor = page.Next
126		}	251		}
127	-	if pending {	252	+	}
128	-	return ErrPendingRecoveryUnsupported	253	+
			254	+	func (c *Coordinator) recoverDeletionTombstone(ctx context.Context, tombstone domain.DeletionTombstone) error {
			255	+	if err := tombstone.Validate(); err != nil {
			256	+	return err
			257	+	}
			258	+	unlock := c.locks.Lock([]string{tombstone.Name})
			259	+	defer unlock()
			260	+	record, exists := c.catalogue.Record(tombstone.Name)
			261	+	includeLegacyName := true
			262	+	if exists {
			263	+	switch {
			264	+	case record.IncarnationID != tombstone.IncarnationID:
			265	+	includeLegacyName = false
			266	+	case record.RecoveryState != domain.RecoveryDeleting:
			267	+	return fmt.Errorf("recovery: deletion tombstone conflicts with non-deleting session %q", tombstone.Name)
			268	+	}
			269	+	}
			270	+	if err := c.repository.QuarantineDeletionSources(ctx, tombstone, includeLegacyName); err != nil {
			271	+	return err
			272	+	}
			273	+	if exists && record.IncarnationID == tombstone.IncarnationID {
			274	+	if err := c.catalogue.Delete(tombstone.Name); err != nil {
			275	+	return err
			276	+	}
129		}	277		}
130	-	return nil	278	+	return c.repository.DeleteDeletionTombstone(ctx, tombstone.IncarnationID)
131		}	279		}

M internal/usecase/recovery/coordinator_test.go +2 -2


@@ -26,7 +26,7 @@ func TestCoordinatorRecoverFailsClosedForPendingWork(t *testing.T) {
26	26		repo := portsmocks.NewMockSnapshotRepository(t)
27	27		repo.EXPECT().ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, recoveryListingBudget).
28	28		Return(ports.DeletionTombstonePage{Done: true}, nil)
29		-	c := NewCoordinator(nil, repo, journalStub{intents: []domain.DiscardIntent{{}}}, nil)
	29	+	c := NewCoordinator(newTransactionCatalogue(), repo, journalStub{intents: []domain.DiscardIntent{{}}}, nil)
30	30		require.ErrorIs(t, c.Recover(context.Background()), ErrPendingRecoveryUnsupported)
31	31		}
32	32

@@ -35,6 +35,6 @@ func TestCoordinatorRecoverPagesTombstonesAndPropagatesErrors(t *testing.T) {
35	35		repo := portsmocks.NewMockSnapshotRepository(t)
36	36		repo.EXPECT().ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, recoveryListingBudget).
37	37		Return(ports.DeletionTombstonePage{}, cause)
38		-	c := NewCoordinator(nil, repo, journalStub{}, nil)
	38	+	c := NewCoordinator(newTransactionCatalogue(), repo, journalStub{}, nil)
39	39		require.ErrorIs(t, c.Recover(context.Background()), cause)
40	40		}


@@ -26,7 +26,7 @@ func TestCoordinatorRecoverFailsClosedForPendingWork(t *testing.T) {
26		repo := portsmocks.NewMockSnapshotRepository(t)	26		repo := portsmocks.NewMockSnapshotRepository(t)
27		repo.EXPECT().ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, recoveryListingBudget).	27		repo.EXPECT().ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, recoveryListingBudget).
28		Return(ports.DeletionTombstonePage{Done: true}, nil)	28		Return(ports.DeletionTombstonePage{Done: true}, nil)
29	-	c := NewCoordinator(nil, repo, journalStub{intents: []domain.DiscardIntent{{}}}, nil)	29	+	c := NewCoordinator(newTransactionCatalogue(), repo, journalStub{intents: []domain.DiscardIntent{{}}}, nil)
30		require.ErrorIs(t, c.Recover(context.Background()), ErrPendingRecoveryUnsupported)	30		require.ErrorIs(t, c.Recover(context.Background()), ErrPendingRecoveryUnsupported)
31		}	31		}
32			32

@@ -35,6 +35,6 @@ func TestCoordinatorRecoverPagesTombstonesAndPropagatesErrors(t *testing.T) {
35		repo := portsmocks.NewMockSnapshotRepository(t)	35		repo := portsmocks.NewMockSnapshotRepository(t)
36		repo.EXPECT().ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, recoveryListingBudget).	36		repo.EXPECT().ListDeletionTombstones(context.Background(), ports.DeletionTombstoneCursor{}, recoveryListingBudget).
37		Return(ports.DeletionTombstonePage{}, cause)	37		Return(ports.DeletionTombstonePage{}, cause)
38	-	c := NewCoordinator(nil, repo, journalStub{}, nil)	38	+	c := NewCoordinator(newTransactionCatalogue(), repo, journalStub{}, nil)
39		require.ErrorIs(t, c.Recover(context.Background()), cause)	39		require.ErrorIs(t, c.Recover(context.Background()), cause)
40		}	40		}

A internal/usecase/recovery/transactions_test.go +228 -0

@@ -0,0 +1,228 @@
	1	+	package recovery
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"context"
	6	+	"errors"
	7	+	"sync"
	8	+	"testing"
	9	+
	10	+	"github.com/bnema/vev/internal/domain"
	11	+	"github.com/bnema/vev/internal/ports"
	12	+	"github.com/stretchr/testify/require"
	13	+	)
	14	+
	15	+	type transactionCatalogue struct {
	16	+	mu sync.Mutex
	17	+	records map[string]domain.CatalogueRecord
	18	+	events []string
	19	+	}
	20	+
	21	+	func newTransactionCatalogue(records ...domain.CatalogueRecord) *transactionCatalogue {
	22	+	catalogue := &transactionCatalogue{records: make(map[string]domain.CatalogueRecord)}
	23	+	for _, record := range records {
	24	+	catalogue.records[record.Name] = record
	25	+	}
	26	+	return catalogue
	27	+	}
	28	+
	29	+	func (c *transactionCatalogue) Records() []domain.CatalogueRecord {
	30	+	c.mu.Lock()
	31	+	defer c.mu.Unlock()
	32	+	out := make([]domain.CatalogueRecord, 0, len(c.records))
	33	+	for _, record := range c.records {
	34	+	out = append(out, record)
	35	+	}
	36	+	return out
	37	+	}
	38	+	func (c *transactionCatalogue) Record(name string) (domain.CatalogueRecord, bool) {
	39	+	c.mu.Lock()
	40	+	defer c.mu.Unlock()
	41	+	record, ok := c.records[name]
	42	+	return record, ok
	43	+	}
	44	+	func (c *transactionCatalogue) Create(record domain.CatalogueRecord) error {
	45	+	c.mu.Lock()
	46	+	defer c.mu.Unlock()
	47	+	if _, exists := c.records[record.Name]; exists {
	48	+	return errors.New("record exists")
	49	+	}
	50	+	c.records[record.Name] = record
	51	+	c.events = append(c.events, "create")
	52	+	return nil
	53	+	}
	54	+	func (c *transactionCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error { return nil }
	55	+	func (c *transactionCatalogue) Replace(name string, record domain.CatalogueRecord) error {
	56	+	c.mu.Lock()
	57	+	defer c.mu.Unlock()
	58	+	if _, exists := c.records[name]; !exists {
	59	+	return errors.New("record missing")
	60	+	}
	61	+	c.records[name] = record
	62	+	c.events = append(c.events, "replace:"+name)
	63	+	return nil
	64	+	}
	65	+	func (c *transactionCatalogue) Rename(oldName string, record domain.CatalogueRecord) error {
	66	+	c.mu.Lock()
	67	+	defer c.mu.Unlock()
	68	+	if _, exists := c.records[oldName]; !exists {
	69	+	return errors.New("record missing")
	70	+	}
	71	+	if _, exists := c.records[record.Name]; exists && record.Name != oldName {
	72	+	return errors.New("record exists")
	73	+	}
	74	+	delete(c.records, oldName)
	75	+	c.records[record.Name] = record
	76	+	c.events = append(c.events, "rename:"+oldName+":"+record.Name)
	77	+	return nil
	78	+	}
	79	+	func (c *transactionCatalogue) Delete(name string) error {
	80	+	c.mu.Lock()
	81	+	defer c.mu.Unlock()
	82	+	delete(c.records, name)
	83	+	c.events = append(c.events, "delete:"+name)
	84	+	return nil
	85	+	}
	86	+	func (c *transactionCatalogue) Close() error { return nil }
	87	+
	88	+	type transactionRepository struct {
	89	+	ports.SnapshotRepository
	90	+	mu sync.Mutex
	91	+	tombstones map[domain.IncarnationID]domain.DeletionTombstone
	92	+	events []string
	93	+	quarantine []bool
	94	+	}
	95	+
	96	+	func newTransactionRepository() *transactionRepository {
	97	+	return &transactionRepository{tombstones: make(map[domain.IncarnationID]domain.DeletionTombstone)}
	98	+	}
	99	+	func (r *transactionRepository) WriteDeletionTombstone(_ context.Context, tombstone domain.DeletionTombstone) error {
	100	+	r.mu.Lock()
	101	+	defer r.mu.Unlock()
	102	+	r.tombstones[tombstone.IncarnationID] = tombstone
	103	+	r.events = append(r.events, "tombstone")
	104	+	return nil
	105	+	}
	106	+	func (r *transactionRepository) ListDeletionTombstones(_ context.Context, cursor ports.DeletionTombstoneCursor, _ ports.MaintenanceBudget) (ports.DeletionTombstonePage, error) {
	107	+	r.mu.Lock()
	108	+	defer r.mu.Unlock()
	109	+	if cursor.After != "" \|\| len(r.tombstones) == 0 {
	110	+	return ports.DeletionTombstonePage{Done: true}, nil
	111	+	}
	112	+	out := make([]domain.DeletionTombstone, 0, len(r.tombstones))
	113	+	for _, tombstone := range r.tombstones {
	114	+	out = append(out, tombstone)
	115	+	}
	116	+	return ports.DeletionTombstonePage{Tombstones: out, Done: true}, nil
	117	+	}
	118	+	func (r *transactionRepository) QuarantineDeletionSources(_ context.Context, _ domain.DeletionTombstone, includeLegacyName bool) error {
	119	+	r.mu.Lock()
	120	+	defer r.mu.Unlock()
	121	+	r.events = append(r.events, "quarantine")
	122	+	r.quarantine = append(r.quarantine, includeLegacyName)
	123	+	return nil
	124	+	}
	125	+	func (r *transactionRepository) DeleteDeletionTombstone(_ context.Context, id domain.IncarnationID) error {
	126	+	r.mu.Lock()
	127	+	defer r.mu.Unlock()
	128	+	delete(r.tombstones, id)
	129	+	r.events = append(r.events, "remove-tombstone")
	130	+	return nil
	131	+	}
	132	+
	133	+	func TestTransactionCrashMatrix(t *testing.T) {
	134	+	ctx := context.Background()
	135	+	t.Run("create", func(t *testing.T) {
	136	+	catalogue := newTransactionCatalogue()
	137	+	coordinator := NewCoordinator(catalogue, newTransactionRepository(), journalStub{}, bytes.NewReader(bytes.Repeat([]byte{1}, 16)))
	138	+	record, err := coordinator.Create(ctx, domain.CatalogueRecord{Name: "work", Cwd: "/tmp"})
	139	+	require.NoError(t, err)
	140	+	require.Equal(t, domain.RecoveryFresh, record.RecoveryState)
	141	+	require.Equal(t, domain.IncarnationID{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1}, record.IncarnationID)
	142	+	require.Equal(t, record, catalogue.records["work"])
	143	+	})
	144	+	t.Run("checkpoint", func(t *testing.T) {
	145	+	record := checkpointRecord()
	146	+	catalogue := newTransactionCatalogue(record)
	147	+	repository := &checkpointRepository{}
	148	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	149	+	next, err := coordinator.PublishCheckpoint(ctx, record.Name, checkpointPublication(record, 1))
	150	+	require.NoError(t, err)
	151	+	require.Equal(t, next, catalogue.records[record.Name])
	152	+	})
	153	+	t.Run("rename", func(t *testing.T) {
	154	+	ref := &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{1}}
	155	+	record := domain.CatalogueRecord{Name: "old", IncarnationID: domain.IncarnationID{2}, RecoveryState: domain.RecoveryHealthy, Committed: ref}
	156	+	catalogue := newTransactionCatalogue(record)
	157	+	coordinator := NewCoordinator(catalogue, newTransactionRepository(), journalStub{}, nil)
	158	+	next, err := coordinator.Rename(ctx, "old", "new")
	159	+	require.NoError(t, err)
	160	+	_, oldExists := catalogue.records["old"]
	161	+	require.False(t, oldExists)
	162	+	require.Equal(t, record.IncarnationID, catalogue.records["new"].IncarnationID)
	163	+	require.Equal(t, record.Committed, next.Committed)
	164	+	})
	165	+	t.Run("delete", func(t *testing.T) {
	166	+	record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{3}, RecoveryState: domain.RecoveryFresh}
	167	+	tombstone := domain.DeletionTombstone{Name: record.Name, IncarnationID: record.IncarnationID}
	168	+	for crashAfter := 1; crashAfter <= 6; crashAfter++ {
	169	+	t.Run(string(rune('0'+crashAfter)), func(t *testing.T) {
	170	+	catalogue := newTransactionCatalogue()
	171	+	repository := newTransactionRepository()
	172	+	switch crashAfter {
	173	+	case 1:
	174	+	deleting := record
	175	+	deleting.RecoveryState = domain.RecoveryDeleting
	176	+	catalogue.records[record.Name] = deleting
	177	+	case 2, 3, 4:
	178	+	deleting := record
	179	+	deleting.RecoveryState = domain.RecoveryDeleting
	180	+	catalogue.records[record.Name] = deleting
	181	+	repository.tombstones[record.IncarnationID] = tombstone
	182	+	case 5:
	183	+	repository.tombstones[record.IncarnationID] = tombstone
	184	+	case 6:
	185	+	// The complete transaction has no durable work left.
	186	+	}
	187	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	188	+	require.NoError(t, coordinator.Recover(ctx))
	189	+	_, exists := catalogue.records[record.Name]
	190	+	require.False(t, exists)
	191	+	require.Empty(t, repository.tombstones)
	192	+	})
	193	+	}
	194	+
	195	+	catalogue := newTransactionCatalogue(record)
	196	+	repository := newTransactionRepository()
	197	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	198	+	require.NoError(t, coordinator.Delete(ctx, "work"))
	199	+	require.Equal(t, []string{"replace:work", "delete:work"}, catalogue.events)
	200	+	require.Equal(t, []string{"tombstone", "quarantine", "remove-tombstone"}, repository.events)
	201	+	require.Equal(t, []bool{true}, repository.quarantine)
	202	+	})
	203	+	}
	204	+
	205	+	func TestDeleteRecoveryAfterCatalogueRemoval(t *testing.T) {
	206	+	ctx := context.Background()
	207	+	old := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{4}}
	208	+	t.Run("absent catalogue record", func(t *testing.T) {
	209	+	catalogue := newTransactionCatalogue()
	210	+	repository := newTransactionRepository()
	211	+	repository.tombstones[old.IncarnationID] = old
	212	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	213	+	require.NoError(t, coordinator.Recover(ctx))
	214	+	require.Empty(t, repository.tombstones)
	215	+	require.Equal(t, []bool{true}, repository.quarantine)
	216	+	})
	217	+	t.Run("reused name preserves new incarnation", func(t *testing.T) {
	218	+	replacement := domain.CatalogueRecord{Name: old.Name, IncarnationID: domain.IncarnationID{9}, RecoveryState: domain.RecoveryFresh}
	219	+	catalogue := newTransactionCatalogue(replacement)
	220	+	repository := newTransactionRepository()
	221	+	repository.tombstones[old.IncarnationID] = old
	222	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	223	+	require.NoError(t, coordinator.Recover(ctx))
	224	+	require.Equal(t, replacement, catalogue.records[old.Name])
	225	+	require.Equal(t, []bool{false}, repository.quarantine)
	226	+	require.Empty(t, repository.tombstones)
	227	+	})
	228	+	}

@@ -0,0 +1,228 @@
	1	+	package recovery
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"context"
	6	+	"errors"
	7	+	"sync"
	8	+	"testing"
	9	+
	10	+	"github.com/bnema/vev/internal/domain"
	11	+	"github.com/bnema/vev/internal/ports"
	12	+	"github.com/stretchr/testify/require"
	13	+	)
	14	+
	15	+	type transactionCatalogue struct {
	16	+	mu sync.Mutex
	17	+	records map[string]domain.CatalogueRecord
	18	+	events []string
	19	+	}
	20	+
	21	+	func newTransactionCatalogue(records ...domain.CatalogueRecord) *transactionCatalogue {
	22	+	catalogue := &transactionCatalogue{records: make(map[string]domain.CatalogueRecord)}
	23	+	for _, record := range records {
	24	+	catalogue.records[record.Name] = record
	25	+	}
	26	+	return catalogue
	27	+	}
	28	+
	29	+	func (c *transactionCatalogue) Records() []domain.CatalogueRecord {
	30	+	c.mu.Lock()
	31	+	defer c.mu.Unlock()
	32	+	out := make([]domain.CatalogueRecord, 0, len(c.records))
	33	+	for _, record := range c.records {
	34	+	out = append(out, record)
	35	+	}
	36	+	return out
	37	+	}
	38	+	func (c *transactionCatalogue) Record(name string) (domain.CatalogueRecord, bool) {
	39	+	c.mu.Lock()
	40	+	defer c.mu.Unlock()
	41	+	record, ok := c.records[name]
	42	+	return record, ok
	43	+	}
	44	+	func (c *transactionCatalogue) Create(record domain.CatalogueRecord) error {
	45	+	c.mu.Lock()
	46	+	defer c.mu.Unlock()
	47	+	if _, exists := c.records[record.Name]; exists {
	48	+	return errors.New("record exists")
	49	+	}
	50	+	c.records[record.Name] = record
	51	+	c.events = append(c.events, "create")
	52	+	return nil
	53	+	}
	54	+	func (c *transactionCatalogue) UpdateMetadata(domain.CatalogueMetadataUpdate) error { return nil }
	55	+	func (c *transactionCatalogue) Replace(name string, record domain.CatalogueRecord) error {
	56	+	c.mu.Lock()
	57	+	defer c.mu.Unlock()
	58	+	if _, exists := c.records[name]; !exists {
	59	+	return errors.New("record missing")
	60	+	}
	61	+	c.records[name] = record
	62	+	c.events = append(c.events, "replace:"+name)
	63	+	return nil
	64	+	}
	65	+	func (c *transactionCatalogue) Rename(oldName string, record domain.CatalogueRecord) error {
	66	+	c.mu.Lock()
	67	+	defer c.mu.Unlock()
	68	+	if _, exists := c.records[oldName]; !exists {
	69	+	return errors.New("record missing")
	70	+	}
	71	+	if _, exists := c.records[record.Name]; exists && record.Name != oldName {
	72	+	return errors.New("record exists")
	73	+	}
	74	+	delete(c.records, oldName)
	75	+	c.records[record.Name] = record
	76	+	c.events = append(c.events, "rename:"+oldName+":"+record.Name)
	77	+	return nil
	78	+	}
	79	+	func (c *transactionCatalogue) Delete(name string) error {
	80	+	c.mu.Lock()
	81	+	defer c.mu.Unlock()
	82	+	delete(c.records, name)
	83	+	c.events = append(c.events, "delete:"+name)
	84	+	return nil
	85	+	}
	86	+	func (c *transactionCatalogue) Close() error { return nil }
	87	+
	88	+	type transactionRepository struct {
	89	+	ports.SnapshotRepository
	90	+	mu sync.Mutex
	91	+	tombstones map[domain.IncarnationID]domain.DeletionTombstone
	92	+	events []string
	93	+	quarantine []bool
	94	+	}
	95	+
	96	+	func newTransactionRepository() *transactionRepository {
	97	+	return &transactionRepository{tombstones: make(map[domain.IncarnationID]domain.DeletionTombstone)}
	98	+	}
	99	+	func (r *transactionRepository) WriteDeletionTombstone(_ context.Context, tombstone domain.DeletionTombstone) error {
	100	+	r.mu.Lock()
	101	+	defer r.mu.Unlock()
	102	+	r.tombstones[tombstone.IncarnationID] = tombstone
	103	+	r.events = append(r.events, "tombstone")
	104	+	return nil
	105	+	}
	106	+	func (r *transactionRepository) ListDeletionTombstones(_ context.Context, cursor ports.DeletionTombstoneCursor, _ ports.MaintenanceBudget) (ports.DeletionTombstonePage, error) {
	107	+	r.mu.Lock()
	108	+	defer r.mu.Unlock()
	109	+	if cursor.After != "" \|\| len(r.tombstones) == 0 {
	110	+	return ports.DeletionTombstonePage{Done: true}, nil
	111	+	}
	112	+	out := make([]domain.DeletionTombstone, 0, len(r.tombstones))
	113	+	for _, tombstone := range r.tombstones {
	114	+	out = append(out, tombstone)
	115	+	}
	116	+	return ports.DeletionTombstonePage{Tombstones: out, Done: true}, nil
	117	+	}
	118	+	func (r *transactionRepository) QuarantineDeletionSources(_ context.Context, _ domain.DeletionTombstone, includeLegacyName bool) error {
	119	+	r.mu.Lock()
	120	+	defer r.mu.Unlock()
	121	+	r.events = append(r.events, "quarantine")
	122	+	r.quarantine = append(r.quarantine, includeLegacyName)
	123	+	return nil
	124	+	}
	125	+	func (r *transactionRepository) DeleteDeletionTombstone(_ context.Context, id domain.IncarnationID) error {
	126	+	r.mu.Lock()
	127	+	defer r.mu.Unlock()
	128	+	delete(r.tombstones, id)
	129	+	r.events = append(r.events, "remove-tombstone")
	130	+	return nil
	131	+	}
	132	+
	133	+	func TestTransactionCrashMatrix(t *testing.T) {
	134	+	ctx := context.Background()
	135	+	t.Run("create", func(t *testing.T) {
	136	+	catalogue := newTransactionCatalogue()
	137	+	coordinator := NewCoordinator(catalogue, newTransactionRepository(), journalStub{}, bytes.NewReader(bytes.Repeat([]byte{1}, 16)))
	138	+	record, err := coordinator.Create(ctx, domain.CatalogueRecord{Name: "work", Cwd: "/tmp"})
	139	+	require.NoError(t, err)
	140	+	require.Equal(t, domain.RecoveryFresh, record.RecoveryState)
	141	+	require.Equal(t, domain.IncarnationID{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1}, record.IncarnationID)
	142	+	require.Equal(t, record, catalogue.records["work"])
	143	+	})
	144	+	t.Run("checkpoint", func(t *testing.T) {
	145	+	record := checkpointRecord()
	146	+	catalogue := newTransactionCatalogue(record)
	147	+	repository := &checkpointRepository{}
	148	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	149	+	next, err := coordinator.PublishCheckpoint(ctx, record.Name, checkpointPublication(record, 1))
	150	+	require.NoError(t, err)
	151	+	require.Equal(t, next, catalogue.records[record.Name])
	152	+	})
	153	+	t.Run("rename", func(t *testing.T) {
	154	+	ref := &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{1}}
	155	+	record := domain.CatalogueRecord{Name: "old", IncarnationID: domain.IncarnationID{2}, RecoveryState: domain.RecoveryHealthy, Committed: ref}
	156	+	catalogue := newTransactionCatalogue(record)
	157	+	coordinator := NewCoordinator(catalogue, newTransactionRepository(), journalStub{}, nil)
	158	+	next, err := coordinator.Rename(ctx, "old", "new")
	159	+	require.NoError(t, err)
	160	+	_, oldExists := catalogue.records["old"]
	161	+	require.False(t, oldExists)
	162	+	require.Equal(t, record.IncarnationID, catalogue.records["new"].IncarnationID)
	163	+	require.Equal(t, record.Committed, next.Committed)
	164	+	})
	165	+	t.Run("delete", func(t *testing.T) {
	166	+	record := domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{3}, RecoveryState: domain.RecoveryFresh}
	167	+	tombstone := domain.DeletionTombstone{Name: record.Name, IncarnationID: record.IncarnationID}
	168	+	for crashAfter := 1; crashAfter <= 6; crashAfter++ {
	169	+	t.Run(string(rune('0'+crashAfter)), func(t *testing.T) {
	170	+	catalogue := newTransactionCatalogue()
	171	+	repository := newTransactionRepository()
	172	+	switch crashAfter {
	173	+	case 1:
	174	+	deleting := record
	175	+	deleting.RecoveryState = domain.RecoveryDeleting
	176	+	catalogue.records[record.Name] = deleting
	177	+	case 2, 3, 4:
	178	+	deleting := record
	179	+	deleting.RecoveryState = domain.RecoveryDeleting
	180	+	catalogue.records[record.Name] = deleting
	181	+	repository.tombstones[record.IncarnationID] = tombstone
	182	+	case 5:
	183	+	repository.tombstones[record.IncarnationID] = tombstone
	184	+	case 6:
	185	+	// The complete transaction has no durable work left.
	186	+	}
	187	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	188	+	require.NoError(t, coordinator.Recover(ctx))
	189	+	_, exists := catalogue.records[record.Name]
	190	+	require.False(t, exists)
	191	+	require.Empty(t, repository.tombstones)
	192	+	})
	193	+	}
	194	+
	195	+	catalogue := newTransactionCatalogue(record)
	196	+	repository := newTransactionRepository()
	197	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	198	+	require.NoError(t, coordinator.Delete(ctx, "work"))
	199	+	require.Equal(t, []string{"replace:work", "delete:work"}, catalogue.events)
	200	+	require.Equal(t, []string{"tombstone", "quarantine", "remove-tombstone"}, repository.events)
	201	+	require.Equal(t, []bool{true}, repository.quarantine)
	202	+	})
	203	+	}
	204	+
	205	+	func TestDeleteRecoveryAfterCatalogueRemoval(t *testing.T) {
	206	+	ctx := context.Background()
	207	+	old := domain.DeletionTombstone{Name: "work", IncarnationID: domain.IncarnationID{4}}
	208	+	t.Run("absent catalogue record", func(t *testing.T) {
	209	+	catalogue := newTransactionCatalogue()
	210	+	repository := newTransactionRepository()
	211	+	repository.tombstones[old.IncarnationID] = old
	212	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	213	+	require.NoError(t, coordinator.Recover(ctx))
	214	+	require.Empty(t, repository.tombstones)
	215	+	require.Equal(t, []bool{true}, repository.quarantine)
	216	+	})
	217	+	t.Run("reused name preserves new incarnation", func(t *testing.T) {
	218	+	replacement := domain.CatalogueRecord{Name: old.Name, IncarnationID: domain.IncarnationID{9}, RecoveryState: domain.RecoveryFresh}
	219	+	catalogue := newTransactionCatalogue(replacement)
	220	+	repository := newTransactionRepository()
	221	+	repository.tombstones[old.IncarnationID] = old
	222	+	coordinator := NewCoordinator(catalogue, repository, journalStub{}, nil)
	223	+	require.NoError(t, coordinator.Recover(ctx))
	224	+	require.Equal(t, replacement, catalogue.records[old.Name])
	225	+	require.Equal(t, []bool{false}, repository.quarantine)
	226	+	require.Empty(t, repository.tombstones)
	227	+	})
	228	+	}