a8af417 fix(daemon): bound defensive shutdown teardown · vev

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


@@ -518,8 +518,8 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
518	518		// worker. The same deadline bounds the worker's final drain, including a
519	519		// repository call that ignores cancellation.
520	520		d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)
521		-	d.shutdownAll(ports.ReasonServerShutdown)
522		-	d.sessWg.Wait()
	521	+	d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
	522	+	d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
523	523		d.waitNotifies()
524	524		if err := d.persist.Close(); err != nil {
525	525		d.log.Warn("closing session persister failed", "err", err)

@@ -550,6 +550,9 @@ func (d Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline snapsho
550	550		return false
551	551		}
552	552		for _, s := range snapshot {
	553	+	// Cancellation and PTY closure must not wait behind a teardown owner that
	554	+	// is blocked in snapshot publication or purge work.
	555	+	s.stopInMemoryLifecycle()
553	556		if err := d.killSessionWithSnapshotDeadline(s, reason, false, deadline); err != nil {
554	557		checkpointIncomplete = true
555	558		d.log.Error("closing session with unpersisted terminal state", "err", err)

@@ -558,6 +561,26 @@ func (d Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline snapsho
558	561		return checkpointIncomplete
559	562		}
560	563
	564	+	// waitSessionWorkersWithSnapshotDeadline joins in-memory workers when possible,
	565	+	// but never extends Serve beyond the shared repository shutdown budget. By this
	566	+	// point connection handlers are joined and every session has been cancelled, so
	567	+	// no new session worker can be registered while the waiter is active.
	568	+	func (d Daemon) waitSessionWorkersWithSnapshotDeadline(deadline snapshotShutdownDeadline) {
	569	+	done := make(chan struct{})
	570	+	go func() {
	571	+	d.sessWg.Wait()
	572	+	close(done)
	573	+	}()
	574	+	if deadline == nil {
	575	+	<-done
	576	+	return
	577	+	}
	578	+	select {
	579	+	case <-done:
	580	+	case <-deadline.Done():
	581	+	}
	582	+	}
	583	+
561	584		// snapshotShutdownDeadline is a single-use shutdown budget shared by forced
562	585		// checkpoints and the final snapshot worker join. Its done channel is never
563	586		// closed by callers, so a detached worker can safely retain session state until


@@ -518,8 +518,8 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
518		// worker. The same deadline bounds the worker's final drain, including a	518		// worker. The same deadline bounds the worker's final drain, including a
519		// repository call that ignores cancellation.	519		// repository call that ignores cancellation.
520		d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)	520		d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)
521	-	d.shutdownAll(ports.ReasonServerShutdown)	521	+	d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
522	-	d.sessWg.Wait()	522	+	d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
523		d.waitNotifies()	523		d.waitNotifies()
524		if err := d.persist.Close(); err != nil {	524		if err := d.persist.Close(); err != nil {
525		d.log.Warn("closing session persister failed", "err", err)	525		d.log.Warn("closing session persister failed", "err", err)

@@ -550,6 +550,9 @@ func (d Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline snapsho
550		return false	550		return false
551		}	551		}
552		for _, s := range snapshot {	552		for _, s := range snapshot {
			553	+	// Cancellation and PTY closure must not wait behind a teardown owner that
			554	+	// is blocked in snapshot publication or purge work.
			555	+	s.stopInMemoryLifecycle()
553		if err := d.killSessionWithSnapshotDeadline(s, reason, false, deadline); err != nil {	556		if err := d.killSessionWithSnapshotDeadline(s, reason, false, deadline); err != nil {
554		checkpointIncomplete = true	557		checkpointIncomplete = true
555		d.log.Error("closing session with unpersisted terminal state", "err", err)	558		d.log.Error("closing session with unpersisted terminal state", "err", err)

@@ -558,6 +561,26 @@ func (d Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline snapsho
558		return checkpointIncomplete	561		return checkpointIncomplete
559		}	562		}
560			563
			564	+	// waitSessionWorkersWithSnapshotDeadline joins in-memory workers when possible,
			565	+	// but never extends Serve beyond the shared repository shutdown budget. By this
			566	+	// point connection handlers are joined and every session has been cancelled, so
			567	+	// no new session worker can be registered while the waiter is active.
			568	+	func (d Daemon) waitSessionWorkersWithSnapshotDeadline(deadline snapshotShutdownDeadline) {
			569	+	done := make(chan struct{})
			570	+	go func() {
			571	+	d.sessWg.Wait()
			572	+	close(done)
			573	+	}()
			574	+	if deadline == nil {
			575	+	<-done
			576	+	return
			577	+	}
			578	+	select {
			579	+	case <-done:
			580	+	case <-deadline.Done():
			581	+	}
			582	+	}
			583	+
561		// snapshotShutdownDeadline is a single-use shutdown budget shared by forced	584		// snapshotShutdownDeadline is a single-use shutdown budget shared by forced
562		// checkpoints and the final snapshot worker join. Its done channel is never	585		// checkpoints and the final snapshot worker join. Its done channel is never
563		// closed by callers, so a detached worker can safely retain session state until	586		// closed by callers, so a detached worker can safely retain session state until

M internal/usecase/daemon/session.go +24 -6


@@ -51,6 +51,7 @@ type session struct {
51	51		teardownDone chan struct{}
52	52		teardownWaiters uint
53	53		teardownChanged chan struct{}
	54	+	lifecycleStopOnce sync.Once
54	55		mu sync.Mutex // guards tabs, active, client, clipFiles, and clipboard queue state
55	56		themeMu sync.Mutex
56	57		tabs []*tab

@@ -913,6 +914,25 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
913	914		return d.killSessionWithSnapshotDeadline(sess, reason, purge, nil)
914	915		}
915	916
	917	+	// stopInMemoryLifecycle cancels every session producer and closes every PTY
	918	+	// independently of teardown ownership. Repository work may remain blocked, but
	919	+	// repeated shutdown passes cannot leave pane readers or launches running.
	920	+	func (s *session) stopInMemoryLifecycle() {
	921	+	if s == nil {
	922	+	return
	923	+	}
	924	+	s.lifecycleStopOnce.Do(func() {
	925	+	s.stopFloatingLaunches()
	926	+	s.cancel()
	927	+	s.mu.Lock()
	928	+	tabs := append([]*tab(nil), s.tabs...)
	929	+	s.mu.Unlock()
	930	+	for _, tb := range tabs {
	931	+	tb.closeAllPanes()
	932	+	}
	933	+	})
	934	+	}
	935	+
916	936		func (s *session) teardownChangeLocked() chan struct{} {
917	937		if s.teardownChanged == nil {
918	938		s.teardownChanged = make(chan struct{})

@@ -1143,11 +1163,10 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
1143	1163		ac.clearCaptureFrames()
1144	1164		}
1145	1165
1146		-	// Prevent queued launches from entering Open, then cancel the parent
1147		-	// context. The launch worker owns any late PTY result and closes it rather
1148		-	// than publishing it into the torn-down tab.
1149		-	sess.stopFloatingLaunches()
1150		-	sess.cancel()
	1166	+	// Prevent queued launches from entering Open, cancel the parent context, and
	1167	+	// close PTYs. Serve may already have performed this phase while repository
	1168	+	// work owned by this teardown was blocked.
	1169	+	sess.stopInMemoryLifecycle()
1151	1170		sess.mu.Lock()
1152	1171		tabs := append([]*tab(nil), sess.tabs...)
1153	1172		clipFiles := sess.clipFiles

@@ -1156,7 +1175,6 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
1156	1175		for _, tb := range tabs {
1157	1176		d.clearDestroyedTabPreview(tb)
1158	1177		d.teardownFloating(tb, ac)
1159		-	tb.closeAllPanes()
1160	1178		}
1161	1179		for _, path := range clipFiles {
1162	1180		if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {

 	teardownDone           chan struct{}
 	teardownWaiters        uint
 	teardownChanged        chan struct{}
+	lifecycleStopOnce      sync.Once
 	mu                     sync.Mutex // guards tabs, active, client, clipFiles, and clipboard queue state
 	themeMu                sync.Mutex
 	tabs                   []*tab
 	return d.killSessionWithSnapshotDeadline(sess, reason, purge, nil)
 }
+// stopInMemoryLifecycle cancels every session producer and closes every PTY
+// independently of teardown ownership. Repository work may remain blocked, but
+// repeated shutdown passes cannot leave pane readers or launches running.
+func (s *session) stopInMemoryLifecycle() {
+	if s == nil {
+		return
+	}
+	s.lifecycleStopOnce.Do(func() {
+		s.stopFloatingLaunches()
+		s.cancel()
+		s.mu.Lock()
+		tabs := append([]*tab(nil), s.tabs...)
+		s.mu.Unlock()
+		for _, tb := range tabs {
+			tb.closeAllPanes()
+		}
+	})
+}
 func (s *session) teardownChangeLocked() chan struct{} {
 	if s.teardownChanged == nil {
 		s.teardownChanged = make(chan struct{})
 		ac.clearCaptureFrames()
 	}
-	// Prevent queued launches from entering Open, then cancel the parent
+	// Prevent queued launches from entering Open, cancel the parent context, and
-	// context. The launch worker owns any late PTY result and closes it rather
+	// close PTYs. Serve may already have performed this phase while repository
-	// than publishing it into the torn-down tab.
+	// work owned by this teardown was blocked.
-	sess.stopFloatingLaunches()
+	sess.stopInMemoryLifecycle()
-	sess.cancel()
 	sess.mu.Lock()
 	tabs := append([]*tab(nil), sess.tabs...)
 	clipFiles := sess.clipFiles
 	for _, tb := range tabs {
 		d.clearDestroyedTabPreview(tb)
 		d.teardownFloating(tb, ac)
-		tb.closeAllPanes()
 	}
 	for _, path := range clipFiles {
 		if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {

M internal/usecase/daemon/snapshot_scheduler_test.go +19 -25


@@ -454,7 +454,6 @@ func TestServeShutdownDeadlineBoundsPaneExitTeardownOwner(t *testing.T) {
454	454		releasePublication := make(chan struct{})
455	455		var releasePublicationOnce sync.Once
456	456		releaseSnapshot := func() { releasePublicationOnce.Do(func() { close(releasePublication) }) }
457		-	t.Cleanup(releaseSnapshot)
458	457		repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
459	458		require.Equal(t, uint64(1), publication.Generation)
460	459		close(publicationStarted)

@@ -483,40 +482,35 @@ func TestServeShutdownDeadlineBoundsPaneExitTeardownOwner(t *testing.T) {
483	482		t.Fatal("pane EOF teardown did not publish its terminal snapshot")
484	483		}
485	484		// The pane-exit owner has an independent final-snapshot timer. Keep it
486		-	// unfired so Serve must use, and remain bounded by, its shared deadline.
487		-	_ = clock.nextFinalTimer(t)
	485	+	// unfired until cleanup so Serve must use, and remain bounded by, its shared
	486	+	// deadline.
	487	+	ownerDeadlineTimer := clock.nextFinalTimer(t)
	488	+
	489	+	// Cleanup is the only place that releases the competing owner. It also fires
	490	+	// that owner's independent budget and joins the real pane reader, proving no
	491	+	// teardown goroutine is leaked without making Serve depend on repository I/O.
	492	+	t.Cleanup(func() {
	493	+	releaseSnapshot()
	494	+	ownerDeadlineTimer.fire()
	495	+	d.sessWg.Wait()
	496	+	})
488	497
489	498		cancel()
490	499		shutdownDeadlineTimer := clock.nextFinalTimer(t)
491		-	awaitTeardownWaiters(t, sess, 1)
492	500		shutdownDeadlineTimer.fire()
493		-	awaitTeardownWaiters(t, sess, 0)
494	501
495		-	sess.snapshotMu.Lock()
496		-	generation := sess.snapshotGeneration
497		-	sess.snapshotMu.Unlock()
498		-	require.Equal(t, uint64(1), generation, "Serve shutdown must not create a competing terminal generation")
499		-	require.Equal(t, int64(snapshotFinalFlushTimeout), clock.finalBudget.Load(), "Serve must spend only its shared snapshot deadline")
500		-
501		-	// Let the ordinary owner commit removal, then join the complete Serve and
502		-	// snapshot-worker lifecycles so no reader, teardown, or worker is leaked.
503		-	releaseSnapshot()
504	502		select {
505	503		case err := <-served:
506	504		require.NoError(t, err)
507	505		case <-time.After(testWaitTimeout):
508		-	t.Fatal("Serve did not return after the pane-exit owner completed")
509		-	}
510		-	d.snapshotWorkerMu.Lock()
511		-	workerDone := d.snapshotWorkerDone
512		-	d.snapshotWorkerMu.Unlock()
513		-	if workerDone != nil {
514		-	select {
515		-	case <-workerDone:
516		-	case <-time.After(testWaitTimeout):
517		-	t.Fatal("snapshot worker leaked after Serve returned")
518		-	}
	506	+	t.Fatal("Serve did not return after its shared deadline while the pane-exit owner remained blocked")
519	507		}
	508	+
	509	+	sess.snapshotMu.Lock()
	510	+	generation := sess.snapshotGeneration
	511	+	sess.snapshotMu.Unlock()
	512	+	require.Equal(t, uint64(1), generation, "Serve shutdown must not create a competing terminal generation")
	513	+	require.Equal(t, int64(snapshotFinalFlushTimeout), clock.finalBudget.Load(), "Serve must spend only its shared snapshot deadline")
520	514		}
521	515
522	516		func awaitTeardownWaiters(t testing.T, sess session, want uint) {

 	releasePublication := make(chan struct{})
 	var releasePublicationOnce sync.Once
 	releaseSnapshot := func() { releasePublicationOnce.Do(func() { close(releasePublication) }) }
-	t.Cleanup(releaseSnapshot)
 	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
 		require.Equal(t, uint64(1), publication.Generation)
 		close(publicationStarted)
 		t.Fatal("pane EOF teardown did not publish its terminal snapshot")
 	}
 	// The pane-exit owner has an independent final-snapshot timer. Keep it
-	// unfired so Serve must use, and remain bounded by, its shared deadline.
+	// unfired until cleanup so Serve must use, and remain bounded by, its shared
-	_ = clock.nextFinalTimer(t)
+	// deadline.
+	ownerDeadlineTimer := clock.nextFinalTimer(t)
+	// Cleanup is the only place that releases the competing owner. It also fires
+	// that owner's independent budget and joins the real pane reader, proving no
+	// teardown goroutine is leaked without making Serve depend on repository I/O.
+	t.Cleanup(func() {
+		releaseSnapshot()
+		ownerDeadlineTimer.fire()
+		d.sessWg.Wait()
+	})
 	cancel()
 	shutdownDeadlineTimer := clock.nextFinalTimer(t)
-	awaitTeardownWaiters(t, sess, 1)
 	shutdownDeadlineTimer.fire()
-	awaitTeardownWaiters(t, sess, 0)
-	sess.snapshotMu.Lock()
-	generation := sess.snapshotGeneration
-	sess.snapshotMu.Unlock()
-	require.Equal(t, uint64(1), generation, "Serve shutdown must not create a competing terminal generation")
-	require.Equal(t, int64(snapshotFinalFlushTimeout), clock.finalBudget.Load(), "Serve must spend only its shared snapshot deadline")
-	// Let the ordinary owner commit removal, then join the complete Serve and
-	// snapshot-worker lifecycles so no reader, teardown, or worker is leaked.
-	releaseSnapshot()
 	select {
 	case err := <-served:
 		require.NoError(t, err)
 	case <-time.After(testWaitTimeout):
-		t.Fatal("Serve did not return after the pane-exit owner completed")
+		t.Fatal("Serve did not return after its shared deadline while the pane-exit owner remained blocked")
-	}
-	d.snapshotWorkerMu.Lock()
-	workerDone := d.snapshotWorkerDone
-	d.snapshotWorkerMu.Unlock()
-	if workerDone != nil {
-		select {
-		case <-workerDone:
-		case <-time.After(testWaitTimeout):
-			t.Fatal("snapshot worker leaked after Serve returned")
-		}
 	}
+	sess.snapshotMu.Lock()
+	generation := sess.snapshotGeneration
+	sess.snapshotMu.Unlock()
+	require.Equal(t, uint64(1), generation, "Serve shutdown must not create a competing terminal generation")
+	require.Equal(t, int64(snapshotFinalFlushTimeout), clock.finalBudget.Load(), "Serve must spend only its shared snapshot deadline")
 }
 func awaitTeardownWaiters(t *testing.T, sess *session, want uint) {