fix(daemon): evaluate ack readiness outside coordinator lock

commit: c6654e67df5ff5f6b9e59f6a4b2c8371bf29edd0
author: brice · 3mo ago · 2026-07-11 19:02 UTC
parent: b906d90
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +38 -12

M internal/usecase/daemon/render_coordinator.go +29 -7


@@ -203,9 +203,7 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) {
203	203		c.mu.Unlock()
204	204		return
205	205		}
206		-	if c.opts.onInvalidate != nil {
207		-	c.opts.onInvalidate(inv)
208		-	}
	206	+	onInvalidate := c.opts.onInvalidate
209	207		c.metrics.invalidations.Add(1)
210	208		wasPending, wasUrgent, wasPreviewPending := c.pending, c.pendingUrgent, c.pendingPreview
211	209		if !wasPending {

@@ -235,6 +233,9 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) {
235	233		c.armed = c.armed \|\| arm
236	234		if !arm \|\| clock == nil {
237	235		c.mu.Unlock()
	236	+	if onInvalidate != nil {
	237	+	onInvalidate(inv)
	238	+	}
238	239		return
239	240		}
240	241		timer := clock.NewTimer(delay)

@@ -243,6 +244,9 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) {
243	244		// synchronously before allocating a worker or its coordination channels.
244	245		if timerC == nil {
245	246		c.mu.Unlock()
	247	+	if onInvalidate != nil {
	248	+	onInvalidate(inv)
	249	+	}
246	250		timer.Stop()
247	251		c.fire(gen, false, true)
248	252		return

@@ -251,6 +255,9 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) {
251	255		done := make(chan struct{})
252	256		c.normalTimer, c.normalCancel, c.normalWorkerDone = timer, cancel, done
253	257		c.mu.Unlock()
	258	+	if onInvalidate != nil {
	259	+	onInvalidate(inv)
	260	+	}
254	261		go func() {
255	262		defer close(done)
256	263		select {

@@ -536,25 +543,36 @@ func (d Daemon) invalidateRenderNow(sess session, ac *attachedClient, reset bo
536	543		}
537	544
538	545		func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) {
	546	+	// Readiness probes enter the attachment send path, which may in turn read
	547	+	// coordinator metadata. Never hold c.mu across that external callback.
539	548		c.mu.Lock()
540		-	if c.torndown \|\| !c.pending \|\| (!watchdog && gen != c.generation) {
	549	+	if !c.fireValidLocked(gen, watchdog) {
541	550		c.mu.Unlock()
542	551		return
543	552		}
544		-	if len(c.syncBatches) != 0 {
	553	+	ackReady := c.opts.ackReady
	554	+	c.mu.Unlock()
	555	+
	556	+	ready := ackReady == nil \|\| ackReady()
	557	+
	558	+	// The callback can detach, replace, start a synchronized batch, or publish
	559	+	// newer work. Revalidate every consumption precondition after it returns.
	560	+	c.mu.Lock()
	561	+	if !c.fireValidLocked(gen, watchdog) {
545	562		c.mu.Unlock()
546	563		return
547	564		}
548	565		w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, watchdog: watchdog}
549		-	preview, previews := c.takePendingPreviewsLocked()
550		-	if c.opts.ackReady != nil && !c.opts.ackReady() {
	566	+	if !ready {
551	567		if deadline {
552	568		c.ackDeferred = true
553	569		}
	570	+	preview, previews := c.takePendingPreviewsLocked()
554	571		c.mu.Unlock()
555	572		c.notifyPreviews(w, preview, previews)
556	573		return
557	574		}
	575	+	preview, previews := c.takePendingPreviewsLocked()
558	576		c.metrics.wakes.Add(1)
559	577		c.metrics.coalesced.Add(uint64(w.coalesced))
560	578		if !w.urgent {

@@ -577,6 +595,10 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) {
577	595		c.notifyPreviews(w, preview, previews)
578	596		}
579	597
	598	+	func (c *renderCoordinator) fireValidLocked(gen uint64, watchdog bool) bool {
	599	+	return !c.torndown && c.pending && (watchdog \|\| gen == c.generation) && len(c.syncBatches) == 0
	600	+	}
	601	+
580	602		// takePendingPreviewsLocked snapshots and consumes the preview delivery for
581	603		// this target generation. Unlike the target primary, this is never ACK-gated.
582	604		func (c *renderCoordinator) takePendingPreviewsLocked() (func(renderWake), []func(renderWake)) {

 		c.mu.Unlock()
 		return
 	}
-	if c.opts.onInvalidate != nil {
+	onInvalidate := c.opts.onInvalidate
-		c.opts.onInvalidate(inv)
-	}
 	c.metrics.invalidations.Add(1)
 	wasPending, wasUrgent, wasPreviewPending := c.pending, c.pendingUrgent, c.pendingPreview
 	if !wasPending {
 	c.armed = c.armed || arm
 	if !arm || clock == nil {
 		c.mu.Unlock()
+		if onInvalidate != nil {
+			onInvalidate(inv)
+		}
 		return
 	}
 	timer := clock.NewTimer(delay)
 	// synchronously before allocating a worker or its coordination channels.
 	if timerC == nil {
 		c.mu.Unlock()
+		if onInvalidate != nil {
+			onInvalidate(inv)
+		}
 		timer.Stop()
 		c.fire(gen, false, true)
 		return
 	done := make(chan struct{})
 	c.normalTimer, c.normalCancel, c.normalWorkerDone = timer, cancel, done
 	c.mu.Unlock()
+	if onInvalidate != nil {
+		onInvalidate(inv)
+	}
 	go func() {
 		defer close(done)
 		select {
 }
 func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) {
+	// Readiness probes enter the attachment send path, which may in turn read
+	// coordinator metadata. Never hold c.mu across that external callback.
 	c.mu.Lock()
-	if c.torndown || !c.pending || (!watchdog && gen != c.generation) {
+	if !c.fireValidLocked(gen, watchdog) {
 		c.mu.Unlock()
 		return
 	}
-	if len(c.syncBatches) != 0 {
+	ackReady := c.opts.ackReady
+	c.mu.Unlock()
+	ready := ackReady == nil || ackReady()
+	// The callback can detach, replace, start a synchronized batch, or publish
+	// newer work. Revalidate every consumption precondition after it returns.
+	c.mu.Lock()
+	if !c.fireValidLocked(gen, watchdog) {
 		c.mu.Unlock()
 		return
 	}
 	w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, watchdog: watchdog}
-	preview, previews := c.takePendingPreviewsLocked()
+	if !ready {
-	if c.opts.ackReady != nil && !c.opts.ackReady() {
 		if deadline {
 			c.ackDeferred = true
 		}
+		preview, previews := c.takePendingPreviewsLocked()
 		c.mu.Unlock()
 		c.notifyPreviews(w, preview, previews)
 		return
 	}
+	preview, previews := c.takePendingPreviewsLocked()
 	c.metrics.wakes.Add(1)
 	c.metrics.coalesced.Add(uint64(w.coalesced))
 	if !w.urgent {
 	c.notifyPreviews(w, preview, previews)
 }
+func (c *renderCoordinator) fireValidLocked(gen uint64, watchdog bool) bool {
+	return !c.torndown && c.pending && (watchdog || gen == c.generation) && len(c.syncBatches) == 0
+}
 // takePendingPreviewsLocked snapshots and consumes the preview delivery for
 // this target generation. Unlike the target primary, this is never ACK-gated.
 func (c *renderCoordinator) takePendingPreviewsLocked() (func(renderWake), []func(renderWake)) {

M internal/usecase/daemon/render_coordinator_test.go +9 -5


@@ -354,12 +354,16 @@ resize:
354	354		select {
355	355		case epoch := <-resizeDone:
356	356		require.Equal(t, uint64(1), epoch)
357		-	select {
358		-	case <-fireDone:
359		-	case <-wakes:
360		-	t.Fatal("fire did not return after wake")
	357	+	require.Equal(t, renderWake{coalesced: 1}, awaitWake(t, wakes))
	358	+	for range 4096 {
	359	+	select {
	360	+	case <-fireDone:
	361	+	return
	362	+	default:
	363	+	runtime.Gosched()
	364	+	}
361	365		}
362		-	return
	366	+	t.Fatal("fire did not return after wake")
363	367		default:
364	368		runtime.Gosched()
365	369		}

 		select {
 		case epoch := <-resizeDone:
 			require.Equal(t, uint64(1), epoch)
-			select {
+			require.Equal(t, renderWake{coalesced: 1}, awaitWake(t, wakes))
-			case <-fireDone:
+			for range 4096 {
-			case <-wakes:
+				select {
-				t.Fatal("fire did not return after wake")
+				case <-fireDone:
+					return
+				default:
+					runtime.Gosched()
+				}
 			}
-			return
+			t.Fatal("fire did not return after wake")
 		default:
 			runtime.Gosched()
 		}