5a197b5 fix(daemon): preserve coordinator sync lifecycle · vev

M internal/usecase/daemon/client.go +3 -0


@@ -637,6 +637,9 @@ func (d Daemon) detachOnSendError(sess session, ac *attachedClient, failed por
637	637		}
638	638		if sess.detachIfCurrent(ac) {
639	639		ac.cancelResizePaint()
	640	+	if rc := sess.renderCoordinator(); rc != nil {
	641	+	rc.noteDetach(ac)
	642	+	}
640	643		d.unregisterPreview(ac)
641	644		if d.parkAttachment(sess, ac) {
642	645		_ = ac.closeCapturedTransport(failed)


@@ -637,6 +637,9 @@ func (d Daemon) detachOnSendError(sess session, ac *attachedClient, failed por
637	}	637		}
638	if sess.detachIfCurrent(ac) {	638		if sess.detachIfCurrent(ac) {
639	ac.cancelResizePaint()	639		ac.cancelResizePaint()
		640	+	if rc := sess.renderCoordinator(); rc != nil {
		641	+	rc.noteDetach(ac)
		642	+	}
640	d.unregisterPreview(ac)	643		d.unregisterPreview(ac)
641	if d.parkAttachment(sess, ac) {	644		if d.parkAttachment(sess, ac) {
642	_ = ac.closeCapturedTransport(failed)	645		_ = ac.closeCapturedTransport(failed)

M internal/usecase/daemon/render.go +20 -6


@@ -46,6 +46,15 @@ func (d Daemon) ptyReader(sess session, tb tab, p pane) {
46	46		p.screen.Write(data)
47	47		p.refreshTerminalTitleLocked()
48	48		isSyncing := p.screen.SyncUpdateActive()
	49	+	var syncGen uint64
	50	+	if wasSyncing != isSyncing {
	51	+	if isSyncing {
	52	+	syncGen = sess.syncGen.Add(1)
	53	+	p.syncGen = syncGen
	54	+	} else {
	55	+	syncGen = p.syncGen
	56	+	}
	57	+	}
49	58		p.mu.Unlock()
50	59		markSnapshotDirty(sess)
51	60		select {

@@ -67,17 +76,22 @@ func (d Daemon) ptyReader(sess session, tb tab, p pane) {
67	76		}
68	77		if rc := sess.renderCoordinator(); rc != nil {
69	78		if wasSyncing != isSyncing {
70		-	p.mu.Lock()
71		-	p.syncGen++
72		-	gen := p.syncGen
73		-	p.mu.Unlock()
	79	+	gen := syncGen
74	80		if isSyncing {
75		-	rc.noteSyncBegin(gen)
	81	+	rc.noteSyncBegin(gen, func() {
	82	+	p.mu.Lock()
	83	+	if p.syncGen == gen && p.screen.SyncUpdateActive() {
	84	+	p.screen.ForceSyncEnd()
	85	+	}
	86	+	p.mu.Unlock()
	87	+	})
	88	+	// The accumulated synchronized batch is the pending render.
	89	+	rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
76	90		} else {
77	91		rc.noteSyncEnd(gen)
78	92		}
79	93		}
80		-	if !isSyncing {
	94	+	if !isSyncing && wasSyncing == isSyncing {
81	95		rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
82	96		}
83	97		}

 			p.screen.Write(data)
 			p.refreshTerminalTitleLocked()
 			isSyncing := p.screen.SyncUpdateActive()
+			var syncGen uint64
+			if wasSyncing != isSyncing {
+				if isSyncing {
+					syncGen = sess.syncGen.Add(1)
+					p.syncGen = syncGen
+				} else {
+					syncGen = p.syncGen
+				}
+			}
 			p.mu.Unlock()
 			markSnapshotDirty(sess)
 			select {
 			}
 			if rc := sess.renderCoordinator(); rc != nil {
 				if wasSyncing != isSyncing {
-					p.mu.Lock()
+					gen := syncGen
-					p.syncGen++
-					gen := p.syncGen
-					p.mu.Unlock()
 					if isSyncing {
-						rc.noteSyncBegin(gen)
+						rc.noteSyncBegin(gen, func() {
+							p.mu.Lock()
+							if p.syncGen == gen && p.screen.SyncUpdateActive() {
+								p.screen.ForceSyncEnd()
+							}
+							p.mu.Unlock()
+						})
+						// The accumulated synchronized batch is the pending render.
+						rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
 					} else {
 						rc.noteSyncEnd(gen)
 					}
 				}
-				if !isSyncing {
+				if !isSyncing && wasSyncing == isSyncing {
 					rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
 				}
 			}

M internal/usecase/daemon/render_coordinator.go +18 -8


@@ -118,6 +118,7 @@ type renderCoordinator struct {
118	118		generation uint64
119	119		armed bool
120	120		syncGeneration uint64
	121	+	syncWatchdog func()
121	122		}
122	123
123	124		// newRenderCoordinator constructs a coordinator bound to opts. It starts no

@@ -177,14 +178,19 @@ func (c *renderCoordinator) notifyAck() {
177	178		}
178	179
179	180		// noteSyncBegin records that a synchronized-output batch opened for gen and
180		-	// arms the completion watchdog.
181		-	func (c *renderCoordinator) noteSyncBegin(gen uint64) {
	181	+	// arms the completion watchdog. force is supplied by the pane owner so the
	182	+	// coordinator can end a wedged VT batch before it composes the pending state.
	183	+	func (c *renderCoordinator) noteSyncBegin(gen uint64, force ...func()) {
182	184		c.mu.Lock()
183	185		if c.torndown {
184	186		c.mu.Unlock()
185	187		return
186	188		}
187	189		c.syncGeneration = gen
	190	+	c.syncWatchdog = nil
	191	+	if len(force) != 0 {
	192	+	c.syncWatchdog = force[0]
	193	+	}
188	194		clock := c.opts.clock
189	195		c.mu.Unlock()
190	196		if clock == nil {

@@ -195,8 +201,16 @@ func (c *renderCoordinator) noteSyncBegin(gen uint64) {
195	201		<-timer.C()
196	202		c.mu.Lock()
197	203		valid := !c.torndown && c.syncGeneration == gen
	204	+	force := c.syncWatchdog
	205	+	if valid {
	206	+	c.syncGeneration = 0
	207	+	c.syncWatchdog = nil
	208	+	}
198	209		c.mu.Unlock()
199	210		if valid {
	211	+	if force != nil {
	212	+	force()
	213	+	}
200	214		c.fireCurrent(true)
201	215		}
202	216		}()

@@ -211,6 +225,7 @@ func (c *renderCoordinator) noteSyncEnd(gen uint64) {
211	225		return
212	226		}
213	227		c.syncGeneration = 0
	228	+	c.syncWatchdog = nil
214	229		c.mu.Unlock()
215	230		c.fireCurrent(false)
216	231		}

@@ -293,11 +308,6 @@ func (c *renderCoordinator) resizeSnapshot() resizeRequestMetadata {
293	308		return c.resize
294	309		}
295	310
296		-	// burstMetrics returns an observational snapshot for benchmark reporting.
297		-	func (c *renderCoordinator) burstMetrics() (invalidations, wakes, coalesced uint64) {
298		-	return c.metrics.invalidations.Load(), c.metrics.wakes.Load(), c.metrics.coalesced.Load()
299		-	}
300		-
301	311		// renderCoordinator returns the coordinator installed for this session.
302	312		func (s session) renderCoordinator() renderCoordinator { return s.coordinator.Load() }
303	313

@@ -332,7 +342,7 @@ func (c *renderCoordinator) fire(gen uint64, watchdog bool) {
332	342		c.mu.Unlock()
333	343		return
334	344		}
335		-	if !watchdog && c.opts.syncActive != nil && c.opts.syncActive() {
	345	+	if !watchdog && c.syncGeneration != 0 {
336	346		c.mu.Unlock()
337	347		return
338	348		}

 	generation     uint64
 	armed          bool
 	syncGeneration uint64
+	syncWatchdog   func()
 }
 // newRenderCoordinator constructs a coordinator bound to opts. It starts no
 }
 // noteSyncBegin records that a synchronized-output batch opened for gen and
-// arms the completion watchdog.
+// arms the completion watchdog. force is supplied by the pane owner so the
-func (c *renderCoordinator) noteSyncBegin(gen uint64) {
+// coordinator can end a wedged VT batch before it composes the pending state.
+func (c *renderCoordinator) noteSyncBegin(gen uint64, force ...func()) {
 	c.mu.Lock()
 	if c.torndown {
 		c.mu.Unlock()
 		return
 	}
 	c.syncGeneration = gen
+	c.syncWatchdog = nil
+	if len(force) != 0 {
+		c.syncWatchdog = force[0]
+	}
 	clock := c.opts.clock
 	c.mu.Unlock()
 	if clock == nil {
 		<-timer.C()
 		c.mu.Lock()
 		valid := !c.torndown && c.syncGeneration == gen
+		force := c.syncWatchdog
+		if valid {
+			c.syncGeneration = 0
+			c.syncWatchdog = nil
+		}
 		c.mu.Unlock()
 		if valid {
+			if force != nil {
+				force()
+			}
 			c.fireCurrent(true)
 		}
 	}()
 		return
 	}
 	c.syncGeneration = 0
+	c.syncWatchdog = nil
 	c.mu.Unlock()
 	c.fireCurrent(false)
 }
 	return c.resize
 }
-// burstMetrics returns an observational snapshot for benchmark reporting.
-func (c *renderCoordinator) burstMetrics() (invalidations, wakes, coalesced uint64) {
-	return c.metrics.invalidations.Load(), c.metrics.wakes.Load(), c.metrics.coalesced.Load()
-}
 // renderCoordinator returns the coordinator installed for this session.
 func (s *session) renderCoordinator() *renderCoordinator { return s.coordinator.Load() }
 		c.mu.Unlock()
 		return
 	}
-	if !watchdog && c.opts.syncActive != nil && c.opts.syncActive() {
+	if !watchdog && c.syncGeneration != 0 {
 		c.mu.Unlock()
 		return
 	}

M internal/usecase/daemon/render_coordinator_test.go +4 -2


@@ -255,15 +255,17 @@ func TestRenderCoordinatorSynchronizedOutput(t *testing.T) {
255	255		requireNoWake(t, h.wakes)
256	256		})
257	257
258		-	t.Run("watchdog forces a flush for a wedged batch", func(t *testing.T) {
	258	+	t.Run("watchdog ends a wedged batch before it flushes", func(t *testing.T) {
259	259		h := newCoordinatorHarness(t)
260	260		h.syncActive.Store(true)
261		-	h.rc.noteSyncBegin(7)
	261	+	forced := make(chan struct{}, 1)
	262	+	h.rc.noteSyncBegin(7, func() { forced <- struct{}{} })
262	263		watchdogs := h.armedTimers(t)
263	264		require.NotEmpty(t, watchdogs, "sync begin must arm the completion watchdog")
264	265		h.rc.invalidate(renderInvalidation{class: invalidateOutput})
265	266
266	267		watchdogs[0].ch <- time.Time{}
	268	+	<-forced
267	269		w := awaitWake(t, h.wakes)
268	270		require.True(t, w.watchdog, "a wedged synchronized batch must be force-flushed by the watchdog")
269	271		requireNoWake(t, h.wakes)

 		requireNoWake(t, h.wakes)
 	})
-	t.Run("watchdog forces a flush for a wedged batch", func(t *testing.T) {
+	t.Run("watchdog ends a wedged batch before it flushes", func(t *testing.T) {
 		h := newCoordinatorHarness(t)
 		h.syncActive.Store(true)
-		h.rc.noteSyncBegin(7)
+		forced := make(chan struct{}, 1)
+		h.rc.noteSyncBegin(7, func() { forced <- struct{}{} })
 		watchdogs := h.armedTimers(t)
 		require.NotEmpty(t, watchdogs, "sync begin must arm the completion watchdog")
 		h.rc.invalidate(renderInvalidation{class: invalidateOutput})
 		watchdogs[0].ch <- time.Time{}
+		<-forced
 		w := awaitWake(t, h.wakes)
 		require.True(t, w.watchdog, "a wedged synchronized batch must be force-flushed by the watchdog")
 		requireNoWake(t, h.wakes)

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


@@ -44,6 +44,9 @@ type session struct {
44	44		mruAt atomic.Uint64
45	45		snapDirty atomic.Bool
46	46		snapEligible atomic.Bool
	47	+	// syncGen makes synchronized-output watchdog generations unique across all
	48	+	// panes in this session.
	49	+	syncGen atomic.Uint64
47	50		// coordinator fans in this session's producer render invalidations.
48	51		coordinator atomic.Pointer[renderCoordinator]
49	52		// clipFiles records clipboard-image-transfer temp file paths (see

@@ -706,6 +709,9 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
706	709		ac := sess.client
707	710		sess.client = nil
708	711		sess.mu.Unlock()
	712	+	if rc := sess.renderCoordinator(); rc != nil {
	713	+	rc.noteSessionTeardown()
	714	+	}
709	715		if ac != nil {
710	716		ac.cancelResizePaint()
711	717		d.unregisterPreview(ac)


@@ -44,6 +44,9 @@ type session struct {
44	mruAt atomic.Uint64	44		mruAt atomic.Uint64
45	snapDirty atomic.Bool	45		snapDirty atomic.Bool
46	snapEligible atomic.Bool	46		snapEligible atomic.Bool
		47	+	// syncGen makes synchronized-output watchdog generations unique across all
		48	+	// panes in this session.
		49	+	syncGen atomic.Uint64
47	// coordinator fans in this session's producer render invalidations.	50		// coordinator fans in this session's producer render invalidations.
48	coordinator atomic.Pointer[renderCoordinator]	51		coordinator atomic.Pointer[renderCoordinator]
49	// clipFiles records clipboard-image-transfer temp file paths (see	52		// clipFiles records clipboard-image-transfer temp file paths (see

@@ -706,6 +709,9 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
706	ac := sess.client	709		ac := sess.client
707	sess.client = nil	710		sess.client = nil
708	sess.mu.Unlock()	711		sess.mu.Unlock()
		712	+	if rc := sess.renderCoordinator(); rc != nil {
		713	+	rc.noteSessionTeardown()
		714	+	}
709	if ac != nil {	715		if ac != nil {
710	ac.cancelResizePaint()	716		ac.cancelResizePaint()
711	d.unregisterPreview(ac)	717		d.unregisterPreview(ac)