Merge pull request 'fix: CI flakes in render coordinator readiness and quicnettest event order' (#16) from fix/ci-race-and-proxy-event-order into main

commit: 9ff586df8173b0dac1c1faf24238b3f4edcda434
author: bnema · 14d ago · 2026-10-08 08:10 UTC
merge of: f67cb47 + fe90d45 · diff against the first parent
signature: unsigned

2 files · +20 -17

M internal/adapters/quicnettest/proxy.go +11 -8


@@ -468,6 +468,17 @@ func (p Proxy) ingest(toServer bool, raw []byte, dst net.UDPAddr) {
468	468		counters.reordered.Add(1)
469	469		}
470	470		due := now.Add(link.BaseLatency + decision.ExtraDelay)
	471	+	// Emit before scheduling: once enqueued, a zero-latency copy can be
	472	+	// delivered and answered before this call returns, and the reply's event
	473	+	// must not be observed ahead of the packet that caused it.
	474	+	p.emit(Event{
	475	+	Direction: directionOf(toServer),
	476	+	Sequence: seq,
	477	+	Bytes: len(raw),
	478	+	At: now,
	479	+	Due: due,
	480	+	Decision: decision,
	481	+	})
471	482		for copyIndex := 0; copyIndex <= decision.Duplicates; copyIndex++ {
472	483		copyDue := due.Add(time.Duration(copyIndex) * link.DuplicateDelay)
473	484		if !p.enqueue(packet{

@@ -480,14 +491,6 @@ func (p Proxy) ingest(toServer bool, raw []byte, dst net.UDPAddr) {
480	491		counters.overflowDrops.Add(1)
481	492		}
482	493		}
483		-	p.emit(Event{
484		-	Direction: directionOf(toServer),
485		-	Sequence: seq,
486		-	Bytes: len(raw),
487		-	At: now,
488		-	Due: due,
489		-	Decision: decision,
490		-	})
491	494		}
492	495
493	496		func (p *Proxy) enqueue(pkt packet) bool {

 		counters.reordered.Add(1)
 	}
 	due := now.Add(link.BaseLatency + decision.ExtraDelay)
+	// Emit before scheduling: once enqueued, a zero-latency copy can be
+	// delivered and answered before this call returns, and the reply's event
+	// must not be observed ahead of the packet that caused it.
+	p.emit(Event{
+		Direction: directionOf(toServer),
+		Sequence:  seq,
+		Bytes:     len(raw),
+		At:        now,
+		Due:       due,
+		Decision:  decision,
+	})
 	for copyIndex := 0; copyIndex <= decision.Duplicates; copyIndex++ {
 		copyDue := due.Add(time.Duration(copyIndex) * link.DuplicateDelay)
 		if !p.enqueue(packet{
 			counters.overflowDrops.Add(1)
 		}
 	}
-	p.emit(Event{
-		Direction: directionOf(toServer),
-		Sequence:  seq,
-		Bytes:     len(raw),
-		At:        now,
-		Due:       due,
-		Decision:  decision,
-	})
 }
 func (p *Proxy) enqueue(pkt packet) bool {

M internal/usecase/daemon/render_coordinator.go +9 -9


@@ -1005,21 +1005,21 @@ func (c renderCoordinator) fireWithTimerTokenAndLease(token timerToken, gen ui
1005	1005		if c.pendingGeneration == 0 {
1006	1006		c.pendingGeneration = 1
1007	1007		}
1008		-	leases := make(map[attachedClient]attachmentLease, len(c.leases))
	1008	+	// lease.ready is guarded by c.mu (markAttachmentReady writes it), so only
	1009	+	// leases already ready under the lock are probed after unlocking.
	1010	+	probe := make(map[attachedClient]attachmentLease, len(c.leases))
1009	1011		for ac, lease := range c.leases {
1010		-	if lease != nil && lease.active {
1011		-	leases[ac] = lease
	1012	+	if lease != nil && lease.active && lease.ready {
	1013	+	probe[ac] = lease
1012	1014		}
1013	1015		}
1014	1016		c.mu.Unlock()
1015	1017
1016		-	ready := make(map[*attachmentLease]bool, len(leases))
1017		-	for ac, lease := range leases {
	1018	+	ready := make(map[*attachmentLease]bool, len(probe))
	1019	+	for ac, lease := range probe {
1018	1020		// Welcome holds attachment.sendMu while its transport Send is in
1019	1021		// flight. Do not probe an incarnation before it is ready.
1020		-	if lease.ready {
1021		-	ready[lease] = c.attachmentAckReady(ac)
1022		-	}
	1022	+	ready[lease] = c.attachmentAckReady(ac)
1023	1023		}
1024	1024
1025	1025		// The readiness callback can detach, replace, or publish a sync batch.

@@ -1041,7 +1041,7 @@ func (c renderCoordinator) fireWithTimerTokenAndLease(token timerToken, gen ui
1041	1041
1042	1042		// Re-snapshot after every external readiness probe. A replacement may have
1043	1043		// retired one of the leases we probed while the callback was unlocked.
1044		-	leases = make(map[attachedClient]attachmentLease, len(c.leases))
	1044	+	leases := make(map[attachedClient]attachmentLease, len(c.leases))
1045	1045		for ac, lease := range c.leases {
1046	1046		if lease != nil && lease.active {
1047	1047		leases[ac] = lease

 	if c.pendingGeneration == 0 {
 		c.pendingGeneration = 1
 	}
-	leases := make(map[*attachedClient]*attachmentLease, len(c.leases))
+	// lease.ready is guarded by c.mu (markAttachmentReady writes it), so only
+	// leases already ready under the lock are probed after unlocking.
+	probe := make(map[*attachedClient]*attachmentLease, len(c.leases))
 	for ac, lease := range c.leases {
-		if lease != nil && lease.active {
+		if lease != nil && lease.active && lease.ready {
-			leases[ac] = lease
+			probe[ac] = lease
 		}
 	}
 	c.mu.Unlock()
-	ready := make(map[*attachmentLease]bool, len(leases))
+	ready := make(map[*attachmentLease]bool, len(probe))
-	for ac, lease := range leases {
+	for ac, lease := range probe {
 		// Welcome holds attachment.sendMu while its transport Send is in
 		// flight. Do not probe an incarnation before it is ready.
-		if lease.ready {
+		ready[lease] = c.attachmentAckReady(ac)
-			ready[lease] = c.attachmentAckReady(ac)
-		}
 	}
 	// The readiness callback can detach, replace, or publish a sync batch.
 	// Re-snapshot after every external readiness probe. A replacement may have
 	// retired one of the leases we probed while the callback was unlocked.
-	leases = make(map[*attachedClient]*attachmentLease, len(c.leases))
+	leases := make(map[*attachedClient]*attachmentLease, len(c.leases))
 	for ac, lease := range c.leases {
 		if lease != nil && lease.active {
 			leases[ac] = lease