808c829 fix(daemon): join restoration before releasing lifecycle ownership · vev

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


@@ -131,10 +131,15 @@ type Daemon struct {
131	131		maintenance maintenanceDependencies
132	132		maintenanceWorkerCancel context.CancelFunc
133	133		maintenanceWorkerDone chan struct{}
134		-	legacySnapshots ports.LegacySnapshotSource
135		-	snapsEnabled bool
136		-	noticeStore ports.NoticeStore
137		-	snapshotJobs chan *snapshotCapture
	134	+	// restoreWorkerDone is the restoration goroutine's ownership signal. Startup
	135	+	// restoration repairs HEADs, promotes fallbacks, and replaces catalogue
	136	+	// records, so it is a durable writer and is guarded by snapshotWorkerMu with
	137	+	// the other two.
	138	+	restoreWorkerDone chan struct{}
	139	+	legacySnapshots ports.LegacySnapshotSource
	140	+	snapsEnabled bool
	141	+	noticeStore ports.NoticeStore
	142	+	snapshotJobs chan *snapshotCapture
138	143		// snapshotWake wakes the repository scheduler when a session becomes dirty
139	144		// or an attempt completes. It is never closed and producers only send
140	145		// non-blockingly.

@@ -483,7 +488,13 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
483	488		d.restoreProcessAllowlist.Store(&allow)
484	489		}
485	490		records := d.catalogueRecords
486		-	if !d.catalogueRecordsProvided {
	491	+	// Only attempt the load when persistence is actually configured
	492	+	// (d.persistEnabled tracks the same store-is-non-nil condition
	493	+	// persist.Persister uses to distinguish "no store" from "load failed").
	494	+	// Without this gate, the default no-persistence Daemon (persist.New(nil))
	495	+	// would log a false-alarm "loading persisted sessions failed" warning on
	496	+	// every construction: nothing failed, persistence simply isn't enabled.
	497	+	if !d.catalogueRecordsProvided && d.persistEnabled {
487	498		var err error
488	499		records, err = d.persist.LoadAll()
489	500		if err != nil {

@@ -541,9 +552,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
541	552		d.sessWg.Go(func() {
542	553		d.snapshotRepositorySaver(d.serveCtx)
543	554		})
544		-	d.sessWg.Go(func() {
545		-	d.restoreSnapshots(d.serveCtx)
546		-	})
	555	+	d.startSnapshotRestoration()
547	556		} else {
548	557		d.closeRestoreDone()
549	558		}

@@ -992,7 +1001,14 @@ func (d *Daemon) waitForTargetRestore(name string) error {
992	1001		case <-shutdown:
993	1002		return &protoErr{ports.ErrServerShutdown, "daemon is shutting down"}
994	1003		}
995		-	case runtimeDegraded, runtimeDeleting, runtimeHealthy:
	1004	+	case runtimeHealthy:
	1005	+	// The catalogue record is healthy; only its runtime registration was
	1006	+	// skipped (shutdown, or a live session claimed the name first). Saying
	1007	+	// "degraded" here would send the operator to durable recovery for a
	1008	+	// session whose durable state is intact.
	1009	+	d.mu.Unlock()
	1010	+	return &protoErr{ports.ErrSessionDegraded, "session was not restored into this daemon: " + name}
	1011	+	case runtimeDegraded, runtimeDeleting:
996	1012		d.mu.Unlock()
997	1013		return &protoErr{ports.ErrSessionDegraded, "session durable state is degraded: " + name}
998	1014		default:

 	maintenance             maintenanceDependencies
 	maintenanceWorkerCancel context.CancelFunc
 	maintenanceWorkerDone   chan struct{}
-	legacySnapshots         ports.LegacySnapshotSource
+	// restoreWorkerDone is the restoration goroutine's ownership signal. Startup
-	snapsEnabled            bool
+	// restoration repairs HEADs, promotes fallbacks, and replaces catalogue
-	noticeStore             ports.NoticeStore
+	// records, so it is a durable writer and is guarded by snapshotWorkerMu with
-	snapshotJobs            chan *snapshotCapture
+	// the other two.
+	restoreWorkerDone chan struct{}
+	legacySnapshots   ports.LegacySnapshotSource
+	snapsEnabled      bool
+	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.
 		d.restoreProcessAllowlist.Store(&allow)
 	}
 	records := d.catalogueRecords
-	if !d.catalogueRecordsProvided {
+	// Only attempt the load when persistence is actually configured
+	// (d.persistEnabled tracks the same store-is-non-nil condition
+	// persist.Persister uses to distinguish "no store" from "load failed").
+	// Without this gate, the default no-persistence Daemon (persist.New(nil))
+	// would log a false-alarm "loading persisted sessions failed" warning on
+	// every construction: nothing failed, persistence simply isn't enabled.
+	if !d.catalogueRecordsProvided && d.persistEnabled {
 		var err error
 		records, err = d.persist.LoadAll()
 		if err != nil {
 		d.sessWg.Go(func() {
 			d.snapshotRepositorySaver(d.serveCtx)
 		})
-		d.sessWg.Go(func() {
+		d.startSnapshotRestoration()
-			d.restoreSnapshots(d.serveCtx)
-		})
 	} else {
 		d.closeRestoreDone()
 	}
 			case <-shutdown:
 				return &protoErr{ports.ErrServerShutdown, "daemon is shutting down"}
 			}
-		case runtimeDegraded, runtimeDeleting, runtimeHealthy:
+		case runtimeHealthy:
+			// The catalogue record is healthy; only its runtime registration was
+			// skipped (shutdown, or a live session claimed the name first). Saying
+			// "degraded" here would send the operator to durable recovery for a
+			// session whose durable state is intact.
+			d.mu.Unlock()
+			return &protoErr{ports.ErrSessionDegraded, "session was not restored into this daemon: " + name}
+		case runtimeDegraded, runtimeDeleting:
 			d.mu.Unlock()
 			return &protoErr{ports.ErrSessionDegraded, "session durable state is degraded: " + name}
 		default:

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


@@ -510,6 +510,32 @@ func TestWaitForTargetRestoreRejectsClosedRestoringChannel(t *testing.T) {
510	510		}
511	511		}
512	512
	513	+	// A healthy catalogue record whose runtime registration was skipped is not
	514	+	// degraded durable state, and must not send the operator to durable recovery.
	515	+	func TestWaitForTargetRestoreDistinguishesHealthyRuntime(t *testing.T) {
	516	+	record := durableRecoveryRecord(0)
	517	+	tests := []struct {
	518	+	name string
	519	+	state runtimeRecoveryState
	520	+	message string
	521	+	}{
	522	+	{name: "healthy record without runtime", state: runtimeHealthy, message: "session was not restored into this daemon: " + record.Name},
	523	+	{name: "degraded durable state", state: runtimeDegraded, message: "session durable state is degraded: " + record.Name},
	524	+	{name: "deleting durable state", state: runtimeDeleting, message: "session durable state is degraded: " + record.Name},
	525	+	}
	526	+	for _, tt := range tests {
	527	+	t.Run(tt.name, func(t *testing.T) {
	528	+	d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
	529	+	d.setStoppedRecovery(record, tt.state)
	530	+
	531	+	var protocolError *protoErr
	532	+	require.ErrorAs(t, d.waitForTargetRestore(record.Name), &protocolError)
	533	+	require.Equal(t, ports.ErrSessionDegraded, protocolError.code)
	534	+	require.Equal(t, tt.message, protocolError.Error())
	535	+	})
	536	+	}
	537	+	}
	538	+
513	539		func TestAttachRejectsDegraded(t *testing.T) {
514	540		record := durableRecoveryRecord(0)
515	541		record.RecoveryState = domain.RecoveryDegraded


@@ -510,6 +510,32 @@ func TestWaitForTargetRestoreRejectsClosedRestoringChannel(t *testing.T) {
510	}	510		}
511	}	511		}
512		512
		513	+	// A healthy catalogue record whose runtime registration was skipped is not
		514	+	// degraded durable state, and must not send the operator to durable recovery.
		515	+	func TestWaitForTargetRestoreDistinguishesHealthyRuntime(t *testing.T) {
		516	+	record := durableRecoveryRecord(0)
		517	+	tests := []struct {
		518	+	name string
		519	+	state runtimeRecoveryState
		520	+	message string
		521	+	}{
		522	+	{name: "healthy record without runtime", state: runtimeHealthy, message: "session was not restored into this daemon: " + record.Name},
		523	+	{name: "degraded durable state", state: runtimeDegraded, message: "session durable state is degraded: " + record.Name},
		524	+	{name: "deleting durable state", state: runtimeDeleting, message: "session durable state is degraded: " + record.Name},
		525	+	}
		526	+	for _, tt := range tests {
		527	+	t.Run(tt.name, func(t *testing.T) {
		528	+	d, _ := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, &durableRecoveryRepository{})
		529	+	d.setStoppedRecovery(record, tt.state)
		530	+
		531	+	var protocolError *protoErr
		532	+	require.ErrorAs(t, d.waitForTargetRestore(record.Name), &protocolError)
		533	+	require.Equal(t, ports.ErrSessionDegraded, protocolError.code)
		534	+	require.Equal(t, tt.message, protocolError.Error())
		535	+	})
		536	+	}
		537	+	}
		538	+
513	func TestAttachRejectsDegraded(t *testing.T) {	539		func TestAttachRejectsDegraded(t *testing.T) {
514	record := durableRecoveryRecord(0)	540		record := durableRecoveryRecord(0)
515	record.RecoveryState = domain.RecoveryDegraded	541		record.RecoveryState = domain.RecoveryDegraded

M internal/usecase/daemon/maintenance.go +10 -3


@@ -119,13 +119,20 @@ func (d *Daemon) runDurableMaintenanceTick(ctx context.Context) {
119	119		if ok && !decision.RetentionResolved {
120	120		deps.repairUncertain[record.IncarnationID] = struct{}{}
121	121		}
122		-	d.log.Info("durable_reconciliation_decision",
123		-	"incarnation", record.IncarnationID.String(),
	122	+	// The catalogue lookup can miss (e.g. an unknown-incarnation finding for a
	123	+	// name the catalogue never held); logging record.IncarnationID.String() in
	124	+	// that case would emit a misleading all-zero incarnation, so the field is
	125	+	// omitted rather than logged as a fabricated identity.
	126	+	args := []any{
124	127		"session", decision.Name,
125	128		"action", decision.Kind,
126	129		"reason_code", decision.ReasonCode,
127	130		"cursor", next.DirectoryCookie,
128		-	)
	131	+	}
	132	+	if ok {
	133	+	args = append([]any{"incarnation", record.IncarnationID.String()}, args...)
	134	+	}
	135	+	d.log.Info("durable_reconciliation_decision", args...)
129	136		}
130	137		d.log.Info("durable_reconciliation_tick", "action", "scan", "cursor", next.DirectoryCookie, "done", next.DirectoryCookie == 0, "err", nil)
131	138		if next.DirectoryCookie != 0 {

 		if ok && !decision.RetentionResolved {
 			deps.repairUncertain[record.IncarnationID] = struct{}{}
 		}
-		d.log.Info("durable_reconciliation_decision",
+		// The catalogue lookup can miss (e.g. an unknown-incarnation finding for a
-			"incarnation", record.IncarnationID.String(),
+		// name the catalogue never held); logging record.IncarnationID.String() in
+		// that case would emit a misleading all-zero incarnation, so the field is
+		// omitted rather than logged as a fabricated identity.
+		args := []any{
 			"session", decision.Name,
 			"action", decision.Kind,
 			"reason_code", decision.ReasonCode,
 			"cursor", next.DirectoryCookie,
-		)
+		}
+		if ok {
+			args = append([]any{"incarnation", record.IncarnationID.String()}, args...)
+		}
+		d.log.Info("durable_reconciliation_decision", args...)
 	}
 	d.log.Info("durable_reconciliation_tick", "action", "scan", "cursor", next.DirectoryCookie, "done", next.DirectoryCookie == 0, "err", nil)
 	if next.DirectoryCookie != 0 {

M internal/usecase/daemon/session_test.go +29 -0


@@ -851,6 +851,35 @@ func TestDaemonLoadsPersistedSessionsAsStopped(t *testing.T) {
851	851		require.Equal(t, uint64(9), d.mruSeq.Load())
852	852		}
853	853
	854	+	// TestDaemonNewWithoutPersistenceLogsNoLoadWarning covers the default,
	855	+	// persistence-free construction path (no WithStore/WithCatalogue option):
	856	+	// d.persist is the persist.New(nil) sentinel, so LoadAll now returns
	857	+	// errPersistenceUnavailable instead of an empty slice. That must not surface
	858	+	// as a "loading persisted sessions failed" warning, since nothing failed —
	859	+	// persistence simply was never configured.
	860	+	func TestDaemonNewWithoutPersistenceLogsNoLoadWarning(t *testing.T) {
	861	+	var logBuf strings.Builder
	862	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(&logBuf, nil)))
	863	+	require.NotNil(t, d)
	864	+	require.False(t, d.persistEnabled)
	865	+	require.Empty(t, d.stopped)
	866	+	require.NotContains(t, logBuf.String(), "loading persisted sessions failed")
	867	+	}
	868	+
	869	+	// TestDaemonNewWithConfiguredStoreStillLogsLoadFailure is the inverse of the
	870	+	// above: when persistence IS configured (a real store via WithStore) and the
	871	+	// load genuinely fails (an undecodable catalogue record here), the warning
	872	+	// must still fire.
	873	+	func TestDaemonNewWithConfiguredStoreStillLogsLoadFailure(t *testing.T) {
	874	+	store, state := newMockStore(t)
	875	+	state.data["bad"] = []byte{0, 1} // not a valid encoded catalogue record
	876	+	var logBuf strings.Builder
	877	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(&logBuf, nil)), WithStore(store))
	878	+	require.NotNil(t, d)
	879	+	require.True(t, d.persistEnabled)
	880	+	require.Contains(t, logBuf.String(), "loading persisted sessions failed")
	881	+	}
	882	+
854	883		func TestTouchMRUConcurrentUpdatesRemainMonotonic(t *testing.T) {
855	884		d := &Daemon{}
856	885		sess := &session{}


@@ -851,6 +851,35 @@ func TestDaemonLoadsPersistedSessionsAsStopped(t *testing.T) {
851	require.Equal(t, uint64(9), d.mruSeq.Load())	851		require.Equal(t, uint64(9), d.mruSeq.Load())
852	}	852		}
853		853
		854	+	// TestDaemonNewWithoutPersistenceLogsNoLoadWarning covers the default,
		855	+	// persistence-free construction path (no WithStore/WithCatalogue option):
		856	+	// d.persist is the persist.New(nil) sentinel, so LoadAll now returns
		857	+	// errPersistenceUnavailable instead of an empty slice. That must not surface
		858	+	// as a "loading persisted sessions failed" warning, since nothing failed —
		859	+	// persistence simply was never configured.
		860	+	func TestDaemonNewWithoutPersistenceLogsNoLoadWarning(t *testing.T) {
		861	+	var logBuf strings.Builder
		862	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(&logBuf, nil)))
		863	+	require.NotNil(t, d)
		864	+	require.False(t, d.persistEnabled)
		865	+	require.Empty(t, d.stopped)
		866	+	require.NotContains(t, logBuf.String(), "loading persisted sessions failed")
		867	+	}
		868	+
		869	+	// TestDaemonNewWithConfiguredStoreStillLogsLoadFailure is the inverse of the
		870	+	// above: when persistence IS configured (a real store via WithStore) and the
		871	+	// load genuinely fails (an undecodable catalogue record here), the warning
		872	+	// must still fire.
		873	+	func TestDaemonNewWithConfiguredStoreStillLogsLoadFailure(t *testing.T) {
		874	+	store, state := newMockStore(t)
		875	+	state.data["bad"] = []byte{0, 1} // not a valid encoded catalogue record
		876	+	var logBuf strings.Builder
		877	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(&logBuf, nil)), WithStore(store))
		878	+	require.NotNil(t, d)
		879	+	require.True(t, d.persistEnabled)
		880	+	require.Contains(t, logBuf.String(), "loading persisted sessions failed")
		881	+	}
		882	+
854	func TestTouchMRUConcurrentUpdatesRemainMonotonic(t *testing.T) {	883		func TestTouchMRUConcurrentUpdatesRemainMonotonic(t *testing.T) {
855	d := &Daemon{}	884		d := &Daemon{}
856	sess := &session{}	885		sess := &session{}

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


@@ -16,6 +16,30 @@ func (d *Daemon) closeRestoreDone() {
16	16		d.restoreOnce.Do(func() { close(d.restoreDone) })
17	17		}
18	18
	19	+	// startSnapshotRestoration launches catalogue-driven restoration as a durable
	20	+	// writer. Restoration repairs HEADs, promotes fallbacks, and replaces catalogue
	21	+	// records, so lifecycle ownership must outlive it exactly as it outlives the
	22	+	// snapshot and maintenance workers. The completion channel is registered before
	23	+	// the goroutine starts, so shutdown can never begin waiting while this writer
	24	+	// is still unregistered.
	25	+	func (d *Daemon) startSnapshotRestoration() {
	26	+	if d == nil {
	27	+	return
	28	+	}
	29	+	d.snapshotWorkerMu.Lock()
	30	+	if d.restoreWorkerDone != nil {
	31	+	d.snapshotWorkerMu.Unlock()
	32	+	return
	33	+	}
	34	+	done := make(chan struct{})
	35	+	d.restoreWorkerDone = done
	36	+	d.snapshotWorkerMu.Unlock()
	37	+	d.sessWg.Go(func() {
	38	+	defer close(done)
	39	+	d.restoreSnapshots(d.serveCtx)
	40	+	})
	41	+	}
	42	+
19	43		func (d *Daemon) restoreSnapshots(ctx context.Context) {
20	44		if d == nil {
21	45		return


@@ -16,6 +16,30 @@ func (d *Daemon) closeRestoreDone() {
16	d.restoreOnce.Do(func() { close(d.restoreDone) })	16		d.restoreOnce.Do(func() { close(d.restoreDone) })
17	}	17		}
18		18
		19	+	// startSnapshotRestoration launches catalogue-driven restoration as a durable
		20	+	// writer. Restoration repairs HEADs, promotes fallbacks, and replaces catalogue
		21	+	// records, so lifecycle ownership must outlive it exactly as it outlives the
		22	+	// snapshot and maintenance workers. The completion channel is registered before
		23	+	// the goroutine starts, so shutdown can never begin waiting while this writer
		24	+	// is still unregistered.
		25	+	func (d *Daemon) startSnapshotRestoration() {
		26	+	if d == nil {
		27	+	return
		28	+	}
		29	+	d.snapshotWorkerMu.Lock()
		30	+	if d.restoreWorkerDone != nil {
		31	+	d.snapshotWorkerMu.Unlock()
		32	+	return
		33	+	}
		34	+	done := make(chan struct{})
		35	+	d.restoreWorkerDone = done
		36	+	d.snapshotWorkerMu.Unlock()
		37	+	d.sessWg.Go(func() {
		38	+	defer close(done)
		39	+	d.restoreSnapshots(d.serveCtx)
		40	+	})
		41	+	}
		42	+
19	func (d *Daemon) restoreSnapshots(ctx context.Context) {	43		func (d *Daemon) restoreSnapshots(ctx context.Context) {
20	if d == nil {	44		if d == nil {
21	return	45		return

M internal/usecase/daemon/snapshot_worker.go +9 -0


@@ -224,6 +224,9 @@ func (d *Daemon) StopDurableWriters(ctx context.Context) []string {
224	224
225	225		// WaitDurableWriters is the unconditional ownership barrier. It has no timeout
226	226		// or status branch: every writer goroutine must exit before Serve may return.
	227	+	// Restoration is joined with the others because it mutates durable state too;
	228	+	// it observes serveCtx cancellation, so waiting for it cannot outlive an
	229	+	// uncooperative repository call any longer than the snapshot worker does.
227	230		func (d *Daemon) WaitDurableWriters() {
228	231		if d == nil {
229	232		return

@@ -231,6 +234,7 @@ func (d *Daemon) WaitDurableWriters() {
231	234		d.snapshotWorkerMu.Lock()
232	235		snapshotDone := d.snapshotWorkerDone
233	236		maintenanceDone := d.maintenanceWorkerDone
	237	+	restoreDone := d.restoreWorkerDone
234	238		d.snapshotWorkerMu.Unlock()
235	239		if snapshotDone != nil {
236	240		<-snapshotDone

@@ -239,6 +243,11 @@ func (d *Daemon) WaitDurableWriters() {
239	243		if maintenanceDone != nil {
240	244		<-maintenanceDone
241	245		}
	246	+	// Restoration never blocks on the workers above (snapshot admission is
	247	+	// non-blocking), so joining it last cannot deadlock.
	248	+	if restoreDone != nil {
	249	+	<-restoreDone
	250	+	}
242	251		}
243	252
244	253		func (d *Daemon) requestDurableWriterStop() (context.CancelFunc, <-chan struct{}) {


@@ -224,6 +224,9 @@ func (d *Daemon) StopDurableWriters(ctx context.Context) []string {
224		224
225	// WaitDurableWriters is the unconditional ownership barrier. It has no timeout	225		// WaitDurableWriters is the unconditional ownership barrier. It has no timeout
226	// or status branch: every writer goroutine must exit before Serve may return.	226		// or status branch: every writer goroutine must exit before Serve may return.
		227	+	// Restoration is joined with the others because it mutates durable state too;
		228	+	// it observes serveCtx cancellation, so waiting for it cannot outlive an
		229	+	// uncooperative repository call any longer than the snapshot worker does.
227	func (d *Daemon) WaitDurableWriters() {	230		func (d *Daemon) WaitDurableWriters() {
228	if d == nil {	231		if d == nil {
229	return	232		return

@@ -231,6 +234,7 @@ func (d *Daemon) WaitDurableWriters() {
231	d.snapshotWorkerMu.Lock()	234		d.snapshotWorkerMu.Lock()
232	snapshotDone := d.snapshotWorkerDone	235		snapshotDone := d.snapshotWorkerDone
233	maintenanceDone := d.maintenanceWorkerDone	236		maintenanceDone := d.maintenanceWorkerDone
		237	+	restoreDone := d.restoreWorkerDone
234	d.snapshotWorkerMu.Unlock()	238		d.snapshotWorkerMu.Unlock()
235	if snapshotDone != nil {	239		if snapshotDone != nil {
236	<-snapshotDone	240		<-snapshotDone

@@ -239,6 +243,11 @@ func (d *Daemon) WaitDurableWriters() {
239	if maintenanceDone != nil {	243		if maintenanceDone != nil {
240	<-maintenanceDone	244		<-maintenanceDone
241	}	245		}
		246	+	// Restoration never blocks on the workers above (snapshot admission is
		247	+	// non-blocking), so joining it last cannot deadlock.
		248	+	if restoreDone != nil {
		249	+	<-restoreDone
		250	+	}
242	}	251		}
243		252
244	func (d *Daemon) requestDurableWriterStop() (context.CancelFunc, <-chan struct{}) {	253		func (d *Daemon) requestDurableWriterStop() (context.CancelFunc, <-chan struct{}) {