d319b30 fix(runtime): address timer lifecycle review · vev

M internal/usecase/daemon/daemon.go +22 -32


@@ -659,6 +659,24 @@ type protoErr struct {
659	659
660	660		func (e *protoErr) Error() string { return e.text }
661	661
	662	+	// finishAttach completes an attachment prepared while d.mu is held. It
	663	+	// publishes the terminal state before releasing d.mu, then performs worker
	664	+	// joins and replacement teardown without the daemon lock.
	665	+	func (d Daemon) finishAttach(sess session, tr ports.Transport, sz domain.Size, term terminalEnv, h ports.Hello) *attachedClient {
	666	+	ac, old, cleanup := d.attachClientDeferred(sess, tr, sz, attachClientOptions{
	667	+	clientID: h.ClientID,
	668	+	resumeCapable: true,
	669	+	maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight),
	670	+	})
	671	+	sess.mu.Lock()
	672	+	sess.terminal = term
	673	+	sess.mu.Unlock()
	674	+	d.mu.Unlock()
	675	+	cleanup.finish()
	676	+	d.detachReplacedClient(old)
	677	+	return ac
	678	+	}
	679	+
662	680		// route resolves a Hello to a session and a freshly attached client, creating
663	681		// the session for ephemeral/new intents.
664	682		func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedClient, error) {

@@ -709,14 +727,7 @@ func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedCl
709	727		}
710	728		}
711	729		d.purgeParkedForSessionLocked(sess)
712		-	ac, old, cleanup := d.attachClientDeferred(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
713		-	sess.mu.Lock()
714		-	sess.terminal = term
715		-	sess.mu.Unlock()
716		-	d.mu.Unlock()
717		-	cleanup.finish()
718		-	d.detachReplacedClient(old)
719		-	return sess, ac, nil
	730	+	return sess, d.finishAttach(sess, tr, sz, term, h), nil
720	731
721	732		case ports.IntentEphemeral:
722	733		name := d.allocEphemeralNameLocked()

@@ -726,14 +737,7 @@ func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedCl
726	737		return nil, nil, err
727	738		}
728	739		d.purgeParkedForSessionLocked(sess)
729		-	ac, old, cleanup := d.attachClientDeferred(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
730		-	sess.mu.Lock()
731		-	sess.terminal = term
732		-	sess.mu.Unlock()
733		-	d.mu.Unlock()
734		-	cleanup.finish()
735		-	d.detachReplacedClient(old)
736		-	return sess, ac, nil
	740	+	return sess, d.finishAttach(sess, tr, sz, term, h), nil
737	741
738	742		case ports.IntentNew:
739	743		if h.Name == "" {

@@ -754,14 +758,7 @@ func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedCl
754	758		return nil, nil, err
755	759		}
756	760		d.purgeParkedForSessionLocked(sess)
757		-	ac, old, cleanup := d.attachClientDeferred(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
758		-	sess.mu.Lock()
759		-	sess.terminal = term
760		-	sess.mu.Unlock()
761		-	d.mu.Unlock()
762		-	cleanup.finish()
763		-	d.detachReplacedClient(old)
764		-	return sess, ac, nil
	761	+	return sess, d.finishAttach(sess, tr, sz, term, h), nil
765	762
766	763		case ports.IntentAttach:
767	764		sess := d.findByNameLocked(h.Name)

@@ -780,14 +777,7 @@ func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedCl
780	777		}
781	778		}
782	779		d.purgeParkedForSessionLocked(sess)
783		-	ac, old, cleanup := d.attachClientDeferred(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
784		-	sess.mu.Lock()
785		-	sess.terminal = term
786		-	sess.mu.Unlock()
787		-	d.mu.Unlock()
788		-	cleanup.finish()
789		-	d.detachReplacedClient(old)
790		-	return sess, ac, nil
	780	+	return sess, d.finishAttach(sess, tr, sz, term, h), nil
791	781
792	782		default:
793	783		d.mu.Unlock()

 func (e *protoErr) Error() string { return e.text }
+// finishAttach completes an attachment prepared while d.mu is held. It
+// publishes the terminal state before releasing d.mu, then performs worker
+// joins and replacement teardown without the daemon lock.
+func (d *Daemon) finishAttach(sess *session, tr ports.Transport, sz domain.Size, term terminalEnv, h ports.Hello) *attachedClient {
+	ac, old, cleanup := d.attachClientDeferred(sess, tr, sz, attachClientOptions{
+		clientID:          h.ClientID,
+		resumeCapable:     true,
+		maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight),
+	})
+	sess.mu.Lock()
+	sess.terminal = term
+	sess.mu.Unlock()
+	d.mu.Unlock()
+	cleanup.finish()
+	d.detachReplacedClient(old)
+	return ac
+}
 // route resolves a Hello to a session and a freshly attached client, creating
 // the session for ephemeral/new intents.
 func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedClient, error) {
 			}
 		}
 		d.purgeParkedForSessionLocked(sess)
-		ac, old, cleanup := d.attachClientDeferred(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
+		return sess, d.finishAttach(sess, tr, sz, term, h), nil
-		sess.mu.Lock()
-		sess.terminal = term
-		sess.mu.Unlock()
-		d.mu.Unlock()
-		cleanup.finish()
-		d.detachReplacedClient(old)
-		return sess, ac, nil
 	case ports.IntentEphemeral:
 		name := d.allocEphemeralNameLocked()
 			return nil, nil, err
 		}
 		d.purgeParkedForSessionLocked(sess)
-		ac, old, cleanup := d.attachClientDeferred(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
+		return sess, d.finishAttach(sess, tr, sz, term, h), nil
-		sess.mu.Lock()
-		sess.terminal = term
-		sess.mu.Unlock()
-		d.mu.Unlock()
-		cleanup.finish()
-		d.detachReplacedClient(old)
-		return sess, ac, nil
 	case ports.IntentNew:
 		if h.Name == "" {
 			return nil, nil, err
 		}
 		d.purgeParkedForSessionLocked(sess)
-		ac, old, cleanup := d.attachClientDeferred(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
+		return sess, d.finishAttach(sess, tr, sz, term, h), nil
-		sess.mu.Lock()
-		sess.terminal = term
-		sess.mu.Unlock()
-		d.mu.Unlock()
-		cleanup.finish()
-		d.detachReplacedClient(old)
-		return sess, ac, nil
 	case ports.IntentAttach:
 		sess := d.findByNameLocked(h.Name)
 			}
 		}
 		d.purgeParkedForSessionLocked(sess)
-		ac, old, cleanup := d.attachClientDeferred(sess, tr, sz, attachClientOptions{clientID: h.ClientID, resumeCapable: true, maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight)})
+		return sess, d.finishAttach(sess, tr, sz, term, h), nil
-		sess.mu.Lock()
-		sess.terminal = term
-		sess.mu.Unlock()
-		d.mu.Unlock()
-		cleanup.finish()
-		d.detachReplacedClient(old)
-		return sess, ac, nil
 	default:
 		d.mu.Unlock()

M internal/usecase/daemon/render_resize_deadline.go +6 -0


@@ -128,6 +128,9 @@ func (c renderCoordinator) scheduleResizeRetryForLease(epoch uint64, source at
128	128		c.mu.Unlock()
129	129		stopAndJoinTimerWorker(old, nil)
130	130		if clock == nil {
	131	+	c.mu.Lock()
	132	+	c.retryTimerOwner().completeLocked(gen)
	133	+	c.mu.Unlock()
131	134		if c.retryCurrentForLease(epoch, source, lease) {
132	135		run()
133	136		}

@@ -139,6 +142,9 @@ func (c renderCoordinator) scheduleResizeRetryForLease(epoch uint64, source at
139	142		// A nil timer channel is the deterministic disabled-clock contract used
140	143		// by headless tests; do not spin retries synchronously.
141	144		stopTimer(timer)
	145	+	c.mu.Lock()
	146	+	c.retryTimerOwner().completeLocked(gen)
	147	+	c.mu.Unlock()
142	148		return
143	149		}
144	150		c.mu.Lock()


@@ -128,6 +128,9 @@ func (c renderCoordinator) scheduleResizeRetryForLease(epoch uint64, source at
128	c.mu.Unlock()	128		c.mu.Unlock()
129	stopAndJoinTimerWorker(old, nil)	129		stopAndJoinTimerWorker(old, nil)
130	if clock == nil {	130		if clock == nil {
		131	+	c.mu.Lock()
		132	+	c.retryTimerOwner().completeLocked(gen)
		133	+	c.mu.Unlock()
131	if c.retryCurrentForLease(epoch, source, lease) {	134		if c.retryCurrentForLease(epoch, source, lease) {
132	run()	135		run()
133	}	136		}

@@ -139,6 +142,9 @@ func (c renderCoordinator) scheduleResizeRetryForLease(epoch uint64, source at
139	// A nil timer channel is the deterministic disabled-clock contract used	142		// A nil timer channel is the deterministic disabled-clock contract used
140	// by headless tests; do not spin retries synchronously.	143		// by headless tests; do not spin retries synchronously.
141	stopTimer(timer)	144		stopTimer(timer)
		145	+	c.mu.Lock()
		146	+	c.retryTimerOwner().completeLocked(gen)
		147	+	c.mu.Unlock()
142	return	148		return
143	}	149		}
144	c.mu.Lock()	150		c.mu.Lock()

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


@@ -19,7 +19,7 @@ func TestTimerOwnershipRejectsStaleTicketsAndCancelsWorkers(t *testing.T) {
19	19
20	20		ticket, stopped := owner.replaceLocked()
21	21		require.Nil(t, stopped.timer)
22		-	first := &portsmocks.MockTimer{}
	22	+	first := portsmocks.NewMockTimer(t)
23	23		first.EXPECT().Stop().Return(true).Once()
24	24		firstCancel, firstDone, ok := owner.publishLocked(ticket, first)
25	25		require.True(t, ok)

@@ -32,7 +32,7 @@ func TestTimerOwnershipRejectsStaleTicketsAndCancelsWorkers(t *testing.T) {
32	32		require.Same(t, first, stopped.timer)
33	33		stopAndJoinTimerWorker(stopped, nil)
34	34
35		-	_, _, ok = owner.publishLocked(ticket, &portsmocks.MockTimer{})
	35	+	_, _, ok = owner.publishLocked(ticket, portsmocks.NewMockTimer(t))
36	36		require.False(t, ok, "a stale generation must not reclaim a replacement lane")
37	37
38	38		require.True(t, owner.completeLocked(generation))


@@ -19,7 +19,7 @@ func TestTimerOwnershipRejectsStaleTicketsAndCancelsWorkers(t *testing.T) {
19			19
20		ticket, stopped := owner.replaceLocked()	20		ticket, stopped := owner.replaceLocked()
21		require.Nil(t, stopped.timer)	21		require.Nil(t, stopped.timer)
22	-	first := &portsmocks.MockTimer{}	22	+	first := portsmocks.NewMockTimer(t)
23		first.EXPECT().Stop().Return(true).Once()	23		first.EXPECT().Stop().Return(true).Once()
24		firstCancel, firstDone, ok := owner.publishLocked(ticket, first)	24		firstCancel, firstDone, ok := owner.publishLocked(ticket, first)
25		require.True(t, ok)	25		require.True(t, ok)

@@ -32,7 +32,7 @@ func TestTimerOwnershipRejectsStaleTicketsAndCancelsWorkers(t *testing.T) {
32		require.Same(t, first, stopped.timer)	32		require.Same(t, first, stopped.timer)
33		stopAndJoinTimerWorker(stopped, nil)	33		stopAndJoinTimerWorker(stopped, nil)
34			34
35	-	_, _, ok = owner.publishLocked(ticket, &portsmocks.MockTimer{})	35	+	_, _, ok = owner.publishLocked(ticket, portsmocks.NewMockTimer(t))
36		require.False(t, ok, "a stale generation must not reclaim a replacement lane")	36		require.False(t, ok, "a stale generation must not reclaim a replacement lane")
37			37
38		require.True(t, owner.completeLocked(generation))	38		require.True(t, owner.completeLocked(generation))