2bc840b fix(recovery): stabilize restore completion states · vev

M internal/persist/migration.go +0 -7


@@ -416,13 +416,6 @@ func installMigratedCatalogue(stateDir string, records []domain.CatalogueRecord)
416	416		return atomicMigrationWrite(filepath.Join(stateDir, migrationDirName, catalogueFormatName), []byte("v1\n"))
417	417		}
418	418
419		-	func decodeLegacyCatalogueV0(data []byte) ([]legacyCatalogueRecordV0, error) {
420		-	record, err := decodeLegacyRecordV0("legacy", data)
421		-	if err != nil {
422		-	return nil, err
423		-	}
424		-	return []legacyCatalogueRecordV0{record}, nil
425		-	}
426	419		func decodeLegacyMap(data map[string][]byte) ([]legacyCatalogueRecordV0, error) {
427	420		out := make([]legacyCatalogueRecordV0, 0, len(data))
428	421		for name, value := range data {


@@ -416,13 +416,6 @@ func installMigratedCatalogue(stateDir string, records []domain.CatalogueRecord)
416		return atomicMigrationWrite(filepath.Join(stateDir, migrationDirName, catalogueFormatName), []byte("v1\n"))	416	return atomicMigrationWrite(filepath.Join(stateDir, migrationDirName, catalogueFormatName), []byte("v1\n"))
417		}	417	}
418			418
419	-	func decodeLegacyCatalogueV0(data []byte) ([]legacyCatalogueRecordV0, error) {
420	-	record, err := decodeLegacyRecordV0("legacy", data)
421	-	if err != nil {
422	-	return nil, err
423	-	}
424	-	return []legacyCatalogueRecordV0{record}, nil
425	-	}
426		func decodeLegacyMap(data map[string][]byte) ([]legacyCatalogueRecordV0, error) {	419	func decodeLegacyMap(data map[string][]byte) ([]legacyCatalogueRecordV0, error) {
427		out := make([]legacyCatalogueRecordV0, 0, len(data))	420	out := make([]legacyCatalogueRecordV0, 0, len(data))
428		for name, value := range data {	421	for name, value := range data {

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


@@ -462,22 +462,22 @@ func (_m MockCheckpointCoordinator) EXPECT() MockCheckpointCoordinator_Expecte
462	462		}
463	463
464	464		// PromoteFallback provides a mock function for the type MockCheckpointCoordinator
465		-	func (_mock *MockCheckpointCoordinator) PromoteFallback(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (domain.CatalogueRecord, error) {
	465	+	func (_mock *MockCheckpointCoordinator) PromoteFallback(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (ports.FallbackPromotionOutcome, error) {
466	466		ret := _mock.Called(context1, s, checkpointRef)
467	467
468	468		if len(ret) == 0 {
469	469		panic("no return value specified for PromoteFallback")
470	470		}
471	471
472		-	var r0 domain.CatalogueRecord
	472	+	var r0 ports.FallbackPromotionOutcome
473	473		var r1 error
474		-	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) (domain.CatalogueRecord, error)); ok {
	474	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) (ports.FallbackPromotionOutcome, error)); ok {
475	475		return returnFunc(context1, s, checkpointRef)
476	476		}
477		-	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) domain.CatalogueRecord); ok {
	477	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) ports.FallbackPromotionOutcome); ok {
478	478		r0 = returnFunc(context1, s, checkpointRef)
479	479		} else {
480		-	r0 = ret.Get(0).(domain.CatalogueRecord)
	480	+	r0 = ret.Get(0).(ports.FallbackPromotionOutcome)
481	481		}
482	482		if returnFunc, ok := ret.Get(1).(func(context.Context, string, domain.CheckpointRef) error); ok {
483	483		r1 = returnFunc(context1, s, checkpointRef)

@@ -523,12 +523,12 @@ func (_c *MockCheckpointCoordinator_PromoteFallback_Call) Run(run func(context1
523	523		return _c
524	524		}
525	525
526		-	func (_c MockCheckpointCoordinator_PromoteFallback_Call) Return(catalogueRecord domain.CatalogueRecord, err error) MockCheckpointCoordinator_PromoteFallback_Call {
527		-	_c.Call.Return(catalogueRecord, err)
	526	+	func (_c MockCheckpointCoordinator_PromoteFallback_Call) Return(fallbackPromotionOutcome ports.FallbackPromotionOutcome, err error) MockCheckpointCoordinator_PromoteFallback_Call {
	527	+	_c.Call.Return(fallbackPromotionOutcome, err)
528	528		return _c
529	529		}
530	530
531		-	func (_c MockCheckpointCoordinator_PromoteFallback_Call) RunAndReturn(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (domain.CatalogueRecord, error)) MockCheckpointCoordinator_PromoteFallback_Call {
	531	+	func (_c MockCheckpointCoordinator_PromoteFallback_Call) RunAndReturn(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (ports.FallbackPromotionOutcome, error)) MockCheckpointCoordinator_PromoteFallback_Call {
532	532		_c.Call.Return(run)
533	533		return _c
534	534		}


@@ -462,22 +462,22 @@ func (_m MockCheckpointCoordinator) EXPECT() MockCheckpointCoordinator_Expecte
462		}	462		}
463			463
464		// PromoteFallback provides a mock function for the type MockCheckpointCoordinator	464		// PromoteFallback provides a mock function for the type MockCheckpointCoordinator
465	-	func (_mock *MockCheckpointCoordinator) PromoteFallback(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (domain.CatalogueRecord, error) {	465	+	func (_mock *MockCheckpointCoordinator) PromoteFallback(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (ports.FallbackPromotionOutcome, error) {
466		ret := _mock.Called(context1, s, checkpointRef)	466		ret := _mock.Called(context1, s, checkpointRef)
467			467
468		if len(ret) == 0 {	468		if len(ret) == 0 {
469		panic("no return value specified for PromoteFallback")	469		panic("no return value specified for PromoteFallback")
470		}	470		}
471			471
472	-	var r0 domain.CatalogueRecord	472	+	var r0 ports.FallbackPromotionOutcome
473		var r1 error	473		var r1 error
474	-	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) (domain.CatalogueRecord, error)); ok {	474	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) (ports.FallbackPromotionOutcome, error)); ok {
475		return returnFunc(context1, s, checkpointRef)	475		return returnFunc(context1, s, checkpointRef)
476		}	476		}
477	-	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) domain.CatalogueRecord); ok {	477	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, domain.CheckpointRef) ports.FallbackPromotionOutcome); ok {
478		r0 = returnFunc(context1, s, checkpointRef)	478		r0 = returnFunc(context1, s, checkpointRef)
479		} else {	479		} else {
480	-	r0 = ret.Get(0).(domain.CatalogueRecord)	480	+	r0 = ret.Get(0).(ports.FallbackPromotionOutcome)
481		}	481		}
482		if returnFunc, ok := ret.Get(1).(func(context.Context, string, domain.CheckpointRef) error); ok {	482		if returnFunc, ok := ret.Get(1).(func(context.Context, string, domain.CheckpointRef) error); ok {
483		r1 = returnFunc(context1, s, checkpointRef)	483		r1 = returnFunc(context1, s, checkpointRef)

@@ -523,12 +523,12 @@ func (_c *MockCheckpointCoordinator_PromoteFallback_Call) Run(run func(context1
523		return _c	523		return _c
524		}	524		}
525			525
526	-	func (_c MockCheckpointCoordinator_PromoteFallback_Call) Return(catalogueRecord domain.CatalogueRecord, err error) MockCheckpointCoordinator_PromoteFallback_Call {	526	+	func (_c MockCheckpointCoordinator_PromoteFallback_Call) Return(fallbackPromotionOutcome ports.FallbackPromotionOutcome, err error) MockCheckpointCoordinator_PromoteFallback_Call {
527	-	_c.Call.Return(catalogueRecord, err)	527	+	_c.Call.Return(fallbackPromotionOutcome, err)
528		return _c	528		return _c
529		}	529		}
530			530
531	-	func (_c MockCheckpointCoordinator_PromoteFallback_Call) RunAndReturn(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (domain.CatalogueRecord, error)) MockCheckpointCoordinator_PromoteFallback_Call {	531	+	func (_c MockCheckpointCoordinator_PromoteFallback_Call) RunAndReturn(run func(context1 context.Context, s string, checkpointRef domain.CheckpointRef) (ports.FallbackPromotionOutcome, error)) MockCheckpointCoordinator_PromoteFallback_Call {
532		_c.Call.Return(run)	532		_c.Call.Return(run)
533		return _c	533		return _c
534		}	534		}

M internal/ports/recovery.go +7 -1


@@ -123,9 +123,15 @@ type RecoveryJournal interface {
123	123		DeleteDiscard(context.Context, domain.IncarnationID) error
124	124		}
125	125
	126	+	type FallbackPromotionOutcome struct {
	127	+	Record domain.CatalogueRecord
	128	+	CatalogueCommitted bool
	129	+	HEADRepairError error
	130	+	}
	131	+
126	132		type CheckpointCoordinator interface {
127	133		PublishCheckpoint(context.Context, string, SnapshotPublication) (domain.CatalogueRecord, error)
128		-	PromoteFallback(context.Context, string, domain.CheckpointRef) (domain.CatalogueRecord, error)
	134	+	PromoteFallback(context.Context, string, domain.CheckpointRef) (FallbackPromotionOutcome, error)
129	135		}
130	136
131	137		type Catalogue interface {

 	DeleteDiscard(context.Context, domain.IncarnationID) error
 }
+type FallbackPromotionOutcome struct {
+	Record             domain.CatalogueRecord
+	CatalogueCommitted bool
+	HEADRepairError    error
+}
 type CheckpointCoordinator interface {
 	PublishCheckpoint(context.Context, string, SnapshotPublication) (domain.CatalogueRecord, error)
-	PromoteFallback(context.Context, string, domain.CheckpointRef) (domain.CatalogueRecord, error)
+	PromoteFallback(context.Context, string, domain.CheckpointRef) (FallbackPromotionOutcome, error)
 }
 type Catalogue interface {

M internal/usecase/daemon/daemon.go +25 -24


@@ -128,12 +128,8 @@ type Daemon struct {
128	128		checkpointRecovery ports.CheckpointCoordinator
129	129		legacySnapshots ports.LegacySnapshotSource
130	130		snapsEnabled bool
131		-	// legacyImportPending holds verified legacy blobs whose source deletion
132		-	// failed, so a later import retries only deletion rather than publication.
133		-	legacyImportMu sync.Mutex
134		-	legacyImportPending map[string][]byte
135		-	noticeStore ports.NoticeStore
136		-	snapshotJobs chan *snapshotCapture
	131	+	noticeStore ports.NoticeStore
	132	+	snapshotJobs chan *snapshotCapture
137	133		// snapshotWake wakes the repository scheduler when a session becomes dirty
138	134		// or an attempt completes. It is never closed and producers only send
139	135		// non-blockingly.

@@ -407,24 +403,23 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
407	403		shell = "/bin/sh"
408	404		}
409	405		d := &Daemon{
410		-	sessions: make(map[domain.SessionID]*session),
411		-	stopped: make(map[string]stoppedSession),
412		-	parked: make(map[uint64]*parkedAttachment),
413		-	ptys: ptys,
414		-	clock: clock,
415		-	log: log,
416		-	baseEnv: os.Environ(),
417		-	shell: shell,
418		-	persist: persist.New(nil),
419		-	dirOrHome: dirOrHome,
420		-	done: make(chan struct{}),
421		-	restoreDone: make(chan struct{}),
422		-	animWake: make(chan struct{}, 1),
423		-	legacyImportPending: make(map[string][]byte),
424		-	snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),
425		-	snapshotWake: make(chan struct{}, 1),
426		-	notices: newNoticeCenter(),
427		-	resumeParkGrace: defaultResumeParkGrace,
	406	+	sessions: make(map[domain.SessionID]*session),
	407	+	stopped: make(map[string]stoppedSession),
	408	+	parked: make(map[uint64]*parkedAttachment),
	409	+	ptys: ptys,
	410	+	clock: clock,
	411	+	log: log,
	412	+	baseEnv: os.Environ(),
	413	+	shell: shell,
	414	+	persist: persist.New(nil),
	415	+	dirOrHome: dirOrHome,
	416	+	done: make(chan struct{}),
	417	+	restoreDone: make(chan struct{}),
	418	+	animWake: make(chan struct{}, 1),
	419	+	snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),
	420	+	snapshotWake: make(chan struct{}, 1),
	421	+	notices: newNoticeCenter(),
	422	+	resumeParkGrace: defaultResumeParkGrace,
428	423		barScripts: &barScriptState{
429	424		cfg: barConfigFromDomain(domain.Defaults().Bar),
430	425		outputs: make(map[domain.SessionID]barScriptOutputs),

@@ -904,6 +899,7 @@ func (d Daemon) finishAttach(sess session, tr ports.Transport, sz domain.Size,
904	899		}
905	900
906	901		func (d *Daemon) waitForTargetRestore(name string) error {
	902	+	var observedClosed chan struct{}
907	903		for {
908	904		d.mu.Lock()
909	905		if d.findByNameLocked(name) != nil {

@@ -925,6 +921,10 @@ func (d *Daemon) waitForTargetRestore(name string) error {
925	921		return &protoErr{ports.ErrSessionDegraded, "session durable state is degraded: " + name}
926	922		case runtimeRestoring:
927	923		done := stopped.restoreDone
	924	+	if done == nil \|\| done == observedClosed {
	925	+	d.mu.Unlock()
	926	+	return &protoErr{ports.ErrSessionDegraded, "session durable state is degraded: " + name}
	927	+	}
928	928		var shutdown <-chan struct{}
929	929		if d.serveCtx != nil {
930	930		shutdown = d.serveCtx.Done()

@@ -932,6 +932,7 @@ func (d *Daemon) waitForTargetRestore(name string) error {
932	932		d.mu.Unlock()
933	933		select {
934	934		case <-done:
	935	+	observedClosed = done
935	936		continue
936	937		case <-shutdown:
937	938		return &protoErr{ports.ErrServerShutdown, "daemon is shutting down"}

 	checkpointRecovery ports.CheckpointCoordinator
 	legacySnapshots    ports.LegacySnapshotSource
 	snapsEnabled       bool
-	// legacyImportPending holds verified legacy blobs whose source deletion
+	noticeStore        ports.NoticeStore
-	// failed, so a later import retries only deletion rather than publication.
+	snapshotJobs       chan *snapshotCapture
-	legacyImportMu      sync.Mutex
-	legacyImportPending map[string][]byte
-	noticeStore         ports.NoticeStore
-	snapshotJobs        chan *snapshotCapture
 	// snapshotWake wakes the repository scheduler when a session becomes dirty
 	// or an attempt completes. It is never closed and producers only send
 	// non-blockingly.
 		shell = "/bin/sh"
 	}
 	d := &Daemon{
-		sessions:            make(map[domain.SessionID]*session),
+		sessions:        make(map[domain.SessionID]*session),
-		stopped:             make(map[string]stoppedSession),
+		stopped:         make(map[string]stoppedSession),
-		parked:              make(map[uint64]*parkedAttachment),
+		parked:          make(map[uint64]*parkedAttachment),
-		ptys:                ptys,
+		ptys:            ptys,
-		clock:               clock,
+		clock:           clock,
-		log:                 log,
+		log:             log,
-		baseEnv:             os.Environ(),
+		baseEnv:         os.Environ(),
-		shell:               shell,
+		shell:           shell,
-		persist:             persist.New(nil),
+		persist:         persist.New(nil),
-		dirOrHome:           dirOrHome,
+		dirOrHome:       dirOrHome,
-		done:                make(chan struct{}),
+		done:            make(chan struct{}),
-		restoreDone:         make(chan struct{}),
+		restoreDone:     make(chan struct{}),
-		animWake:            make(chan struct{}, 1),
+		animWake:        make(chan struct{}, 1),
-		legacyImportPending: make(map[string][]byte),
+		snapshotJobs:    make(chan *snapshotCapture, snapshotQueueCapacity),
-		snapshotJobs:        make(chan *snapshotCapture, snapshotQueueCapacity),
+		snapshotWake:    make(chan struct{}, 1),
-		snapshotWake:        make(chan struct{}, 1),
+		notices:         newNoticeCenter(),
-		notices:             newNoticeCenter(),
+		resumeParkGrace: defaultResumeParkGrace,
-		resumeParkGrace:     defaultResumeParkGrace,
 		barScripts: &barScriptState{
 			cfg:         barConfigFromDomain(domain.Defaults().Bar),
 			outputs:     make(map[domain.SessionID]barScriptOutputs),
 }
 func (d *Daemon) waitForTargetRestore(name string) error {
+	var observedClosed chan struct{}
 	for {
 		d.mu.Lock()
 		if d.findByNameLocked(name) != nil {
 			return &protoErr{ports.ErrSessionDegraded, "session durable state is degraded: " + name}
 		case runtimeRestoring:
 			done := stopped.restoreDone
+			if done == nil || done == observedClosed {
+				d.mu.Unlock()
+				return &protoErr{ports.ErrSessionDegraded, "session durable state is degraded: " + name}
+			}
 			var shutdown <-chan struct{}
 			if d.serveCtx != nil {
 				shutdown = d.serveCtx.Done()
 			d.mu.Unlock()
 			select {
 			case <-done:
+				observedClosed = done
 				continue
 			case <-shutdown:
 				return &protoErr{ports.ErrServerShutdown, "daemon is shutting down"}

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


@@ -70,6 +70,8 @@ type durableRecoveryRepository struct {
70	70		errors map[string]error
71	71		loads map[string]int
72	72		repairs map[string]domain.CheckpointRef
	73	+	repairErr error
	74	+	repairCalls int
73	75		}
74	76
75	77		func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {

@@ -89,6 +91,10 @@ func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.
89	91		func (r *durableRecoveryRepository) RepairHEAD(_ context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
90	92		r.mu.Lock()
91	93		defer r.mu.Unlock()
	94	+	r.repairCalls++
	95	+	if r.repairErr != nil {
	96	+	return r.repairErr
	97	+	}
92	98		r.repairs[id.String()] = ref
93	99		return nil
94	100		}

@@ -234,6 +240,40 @@ func TestCatalogueRestoreIndependent(t *testing.T) {
234	240		require.Equal(t, fallbackRef, fallbackRepository.repairs[fallback.IncarnationID.String()])
235	241		}
236	242
	243	+	func TestCatalogueRestoreKeepsCommittedFallbackHealthyWhenHEADRepairFails(t *testing.T) {
	244	+	record := durableRecoveryRecord(2)
	245	+	fallbackRef := domain.CheckpointRef{Generation: record.Committed.Generation - 1}
	246	+	fallbackRecord := record
	247	+	fallbackRecord.Committed = &fallbackRef
	248	+	generation := validGeneration(t, fallbackRecord)
	249	+	fallbackRef.ManifestDigest = sha256.Sum256(generation.Manifest)
	250	+	record.Fallbacks[0] = &fallbackRef
	251	+	repository := &durableRecoveryRepository{
	252	+	generations: map[string]ports.SnapshotGeneration{record.Name: generation},
	253	+	errors: make(map[string]error),
	254	+	loads: make(map[string]int),
	255	+	repairs: make(map[string]domain.CheckpointRef),
	256	+	repairErr: errors.New("head repair unavailable"),
	257	+	}
	258	+
	259	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
	260	+	d.restoreCatalogue(context.Background(), catalogue.Records())
	261	+
	262	+	committed, ok := catalogue.Record(record.Name)
	263	+	require.True(t, ok)
	264	+	require.Equal(t, domain.RecoveryHealthy, committed.RecoveryState)
	265	+	require.Equal(t, fallbackRef, *committed.Committed)
	266	+	require.Equal(t, runtimeHealthy, d.stopped[record.Name].state)
	267	+	require.Equal(t, 1, repository.repairCalls)
	268	+
	269	+	repository.repairErr = nil
	270	+	next, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{committed}, repository)
	271	+	next.restoreCatalogue(context.Background(), catalogue.Records())
	272	+	require.Equal(t, runtimeHealthy, next.stopped[record.Name].state)
	273	+	require.Equal(t, 2, repository.repairCalls, "the next startup retries the pending HEAD repair")
	274	+	require.Equal(t, fallbackRef, repository.repairs[record.IncarnationID.String()])
	275	+	}
	276	+
237	277		func TestCatalogueRestoreAggregateOver4096(t *testing.T) {
238	278		const count = 4200
239	279		records := make([]domain.CatalogueRecord, count)

@@ -280,6 +320,18 @@ func TestCatalogueRestoreIncarnationMismatch(t *testing.T) {
280	320		require.Equal(t, domain.RecoveryDegraded, got.RecoveryState)
281	321		}
282	322
	323	+	type cancellationRecoveryRepository struct {
	324	+	ports.SnapshotRepository
	325	+	started chan struct{}
	326	+	once sync.Once
	327	+	}
	328	+
	329	+	func (r *cancellationRecoveryRepository) LoadCheckpoint(ctx context.Context, _ domain.IncarnationID, _ string, _ ports.CheckpointRef) (ports.SnapshotGeneration, error) {
	330	+	r.once.Do(func() { close(r.started) })
	331	+	<-ctx.Done()
	332	+	return ports.SnapshotGeneration{}, ctx.Err()
	333	+	}
	334	+
283	335		type recoveryCountingPTYFactory struct {
284	336		calls atomic.Int32
285	337		}

@@ -335,6 +387,60 @@ func TestAttachWaitsForRestore(t *testing.T) {
335	387		require.Zero(t, factory.calls.Load())
336	388		}
337	389
	390	+	func TestRestoreCancellationTransitionsBeforeAttachCompletion(t *testing.T) {
	391	+	record := durableRecoveryRecord(0)
	392	+	repository := &cancellationRecoveryRepository{started: make(chan struct{})}
	393	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
	394	+	restoreCtx, cancelRestore := context.WithCancel(context.Background())
	395	+	restored := make(chan struct{})
	396	+	go func() {
	397	+	d.restoreCatalogue(restoreCtx, catalogue.Records())
	398	+	close(restored)
	399	+	}()
	400	+	<-repository.started
	401	+
	402	+	attached := make(chan error, 1)
	403	+	go func() {
	404	+	_, _, err := d.route(ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: record.Name}, nil)
	405	+	attached <- err
	406	+	}()
	407	+	cancelRestore()
	408	+
	409	+	var protocolError *protoErr
	410	+	require.ErrorAs(t, <-attached, &protocolError)
	411	+	require.Equal(t, ports.ErrSessionDegraded, protocolError.code)
	412	+	<-restored
	413	+	entry := d.stopped[record.Name]
	414	+	require.Equal(t, runtimeDegraded, entry.state)
	415	+	require.Equal(t, domain.RecoveryDegraded, entry.record.RecoveryState)
	416	+	require.Equal(t, "restore interrupted", entry.record.DegradedReason)
	417	+	select {
	418	+	case <-entry.restoreDone:
	419	+	default:
	420	+	t.Fatal("restore completion was not signaled")
	421	+	}
	422	+	committed, _ := catalogue.Record(record.Name)
	423	+	require.Equal(t, entry.record, committed)
	424	+	}
	425	+
	426	+	func TestWaitForTargetRestoreRejectsClosedRestoringChannel(t *testing.T) {
	427	+	record := durableRecoveryRecord(0)
	428	+	d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
	429	+	closeRuntimeRestoreDone(d.stopped[record.Name].restoreDone)
	430	+
	431	+	result := make(chan error, 1)
	432	+	go func() { result <- d.waitForTargetRestore(record.Name) }()
	433	+
	434	+	select {
	435	+	case err := <-result:
	436	+	var protocolError *protoErr
	437	+	require.ErrorAs(t, err, &protocolError)
	438	+	require.Equal(t, ports.ErrSessionDegraded, protocolError.code)
	439	+	case <-time.After(time.Second):
	440	+	t.Fatal("waiter spun on a closed restore channel")
	441	+	}
	442	+	}
	443	+
338	444		func TestAttachRejectsDegraded(t *testing.T) {
339	445		record := durableRecoveryRecord(0)
340	446		record.RecoveryState = domain.RecoveryDegraded


@@ -70,6 +70,8 @@ type durableRecoveryRepository struct {
70	errors map[string]error	70		errors map[string]error
71	loads map[string]int	71		loads map[string]int
72	repairs map[string]domain.CheckpointRef	72		repairs map[string]domain.CheckpointRef
		73	+	repairErr error
		74	+	repairCalls int
73	}	75		}
74		76
75	func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {	77		func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {

@@ -89,6 +91,10 @@ func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.
89	func (r *durableRecoveryRepository) RepairHEAD(_ context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {	91		func (r *durableRecoveryRepository) RepairHEAD(_ context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
90	r.mu.Lock()	92		r.mu.Lock()
91	defer r.mu.Unlock()	93		defer r.mu.Unlock()
		94	+	r.repairCalls++
		95	+	if r.repairErr != nil {
		96	+	return r.repairErr
		97	+	}
92	r.repairs[id.String()] = ref	98		r.repairs[id.String()] = ref
93	return nil	99		return nil
94	}	100		}

@@ -234,6 +240,40 @@ func TestCatalogueRestoreIndependent(t *testing.T) {
234	require.Equal(t, fallbackRef, fallbackRepository.repairs[fallback.IncarnationID.String()])	240		require.Equal(t, fallbackRef, fallbackRepository.repairs[fallback.IncarnationID.String()])
235	}	241		}
236		242
		243	+	func TestCatalogueRestoreKeepsCommittedFallbackHealthyWhenHEADRepairFails(t *testing.T) {
		244	+	record := durableRecoveryRecord(2)
		245	+	fallbackRef := domain.CheckpointRef{Generation: record.Committed.Generation - 1}
		246	+	fallbackRecord := record
		247	+	fallbackRecord.Committed = &fallbackRef
		248	+	generation := validGeneration(t, fallbackRecord)
		249	+	fallbackRef.ManifestDigest = sha256.Sum256(generation.Manifest)
		250	+	record.Fallbacks[0] = &fallbackRef
		251	+	repository := &durableRecoveryRepository{
		252	+	generations: map[string]ports.SnapshotGeneration{record.Name: generation},
		253	+	errors: make(map[string]error),
		254	+	loads: make(map[string]int),
		255	+	repairs: make(map[string]domain.CheckpointRef),
		256	+	repairErr: errors.New("head repair unavailable"),
		257	+	}
		258	+
		259	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
		260	+	d.restoreCatalogue(context.Background(), catalogue.Records())
		261	+
		262	+	committed, ok := catalogue.Record(record.Name)
		263	+	require.True(t, ok)
		264	+	require.Equal(t, domain.RecoveryHealthy, committed.RecoveryState)
		265	+	require.Equal(t, fallbackRef, *committed.Committed)
		266	+	require.Equal(t, runtimeHealthy, d.stopped[record.Name].state)
		267	+	require.Equal(t, 1, repository.repairCalls)
		268	+
		269	+	repository.repairErr = nil
		270	+	next, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{committed}, repository)
		271	+	next.restoreCatalogue(context.Background(), catalogue.Records())
		272	+	require.Equal(t, runtimeHealthy, next.stopped[record.Name].state)
		273	+	require.Equal(t, 2, repository.repairCalls, "the next startup retries the pending HEAD repair")
		274	+	require.Equal(t, fallbackRef, repository.repairs[record.IncarnationID.String()])
		275	+	}
		276	+
237	func TestCatalogueRestoreAggregateOver4096(t *testing.T) {	277		func TestCatalogueRestoreAggregateOver4096(t *testing.T) {
238	const count = 4200	278		const count = 4200
239	records := make([]domain.CatalogueRecord, count)	279		records := make([]domain.CatalogueRecord, count)

@@ -280,6 +320,18 @@ func TestCatalogueRestoreIncarnationMismatch(t *testing.T) {
280	require.Equal(t, domain.RecoveryDegraded, got.RecoveryState)	320		require.Equal(t, domain.RecoveryDegraded, got.RecoveryState)
281	}	321		}
282		322
		323	+	type cancellationRecoveryRepository struct {
		324	+	ports.SnapshotRepository
		325	+	started chan struct{}
		326	+	once sync.Once
		327	+	}
		328	+
		329	+	func (r *cancellationRecoveryRepository) LoadCheckpoint(ctx context.Context, _ domain.IncarnationID, _ string, _ ports.CheckpointRef) (ports.SnapshotGeneration, error) {
		330	+	r.once.Do(func() { close(r.started) })
		331	+	<-ctx.Done()
		332	+	return ports.SnapshotGeneration{}, ctx.Err()
		333	+	}
		334	+
283	type recoveryCountingPTYFactory struct {	335		type recoveryCountingPTYFactory struct {
284	calls atomic.Int32	336		calls atomic.Int32
285	}	337		}

@@ -335,6 +387,60 @@ func TestAttachWaitsForRestore(t *testing.T) {
335	require.Zero(t, factory.calls.Load())	387		require.Zero(t, factory.calls.Load())
336	}	388		}
337		389
		390	+	func TestRestoreCancellationTransitionsBeforeAttachCompletion(t *testing.T) {
		391	+	record := durableRecoveryRecord(0)
		392	+	repository := &cancellationRecoveryRepository{started: make(chan struct{})}
		393	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
		394	+	restoreCtx, cancelRestore := context.WithCancel(context.Background())
		395	+	restored := make(chan struct{})
		396	+	go func() {
		397	+	d.restoreCatalogue(restoreCtx, catalogue.Records())
		398	+	close(restored)
		399	+	}()
		400	+	<-repository.started
		401	+
		402	+	attached := make(chan error, 1)
		403	+	go func() {
		404	+	_, _, err := d.route(ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: record.Name}, nil)
		405	+	attached <- err
		406	+	}()
		407	+	cancelRestore()
		408	+
		409	+	var protocolError *protoErr
		410	+	require.ErrorAs(t, <-attached, &protocolError)
		411	+	require.Equal(t, ports.ErrSessionDegraded, protocolError.code)
		412	+	<-restored
		413	+	entry := d.stopped[record.Name]
		414	+	require.Equal(t, runtimeDegraded, entry.state)
		415	+	require.Equal(t, domain.RecoveryDegraded, entry.record.RecoveryState)
		416	+	require.Equal(t, "restore interrupted", entry.record.DegradedReason)
		417	+	select {
		418	+	case <-entry.restoreDone:
		419	+	default:
		420	+	t.Fatal("restore completion was not signaled")
		421	+	}
		422	+	committed, _ := catalogue.Record(record.Name)
		423	+	require.Equal(t, entry.record, committed)
		424	+	}
		425	+
		426	+	func TestWaitForTargetRestoreRejectsClosedRestoringChannel(t *testing.T) {
		427	+	record := durableRecoveryRecord(0)
		428	+	d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
		429	+	closeRuntimeRestoreDone(d.stopped[record.Name].restoreDone)
		430	+
		431	+	result := make(chan error, 1)
		432	+	go func() { result <- d.waitForTargetRestore(record.Name) }()
		433	+
		434	+	select {
		435	+	case err := <-result:
		436	+	var protocolError *protoErr
		437	+	require.ErrorAs(t, err, &protocolError)
		438	+	require.Equal(t, ports.ErrSessionDegraded, protocolError.code)
		439	+	case <-time.After(time.Second):
		440	+	t.Fatal("waiter spun on a closed restore channel")
		441	+	}
		442	+	}
		443	+
338	func TestAttachRejectsDegraded(t *testing.T) {	444		func TestAttachRejectsDegraded(t *testing.T) {
339	record := durableRecoveryRecord(0)	445		record := durableRecoveryRecord(0)
340	record.RecoveryState = domain.RecoveryDegraded	446		record.RecoveryState = domain.RecoveryDegraded

M internal/usecase/daemon/picker_test.go +40 -0


@@ -17,6 +17,7 @@ import (
17	17		"github.com/bnema/vev/internal/usecase/keys"
18	18		"github.com/bnema/vev/internal/usecase/layout"
19	19		"github.com/bnema/vev/internal/usecase/picker"
	20	+	recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
20	21		themeui "github.com/bnema/vev/internal/usecase/theme"
21	22		"github.com/bnema/vev/pkg/renderer"
22	23		"github.com/bnema/vev/pkg/vt"

@@ -287,6 +288,45 @@ func TestPickerWaitsForRestoringTargetBeforeSwitching(t *testing.T) {
287	288		require.Same(t, target, ac.currentSession())
288	289		}
289	290
	291	+	func TestRestoreCancellationTransitionsBeforePickerCompletion(t *testing.T) {
	292	+	p, release := newBlockingPTY(t)
	293	+	defer release()
	294	+	d, from, ac, _ := newManualSessionWithPTYs(t, p)
	295	+	record := durableRecoveryRecord(0)
	296	+	record.Name = "restoring"
	297	+	repository := &cancellationRecoveryRepository{started: make(chan struct{})}
	298	+	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
	299	+	coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil, nil)
	300	+	WithCatalogue(catalogue, []domain.CatalogueRecord{record})(d)
	301	+	WithSnapshotRepository(repository, nil)(d)
	302	+	WithCheckpointCoordinator(coordinator)(d)
	303	+	restoreCtx, cancelRestore := context.WithCancel(context.Background())
	304	+	restored := make(chan struct{})
	305	+	go func() {
	306	+	d.restoreCatalogue(restoreCtx, catalogue.Records())
	307	+	close(restored)
	308	+	}()
	309	+	<-repository.started
	310	+
	311	+	result := make(chan error, 1)
	312	+	go func() {
	313	+	result <- d.switchToTarget(from, ac, picker.Target{Name: record.Name, Stopped: true})
	314	+	}()
	315	+	cancelRestore()
	316	+
	317	+	require.Error(t, <-result)
	318	+	<-restored
	319	+	require.Same(t, from, ac.currentSession())
	320	+	entry := d.stopped[record.Name]
	321	+	require.Equal(t, runtimeDegraded, entry.state)
	322	+	require.Equal(t, "restore interrupted", entry.record.DegradedReason)
	323	+	select {
	324	+	case <-entry.restoreDone:
	325	+	default:
	326	+	t.Fatal("restore completion was not signaled")
	327	+	}
	328	+	}
	329	+
290	330		func TestPickerRejectsCatalogueTargetsWithoutFreshRuntime(t *testing.T) {
291	331		tests := []struct {
292	332		name string


@@ -17,6 +17,7 @@ import (
17	"github.com/bnema/vev/internal/usecase/keys"	17		"github.com/bnema/vev/internal/usecase/keys"
18	"github.com/bnema/vev/internal/usecase/layout"	18		"github.com/bnema/vev/internal/usecase/layout"
19	"github.com/bnema/vev/internal/usecase/picker"	19		"github.com/bnema/vev/internal/usecase/picker"
		20	+	recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
20	themeui "github.com/bnema/vev/internal/usecase/theme"	21		themeui "github.com/bnema/vev/internal/usecase/theme"
21	"github.com/bnema/vev/pkg/renderer"	22		"github.com/bnema/vev/pkg/renderer"
22	"github.com/bnema/vev/pkg/vt"	23		"github.com/bnema/vev/pkg/vt"

@@ -287,6 +288,45 @@ func TestPickerWaitsForRestoringTargetBeforeSwitching(t *testing.T) {
287	require.Same(t, target, ac.currentSession())	288		require.Same(t, target, ac.currentSession())
288	}	289		}
289		290
		291	+	func TestRestoreCancellationTransitionsBeforePickerCompletion(t *testing.T) {
		292	+	p, release := newBlockingPTY(t)
		293	+	defer release()
		294	+	d, from, ac, _ := newManualSessionWithPTYs(t, p)
		295	+	record := durableRecoveryRecord(0)
		296	+	record.Name = "restoring"
		297	+	repository := &cancellationRecoveryRepository{started: make(chan struct{})}
		298	+	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
		299	+	coordinator := recoveryusecase.NewCoordinator(catalogue, repository, nil, nil)
		300	+	WithCatalogue(catalogue, []domain.CatalogueRecord{record})(d)
		301	+	WithSnapshotRepository(repository, nil)(d)
		302	+	WithCheckpointCoordinator(coordinator)(d)
		303	+	restoreCtx, cancelRestore := context.WithCancel(context.Background())
		304	+	restored := make(chan struct{})
		305	+	go func() {
		306	+	d.restoreCatalogue(restoreCtx, catalogue.Records())
		307	+	close(restored)
		308	+	}()
		309	+	<-repository.started
		310	+
		311	+	result := make(chan error, 1)
		312	+	go func() {
		313	+	result <- d.switchToTarget(from, ac, picker.Target{Name: record.Name, Stopped: true})
		314	+	}()
		315	+	cancelRestore()
		316	+
		317	+	require.Error(t, <-result)
		318	+	<-restored
		319	+	require.Same(t, from, ac.currentSession())
		320	+	entry := d.stopped[record.Name]
		321	+	require.Equal(t, runtimeDegraded, entry.state)
		322	+	require.Equal(t, "restore interrupted", entry.record.DegradedReason)
		323	+	select {
		324	+	case <-entry.restoreDone:
		325	+	default:
		326	+	t.Fatal("restore completion was not signaled")
		327	+	}
		328	+	}
		329	+
290	func TestPickerRejectsCatalogueTargetsWithoutFreshRuntime(t *testing.T) {	330		func TestPickerRejectsCatalogueTargetsWithoutFreshRuntime(t *testing.T) {
291	tests := []struct {	331		tests := []struct {
292	name string	332		name string

M internal/usecase/daemon/snapshot_restore_incremental.go +48 -17


@@ -58,10 +58,7 @@ func (d *Daemon) restoreCatalogue(ctx context.Context, records []domain.Catalogu
58	58		d.ensureCatalogueRegistryEntry(record)
59	59		done := d.recordRestoreDone(record.Name)
60	60		err := d.restoreRecord(ctx, record)
61		-	if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
62		-	d.markRecordDegraded(record, err)
63		-	}
64		-	closeRuntimeRestoreDone(done)
	61	+	d.finishRecordRestore(record, err, done)
65	62		}
66	63		})
67	64		}

@@ -126,21 +123,44 @@ func (d *Daemon) setStoppedRecovery(record domain.CatalogueRecord, state runtime
126	123		d.stopped[record.Name] = entry
127	124		}
128	125
129		-	func (d *Daemon) markRecordDegraded(record domain.CatalogueRecord, restoreErr error) {
130		-	if d.catalogue != nil {
131		-	if current, ok := d.catalogue.Record(record.Name); ok && current.IncarnationID == record.IncarnationID {
132		-	record = current
	126	+	func (d *Daemon) finishRecordRestore(record domain.CatalogueRecord, restoreErr error, done chan struct{}) {
	127	+	if restoreErr != nil {
	128	+	if d.catalogue != nil {
	129	+	if current, ok := d.catalogue.Record(record.Name); ok && current.IncarnationID == record.IncarnationID {
	130	+	record = current
	131	+	}
	132	+	}
	133	+	record.RecoveryState = domain.RecoveryDegraded
	134	+	record.DegradedReason = "checkpoint validation failed"
	135	+	if errors.Is(restoreErr, context.Canceled) \|\| errors.Is(restoreErr, context.DeadlineExceeded) {
	136	+	record.DegradedReason = "restore interrupted"
	137	+	}
	138	+	if d.catalogue != nil {
	139	+	if err := d.catalogue.Replace(record.Name, record); err != nil {
	140	+	d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
	141	+	}
133	142		}
134	143		}
135		-	record.RecoveryState = domain.RecoveryDegraded
136		-	record.DegradedReason = "checkpoint validation failed"
137		-	if d.catalogue != nil {
138		-	if err := d.catalogue.Replace(record.Name, record); err != nil {
139		-	d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
	144	+
	145	+	d.mu.Lock()
	146	+	entry, ok := d.stopped[record.Name]
	147	+	if ok {
	148	+	if restoreErr != nil {
	149	+	entry.record = record
	150	+	entry.state = runtimeDegraded
	151	+	entry.cwd = record.Cwd
	152	+	entry.createdAt = record.CreatedAt
	153	+	entry.incarnation = record.IncarnationID
	154	+	entry.lastUsedSeq = record.LastUsedSeq
	155	+	entry.tabNames = append([]string(nil), record.TabNames...)
	156	+	d.stopped[record.Name] = entry
140	157		}
	158	+	closeRuntimeRestoreDone(done)
	159	+	}
	160	+	d.mu.Unlock()
	161	+	if restoreErr != nil {
	162	+	d.log.Warn("session checkpoint restore failed", "session", record.Name, "err", restoreErr)
141	163		}
142		-	d.setStoppedRecovery(record, runtimeDegraded)
143		-	d.log.Warn("session checkpoint restore failed", "session", record.Name, "err", restoreErr)
144	164		}
145	165
146	166		func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecord) error {

@@ -174,6 +194,9 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
174	194		for i, candidate := range candidates {
175	195		generation, err := d.snapshotRepository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, candidate)
176	196		if err != nil {
	197	+	if ctxErr := ctx.Err(); ctxErr != nil {
	198	+	return ctxErr
	199	+	}
177	200		continue
178	201		}
179	202		snapshot, err := sessionFromGeneration(generation)

@@ -191,12 +214,20 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
191	214		if d.checkpointRecovery == nil {
192	215		return errors.New("snapshot: checkpoint coordinator unavailable for fallback promotion")
193	216		}
194		-	var err error
195		-	record, err = d.checkpointRecovery.PromoteFallback(ctx, record.Name, selected)
	217	+	outcome, err := d.checkpointRecovery.PromoteFallback(ctx, record.Name, selected)
196	218		if err != nil {
197	219		return fmt.Errorf("snapshot: promote fallback: %w", err)
198	220		}
	221	+	if !outcome.CatalogueCommitted {
	222	+	return errors.New("snapshot: fallback promotion did not commit catalogue")
	223	+	}
	224	+	record = outcome.Record
	225	+	if outcome.HEADRepairError != nil {
	226	+	d.log.Warn("fallback committed with HEAD repair pending", "session", record.Name, "err", outcome.HEADRepairError)
	227	+	}
199	228		d.log.Warn("restored session from fallback checkpoint", "session", record.Name, "generation", selected.Generation)
	229	+	} else if err := d.snapshotRepository.RepairHEAD(ctx, record.IncarnationID, selected); err != nil {
	230	+	d.log.Warn("checkpoint HEAD repair pending", "session", record.Name, "err", err)
200	231		}
201	232
202	233		if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation); err != nil {


@@ -58,10 +58,7 @@ func (d *Daemon) restoreCatalogue(ctx context.Context, records []domain.Catalogu
58		d.ensureCatalogueRegistryEntry(record)	58		d.ensureCatalogueRegistryEntry(record)
59		done := d.recordRestoreDone(record.Name)	59		done := d.recordRestoreDone(record.Name)
60		err := d.restoreRecord(ctx, record)	60		err := d.restoreRecord(ctx, record)
61	-	if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {	61	+	d.finishRecordRestore(record, err, done)
62	-	d.markRecordDegraded(record, err)
63	-	}
64	-	closeRuntimeRestoreDone(done)
65		}	62		}
66		})	63		})
67		}	64		}

@@ -126,21 +123,44 @@ func (d *Daemon) setStoppedRecovery(record domain.CatalogueRecord, state runtime
126		d.stopped[record.Name] = entry	123		d.stopped[record.Name] = entry
127		}	124		}
128			125
129	-	func (d *Daemon) markRecordDegraded(record domain.CatalogueRecord, restoreErr error) {	126	+	func (d *Daemon) finishRecordRestore(record domain.CatalogueRecord, restoreErr error, done chan struct{}) {
130	-	if d.catalogue != nil {	127	+	if restoreErr != nil {
131	-	if current, ok := d.catalogue.Record(record.Name); ok && current.IncarnationID == record.IncarnationID {	128	+	if d.catalogue != nil {
132	-	record = current	129	+	if current, ok := d.catalogue.Record(record.Name); ok && current.IncarnationID == record.IncarnationID {
			130	+	record = current
			131	+	}
			132	+	}
			133	+	record.RecoveryState = domain.RecoveryDegraded
			134	+	record.DegradedReason = "checkpoint validation failed"
			135	+	if errors.Is(restoreErr, context.Canceled) \|\| errors.Is(restoreErr, context.DeadlineExceeded) {
			136	+	record.DegradedReason = "restore interrupted"
			137	+	}
			138	+	if d.catalogue != nil {
			139	+	if err := d.catalogue.Replace(record.Name, record); err != nil {
			140	+	d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
			141	+	}
133		}	142		}
134		}	143		}
135	-	record.RecoveryState = domain.RecoveryDegraded	144	+
136	-	record.DegradedReason = "checkpoint validation failed"	145	+	d.mu.Lock()
137	-	if d.catalogue != nil {	146	+	entry, ok := d.stopped[record.Name]
138	-	if err := d.catalogue.Replace(record.Name, record); err != nil {	147	+	if ok {
139	-	d.log.Error("marking session degraded failed", "session", record.Name, "err", err)	148	+	if restoreErr != nil {
			149	+	entry.record = record
			150	+	entry.state = runtimeDegraded
			151	+	entry.cwd = record.Cwd
			152	+	entry.createdAt = record.CreatedAt
			153	+	entry.incarnation = record.IncarnationID
			154	+	entry.lastUsedSeq = record.LastUsedSeq
			155	+	entry.tabNames = append([]string(nil), record.TabNames...)
			156	+	d.stopped[record.Name] = entry
140		}	157		}
			158	+	closeRuntimeRestoreDone(done)
			159	+	}
			160	+	d.mu.Unlock()
			161	+	if restoreErr != nil {
			162	+	d.log.Warn("session checkpoint restore failed", "session", record.Name, "err", restoreErr)
141		}	163		}
142	-	d.setStoppedRecovery(record, runtimeDegraded)
143	-	d.log.Warn("session checkpoint restore failed", "session", record.Name, "err", restoreErr)
144		}	164		}
145			165
146		func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecord) error {	166		func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecord) error {

@@ -174,6 +194,9 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
174		for i, candidate := range candidates {	194		for i, candidate := range candidates {
175		generation, err := d.snapshotRepository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, candidate)	195		generation, err := d.snapshotRepository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, candidate)
176		if err != nil {	196		if err != nil {
			197	+	if ctxErr := ctx.Err(); ctxErr != nil {
			198	+	return ctxErr
			199	+	}
177		continue	200		continue
178		}	201		}
179		snapshot, err := sessionFromGeneration(generation)	202		snapshot, err := sessionFromGeneration(generation)

@@ -191,12 +214,20 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
191		if d.checkpointRecovery == nil {	214		if d.checkpointRecovery == nil {
192		return errors.New("snapshot: checkpoint coordinator unavailable for fallback promotion")	215		return errors.New("snapshot: checkpoint coordinator unavailable for fallback promotion")
193		}	216		}
194	-	var err error	217	+	outcome, err := d.checkpointRecovery.PromoteFallback(ctx, record.Name, selected)
195	-	record, err = d.checkpointRecovery.PromoteFallback(ctx, record.Name, selected)
196		if err != nil {	218		if err != nil {
197		return fmt.Errorf("snapshot: promote fallback: %w", err)	219		return fmt.Errorf("snapshot: promote fallback: %w", err)
198		}	220		}
			221	+	if !outcome.CatalogueCommitted {
			222	+	return errors.New("snapshot: fallback promotion did not commit catalogue")
			223	+	}
			224	+	record = outcome.Record
			225	+	if outcome.HEADRepairError != nil {
			226	+	d.log.Warn("fallback committed with HEAD repair pending", "session", record.Name, "err", outcome.HEADRepairError)
			227	+	}
199		d.log.Warn("restored session from fallback checkpoint", "session", record.Name, "generation", selected.Generation)	228		d.log.Warn("restored session from fallback checkpoint", "session", record.Name, "generation", selected.Generation)
			229	+	} else if err := d.snapshotRepository.RepairHEAD(ctx, record.IncarnationID, selected); err != nil {
			230	+	d.log.Warn("checkpoint HEAD repair pending", "session", record.Name, "err", err)
200		}	231		}
201			232
202		if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation); err != nil {	233		if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation); err != nil {

M internal/usecase/recovery/checkpoint.go +10 -11


@@ -64,16 +64,16 @@ func (c *Coordinator) PublishCheckpoint(ctx context.Context, name string, public
64	64		// PromoteFallback validates a directly indexed fallback and any alternatives,
65	65		// commits the bounded validated set, and repairs HEAD only after the catalogue
66	66		// commit is durable.
67		-	func (c *Coordinator) PromoteFallback(ctx context.Context, name string, ref domain.CheckpointRef) (domain.CatalogueRecord, error) {
	67	+	func (c *Coordinator) PromoteFallback(ctx context.Context, name string, ref domain.CheckpointRef) (ports.FallbackPromotionOutcome, error) {
68	68		if c == nil \|\| c.catalogue == nil \|\| c.repository == nil \|\| c.locks == nil {
69		-	return domain.CatalogueRecord{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")
	69	+	return ports.FallbackPromotionOutcome{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")
70	70		}
71	71		unlock := c.locks.Lock([]string{name})
72	72		defer unlock()
73	73
74	74		record, ok := c.catalogue.Record(name)
75	75		if !ok {
76		-	return domain.CatalogueRecord{}, ErrCheckpointRecordNotFound
	76	+	return ports.FallbackPromotionOutcome{}, ErrCheckpointRecordNotFound
77	77		}
78	78		found := false
79	79		for _, fallback := range record.Fallbacks {

@@ -83,10 +83,10 @@ func (c *Coordinator) PromoteFallback(ctx context.Context, name string, ref doma
83	83		}
84	84		}
85	85		if !found {
86		-	return domain.CatalogueRecord{}, ErrFallbackNotFound
	86	+	return ports.FallbackPromotionOutcome{}, ErrFallbackNotFound
87	87		}
88	88		if err := c.validateCheckpoint(ctx, record, ref); err != nil {
89		-	return domain.CatalogueRecord{}, err
	89	+	return ports.FallbackPromotionOutcome{}, err
90	90		}
91	91
92	92		alternatives := make([]domain.CheckpointRef, 0, 2)

@@ -111,15 +111,14 @@ func (c *Coordinator) PromoteFallback(ctx context.Context, name string, ref doma
111	111		next.Fallbacks[i] = copyCheckpointRef(&alternatives[i])
112	112		}
113	113		if err := next.Validate(); err != nil {
114		-	return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid fallback promotion: %w", err)
	114	+	return ports.FallbackPromotionOutcome{}, fmt.Errorf("recovery: invalid fallback promotion: %w", err)
115	115		}
116	116		if err := c.catalogue.Replace(name, next); err != nil {
117		-	return domain.CatalogueRecord{}, err
118		-	}
119		-	if err := c.repository.RepairHEAD(ctx, record.IncarnationID, ref); err != nil {
120		-	return next, err
	117	+	return ports.FallbackPromotionOutcome{}, err
121	118		}
122		-	return next, nil
	119	+	outcome := ports.FallbackPromotionOutcome{Record: next, CatalogueCommitted: true}
	120	+	outcome.HEADRepairError = c.repository.RepairHEAD(ctx, record.IncarnationID, ref)
	121	+	return outcome, nil
123	122		}
124	123
125	124		func (c *Coordinator) validateCheckpoint(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef) error {

 // PromoteFallback validates a directly indexed fallback and any alternatives,
 // commits the bounded validated set, and repairs HEAD only after the catalogue
 // commit is durable.
-func (c *Coordinator) PromoteFallback(ctx context.Context, name string, ref domain.CheckpointRef) (domain.CatalogueRecord, error) {
+func (c *Coordinator) PromoteFallback(ctx context.Context, name string, ref domain.CheckpointRef) (ports.FallbackPromotionOutcome, error) {
 	if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil {
-		return domain.CatalogueRecord{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")
+		return ports.FallbackPromotionOutcome{}, errors.New("recovery: incomplete checkpoint coordinator dependencies")
 	}
 	unlock := c.locks.Lock([]string{name})
 	defer unlock()
 	record, ok := c.catalogue.Record(name)
 	if !ok {
-		return domain.CatalogueRecord{}, ErrCheckpointRecordNotFound
+		return ports.FallbackPromotionOutcome{}, ErrCheckpointRecordNotFound
 	}
 	found := false
 	for _, fallback := range record.Fallbacks {
 		}
 	}
 	if !found {
-		return domain.CatalogueRecord{}, ErrFallbackNotFound
+		return ports.FallbackPromotionOutcome{}, ErrFallbackNotFound
 	}
 	if err := c.validateCheckpoint(ctx, record, ref); err != nil {
-		return domain.CatalogueRecord{}, err
+		return ports.FallbackPromotionOutcome{}, err
 	}
 	alternatives := make([]domain.CheckpointRef, 0, 2)
 		next.Fallbacks[i] = copyCheckpointRef(&alternatives[i])
 	}
 	if err := next.Validate(); err != nil {
-		return domain.CatalogueRecord{}, fmt.Errorf("recovery: invalid fallback promotion: %w", err)
+		return ports.FallbackPromotionOutcome{}, fmt.Errorf("recovery: invalid fallback promotion: %w", err)
 	}
 	if err := c.catalogue.Replace(name, next); err != nil {
-		return domain.CatalogueRecord{}, err
+		return ports.FallbackPromotionOutcome{}, err
-	}
-	if err := c.repository.RepairHEAD(ctx, record.IncarnationID, ref); err != nil {
-		return next, err
 	}
-	return next, nil
+	outcome := ports.FallbackPromotionOutcome{Record: next, CatalogueCommitted: true}
+	outcome.HEADRepairError = c.repository.RepairHEAD(ctx, record.IncarnationID, ref)
+	return outcome, nil
 }
 func (c *Coordinator) validateCheckpoint(ctx context.Context, record domain.CatalogueRecord, ref domain.CheckpointRef) error {

M internal/usecase/recovery/checkpoint_test.go +29 -3


@@ -56,6 +56,7 @@ type checkpointRepository struct {
56	56		loadErrors map[uint64]error
57	57		loads []uint64
58	58		repairs []domain.CheckpointRef
	59	+	repairErr error
59	60		events *[]string
60	61		}
61	62

@@ -81,7 +82,7 @@ func (r *checkpointRepository) LoadCheckpoint(_ context.Context, _ domain.Incarn
81	82
82	83		func (r *checkpointRepository) RepairHEAD(_ context.Context, _ domain.IncarnationID, ref ports.CheckpointRef) error {
83	84		r.repairs = append(r.repairs, ref)
84		-	return nil
	85	+	return r.repairErr
85	86		}
86	87
87	88		func checkpointRecord() domain.CatalogueRecord {

@@ -173,13 +174,38 @@ func TestCheckpointCommitThirdPromotesOnlyValidatedDirectFallbacks(t *testing.T)
173	174
174	175		promoted, err := coordinator.PromoteFallback(context.Background(), record.Name, refs[1])
175	176		require.NoError(t, err)
176		-	require.Equal(t, refs[1], *promoted.Committed)
177		-	require.Equal(t, []uint64{1}, populatedGenerations(promoted.Fallbacks))
	177	+	require.Equal(t, refs[1], *promoted.Record.Committed)
	178	+	require.Equal(t, []uint64{1}, populatedGenerations(promoted.Record.Fallbacks))
178	179		require.Equal(t, []uint64{2, 1}, repository.loads, "only the selected and older catalogue-indexed fallbacks are read directly")
179	180		require.Equal(t, []domain.CheckpointRef{refs[1]}, repository.repairs)
180	181		require.Equal(t, 1, catalogue.replaces)
181	182		}
182	183
	184	+	func TestPromoteFallbackReportsPostCommitHEADRepairFailure(t *testing.T) {
	185	+	selected := domain.CheckpointRef{Generation: 2, ManifestDigest: sha256.Sum256([]byte("second"))}
	186	+	record := checkpointRecord()
	187	+	record.RecoveryState = domain.RecoveryHealthy
	188	+	record.Committed = &domain.CheckpointRef{Generation: 3, ManifestDigest: sha256.Sum256([]byte("third"))}
	189	+	record.Fallbacks[0] = &selected
	190	+	catalogue := &checkpointCatalogue{record: record}
	191	+	repairErr := errors.New("head repair failed")
	192	+	repository := &checkpointRepository{
	193	+	generations: map[uint64]ports.SnapshotGeneration{
	194	+	2: {IncarnationID: record.IncarnationID, Name: record.Name, Generation: 2, Manifest: []byte("second")},
	195	+	},
	196	+	repairErr: repairErr,
	197	+	}
	198	+	coordinator := NewCoordinator(catalogue, repository, nil, nil)
	199	+
	200	+	outcome, err := coordinator.PromoteFallback(context.Background(), record.Name, selected)
	201	+	require.NoError(t, err)
	202	+	require.True(t, outcome.CatalogueCommitted)
	203	+	require.ErrorIs(t, outcome.HEADRepairError, repairErr)
	204	+	require.Equal(t, selected, *outcome.Record.Committed)
	205	+	require.Equal(t, domain.RecoveryHealthy, catalogue.record.RecoveryState)
	206	+	require.Equal(t, selected, *catalogue.record.Committed)
	207	+	}
	208	+
183	209		func TestCheckpointCommitPublishOrphan(t *testing.T) {
184	210		prior := checkpointRecord()
185	211		priorRef := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}}

 	loadErrors   map[uint64]error
 	loads        []uint64
 	repairs      []domain.CheckpointRef
+	repairErr    error
 	events       *[]string
 }
 func (r *checkpointRepository) RepairHEAD(_ context.Context, _ domain.IncarnationID, ref ports.CheckpointRef) error {
 	r.repairs = append(r.repairs, ref)
-	return nil
+	return r.repairErr
 }
 func checkpointRecord() domain.CatalogueRecord {
 	promoted, err := coordinator.PromoteFallback(context.Background(), record.Name, refs[1])
 	require.NoError(t, err)
-	require.Equal(t, refs[1], *promoted.Committed)
+	require.Equal(t, refs[1], *promoted.Record.Committed)
-	require.Equal(t, []uint64{1}, populatedGenerations(promoted.Fallbacks))
+	require.Equal(t, []uint64{1}, populatedGenerations(promoted.Record.Fallbacks))
 	require.Equal(t, []uint64{2, 1}, repository.loads, "only the selected and older catalogue-indexed fallbacks are read directly")
 	require.Equal(t, []domain.CheckpointRef{refs[1]}, repository.repairs)
 	require.Equal(t, 1, catalogue.replaces)
 }
+func TestPromoteFallbackReportsPostCommitHEADRepairFailure(t *testing.T) {
+	selected := domain.CheckpointRef{Generation: 2, ManifestDigest: sha256.Sum256([]byte("second"))}
+	record := checkpointRecord()
+	record.RecoveryState = domain.RecoveryHealthy
+	record.Committed = &domain.CheckpointRef{Generation: 3, ManifestDigest: sha256.Sum256([]byte("third"))}
+	record.Fallbacks[0] = &selected
+	catalogue := &checkpointCatalogue{record: record}
+	repairErr := errors.New("head repair failed")
+	repository := &checkpointRepository{
+		generations: map[uint64]ports.SnapshotGeneration{
+			2: {IncarnationID: record.IncarnationID, Name: record.Name, Generation: 2, Manifest: []byte("second")},
+		},
+		repairErr: repairErr,
+	}
+	coordinator := NewCoordinator(catalogue, repository, nil, nil)
+	outcome, err := coordinator.PromoteFallback(context.Background(), record.Name, selected)
+	require.NoError(t, err)
+	require.True(t, outcome.CatalogueCommitted)
+	require.ErrorIs(t, outcome.HEADRepairError, repairErr)
+	require.Equal(t, selected, *outcome.Record.Committed)
+	require.Equal(t, domain.RecoveryHealthy, catalogue.record.RecoveryState)
+	require.Equal(t, selected, *catalogue.record.Committed)
+}
 func TestCheckpointCommitPublishOrphan(t *testing.T) {
 	prior := checkpointRecord()
 	priorRef := domain.CheckpointRef{Generation: 7, ManifestDigest: [32]byte{7}}