fix(daemon): retain ownership until snapshot writers stop

commit: df0f462a0e3044177591a40e7f175a513df4da67
author: brice · 2mo ago · 2026-07-26 01:53 UTC
parent: 3c189af
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +310 -40

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


@@ -529,6 +529,198 @@ func TestKillDaemonWaitsForOwnershipTransfer(t *testing.T) {
529	529		require.NoError(t, <-done)
530	530		}
531	531
	532	+	func TestLifecycleOwnershipOutlivesSnapshotWriter(t *testing.T) {
	533	+	result := runBlockedSnapshotWriterShutdown(t)
	534	+	assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
	535	+	assertLifecycleResultPending(t, result.wrapperReturned, "lifecycle wrapper return while snapshot writer is blocked")
	536	+	owner, err := lifecycle.TryAcquire(result.runtimeDir)
	537	+	require.ErrorIs(t, err, lifecycle.ErrBusy)
	538	+
	539	+	result.releaseWriter()
	540	+	awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
	541	+	require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
	542	+	owner, err = lifecycle.TryAcquire(result.runtimeDir)
	543	+	require.NoError(t, err)
	544	+	require.NoError(t, owner.Release())
	545	+	}
	546	+
	547	+	func TestLifecycleCallbackWaitsForEveryWriter(t *testing.T) {
	548	+	result := runBlockedSnapshotWriterShutdown(t)
	549	+	assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
	550	+
	551	+	result.releaseWriter()
	552	+	awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
	553	+	awaitLifecycleStage(t, result.callbackReturned, "Serve callback return after snapshot writer")
	554	+	require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
	555	+	}
	556	+
	557	+	func TestFinalCheckpointTimeoutKeepsOwner(t *testing.T) {
	558	+	result := runBlockedSnapshotWriterShutdown(t)
	559	+	assertLifecycleResultPending(t, result.wrapperReturned, "ownership release after checkpoint timeout with writer alive")
	560	+	_, err := lifecycle.TryAcquire(result.runtimeDir)
	561	+	require.ErrorIs(t, err, lifecycle.ErrBusy)
	562	+
	563	+	result.releaseWriter()
	564	+	require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
	565	+	}
	566	+
	567	+	type blockedSnapshotWriterShutdown struct {
	568	+	runtimeDir string
	569	+	callbackReturned <-chan struct{}
	570	+	wrapperReturned <-chan error
	571	+	publishReturned <-chan struct{}
	572	+	releaseWriter func()
	573	+	}
	574	+
	575	+	func runBlockedSnapshotWriterShutdown(t *testing.T) blockedSnapshotWriterShutdown {
	576	+	t.Helper()
	577	+	runtimeDir := filepath.Join(t.TempDir(), "runtime")
	578	+	stateDir := filepath.Join(t.TempDir(), "state")
	579	+	shutdownClock := newLifecycleShutdownClock()
	580	+	repository := newLifecycleBlockingRepository()
	581	+	ctx, cancel := context.WithCancel(context.Background())
	582	+	t.Cleanup(cancel)
	583	+
	584	+	listenerReady := make(chan struct{})
	585	+	listenerClosed := make(chan struct{})
	586	+	callbackReturned := make(chan struct{})
	587	+	wrapperReturned := make(chan error, 1)
	588	+	go func() {
	589	+	wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
	590	+	defer close(callbackReturned)
	591	+	listener, err := ipc.Listen(runtimeDir)
	592	+	if err != nil {
	593	+	return err
	594	+	}
	595	+	observed := &lifecycleObservedListener{Listener: listener, closed: listenerClosed}
	596	+	close(listenerReady)
	597	+	d := daemon.New(
	598	+	pty.NewFactory(),
	599	+	shutdownClock,
	600	+	discardLog(),
	601	+	daemon.WithShell("/bin/cat", nil),
	602	+	daemon.WithSnapshotRepository(repository, nil),
	603	+	)
	604	+	return d.Serve(ctx, observed)
	605	+	})
	606	+	}()
	607	+
	608	+	awaitLifecycleStage(t, listenerReady, "daemon listener")
	609	+	tr, _ := attach(t, runtimeDir, ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24})
	610	+	require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("dirty state\n")})}))
	611	+	awaitLifecycleStage(t, repository.entered, "snapshot publication")
	612	+	require.NoError(t, tr.Close())
	613	+	require.Eventually(t, func() bool {
	614	+	sessions := listRemoteSessions(t, runtimeDir)
	615	+	return len(sessions.Sessions) == 1 && !sessions.Sessions[0].Attached
	616	+	}, time.Second, 10*time.Millisecond)
	617	+	cancel()
	618	+	awaitLifecycleStage(t, listenerClosed, "listener close")
	619	+
	620	+	shutdownClock.nextTimer(t).fire()
	621	+	select {
	622	+	case <-callbackReturned:
	623	+	t.Fatal("Serve callback returned after checkpoint timeout while snapshot writer was blocked")
	624	+	case <-time.After(100 * time.Millisecond):
	625	+	}
	626	+	assertLifecycleStagePending(t, callbackReturned, "Serve callback return after checkpoint timeout")
	627	+	assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return after checkpoint timeout")
	628	+
	629	+	var releaseOnce sync.Once
	630	+	releaseWriter := func() { releaseOnce.Do(func() { close(repository.release) }) }
	631	+	t.Cleanup(releaseWriter)
	632	+	return blockedSnapshotWriterShutdown{
	633	+	runtimeDir: runtimeDir,
	634	+	callbackReturned: callbackReturned,
	635	+	wrapperReturned: wrapperReturned,
	636	+	publishReturned: repository.returned,
	637	+	releaseWriter: releaseWriter,
	638	+	}
	639	+	}
	640	+
	641	+	type lifecycleBlockingRepository struct {
	642	+	ports.SnapshotRepository
	643	+	entered chan struct{}
	644	+	release chan struct{}
	645	+	returned chan struct{}
	646	+	enterOnce sync.Once
	647	+	returnOnce sync.Once
	648	+	}
	649	+
	650	+	func newLifecycleBlockingRepository() *lifecycleBlockingRepository {
	651	+	return &lifecycleBlockingRepository{
	652	+	entered: make(chan struct{}),
	653	+	release: make(chan struct{}),
	654	+	returned: make(chan struct{}),
	655	+	}
	656	+	}
	657	+
	658	+	func (r *lifecycleBlockingRepository) Publish(context.Context, ports.SnapshotPublication) error {
	659	+	r.enterOnce.Do(func() { close(r.entered) })
	660	+	<-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
	661	+	r.returnOnce.Do(func() { close(r.returned) })
	662	+	return nil
	663	+	}
	664	+
	665	+	type lifecycleShutdownClock struct {
	666	+	base ports.Clock
	667	+	timers chan *lifecycleManualTimer
	668	+	}
	669	+
	670	+	func newLifecycleShutdownClock() *lifecycleShutdownClock {
	671	+	return &lifecycleShutdownClock{base: clock.New(), timers: make(chan *lifecycleManualTimer, 4)}
	672	+	}
	673	+
	674	+	func (c *lifecycleShutdownClock) Now() time.Time { return c.base.Now() }
	675	+	func (c *lifecycleShutdownClock) NewTimer(delay time.Duration) ports.Timer {
	676	+	if delay != time.Second {
	677	+	return c.base.NewTimer(delay)
	678	+	}
	679	+	timer := &lifecycleManualTimer{ch: make(chan time.Time, 1)}
	680	+	c.timers <- timer
	681	+	return timer
	682	+	}
	683	+
	684	+	func (c lifecycleShutdownClock) nextTimer(t testing.T) *lifecycleManualTimer {
	685	+	t.Helper()
	686	+	select {
	687	+	case timer := <-c.timers:
	688	+	return timer
	689	+	case <-time.After(3 * time.Second):
	690	+	t.Fatal("timed out waiting for shutdown deadline timer")
	691	+	return nil
	692	+	}
	693	+	}
	694	+
	695	+	type lifecycleManualTimer struct {
	696	+	ch chan time.Time
	697	+	stopped atomic.Bool
	698	+	}
	699	+
	700	+	func (t *lifecycleManualTimer) C() <-chan time.Time { return t.ch }
	701	+	func (t *lifecycleManualTimer) Reset(time.Duration) bool {
	702	+	t.stopped.Store(false)
	703	+	return false
	704	+	}
	705	+	func (t *lifecycleManualTimer) Stop() bool { return !t.stopped.Swap(true) }
	706	+	func (t *lifecycleManualTimer) fire() {
	707	+	if !t.stopped.Load() {
	708	+	t.ch <- time.Now()
	709	+	}
	710	+	}
	711	+
	712	+	type lifecycleObservedListener struct {
	713	+	ports.Listener
	714	+	closed chan struct{}
	715	+	once sync.Once
	716	+	}
	717	+
	718	+	func (l *lifecycleObservedListener) Close() error {
	719	+	err := l.Listener.Close()
	720	+	l.once.Do(func() { close(l.closed) })
	721	+	return err
	722	+	}
	723	+
532	724		func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {
533	725		runtimeDir := filepath.Join(t.TempDir(), "runtime")
534	726		stateDir := filepath.Join(t.TempDir(), "state")

@@ -690,6 +882,15 @@ func assertLifecycleStagePending(t *testing.T, stage <-chan struct{}, name strin
690	882		}
691	883		}
692	884
	885	+	func assertLifecycleResultPending(t *testing.T, result <-chan error, name string) {
	886	+	t.Helper()
	887	+	select {
	888	+	case err := <-result:
	889	+	t.Fatalf("unexpected %s: %v", name, err)
	890	+	default:
	891	+	}
	892	+	}
	893	+
693	894		func assertNewDaemonStartupPending(t *testing.T, durableOpen, listen <-chan struct{}, heldAt string) {
694	895		t.Helper()
695	896		assertLifecycleStagePending(t, durableOpen, "new durable open while old daemon held at "+heldAt)


@@ -529,6 +529,198 @@ func TestKillDaemonWaitsForOwnershipTransfer(t *testing.T) {
529	require.NoError(t, <-done)	529		require.NoError(t, <-done)
530	}	530		}
531		531
		532	+	func TestLifecycleOwnershipOutlivesSnapshotWriter(t *testing.T) {
		533	+	result := runBlockedSnapshotWriterShutdown(t)
		534	+	assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
		535	+	assertLifecycleResultPending(t, result.wrapperReturned, "lifecycle wrapper return while snapshot writer is blocked")
		536	+	owner, err := lifecycle.TryAcquire(result.runtimeDir)
		537	+	require.ErrorIs(t, err, lifecycle.ErrBusy)
		538	+
		539	+	result.releaseWriter()
		540	+	awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
		541	+	require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
		542	+	owner, err = lifecycle.TryAcquire(result.runtimeDir)
		543	+	require.NoError(t, err)
		544	+	require.NoError(t, owner.Release())
		545	+	}
		546	+
		547	+	func TestLifecycleCallbackWaitsForEveryWriter(t *testing.T) {
		548	+	result := runBlockedSnapshotWriterShutdown(t)
		549	+	assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
		550	+
		551	+	result.releaseWriter()
		552	+	awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
		553	+	awaitLifecycleStage(t, result.callbackReturned, "Serve callback return after snapshot writer")
		554	+	require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
		555	+	}
		556	+
		557	+	func TestFinalCheckpointTimeoutKeepsOwner(t *testing.T) {
		558	+	result := runBlockedSnapshotWriterShutdown(t)
		559	+	assertLifecycleResultPending(t, result.wrapperReturned, "ownership release after checkpoint timeout with writer alive")
		560	+	_, err := lifecycle.TryAcquire(result.runtimeDir)
		561	+	require.ErrorIs(t, err, lifecycle.ErrBusy)
		562	+
		563	+	result.releaseWriter()
		564	+	require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
		565	+	}
		566	+
		567	+	type blockedSnapshotWriterShutdown struct {
		568	+	runtimeDir string
		569	+	callbackReturned <-chan struct{}
		570	+	wrapperReturned <-chan error
		571	+	publishReturned <-chan struct{}
		572	+	releaseWriter func()
		573	+	}
		574	+
		575	+	func runBlockedSnapshotWriterShutdown(t *testing.T) blockedSnapshotWriterShutdown {
		576	+	t.Helper()
		577	+	runtimeDir := filepath.Join(t.TempDir(), "runtime")
		578	+	stateDir := filepath.Join(t.TempDir(), "state")
		579	+	shutdownClock := newLifecycleShutdownClock()
		580	+	repository := newLifecycleBlockingRepository()
		581	+	ctx, cancel := context.WithCancel(context.Background())
		582	+	t.Cleanup(cancel)
		583	+
		584	+	listenerReady := make(chan struct{})
		585	+	listenerClosed := make(chan struct{})
		586	+	callbackReturned := make(chan struct{})
		587	+	wrapperReturned := make(chan error, 1)
		588	+	go func() {
		589	+	wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
		590	+	defer close(callbackReturned)
		591	+	listener, err := ipc.Listen(runtimeDir)
		592	+	if err != nil {
		593	+	return err
		594	+	}
		595	+	observed := &lifecycleObservedListener{Listener: listener, closed: listenerClosed}
		596	+	close(listenerReady)
		597	+	d := daemon.New(
		598	+	pty.NewFactory(),
		599	+	shutdownClock,
		600	+	discardLog(),
		601	+	daemon.WithShell("/bin/cat", nil),
		602	+	daemon.WithSnapshotRepository(repository, nil),
		603	+	)
		604	+	return d.Serve(ctx, observed)
		605	+	})
		606	+	}()
		607	+
		608	+	awaitLifecycleStage(t, listenerReady, "daemon listener")
		609	+	tr, _ := attach(t, runtimeDir, ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24})
		610	+	require.NoError(t, tr.Send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("dirty state\n")})}))
		611	+	awaitLifecycleStage(t, repository.entered, "snapshot publication")
		612	+	require.NoError(t, tr.Close())
		613	+	require.Eventually(t, func() bool {
		614	+	sessions := listRemoteSessions(t, runtimeDir)
		615	+	return len(sessions.Sessions) == 1 && !sessions.Sessions[0].Attached
		616	+	}, time.Second, 10*time.Millisecond)
		617	+	cancel()
		618	+	awaitLifecycleStage(t, listenerClosed, "listener close")
		619	+
		620	+	shutdownClock.nextTimer(t).fire()
		621	+	select {
		622	+	case <-callbackReturned:
		623	+	t.Fatal("Serve callback returned after checkpoint timeout while snapshot writer was blocked")
		624	+	case <-time.After(100 * time.Millisecond):
		625	+	}
		626	+	assertLifecycleStagePending(t, callbackReturned, "Serve callback return after checkpoint timeout")
		627	+	assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return after checkpoint timeout")
		628	+
		629	+	var releaseOnce sync.Once
		630	+	releaseWriter := func() { releaseOnce.Do(func() { close(repository.release) }) }
		631	+	t.Cleanup(releaseWriter)
		632	+	return blockedSnapshotWriterShutdown{
		633	+	runtimeDir: runtimeDir,
		634	+	callbackReturned: callbackReturned,
		635	+	wrapperReturned: wrapperReturned,
		636	+	publishReturned: repository.returned,
		637	+	releaseWriter: releaseWriter,
		638	+	}
		639	+	}
		640	+
		641	+	type lifecycleBlockingRepository struct {
		642	+	ports.SnapshotRepository
		643	+	entered chan struct{}
		644	+	release chan struct{}
		645	+	returned chan struct{}
		646	+	enterOnce sync.Once
		647	+	returnOnce sync.Once
		648	+	}
		649	+
		650	+	func newLifecycleBlockingRepository() *lifecycleBlockingRepository {
		651	+	return &lifecycleBlockingRepository{
		652	+	entered: make(chan struct{}),
		653	+	release: make(chan struct{}),
		654	+	returned: make(chan struct{}),
		655	+	}
		656	+	}
		657	+
		658	+	func (r *lifecycleBlockingRepository) Publish(context.Context, ports.SnapshotPublication) error {
		659	+	r.enterOnce.Do(func() { close(r.entered) })
		660	+	<-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
		661	+	r.returnOnce.Do(func() { close(r.returned) })
		662	+	return nil
		663	+	}
		664	+
		665	+	type lifecycleShutdownClock struct {
		666	+	base ports.Clock
		667	+	timers chan *lifecycleManualTimer
		668	+	}
		669	+
		670	+	func newLifecycleShutdownClock() *lifecycleShutdownClock {
		671	+	return &lifecycleShutdownClock{base: clock.New(), timers: make(chan *lifecycleManualTimer, 4)}
		672	+	}
		673	+
		674	+	func (c *lifecycleShutdownClock) Now() time.Time { return c.base.Now() }
		675	+	func (c *lifecycleShutdownClock) NewTimer(delay time.Duration) ports.Timer {
		676	+	if delay != time.Second {
		677	+	return c.base.NewTimer(delay)
		678	+	}
		679	+	timer := &lifecycleManualTimer{ch: make(chan time.Time, 1)}
		680	+	c.timers <- timer
		681	+	return timer
		682	+	}
		683	+
		684	+	func (c lifecycleShutdownClock) nextTimer(t testing.T) *lifecycleManualTimer {
		685	+	t.Helper()
		686	+	select {
		687	+	case timer := <-c.timers:
		688	+	return timer
		689	+	case <-time.After(3 * time.Second):
		690	+	t.Fatal("timed out waiting for shutdown deadline timer")
		691	+	return nil
		692	+	}
		693	+	}
		694	+
		695	+	type lifecycleManualTimer struct {
		696	+	ch chan time.Time
		697	+	stopped atomic.Bool
		698	+	}
		699	+
		700	+	func (t *lifecycleManualTimer) C() <-chan time.Time { return t.ch }
		701	+	func (t *lifecycleManualTimer) Reset(time.Duration) bool {
		702	+	t.stopped.Store(false)
		703	+	return false
		704	+	}
		705	+	func (t *lifecycleManualTimer) Stop() bool { return !t.stopped.Swap(true) }
		706	+	func (t *lifecycleManualTimer) fire() {
		707	+	if !t.stopped.Load() {
		708	+	t.ch <- time.Now()
		709	+	}
		710	+	}
		711	+
		712	+	type lifecycleObservedListener struct {
		713	+	ports.Listener
		714	+	closed chan struct{}
		715	+	once sync.Once
		716	+	}
		717	+
		718	+	func (l *lifecycleObservedListener) Close() error {
		719	+	err := l.Listener.Close()
		720	+	l.once.Do(func() { close(l.closed) })
		721	+	return err
		722	+	}
		723	+
532	func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {	724		func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {
533	runtimeDir := filepath.Join(t.TempDir(), "runtime")	725		runtimeDir := filepath.Join(t.TempDir(), "runtime")
534	stateDir := filepath.Join(t.TempDir(), "state")	726		stateDir := filepath.Join(t.TempDir(), "state")

@@ -690,6 +882,15 @@ func assertLifecycleStagePending(t *testing.T, stage <-chan struct{}, name strin
690	}	882		}
691	}	883		}
692		884
		885	+	func assertLifecycleResultPending(t *testing.T, result <-chan error, name string) {
		886	+	t.Helper()
		887	+	select {
		888	+	case err := <-result:
		889	+	t.Fatalf("unexpected %s: %v", name, err)
		890	+	default:
		891	+	}
		892	+	}
		893	+
693	func assertNewDaemonStartupPending(t *testing.T, durableOpen, listen <-chan struct{}, heldAt string) {	894		func assertNewDaemonStartupPending(t *testing.T, durableOpen, listen <-chan struct{}, heldAt string) {
694	t.Helper()	895		t.Helper()
695	assertLifecycleStagePending(t, durableOpen, "new durable open while old daemon held at "+heldAt)	896		assertLifecycleStagePending(t, durableOpen, "new durable open while old daemon held at "+heldAt)

M internal/usecase/daemon/daemon.go +3 -2


@@ -586,9 +586,10 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
586	586		d.connWg.Wait()
587	587		d.attachmentCleanupWg.Wait()
588	588		// Forced terminal checkpoints run before shutdown stops producers or the
589		-	// worker. The same deadline bounds the worker's final drain, including a
590		-	// repository call that ignores cancellation.
	589	+	// worker. The deadline may abandon checkpoint success and cancel cooperative
	590	+	// I/O, but ownership remains pinned until every writer has actually exited.
591	591		d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)
	592	+	d.WaitDurableWriters()
592	593		d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
593	594		d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
594	595		d.waitNotifies()

 	d.connWg.Wait()
 	d.attachmentCleanupWg.Wait()
 	// Forced terminal checkpoints run before shutdown stops producers or the
-	// worker. The same deadline bounds the worker's final drain, including a
+	// worker. The deadline may abandon checkpoint success and cancel cooperative
-	// repository call that ignores cancellation.
+	// I/O, but ownership remains pinned until every writer has actually exited.
 	d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)
+	d.WaitDurableWriters()
 	d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
 	d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
 	d.waitNotifies()

M internal/usecase/daemon/snapshot_scheduler_test.go +24 -15


@@ -285,7 +285,7 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
285	285		}
286	286		}
287	287
288		-	func TestServeShutdownDeadlineBoundsPaneExitTeardownOwner(t *testing.T) {
	288	+	func TestServeShutdownDeadlineStillJoinsPaneExitWriter(t *testing.T) {
289	289		pty, releasePTY := newBlockingPTY(t)
290	290		clock := newServeShutdownClock()
291	291		d := newTestDaemon(t, newFactory(t, pty), clock)

@@ -328,9 +328,9 @@ func TestServeShutdownDeadlineBoundsPaneExitTeardownOwner(t *testing.T) {
328	328		// deadline.
329	329		ownerDeadlineTimer := clock.nextFinalTimer(t)
330	330
331		-	// Cleanup is the only place that releases the competing owner. It also fires
332		-	// that owner's independent budget and joins the real pane reader, proving no
333		-	// teardown goroutine is leaked without making Serve depend on repository I/O.
	331	+	// Cleanup fires the pane-exit owner's independent budget and joins the real
	332	+	// pane reader. Serve itself must remain owned until the repository writer is
	333	+	// released below, even after its checkpoint deadline expires.
334	334		t.Cleanup(func() {
335	335		releaseSnapshot()
336	336		ownerDeadlineTimer.fire()

@@ -341,11 +341,17 @@ func TestServeShutdownDeadlineBoundsPaneExitTeardownOwner(t *testing.T) {
341	341		shutdownDeadlineTimer := clock.nextFinalTimer(t)
342	342		shutdownDeadlineTimer.fire()
343	343
	344	+	select {
	345	+	case err := <-served:
	346	+	t.Fatalf("Serve returned with a durable writer alive: %v", err)
	347	+	default:
	348	+	}
	349	+	releaseSnapshot()
344	350		select {
345	351		case err := <-served:
346	352		require.NoError(t, err)
347	353		case <-time.After(testWaitTimeout):
348		-	t.Fatal("Serve did not return after its shared deadline while the pane-exit owner remained blocked")
	354	+	t.Fatal("Serve did not return after the durable writer exited")
349	355		}
350	356
351	357		sess.snapshotMu.Lock()

@@ -488,7 +494,7 @@ func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {
488	494		}
489	495		}
490	496
491		-	func TestServeShutdownDeadlineDoesNotWaitTwiceForUncooperativeSnapshotRepository(t *testing.T) {
	497	+	func TestServeShutdownDeadlineStillJoinsUncooperativeSnapshotRepository(t *testing.T) {
492	498		p, releasePTY := newBlockingPTY(t)
493	499		defer releasePTY()
494	500		clock := newServeShutdownClock()

@@ -544,21 +550,24 @@ func TestServeShutdownDeadlineDoesNotWaitTwiceForUncooperativeSnapshotRepository
544	550		workerDone := d.snapshotWorkerDone
545	551		d.snapshotWorkerMu.Unlock()
546	552		require.NotNil(t, workerDone)
547		-	defer func() {
548		-	close(releaseTerminalPublication)
549		-	select {
550		-	case <-workerDone:
551		-	case <-time.After(testWaitTimeout):
552		-	t.Error("snapshot worker leaked after its repository call was released")
553		-	}
554		-	}()
555	553		clock.nextFinalTimer(t).fire()
556	554
	555	+	select {
	556	+	case err := <-served:
	557	+	t.Fatalf("Serve returned with an uncooperative durable writer alive: %v", err)
	558	+	default:
	559	+	}
	560	+	close(releaseTerminalPublication)
	561	+	select {
	562	+	case <-workerDone:
	563	+	case <-time.After(testWaitTimeout):
	564	+	t.Fatal("snapshot worker leaked after its repository call was released")
	565	+	}
557	566		select {
558	567		case err := <-served:
559	568		require.NoError(t, err)
560	569		case <-time.After(testWaitTimeout):
561		-	t.Fatal("Serve waited beyond the terminal checkpoint deadline")
	570	+	t.Fatal("Serve did not return after the uncooperative writer exited")
562	571		}
563	572		require.LessOrEqual(t, clock.finalBudget.Load(), int64(snapshotFinalFlushTimeout), "shutdown must consume at most one snapshot deadline")
564	573		require.Equal(t, uint64(1), lastValid.Load(), "the last valid checkpoint remains available after the terminal attempt times out")

 	}
 }
-func TestServeShutdownDeadlineBoundsPaneExitTeardownOwner(t *testing.T) {
+func TestServeShutdownDeadlineStillJoinsPaneExitWriter(t *testing.T) {
 	pty, releasePTY := newBlockingPTY(t)
 	clock := newServeShutdownClock()
 	d := newTestDaemon(t, newFactory(t, pty), clock)
 	// deadline.
 	ownerDeadlineTimer := clock.nextFinalTimer(t)
-	// Cleanup is the only place that releases the competing owner. It also fires
+	// Cleanup fires the pane-exit owner's independent budget and joins the real
-	// that owner's independent budget and joins the real pane reader, proving no
+	// pane reader. Serve itself must remain owned until the repository writer is
-	// teardown goroutine is leaked without making Serve depend on repository I/O.
+	// released below, even after its checkpoint deadline expires.
 	t.Cleanup(func() {
 		releaseSnapshot()
 		ownerDeadlineTimer.fire()
 	shutdownDeadlineTimer := clock.nextFinalTimer(t)
 	shutdownDeadlineTimer.fire()
+	select {
+	case err := <-served:
+		t.Fatalf("Serve returned with a durable writer alive: %v", err)
+	default:
+	}
+	releaseSnapshot()
 	select {
 	case err := <-served:
 		require.NoError(t, err)
 	case <-time.After(testWaitTimeout):
-		t.Fatal("Serve did not return after its shared deadline while the pane-exit owner remained blocked")
+		t.Fatal("Serve did not return after the durable writer exited")
 	}
 	sess.snapshotMu.Lock()
 	}
 }
-func TestServeShutdownDeadlineDoesNotWaitTwiceForUncooperativeSnapshotRepository(t *testing.T) {
+func TestServeShutdownDeadlineStillJoinsUncooperativeSnapshotRepository(t *testing.T) {
 	p, releasePTY := newBlockingPTY(t)
 	defer releasePTY()
 	clock := newServeShutdownClock()
 	workerDone := d.snapshotWorkerDone
 	d.snapshotWorkerMu.Unlock()
 	require.NotNil(t, workerDone)
-	defer func() {
-		close(releaseTerminalPublication)
-		select {
-		case <-workerDone:
-		case <-time.After(testWaitTimeout):
-			t.Error("snapshot worker leaked after its repository call was released")
-		}
-	}()
 	clock.nextFinalTimer(t).fire()
+	select {
+	case err := <-served:
+		t.Fatalf("Serve returned with an uncooperative durable writer alive: %v", err)
+	default:
+	}
+	close(releaseTerminalPublication)
+	select {
+	case <-workerDone:
+	case <-time.After(testWaitTimeout):
+		t.Fatal("snapshot worker leaked after its repository call was released")
+	}
 	select {
 	case err := <-served:
 		require.NoError(t, err)
 	case <-time.After(testWaitTimeout):
-		t.Fatal("Serve waited beyond the terminal checkpoint deadline")
+		t.Fatal("Serve did not return after the uncooperative writer exited")
 	}
 	require.LessOrEqual(t, clock.finalBudget.Load(), int64(snapshotFinalFlushTimeout), "shutdown must consume at most one snapshot deadline")
 	require.Equal(t, uint64(1), lastValid.Load(), "the last valid checkpoint remains available after the terminal attempt times out")

M internal/usecase/daemon/snapshot_worker.go +82 -23


@@ -2,6 +2,7 @@ package daemon
2	2
3	3		import (
4	4		"context"
	5	+	"sort"
5	6
6	7		"github.com/bnema/vev/internal/domain"
7	8		)

@@ -193,44 +194,102 @@ func (d *Daemon) stopSnapshotEncodeWorker() {
193	194		if d == nil {
194	195		return
195	196		}
196		-	deadline := newSnapshotShutdownDeadline(d.clock)
197		-	defer deadline.stop()
198		-	d.stopSnapshotEncodeWorkerWithDeadline(deadline)
	197	+	_ = d.StopDurableWriters(context.Background())
	198	+	d.WaitDurableWriters()
199	199		}
200	200
201		-	// stopSnapshotEncodeWorkerWithDeadline performs the one deadline-aware join
202		-	// for a Serve shutdown. Queues are never closed, and detached writers only
203		-	// retain immutable captures plus contexts/channels that remain live, so an
204		-	// uncooperative repository call cannot race teardown or block process exit.
205		-	func (d Daemon) stopSnapshotEncodeWorkerWithDeadline(deadline snapshotShutdownDeadline) {
	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.
	204	+	func (d *Daemon) StopDurableWriters(ctx context.Context) []string {
206	205		if d == nil {
207		-	return
	206	+	return nil
208	207		}
209		-	d.snapshotWorkerMu.Lock()
210		-	cancel := d.snapshotWorkerCancel
211		-	if cancel == nil \|\| d.snapshotWorkerClosing {
212		-	d.snapshotWorkerMu.Unlock()
	208	+	cancel, done := d.requestDurableWriterStop()
	209	+	if done == nil {
	210	+	return nil
	211	+	}
	212	+	if ctx == nil {
	213	+	ctx = context.Background()
	214	+	}
	215	+	select {
	216	+	case <-done:
	217	+	return nil
	218	+	case <-ctx.Done():
	219	+	cancel()
	220	+	return d.durableWriterFailureNames()
	221	+	}
	222	+	}
	223	+
	224	+	// WaitDurableWriters is the unconditional ownership barrier. It has no timeout
	225	+	// or status branch: every writer goroutine must exit before Serve may return.
	226	+	func (d *Daemon) WaitDurableWriters() {
	227	+	if d == nil {
213	228		return
214	229		}
215		-	// Stop accepting producers, then let the single worker drain its bounded
216		-	// queue. A non-context-aware store may still block indefinitely, so the
217		-	// shared shutdown deadline bounds the one and only join.
218		-	d.snapshotWorkerClosing = true
219		-	flush := d.snapshotWorkerFlush
	230	+	d.snapshotWorkerMu.Lock()
220	231		done := d.snapshotWorkerDone
221		-	close(flush)
222	232		d.snapshotWorkerMu.Unlock()
	233	+	if done == nil {
	234	+	return
	235	+	}
	236	+	<-done
	237	+	d.finishStoppedSnapshotWorker(false)
	238	+	}
	239	+
	240	+	func (d *Daemon) requestDurableWriterStop() (context.CancelFunc, <-chan struct{}) {
	241	+	d.snapshotWorkerMu.Lock()
	242	+	defer d.snapshotWorkerMu.Unlock()
	243	+	cancel := d.snapshotWorkerCancel
	244	+	if cancel == nil {
	245	+	return nil, nil
	246	+	}
	247	+	if !d.snapshotWorkerClosing {
	248	+	d.snapshotWorkerClosing = true
	249	+	close(d.snapshotWorkerFlush)
	250	+	}
	251	+	return cancel, d.snapshotWorkerDone
	252	+	}
223	253
	254	+	func (d *Daemon) durableWriterFailureNames() []string {
	255	+	d.snapshotWorkerMu.Lock()
	256	+	defer d.snapshotWorkerMu.Unlock()
	257	+	seen := make(map[string]struct{})
	258	+	add := func(capture *snapshotCapture) {
	259	+	if capture != nil && capture.name != "" {
	260	+	seen[capture.name] = struct{}{}
	261	+	}
	262	+	}
	263	+	add(d.snapshotWorkerInFlight)
	264	+	for _, capture := range d.snapshotFinalJobs {
	265	+	add(capture)
	266	+	}
	267	+	names := make([]string, 0, len(seen))
	268	+	for name := range seen {
	269	+	names = append(names, name)
	270	+	}
	271	+	sort.Strings(names)
	272	+	return names
	273	+	}
	274	+
	275	+	// stopSnapshotEncodeWorkerWithDeadline bounds the final checkpoint attempt,
	276	+	// but deliberately does not bound the ownership join that follows it.
	277	+	func (d Daemon) stopSnapshotEncodeWorkerWithDeadline(deadline snapshotShutdownDeadline) {
	278	+	if d == nil {
	279	+	return
	280	+	}
	281	+	cancel, done := d.requestDurableWriterStop()
	282	+	if done == nil {
	283	+	return
	284	+	}
224	285		if deadline == nil {
225		-	deadline = newSnapshotShutdownDeadline(d.clock)
226		-	defer deadline.stop()
	286	+	<-done
	287	+	return
227	288		}
228	289		select {
229	290		case <-done:
230		-	d.finishStoppedSnapshotWorker(false)
231	291		case <-deadline.Done():
232	292		cancel()
233		-	d.finishStoppedSnapshotWorker(true)
234	293		}
235	294		}
236	295


@@ -2,6 +2,7 @@ package daemon
2			2
3		import (	3		import (
4		"context"	4		"context"
			5	+	"sort"
5			6
6		"github.com/bnema/vev/internal/domain"	7		"github.com/bnema/vev/internal/domain"
7		)	8		)

@@ -193,44 +194,102 @@ func (d *Daemon) stopSnapshotEncodeWorker() {
193		if d == nil {	194		if d == nil {
194		return	195		return
195		}	196		}
196	-	deadline := newSnapshotShutdownDeadline(d.clock)	197	+	_ = d.StopDurableWriters(context.Background())
197	-	defer deadline.stop()	198	+	d.WaitDurableWriters()
198	-	d.stopSnapshotEncodeWorkerWithDeadline(deadline)
199		}	199		}
200			200
201	-	// stopSnapshotEncodeWorkerWithDeadline performs the one deadline-aware join	201	+	// StopDurableWriters stops admission and requests the worker's final drain.
202	-	// for a Serve shutdown. Queues are never closed, and detached writers only	202	+	// The context bounds checkpoint success only: cancellation asks cooperative
203	-	// retain immutable captures plus contexts/channels that remain live, so an	203	+	// repository calls to stop, but it never detaches the worker from its owner.
204	-	// uncooperative repository call cannot race teardown or block process exit.	204	+	func (d *Daemon) StopDurableWriters(ctx context.Context) []string {
205	-	func (d Daemon) stopSnapshotEncodeWorkerWithDeadline(deadline snapshotShutdownDeadline) {
206		if d == nil {	205		if d == nil {
207	-	return	206	+	return nil
208		}	207		}
209	-	d.snapshotWorkerMu.Lock()	208	+	cancel, done := d.requestDurableWriterStop()
210	-	cancel := d.snapshotWorkerCancel	209	+	if done == nil {
211	-	if cancel == nil \|\| d.snapshotWorkerClosing {	210	+	return nil
212	-	d.snapshotWorkerMu.Unlock()	211	+	}
			212	+	if ctx == nil {
			213	+	ctx = context.Background()
			214	+	}
			215	+	select {
			216	+	case <-done:
			217	+	return nil
			218	+	case <-ctx.Done():
			219	+	cancel()
			220	+	return d.durableWriterFailureNames()
			221	+	}
			222	+	}
			223	+
			224	+	// WaitDurableWriters is the unconditional ownership barrier. It has no timeout
			225	+	// or status branch: every writer goroutine must exit before Serve may return.
			226	+	func (d *Daemon) WaitDurableWriters() {
			227	+	if d == nil {
213		return	228		return
214		}	229		}
215	-	// Stop accepting producers, then let the single worker drain its bounded	230	+	d.snapshotWorkerMu.Lock()
216	-	// queue. A non-context-aware store may still block indefinitely, so the
217	-	// shared shutdown deadline bounds the one and only join.
218	-	d.snapshotWorkerClosing = true
219	-	flush := d.snapshotWorkerFlush
220		done := d.snapshotWorkerDone	231		done := d.snapshotWorkerDone
221	-	close(flush)
222		d.snapshotWorkerMu.Unlock()	232		d.snapshotWorkerMu.Unlock()
			233	+	if done == nil {
			234	+	return
			235	+	}
			236	+	<-done
			237	+	d.finishStoppedSnapshotWorker(false)
			238	+	}
			239	+
			240	+	func (d *Daemon) requestDurableWriterStop() (context.CancelFunc, <-chan struct{}) {
			241	+	d.snapshotWorkerMu.Lock()
			242	+	defer d.snapshotWorkerMu.Unlock()
			243	+	cancel := d.snapshotWorkerCancel
			244	+	if cancel == nil {
			245	+	return nil, nil
			246	+	}
			247	+	if !d.snapshotWorkerClosing {
			248	+	d.snapshotWorkerClosing = true
			249	+	close(d.snapshotWorkerFlush)
			250	+	}
			251	+	return cancel, d.snapshotWorkerDone
			252	+	}
223			253
			254	+	func (d *Daemon) durableWriterFailureNames() []string {
			255	+	d.snapshotWorkerMu.Lock()
			256	+	defer d.snapshotWorkerMu.Unlock()
			257	+	seen := make(map[string]struct{})
			258	+	add := func(capture *snapshotCapture) {
			259	+	if capture != nil && capture.name != "" {
			260	+	seen[capture.name] = struct{}{}
			261	+	}
			262	+	}
			263	+	add(d.snapshotWorkerInFlight)
			264	+	for _, capture := range d.snapshotFinalJobs {
			265	+	add(capture)
			266	+	}
			267	+	names := make([]string, 0, len(seen))
			268	+	for name := range seen {
			269	+	names = append(names, name)
			270	+	}
			271	+	sort.Strings(names)
			272	+	return names
			273	+	}
			274	+
			275	+	// stopSnapshotEncodeWorkerWithDeadline bounds the final checkpoint attempt,
			276	+	// but deliberately does not bound the ownership join that follows it.
			277	+	func (d Daemon) stopSnapshotEncodeWorkerWithDeadline(deadline snapshotShutdownDeadline) {
			278	+	if d == nil {
			279	+	return
			280	+	}
			281	+	cancel, done := d.requestDurableWriterStop()
			282	+	if done == nil {
			283	+	return
			284	+	}
224		if deadline == nil {	285		if deadline == nil {
225	-	deadline = newSnapshotShutdownDeadline(d.clock)	286	+	<-done
226	-	defer deadline.stop()	287	+	return
227		}	288		}
228		select {	289		select {
229		case <-done:	290		case <-done:
230	-	d.finishStoppedSnapshotWorker(false)
231		case <-deadline.Done():	291		case <-deadline.Done():
232		cancel()	292		cancel()
233	-	d.finishStoppedSnapshotWorker(true)
234		}	293		}
235		}	294		}
236			295