fix(obs): order ACK blocked span publication

commit: aefef6879278227c04a59b5bb265d15cbcc4089d
author: brice · 3mo ago · 2026-07-13 09:49 UTC
parent: 1f305e9
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +159 -39

M internal/usecase/daemon/render_coordinator.go +88 -39


@@ -108,6 +108,56 @@ type renderCoordinatorBurstMetrics struct {
108	108		coalesced atomic.Uint64
109	109		}
110	110
	111	+	// ackBlockedSpan hands an ACK-blocked endpoint pair between coordinator
	112	+	// transitions without holding c.mu while the observer runs. A consuming ACK
	113	+	// may arrive while Start is blocked in observer I/O; finish records that
	114	+	// closure and publishStart emits it only after Start has returned.
	115	+	type ackBlockedSpan struct {
	116	+	observer ports.RuntimeObserver
	117	+	correlation ports.RuntimeCorrelation
	118	+
	119	+	mu sync.Mutex
	120	+	startPublished bool
	121	+	finished bool
	122	+	endPending bool
	123	+	endValid bool
	124	+	}
	125	+
	126	+	func newACKBlockedSpan(observer ports.RuntimeObserver) *ackBlockedSpan {
	127	+	return &ackBlockedSpan{observer: observer, correlation: ports.NewRuntimeCorrelation()}
	128	+	}
	129	+
	130	+	func (s *ackBlockedSpan) publishStart() {
	131	+	s.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", s.correlation, ports.RuntimeACKBlockedStart, 0, true))
	132	+
	133	+	s.mu.Lock()
	134	+	s.startPublished = true
	135	+	endPending, endValid := s.endPending, s.endValid
	136	+	s.endPending = false
	137	+	s.mu.Unlock()
	138	+	if endPending {
	139	+	s.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", s.correlation, ports.RuntimeACKBlockedEnd, 0, endValid))
	140	+	}
	141	+	}
	142	+
	143	+	// finish closes a span at most once. It never invokes the observer before the
	144	+	// blocked start has completed publication.
	145	+	func (s *ackBlockedSpan) finish(valid bool) {
	146	+	s.mu.Lock()
	147	+	if s.finished {
	148	+	s.mu.Unlock()
	149	+	return
	150	+	}
	151	+	s.finished = true
	152	+	if !s.startPublished {
	153	+	s.endPending, s.endValid = true, valid
	154	+	s.mu.Unlock()
	155	+	return
	156	+	}
	157	+	s.mu.Unlock()
	158	+	s.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", s.correlation, ports.RuntimeACKBlockedEnd, 0, valid))
	159	+	}
	160	+
111	161		// renderCoordinatorBurstMetricsSnapshot is an immutable, internal view for
112	162		// benchmark reporting. Scheduling only writes the atomic counters above.
113	163		type renderCoordinatorBurstMetricsSnapshot struct {

@@ -136,10 +186,9 @@ type renderCoordinator struct {
136	186		// ackDeferred records that an expired deadline or watchdog was blocked by
137	187		// output-window capacity. ACK notifications may flush only this state.
138	188		ackDeferred bool
139		-	// ackBlocked retains one correlation from the first rejected ACK-capacity
140		-	// probe until notifyAck releases that exact blocked interval.
141		-	ackBlocked bool
142		-	ackCorrelation ports.RuntimeCorrelation
	189	+	// ackBlocked retains the first rejected ACK-capacity probe until the
	190	+	// matching consume or lifecycle closure releases that exact interval.
	191	+	ackBlocked *ackBlockedSpan
143	192		// deadlineDue closes the handoff between a timer worker receiving its tick
144	193		// and fire observing unavailable ACK capacity.
145	194		deadlineDue bool

@@ -398,11 +447,11 @@ func (c *renderCoordinator) notifyAck() {
398	447		return
399	448		}
400	449		gen := c.generation
401		-	observer, correlation, blocked := c.opts.observer, c.ackCorrelation, c.ackBlocked
402		-	c.ackBlocked = false
	450	+	blocked := c.ackBlocked
	451	+	c.ackBlocked = nil
403	452		c.mu.Unlock()
404		-	if blocked && observer != nil {
405		-	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, ports.RuntimeACKBlockedEnd, 0, true))
	453	+	if blocked != nil {
	454	+	blocked.finish(true)
406	455		}
407	456		c.fire(gen, false, false)
408	457		}

@@ -662,8 +711,9 @@ func (c renderCoordinator) noteDetach(ac attachedClient) {
662	711		c.mu.Lock()
663	712		var timer, resizeTimer, retryTimer ports.Timer
664	713		var observer ports.RuntimeObserver
665		-	var queueCorrelation, ackCorrelation ports.RuntimeCorrelation
666		-	var queueMarked, ackBlocked bool
	714	+	var queueCorrelation ports.RuntimeCorrelation
	715	+	var queueMarked bool
	716	+	var ackBlocked *ackBlockedSpan
667	717		if c.attachment == ac {
668	718		c.advanceAttachmentEpochLocked(ac)
669	719		c.attachment = nil

@@ -690,8 +740,8 @@ func (c renderCoordinator) noteDetach(ac attachedClient) {
690	740		retryTimer, c.retryTimer = c.retryTimer, nil
691	741		c.retryGen++
692	742		observer, queueCorrelation, queueMarked = c.opts.observer, c.queueCorrelation, c.queueMarked
693		-	ackCorrelation, ackBlocked = c.ackCorrelation, c.ackBlocked
694		-	c.queueMarked, c.ackBlocked = false, false
	743	+	ackBlocked = c.ackBlocked
	744	+	c.queueMarked, c.ackBlocked = false, nil
695	745		}
696	746		c.mu.Unlock()
697	747		stopTimer(timer)

@@ -700,8 +750,8 @@ func (c renderCoordinator) noteDetach(ac attachedClient) {
700	750		if queueMarked && observer != nil {
701	751		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
702	752		}
703		-	if ackBlocked && observer != nil {
704		-	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
	753	+	if ackBlocked != nil {
	754	+	ackBlocked.finish(false)
705	755		}
706	756		}
707	757

@@ -714,8 +764,9 @@ func (c renderCoordinator) noteReplace(old, replacement attachedClient, readin
714	764		}
715	765		c.mu.Lock()
716	766		var observer ports.RuntimeObserver
717		-	var queueCorrelation, ackCorrelation ports.RuntimeCorrelation
718		-	var queueMarked, ackBlocked bool
	767	+	var queueCorrelation ports.RuntimeCorrelation
	768	+	var queueMarked bool
	769	+	var ackBlocked *ackBlockedSpan
719	770		// A coordinator may be installed while replacing a legacy attachment
720	771		// which predated coordinator ownership. In that case nil has no pending
721	772		// identity to invalidate, and the replacement becomes the first bound

@@ -750,8 +801,8 @@ func (c renderCoordinator) noteReplace(old, replacement attachedClient, readin
750	801		retryTimer, c.retryTimer = c.retryTimer, nil
751	802		c.retryGen++
752	803		observer, queueCorrelation, queueMarked = c.opts.observer, c.queueCorrelation, c.queueMarked
753		-	ackCorrelation, ackBlocked = c.ackCorrelation, c.ackBlocked
754		-	c.queueMarked, c.ackBlocked = false, false
	804	+	ackBlocked = c.ackBlocked
	805	+	c.queueMarked, c.ackBlocked = false, nil
755	806		}
756	807		c.mu.Unlock()
757	808		stopTimer(timer)

@@ -760,8 +811,8 @@ func (c renderCoordinator) noteReplace(old, replacement attachedClient, readin
760	811		if queueMarked && observer != nil {
761	812		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
762	813		}
763		-	if ackBlocked && observer != nil {
764		-	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
	814	+	if ackBlocked != nil {
	815	+	ackBlocked.finish(false)
765	816		}
766	817		}
767	818

@@ -798,14 +849,14 @@ func (c *renderCoordinator) noteSessionTeardown() {
798	849		timer := c.detachNormalTimerLocked()
799	850		timers := c.detachSyncTimersLocked()
800	851		observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
801		-	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
802		-	c.queueMarked, c.ackBlocked = false, false
	852	+	ackBlocked := c.ackBlocked
	853	+	c.queueMarked, c.ackBlocked = false, nil
803	854		c.mu.Unlock()
804	855		if queueMarked && observer != nil {
805	856		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
806	857		}
807		-	if ackBlocked && observer != nil {
808		-	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
	858	+	if ackBlocked != nil {
	859	+	ackBlocked.finish(false)
809	860		}
810	861		stopTimer(timer)
811	862		stopTimer(resizeTimer)

@@ -1111,23 +1162,21 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) {
1111	1162		// notification rather than accumulating ackDeferred forever.
1112	1163		headlessPreviewOnly := c.detached && c.attachment == nil
1113	1164		if !headlessPreviewOnly && (!ready \|\| (c.attachment != nil && !c.attachmentReady)) {
1114		-	var ackStartObserver ports.RuntimeObserver
1115		-	var ackStartCorrelation ports.RuntimeCorrelation
	1165	+	var ackStart *ackBlockedSpan
1116	1166		if !ready && deadline {
1117	1167		c.ackDeferred = true
1118		-	if !c.ackBlocked && c.opts.observer != nil {
1119		-	c.ackCorrelation = ports.NewRuntimeCorrelation()
1120		-	c.ackBlocked = true
1121		-	// Capture the immutable mark while coordinator state is owned, but
1122		-	// invoke the potentially blocking observer only after c.mu is free.
1123		-	ackStartObserver = c.opts.observer
1124		-	ackStartCorrelation = c.ackCorrelation
	1168	+	if c.ackBlocked == nil && c.opts.observer != nil {
	1169	+	// Install the span while coordinator state is owned. Its Start
	1170	+	// publication runs unlocked, and its own handoff defers End until
	1171	+	// that call has returned.
	1172	+	ackStart = newACKBlockedSpan(c.opts.observer)
	1173	+	c.ackBlocked = ackStart
1125	1174		}
1126	1175		}
1127	1176		preview, previews := c.takePendingPreviewsLocked()
1128	1177		c.mu.Unlock()
1129		-	if ackStartObserver != nil {
1130		-	ackStartObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackStartCorrelation, ports.RuntimeACKBlockedStart, 0, true))
	1178	+	if ackStart != nil {
	1179	+	ackStart.publishStart()
1131	1180		}
1132	1181		c.notifyPreviews(w, preview, previews)
1133	1182		return

@@ -1148,16 +1197,16 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) {
1148	1197		timer := c.detachNormalTimerLocked()
1149	1198		c.pending, c.pendingReset, c.pendingUrgent, c.ackDeferred, c.deadlineDue, c.coalesced, c.armed = false, false, false, false, false, 0, false
1150	1199		observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
1151		-	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
1152		-	c.queueMarked, c.ackBlocked = false, false
	1200	+	ackBlocked := c.ackBlocked
	1201	+	c.queueMarked, c.ackBlocked = false, nil
1153	1202		wake := c.opts.wake
1154	1203		c.mu.Unlock()
1155	1204		stopTimer(timer)
1156	1205		if queueMarked && observer != nil {
1157	1206		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, true))
1158	1207		}
1159		-	if ackBlocked && observer != nil {
1160		-	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, true))
	1208	+	if ackBlocked != nil {
	1209	+	ackBlocked.finish(true)
1161	1210		}
1162	1211		if wake != nil && !headlessPreviewOnly {
1163	1212		wake(w)

 	coalesced     atomic.Uint64
 }
+// ackBlockedSpan hands an ACK-blocked endpoint pair between coordinator
+// transitions without holding c.mu while the observer runs. A consuming ACK
+// may arrive while Start is blocked in observer I/O; finish records that
+// closure and publishStart emits it only after Start has returned.
+type ackBlockedSpan struct {
+	observer    ports.RuntimeObserver
+	correlation ports.RuntimeCorrelation
+	mu             sync.Mutex
+	startPublished bool
+	finished       bool
+	endPending     bool
+	endValid       bool
+}
+func newACKBlockedSpan(observer ports.RuntimeObserver) *ackBlockedSpan {
+	return &ackBlockedSpan{observer: observer, correlation: ports.NewRuntimeCorrelation()}
+}
+func (s *ackBlockedSpan) publishStart() {
+	s.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", s.correlation, ports.RuntimeACKBlockedStart, 0, true))
+	s.mu.Lock()
+	s.startPublished = true
+	endPending, endValid := s.endPending, s.endValid
+	s.endPending = false
+	s.mu.Unlock()
+	if endPending {
+		s.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", s.correlation, ports.RuntimeACKBlockedEnd, 0, endValid))
+	}
+}
+// finish closes a span at most once. It never invokes the observer before the
+// blocked start has completed publication.
+func (s *ackBlockedSpan) finish(valid bool) {
+	s.mu.Lock()
+	if s.finished {
+		s.mu.Unlock()
+		return
+	}
+	s.finished = true
+	if !s.startPublished {
+		s.endPending, s.endValid = true, valid
+		s.mu.Unlock()
+		return
+	}
+	s.mu.Unlock()
+	s.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", s.correlation, ports.RuntimeACKBlockedEnd, 0, valid))
+}
 // renderCoordinatorBurstMetricsSnapshot is an immutable, internal view for
 // benchmark reporting. Scheduling only writes the atomic counters above.
 type renderCoordinatorBurstMetricsSnapshot struct {
 	// ackDeferred records that an expired deadline or watchdog was blocked by
 	// output-window capacity. ACK notifications may flush only this state.
 	ackDeferred bool
-	// ackBlocked retains one correlation from the first rejected ACK-capacity
+	// ackBlocked retains the first rejected ACK-capacity probe until the
-	// probe until notifyAck releases that exact blocked interval.
+	// matching consume or lifecycle closure releases that exact interval.
-	ackBlocked     bool
+	ackBlocked *ackBlockedSpan
-	ackCorrelation ports.RuntimeCorrelation
 	// deadlineDue closes the handoff between a timer worker receiving its tick
 	// and fire observing unavailable ACK capacity.
 	deadlineDue bool
 		return
 	}
 	gen := c.generation
-	observer, correlation, blocked := c.opts.observer, c.ackCorrelation, c.ackBlocked
+	blocked := c.ackBlocked
-	c.ackBlocked = false
+	c.ackBlocked = nil
 	c.mu.Unlock()
-	if blocked && observer != nil {
+	if blocked != nil {
-		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, ports.RuntimeACKBlockedEnd, 0, true))
+		blocked.finish(true)
 	}
 	c.fire(gen, false, false)
 }
 	c.mu.Lock()
 	var timer, resizeTimer, retryTimer ports.Timer
 	var observer ports.RuntimeObserver
-	var queueCorrelation, ackCorrelation ports.RuntimeCorrelation
+	var queueCorrelation ports.RuntimeCorrelation
-	var queueMarked, ackBlocked bool
+	var queueMarked bool
+	var ackBlocked *ackBlockedSpan
 	if c.attachment == ac {
 		c.advanceAttachmentEpochLocked(ac)
 		c.attachment = nil
 		retryTimer, c.retryTimer = c.retryTimer, nil
 		c.retryGen++
 		observer, queueCorrelation, queueMarked = c.opts.observer, c.queueCorrelation, c.queueMarked
-		ackCorrelation, ackBlocked = c.ackCorrelation, c.ackBlocked
+		ackBlocked = c.ackBlocked
-		c.queueMarked, c.ackBlocked = false, false
+		c.queueMarked, c.ackBlocked = false, nil
 	}
 	c.mu.Unlock()
 	stopTimer(timer)
 	if queueMarked && observer != nil {
 		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
 	}
-	if ackBlocked && observer != nil {
+	if ackBlocked != nil {
-		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
+		ackBlocked.finish(false)
 	}
 }
 	}
 	c.mu.Lock()
 	var observer ports.RuntimeObserver
-	var queueCorrelation, ackCorrelation ports.RuntimeCorrelation
+	var queueCorrelation ports.RuntimeCorrelation
-	var queueMarked, ackBlocked bool
+	var queueMarked bool
+	var ackBlocked *ackBlockedSpan
 	// A coordinator may be installed while replacing a legacy attachment
 	// which predated coordinator ownership. In that case nil has no pending
 	// identity to invalidate, and the replacement becomes the first bound
 		retryTimer, c.retryTimer = c.retryTimer, nil
 		c.retryGen++
 		observer, queueCorrelation, queueMarked = c.opts.observer, c.queueCorrelation, c.queueMarked
-		ackCorrelation, ackBlocked = c.ackCorrelation, c.ackBlocked
+		ackBlocked = c.ackBlocked
-		c.queueMarked, c.ackBlocked = false, false
+		c.queueMarked, c.ackBlocked = false, nil
 	}
 	c.mu.Unlock()
 	stopTimer(timer)
 	if queueMarked && observer != nil {
 		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
 	}
-	if ackBlocked && observer != nil {
+	if ackBlocked != nil {
-		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
+		ackBlocked.finish(false)
 	}
 }
 	timer := c.detachNormalTimerLocked()
 	timers := c.detachSyncTimersLocked()
 	observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
-	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
+	ackBlocked := c.ackBlocked
-	c.queueMarked, c.ackBlocked = false, false
+	c.queueMarked, c.ackBlocked = false, nil
 	c.mu.Unlock()
 	if queueMarked && observer != nil {
 		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
 	}
-	if ackBlocked && observer != nil {
+	if ackBlocked != nil {
-		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
+		ackBlocked.finish(false)
 	}
 	stopTimer(timer)
 	stopTimer(resizeTimer)
 	// notification rather than accumulating ackDeferred forever.
 	headlessPreviewOnly := c.detached && c.attachment == nil
 	if !headlessPreviewOnly && (!ready || (c.attachment != nil && !c.attachmentReady)) {
-		var ackStartObserver ports.RuntimeObserver
+		var ackStart *ackBlockedSpan
-		var ackStartCorrelation ports.RuntimeCorrelation
 		if !ready && deadline {
 			c.ackDeferred = true
-			if !c.ackBlocked && c.opts.observer != nil {
+			if c.ackBlocked == nil && c.opts.observer != nil {
-				c.ackCorrelation = ports.NewRuntimeCorrelation()
+				// Install the span while coordinator state is owned. Its Start
-				c.ackBlocked = true
+				// publication runs unlocked, and its own handoff defers End until
-				// Capture the immutable mark while coordinator state is owned, but
+				// that call has returned.
-				// invoke the potentially blocking observer only after c.mu is free.
+				ackStart = newACKBlockedSpan(c.opts.observer)
-				ackStartObserver = c.opts.observer
+				c.ackBlocked = ackStart
-				ackStartCorrelation = c.ackCorrelation
 			}
 		}
 		preview, previews := c.takePendingPreviewsLocked()
 		c.mu.Unlock()
-		if ackStartObserver != nil {
+		if ackStart != nil {
-			ackStartObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackStartCorrelation, ports.RuntimeACKBlockedStart, 0, true))
+			ackStart.publishStart()
 		}
 		c.notifyPreviews(w, preview, previews)
 		return
 	timer := c.detachNormalTimerLocked()
 	c.pending, c.pendingReset, c.pendingUrgent, c.ackDeferred, c.deadlineDue, c.coalesced, c.armed = false, false, false, false, false, 0, false
 	observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
-	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
+	ackBlocked := c.ackBlocked
-	c.queueMarked, c.ackBlocked = false, false
+	c.queueMarked, c.ackBlocked = false, nil
 	wake := c.opts.wake
 	c.mu.Unlock()
 	stopTimer(timer)
 	if queueMarked && observer != nil {
 		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, true))
 	}
-	if ackBlocked && observer != nil {
+	if ackBlocked != nil {
-		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, true))
+		ackBlocked.finish(true)
 	}
 	if wake != nil && !headlessPreviewOnly {
 		wake(w)

M internal/usecase/daemon/transport_observability_test.go +71 -0


@@ -253,3 +253,74 @@ func TestTransportObservabilityBlockedCoordinatorObserverReleasesLock(t *testing
253	253		close(observer.release)
254	254		awaitDaemonObserver(t, fired, "ACK fire completion")
255	255		}
	256	+
	257	+	type ackSpanInterleavingObserver struct {
	258	+	mu sync.Mutex
	259	+	marks []ports.RuntimeMark
	260	+	startEntered chan struct{}
	261	+	releaseStart chan struct{}
	262	+	}
	263	+
	264	+	func newACKSpanInterleavingObserver() *ackSpanInterleavingObserver {
	265	+	return &ackSpanInterleavingObserver{startEntered: make(chan struct{}), releaseStart: make(chan struct{})}
	266	+	}
	267	+
	268	+	func (o *ackSpanInterleavingObserver) ObserveRuntime(mark ports.RuntimeMark) {
	269	+	if mark.Kind == ports.RuntimeACKBlockedStart {
	270	+	close(o.startEntered)
	271	+	<-o.releaseStart
	272	+	}
	273	+	o.mu.Lock()
	274	+	o.marks = append(o.marks, mark)
	275	+	o.mu.Unlock()
	276	+	}
	277	+
	278	+	func (o *ackSpanInterleavingObserver) snapshot() []ports.RuntimeMark {
	279	+	o.mu.Lock()
	280	+	defer o.mu.Unlock()
	281	+	return append([]ports.RuntimeMark(nil), o.marks...)
	282	+	}
	283	+
	284	+	// An ACK can consume deferred work while the start mark is still waiting on
	285	+	// observer I/O. The End must wait for that Start publication, but the ACK
	286	+	// itself must not wait for either observer call.
	287	+	func TestTransportObservabilityACKBlockedStartPublishesBeforeEnd(t *testing.T) {
	288	+	observer := newACKSpanInterleavingObserver()
	289	+	ready := false
	290	+	rc := newRenderCoordinator(renderCoordinatorOptions{
	291	+	observer: observer,
	292	+	ackReady: func() bool { return ready },
	293	+	})
	294	+	rc.mu.Lock()
	295	+	rc.pending = true
	296	+	gen := rc.generation
	297	+	rc.mu.Unlock()
	298	+
	299	+	fireDone := make(chan struct{})
	300	+	go func() {
	301	+	rc.fire(gen, false, true)
	302	+	close(fireDone)
	303	+	}()
	304	+	awaitDaemonObserver(t, observer.startEntered, "ACK-blocked start observer")
	305	+
	306	+	ready = true
	307	+	ackDone := make(chan struct{})
	308	+	go func() {
	309	+	rc.notifyAck()
	310	+	close(ackDone)
	311	+	}()
	312	+	awaitDaemonObserver(t, ackDone, "nonblocking ACK progress")
	313	+	if marks := observer.snapshot(); len(marks) != 0 {
	314	+	t.Fatalf("ACK-blocked end published before its blocked start: %#v", marks)
	315	+	}
	316	+
	317	+	close(observer.releaseStart)
	318	+	awaitDaemonObserver(t, fireDone, "ACK-blocked start completion")
	319	+	marks := observer.snapshot()
	320	+	if len(marks) != 2 \|\| marks[0].Kind != ports.RuntimeACKBlockedStart \|\| marks[1].Kind != ports.RuntimeACKBlockedEnd {
	321	+	t.Fatalf("ACK-blocked marks = %#v, want one ordered start/end pair", marks)
	322	+	}
	323	+	if marks[0].Sequence != marks[1].Sequence \|\| marks[0].RequestID != marks[1].RequestID \|\| marks[0].Epoch != marks[1].Epoch {
	324	+	t.Fatalf("ACK-blocked span correlation changed: start=%+v end=%+v", marks[0], marks[1])
	325	+	}
	326	+	}


@@ -253,3 +253,74 @@ func TestTransportObservabilityBlockedCoordinatorObserverReleasesLock(t *testing
253	close(observer.release)	253		close(observer.release)
254	awaitDaemonObserver(t, fired, "ACK fire completion")	254		awaitDaemonObserver(t, fired, "ACK fire completion")
255	}	255		}
		256	+
		257	+	type ackSpanInterleavingObserver struct {
		258	+	mu sync.Mutex
		259	+	marks []ports.RuntimeMark
		260	+	startEntered chan struct{}
		261	+	releaseStart chan struct{}
		262	+	}
		263	+
		264	+	func newACKSpanInterleavingObserver() *ackSpanInterleavingObserver {
		265	+	return &ackSpanInterleavingObserver{startEntered: make(chan struct{}), releaseStart: make(chan struct{})}
		266	+	}
		267	+
		268	+	func (o *ackSpanInterleavingObserver) ObserveRuntime(mark ports.RuntimeMark) {
		269	+	if mark.Kind == ports.RuntimeACKBlockedStart {
		270	+	close(o.startEntered)
		271	+	<-o.releaseStart
		272	+	}
		273	+	o.mu.Lock()
		274	+	o.marks = append(o.marks, mark)
		275	+	o.mu.Unlock()
		276	+	}
		277	+
		278	+	func (o *ackSpanInterleavingObserver) snapshot() []ports.RuntimeMark {
		279	+	o.mu.Lock()
		280	+	defer o.mu.Unlock()
		281	+	return append([]ports.RuntimeMark(nil), o.marks...)
		282	+	}
		283	+
		284	+	// An ACK can consume deferred work while the start mark is still waiting on
		285	+	// observer I/O. The End must wait for that Start publication, but the ACK
		286	+	// itself must not wait for either observer call.
		287	+	func TestTransportObservabilityACKBlockedStartPublishesBeforeEnd(t *testing.T) {
		288	+	observer := newACKSpanInterleavingObserver()
		289	+	ready := false
		290	+	rc := newRenderCoordinator(renderCoordinatorOptions{
		291	+	observer: observer,
		292	+	ackReady: func() bool { return ready },
		293	+	})
		294	+	rc.mu.Lock()
		295	+	rc.pending = true
		296	+	gen := rc.generation
		297	+	rc.mu.Unlock()
		298	+
		299	+	fireDone := make(chan struct{})
		300	+	go func() {
		301	+	rc.fire(gen, false, true)
		302	+	close(fireDone)
		303	+	}()
		304	+	awaitDaemonObserver(t, observer.startEntered, "ACK-blocked start observer")
		305	+
		306	+	ready = true
		307	+	ackDone := make(chan struct{})
		308	+	go func() {
		309	+	rc.notifyAck()
		310	+	close(ackDone)
		311	+	}()
		312	+	awaitDaemonObserver(t, ackDone, "nonblocking ACK progress")
		313	+	if marks := observer.snapshot(); len(marks) != 0 {
		314	+	t.Fatalf("ACK-blocked end published before its blocked start: %#v", marks)
		315	+	}
		316	+
		317	+	close(observer.releaseStart)
		318	+	awaitDaemonObserver(t, fireDone, "ACK-blocked start completion")
		319	+	marks := observer.snapshot()
		320	+	if len(marks) != 2 \|\| marks[0].Kind != ports.RuntimeACKBlockedStart \|\| marks[1].Kind != ports.RuntimeACKBlockedEnd {
		321	+	t.Fatalf("ACK-blocked marks = %#v, want one ordered start/end pair", marks)
		322	+	}
		323	+	if marks[0].Sequence != marks[1].Sequence \|\| marks[0].RequestID != marks[1].RequestID \|\| marks[0].Epoch != marks[1].Epoch {
		324	+	t.Fatalf("ACK-blocked span correlation changed: start=%+v end=%+v", marks[0], marks[1])
		325	+	}
		326	+	}