fix(daemon): join maintenance before owner release

commit: 18a09bbfb00cc592e9c9c2c0419500bf613e1fb8
author: brice · 2mo ago · 2026-07-26 02:23 UTC
parent: 9297910
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +145 -20

M internal/app/daemon_integration_test.go +95 -0


@@ -529,6 +529,101 @@ func TestKillDaemonWaitsForOwnershipTransfer(t *testing.T) {
529	529		require.NoError(t, <-done)
530	530		}
531	531
	532	+	func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
	533	+	runtimeDir := filepath.Join(t.TempDir(), "runtime")
	534	+	stateDir := filepath.Join(t.TempDir(), "state")
	535	+	catalogue := &lifecycleObservedCatalogue{
	536	+	Catalogue: newTestPersister(t, stateDir),
	537	+	closed: make(chan struct{}),
	538	+	}
	539	+	repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
	540	+	shutdownClock := newLifecycleShutdownClock()
	541	+	reconciler := &lifecycleBlockingReconciler{
	542	+	entered: make(chan struct{}),
	543	+	release: make(chan struct{}),
	544	+	returned: make(chan struct{}),
	545	+	}
	546	+	ctx, cancel := context.WithCancel(context.Background())
	547	+	t.Cleanup(cancel)
	548	+	var releaseOnce sync.Once
	549	+	releaseMaintenance := func() { releaseOnce.Do(func() { close(reconciler.release) }) }
	550	+	t.Cleanup(releaseMaintenance)
	551	+
	552	+	listenerReady := make(chan struct{})
	553	+	callbackReturned := make(chan struct{})
	554	+	wrapperReturned := make(chan error, 1)
	555	+	go func() {
	556	+	wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
	557	+	defer close(callbackReturned)
	558	+	listener, err := ipc.Listen(runtimeDir)
	559	+	if err != nil {
	560	+	return err
	561	+	}
	562	+	close(listenerReady)
	563	+	d := daemon.New(
	564	+	pty.NewFactory(),
	565	+	shutdownClock,
	566	+	discardLog(),
	567	+	daemon.WithSnapshotRepository(repository, nil),
	568	+	daemon.WithDurableMaintenance(catalogue, repository, reconciler, nil),
	569	+	daemon.WithCatalogue(catalogue, nil),
	570	+	)
	571	+	return d.Serve(ctx, listener)
	572	+	})
	573	+	}()
	574	+
	575	+	awaitLifecycleStage(t, listenerReady, "daemon listener")
	576	+	awaitLifecycleStage(t, reconciler.entered, "maintenance repository call")
	577	+	cancel()
	578	+	shutdownClock.nextTimer(t).fire()
	579	+	select {
	580	+	case <-callbackReturned:
	581	+	t.Fatal("Serve callback returned after shutdown deadline while maintenance was blocked")
	582	+	case <-time.After(100 * time.Millisecond):
	583	+	}
	584	+	assertLifecycleStagePending(t, callbackReturned, "Serve callback return while maintenance is blocked")
	585	+	assertLifecycleStagePending(t, catalogue.closed, "catalogue close while maintenance is blocked")
	586	+	assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while maintenance is blocked")
	587	+	_, err := lifecycle.TryAcquire(runtimeDir)
	588	+	require.ErrorIs(t, err, lifecycle.ErrBusy)
	589	+
	590	+	releaseMaintenance()
	591	+	awaitLifecycleStage(t, reconciler.returned, "maintenance repository return")
	592	+	awaitLifecycleStage(t, catalogue.closed, "catalogue close after maintenance")
	593	+	awaitLifecycleStage(t, callbackReturned, "Serve callback return after catalogue close")
	594	+	require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
	595	+	owner, err := lifecycle.TryAcquire(runtimeDir)
	596	+	require.NoError(t, err)
	597	+	require.NoError(t, owner.Release())
	598	+	}
	599	+
	600	+	type lifecycleBlockingReconciler struct {
	601	+	entered chan struct{}
	602	+	release chan struct{}
	603	+	returned chan struct{}
	604	+	enterOnce sync.Once
	605	+	returnOnce sync.Once
	606	+	}
	607	+
	608	+	func (r *lifecycleBlockingReconciler) Step(context.Context, ports.ReconcileCursor) (ports.ReconcileCursor, []ports.ReconcileDecision, error) {
	609	+	r.enterOnce.Do(func() { close(r.entered) })
	610	+	<-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
	611	+	r.returnOnce.Do(func() { close(r.returned) })
	612	+	return ports.ReconcileCursor{}, nil, nil
	613	+	}
	614	+
	615	+	type lifecycleObservedCatalogue struct {
	616	+	ports.Catalogue
	617	+	closed chan struct{}
	618	+	closeOnce sync.Once
	619	+	}
	620	+
	621	+	func (c *lifecycleObservedCatalogue) Close() error {
	622	+	err := c.Catalogue.Close()
	623	+	c.closeOnce.Do(func() { close(c.closed) })
	624	+	return err
	625	+	}
	626	+
532	627		func TestLifecycleOwnershipOutlivesSnapshotWriter(t *testing.T) {
533	628		result := runBlockedSnapshotWriterShutdown(t)
534	629		assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")


@@ -529,6 +529,101 @@ func TestKillDaemonWaitsForOwnershipTransfer(t *testing.T) {
529	require.NoError(t, <-done)	529		require.NoError(t, <-done)
530	}	530		}
531		531
		532	+	func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
		533	+	runtimeDir := filepath.Join(t.TempDir(), "runtime")
		534	+	stateDir := filepath.Join(t.TempDir(), "state")
		535	+	catalogue := &lifecycleObservedCatalogue{
		536	+	Catalogue: newTestPersister(t, stateDir),
		537	+	closed: make(chan struct{}),
		538	+	}
		539	+	repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
		540	+	shutdownClock := newLifecycleShutdownClock()
		541	+	reconciler := &lifecycleBlockingReconciler{
		542	+	entered: make(chan struct{}),
		543	+	release: make(chan struct{}),
		544	+	returned: make(chan struct{}),
		545	+	}
		546	+	ctx, cancel := context.WithCancel(context.Background())
		547	+	t.Cleanup(cancel)
		548	+	var releaseOnce sync.Once
		549	+	releaseMaintenance := func() { releaseOnce.Do(func() { close(reconciler.release) }) }
		550	+	t.Cleanup(releaseMaintenance)
		551	+
		552	+	listenerReady := make(chan struct{})
		553	+	callbackReturned := make(chan struct{})
		554	+	wrapperReturned := make(chan error, 1)
		555	+	go func() {
		556	+	wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
		557	+	defer close(callbackReturned)
		558	+	listener, err := ipc.Listen(runtimeDir)
		559	+	if err != nil {
		560	+	return err
		561	+	}
		562	+	close(listenerReady)
		563	+	d := daemon.New(
		564	+	pty.NewFactory(),
		565	+	shutdownClock,
		566	+	discardLog(),
		567	+	daemon.WithSnapshotRepository(repository, nil),
		568	+	daemon.WithDurableMaintenance(catalogue, repository, reconciler, nil),
		569	+	daemon.WithCatalogue(catalogue, nil),
		570	+	)
		571	+	return d.Serve(ctx, listener)
		572	+	})
		573	+	}()
		574	+
		575	+	awaitLifecycleStage(t, listenerReady, "daemon listener")
		576	+	awaitLifecycleStage(t, reconciler.entered, "maintenance repository call")
		577	+	cancel()
		578	+	shutdownClock.nextTimer(t).fire()
		579	+	select {
		580	+	case <-callbackReturned:
		581	+	t.Fatal("Serve callback returned after shutdown deadline while maintenance was blocked")
		582	+	case <-time.After(100 * time.Millisecond):
		583	+	}
		584	+	assertLifecycleStagePending(t, callbackReturned, "Serve callback return while maintenance is blocked")
		585	+	assertLifecycleStagePending(t, catalogue.closed, "catalogue close while maintenance is blocked")
		586	+	assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while maintenance is blocked")
		587	+	_, err := lifecycle.TryAcquire(runtimeDir)
		588	+	require.ErrorIs(t, err, lifecycle.ErrBusy)
		589	+
		590	+	releaseMaintenance()
		591	+	awaitLifecycleStage(t, reconciler.returned, "maintenance repository return")
		592	+	awaitLifecycleStage(t, catalogue.closed, "catalogue close after maintenance")
		593	+	awaitLifecycleStage(t, callbackReturned, "Serve callback return after catalogue close")
		594	+	require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
		595	+	owner, err := lifecycle.TryAcquire(runtimeDir)
		596	+	require.NoError(t, err)
		597	+	require.NoError(t, owner.Release())
		598	+	}
		599	+
		600	+	type lifecycleBlockingReconciler struct {
		601	+	entered chan struct{}
		602	+	release chan struct{}
		603	+	returned chan struct{}
		604	+	enterOnce sync.Once
		605	+	returnOnce sync.Once
		606	+	}
		607	+
		608	+	func (r *lifecycleBlockingReconciler) Step(context.Context, ports.ReconcileCursor) (ports.ReconcileCursor, []ports.ReconcileDecision, error) {
		609	+	r.enterOnce.Do(func() { close(r.entered) })
		610	+	<-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
		611	+	r.returnOnce.Do(func() { close(r.returned) })
		612	+	return ports.ReconcileCursor{}, nil, nil
		613	+	}
		614	+
		615	+	type lifecycleObservedCatalogue struct {
		616	+	ports.Catalogue
		617	+	closed chan struct{}
		618	+	closeOnce sync.Once
		619	+	}
		620	+
		621	+	func (c *lifecycleObservedCatalogue) Close() error {
		622	+	err := c.Catalogue.Close()
		623	+	c.closeOnce.Do(func() { close(c.closed) })
		624	+	return err
		625	+	}
		626	+
532	func TestLifecycleOwnershipOutlivesSnapshotWriter(t *testing.T) {	627		func TestLifecycleOwnershipOutlivesSnapshotWriter(t *testing.T) {
533	result := runBlockedSnapshotWriterShutdown(t)	628		result := runBlockedSnapshotWriterShutdown(t)
534	assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")	629		assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")

M internal/usecase/daemon/daemon.go +12 -12


@@ -124,15 +124,17 @@ type Daemon struct {
124	124		catalogueRecordsProvided bool
125	125		// snapshotRepository is the sole checkpoint contract. legacySnapshots is
126	126		// read-only migration input and is never used for new writes.
127		-	snapshotRepository ports.SnapshotRepository
128		-	checkpointRecovery ports.CheckpointCoordinator
129		-	lifecycleRecovery ports.SessionLifecycleCoordinator
130		-	degradedRecovery ports.DegradedRecoveryCoordinator
131		-	maintenance maintenanceDependencies
132		-	legacySnapshots ports.LegacySnapshotSource
133		-	snapsEnabled bool
134		-	noticeStore ports.NoticeStore
135		-	snapshotJobs chan *snapshotCapture
	127	+	snapshotRepository ports.SnapshotRepository
	128	+	checkpointRecovery ports.CheckpointCoordinator
	129	+	lifecycleRecovery ports.SessionLifecycleCoordinator
	130	+	degradedRecovery ports.DegradedRecoveryCoordinator
	131	+	maintenance maintenanceDependencies
	132	+	maintenanceWorkerCancel context.CancelFunc
	133	+	maintenanceWorkerDone chan struct{}
	134	+	legacySnapshots ports.LegacySnapshotSource
	135	+	snapsEnabled bool
	136	+	noticeStore ports.NoticeStore
	137	+	snapshotJobs chan *snapshotCapture
136	138		// snapshotWake wakes the repository scheduler when a session becomes dirty
137	139		// or an attempt completes. It is never closed and producers only send
138	140		// non-blockingly.

@@ -546,9 +548,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
546	548		d.closeRestoreDone()
547	549		}
548	550		if d.maintenance.reconciler != nil {
549		-	d.sessWg.Go(func() {
550		-	d.runDurableMaintenance(d.serveCtx)
551		-	})
	551	+	d.startDurableMaintenance()
552	552		}
553	553
554	554		// Break the accept loop when either the parent context is cancelled or the

 	catalogueRecordsProvided       bool
 	// snapshotRepository is the sole checkpoint contract. legacySnapshots is
 	// read-only migration input and is never used for new writes.
-	snapshotRepository ports.SnapshotRepository
+	snapshotRepository      ports.SnapshotRepository
-	checkpointRecovery ports.CheckpointCoordinator
+	checkpointRecovery      ports.CheckpointCoordinator
-	lifecycleRecovery  ports.SessionLifecycleCoordinator
+	lifecycleRecovery       ports.SessionLifecycleCoordinator
-	degradedRecovery   ports.DegradedRecoveryCoordinator
+	degradedRecovery        ports.DegradedRecoveryCoordinator
-	maintenance        maintenanceDependencies
+	maintenance             maintenanceDependencies
-	legacySnapshots    ports.LegacySnapshotSource
+	maintenanceWorkerCancel context.CancelFunc
-	snapsEnabled       bool
+	maintenanceWorkerDone   chan struct{}
-	noticeStore        ports.NoticeStore
+	legacySnapshots         ports.LegacySnapshotSource
-	snapshotJobs       chan *snapshotCapture
+	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.closeRestoreDone()
 	}
 	if d.maintenance.reconciler != nil {
-		d.sessWg.Go(func() {
+		d.startDurableMaintenance()
-			d.runDurableMaintenance(d.serveCtx)
-		})
 	}
 	// Break the accept loop when either the parent context is cancelled or the

M internal/usecase/daemon/maintenance.go +23 -0


@@ -58,6 +58,29 @@ func retentionPlan(record domain.CatalogueRecord, unresolved, restoring bool) po
58	58		return plan
59	59		}
60	60
	61	+	func (d *Daemon) startDurableMaintenance() {
	62	+	if d == nil {
	63	+	return
	64	+	}
	65	+	d.snapshotWorkerMu.Lock()
	66	+	if d.maintenanceWorkerCancel != nil {
	67	+	d.snapshotWorkerMu.Unlock()
	68	+	return
	69	+	}
	70	+	ctx, cancel := context.WithCancel(d.serveCtx)
	71	+	d.maintenanceWorkerCancel = cancel
	72	+	d.maintenanceWorkerDone = make(chan struct{})
	73	+	done := d.maintenanceWorkerDone
	74	+	d.snapshotWorkerMu.Unlock()
	75	+
	76	+	// Register the completion channel before launch so shutdown can never begin
	77	+	// waiting while this durable writer is still unregistered.
	78	+	go func() {
	79	+	defer close(done)
	80	+	d.runDurableMaintenance(ctx)
	81	+	}()
	82	+	}
	83	+
61	84		func (d *Daemon) runDurableMaintenance(ctx context.Context) {
62	85		select {
63	86		case <-ctx.Done():


@@ -58,6 +58,29 @@ func retentionPlan(record domain.CatalogueRecord, unresolved, restoring bool) po
58	return plan	58		return plan
59	}	59		}
60		60
		61	+	func (d *Daemon) startDurableMaintenance() {
		62	+	if d == nil {
		63	+	return
		64	+	}
		65	+	d.snapshotWorkerMu.Lock()
		66	+	if d.maintenanceWorkerCancel != nil {
		67	+	d.snapshotWorkerMu.Unlock()
		68	+	return
		69	+	}
		70	+	ctx, cancel := context.WithCancel(d.serveCtx)
		71	+	d.maintenanceWorkerCancel = cancel
		72	+	d.maintenanceWorkerDone = make(chan struct{})
		73	+	done := d.maintenanceWorkerDone
		74	+	d.snapshotWorkerMu.Unlock()
		75	+
		76	+	// Register the completion channel before launch so shutdown can never begin
		77	+	// waiting while this durable writer is still unregistered.
		78	+	go func() {
		79	+	defer close(done)
		80	+	d.runDurableMaintenance(ctx)
		81	+	}()
		82	+	}
		83	+
61	func (d *Daemon) runDurableMaintenance(ctx context.Context) {	84		func (d *Daemon) runDurableMaintenance(ctx context.Context) {
62	select {	85		select {
63	case <-ctx.Done():	86		case <-ctx.Done():

M internal/usecase/daemon/snapshot_worker.go +15 -8


@@ -198,9 +198,10 @@ func (d *Daemon) stopSnapshotEncodeWorker() {
198	198		d.WaitDurableWriters()
199	199		}
200	200
201		-	// StopDurableWriters stops admission and requests the worker's final drain.
202		-	// The context bounds checkpoint success only: cancellation asks cooperative
203		-	// repository calls to stop, but it never detaches the worker from its owner.
	201	+	// StopDurableWriters stops maintenance scheduling, cancels cooperative
	202	+	// maintenance calls, stops snapshot admission, and requests the snapshot
	203	+	// worker's final drain. The context bounds checkpoint success only:
	204	+	// cancellation never detaches a writer from its owner.
204	205		func (d *Daemon) StopDurableWriters(ctx context.Context) []string {
205	206		if d == nil {
206	207		return nil

@@ -228,18 +229,24 @@ func (d *Daemon) WaitDurableWriters() {
228	229		return
229	230		}
230	231		d.snapshotWorkerMu.Lock()
231		-	done := d.snapshotWorkerDone
	232	+	snapshotDone := d.snapshotWorkerDone
	233	+	maintenanceDone := d.maintenanceWorkerDone
232	234		d.snapshotWorkerMu.Unlock()
233		-	if done == nil {
234		-	return
	235	+	if snapshotDone != nil {
	236	+	<-snapshotDone
	237	+	d.finishStoppedSnapshotWorker(false)
	238	+	}
	239	+	if maintenanceDone != nil {
	240	+	<-maintenanceDone
235	241		}
236		-	<-done
237		-	d.finishStoppedSnapshotWorker(false)
238	242		}
239	243
240	244		func (d *Daemon) requestDurableWriterStop() (context.CancelFunc, <-chan struct{}) {
241	245		d.snapshotWorkerMu.Lock()
242	246		defer d.snapshotWorkerMu.Unlock()
	247	+	if d.maintenanceWorkerCancel != nil {
	248	+	d.maintenanceWorkerCancel()
	249	+	}
243	250		cancel := d.snapshotWorkerCancel
244	251		if cancel == nil {
245	252		return nil, nil

 	d.WaitDurableWriters()
 }
-// StopDurableWriters stops admission and requests the worker's final drain.
+// StopDurableWriters stops maintenance scheduling, cancels cooperative
-// The context bounds checkpoint success only: cancellation asks cooperative
+// maintenance calls, stops snapshot admission, and requests the snapshot
-// repository calls to stop, but it never detaches the worker from its owner.
+// worker's final drain. The context bounds checkpoint success only:
+// cancellation never detaches a writer from its owner.
 func (d *Daemon) StopDurableWriters(ctx context.Context) []string {
 	if d == nil {
 		return nil
 		return
 	}
 	d.snapshotWorkerMu.Lock()
-	done := d.snapshotWorkerDone
+	snapshotDone := d.snapshotWorkerDone
+	maintenanceDone := d.maintenanceWorkerDone
 	d.snapshotWorkerMu.Unlock()
-	if done == nil {
+	if snapshotDone != nil {
-		return
+		<-snapshotDone
+		d.finishStoppedSnapshotWorker(false)
+	}
+	if maintenanceDone != nil {
+		<-maintenanceDone
 	}
-	<-done
-	d.finishStoppedSnapshotWorker(false)
 }
 func (d *Daemon) requestDurableWriterStop() (context.CancelFunc, <-chan struct{}) {
 	d.snapshotWorkerMu.Lock()
 	defer d.snapshotWorkerMu.Unlock()
+	if d.maintenanceWorkerCancel != nil {
+		d.maintenanceWorkerCancel()
+	}
 	cancel := d.snapshotWorkerCancel
 	if cancel == nil {
 		return nil, nil