26a180f fix(dgram): make constrained-link simulation deterministic · vev

M internal/adapters/dgram/control.go +15 -9


@@ -38,14 +38,14 @@ func (t *Transport) queueACK(seq uint64) {
38	38		}
39	39		select {
40	40		case t.ackWake <- struct{}{}:
	41	+	t.mu.Lock()
	42	+	hook := t.afterACKWakeAccepted
	43	+	t.mu.Unlock()
	44	+	if hook != nil {
	45	+	hook()
	46	+	}
41	47		default:
42	48		}
43		-	t.mu.Lock()
44		-	afterACKQueued := t.afterACKQueued
45		-	t.mu.Unlock()
46		-	if afterACKQueued != nil {
47		-	afterACKQueued()
48		-	}
49	49		}
50	50
51	51		// queueControl is deliberately lossy when the bounded control queue is full:

@@ -108,10 +108,10 @@ func (t *Transport) ackScheduleLoop() {
108	108		ackTimerC = ackTimer.C()
109	109		}
110	110		t.mu.Lock()
111		-	afterACKScheduled := t.afterACKScheduled
	111	+	hook := t.afterACKScheduled
112	112		t.mu.Unlock()
113		-	if afterACKScheduled != nil {
114		-	afterACKScheduled()
	113	+	if hook != nil {
	114	+	hook()
115	115		}
116	116		}
117	117		dispatchACK := func() {

@@ -121,6 +121,12 @@ func (t *Transport) ackScheduleLoop() {
121	121		}
122	122		select {
123	123		case t.ackSend <- seq:
	124	+	t.mu.Lock()
	125	+	hook := t.afterACKDispatched
	126	+	t.mu.Unlock()
	127	+	if hook != nil {
	128	+	hook()
	129	+	}
124	130		default:
125	131		// Preserve the cumulative maximum until the bounded ACK sender has
126	132		// capacity again.

 	}
 	select {
 	case t.ackWake <- struct{}{}:
+		t.mu.Lock()
+		hook := t.afterACKWakeAccepted
+		t.mu.Unlock()
+		if hook != nil {
+			hook()
+		}
 	default:
 	}
-	t.mu.Lock()
-	afterACKQueued := t.afterACKQueued
-	t.mu.Unlock()
-	if afterACKQueued != nil {
-		afterACKQueued()
-	}
 }
 // queueControl is deliberately lossy when the bounded control queue is full:
 			ackTimerC = ackTimer.C()
 		}
 		t.mu.Lock()
-		afterACKScheduled := t.afterACKScheduled
+		hook := t.afterACKScheduled
 		t.mu.Unlock()
-		if afterACKScheduled != nil {
+		if hook != nil {
-			afterACKScheduled()
+			hook()
 		}
 	}
 	dispatchACK := func() {
 		}
 		select {
 		case t.ackSend <- seq:
+			t.mu.Lock()
+			hook := t.afterACKDispatched
+			t.mu.Unlock()
+			if hook != nil {
+				hook()
+			}
 		default:
 			// Preserve the cumulative maximum until the bounded ACK sender has
 			// capacity again.

M internal/adapters/dgram/network_sim_test.go +101 -22


@@ -387,13 +387,18 @@ func TestTransportFloodClassification(t *testing.T) {
387	387		}
388	388		defer closeFloodTransports(a, b)
389	389		waitForManualTimers(t, clk, 6)
	390	+	for range 6 {
	391	+	awaitSignal(t, clk.timerCreated, "initial transport timer creation")
	392	+	}
390	393		var contacted atomic.Bool
391		-	processed := make(chan struct{})
	394	+	// All fixture hooks are lossy notifications: transport read/control loops
	395	+	// must never wait for the test goroutine to be scheduled.
	396	+	processed := make(chan struct{}, 256)
392	397		primerWritten := make(chan struct{}, 1)
393		-	primerACKScheduled := make(chan struct{}, 1)
394	398		primerAcknowledged := make(chan struct{}, 1)
395		-	ackQueued := make(chan struct{}, 128)
	399	+	ackWakeAccepted := make(chan struct{}, 128)
396	400		ackScheduled := make(chan struct{}, 128)
	401	+	ackDispatched := make(chan struct{}, 128)
397	402		ackWritten := make(chan struct{}, 128)
398	403		aPC.replaceAfterWrite(func() {
399	404		select {

@@ -412,22 +417,39 @@ func TestTransportFloodClassification(t *testing.T) {
412	417		}
413	418		})
414	419		a.mu.Lock()
415		-	a.afterPacketProcessed = func() { processed <- struct{}{} }
	420	+	a.afterPacketProcessed = func() {
	421	+	select {
	422	+	case processed <- struct{}{}:
	423	+	default:
	424	+	}
	425	+	}
416	426		a.mu.Unlock()
417	427		b.mu.Lock()
418	428		b.afterAuthenticatedPacket = func() { contacted.Store(true) }
419		-	b.afterPacketProcessed = func() { processed <- struct{}{} }
420		-	b.afterACKScheduled = func() {
	429	+	b.afterPacketProcessed = func() {
421	430		select {
422		-	case primerACKScheduled <- struct{}{}:
	431	+	case processed <- struct{}{}:
423	432		default:
424	433		}
	434	+	}
	435	+	b.afterACKWakeAccepted = func() {
	436	+	select {
	437	+	case ackWakeAccepted <- struct{}{}:
	438	+	default:
	439	+	}
	440	+	}
	441	+	b.afterACKScheduled = func() {
425	442		select {
426	443		case ackScheduled <- struct{}{}:
427	444		default:
428	445		}
429	446		}
430		-	b.afterACKQueued = func() { ackQueued <- struct{}{} }
	447	+	b.afterACKDispatched = func() {
	448	+	select {
	449	+	case ackDispatched <- struct{}{}:
	450	+	default:
	451	+	}
	452	+	}
431	453		b.mu.Unlock()
432	454		primer := ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("primer")})
433	455		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: primer}); err != nil {

@@ -443,7 +465,8 @@ func TestTransportFloodClassification(t *testing.T) {
443	465		awaitingPrimerACK = false
444	466		}
445	467		if flushAndAwait(t, forward, bPC, processed) > 0 {
446		-	awaitSignal(t, primerACKScheduled, "primer ACK scheduling")
	468	+	awaitSignal(t, ackWakeAccepted, "primer accepted ACK wake")
	469	+	awaitSignal(t, ackScheduled, "primer ACK scheduling")
447	470		awaitingPrimerACK = true
448	471		}
449	472		flushAndAwait(t, reverse, aPC, processed)

@@ -457,15 +480,46 @@ func TestTransportFloodClassification(t *testing.T) {
457	480		if !primerDrained {
458	481		t.Fatal("clean RTT primer did not drain")
459	482		}
460		-	<-ackQueued
461		-	<-ackScheduled
462	483		a.mu.Lock()
463	484		initialCwnd := a.congestion.cwndBytes
464	485		a.mu.Unlock()
	486	+	timerWork := make(chan struct{}, 128)
465	487		dataPaceStarted := make(chan struct{}, 128)
466		-	dataWritten := make(chan struct{}, 1)
	488	+	dataPaceWait := make(chan time.Time, 128)
	489	+	dataJobDone := make(chan struct{}, 128)
	490	+	dataWritten := make(chan struct{}, 128)
	491	+	notifyTimerWork := func() {
	492	+	select {
	493	+	case timerWork <- struct{}{}:
	494	+	default:
	495	+	}
	496	+	}
	497	+	for _, tr := range []*Transport{a, b} {
	498	+	tr.mu.Lock()
	499	+	tr.afterRetransmitTimer = notifyTimerWork
	500	+	tr.afterHealthTimer = notifyTimerWork
	501	+	tr.afterHeartbeatTimer = notifyTimerWork
	502	+	tr.mu.Unlock()
	503	+	}
467	504		a.mu.Lock()
468		-	a.beforeDataPace = func() { dataPaceStarted <- struct{}{} }
	505	+	a.beforeDataPace = func() {
	506	+	select {
	507	+	case dataPaceStarted <- struct{}{}:
	508	+	default:
	509	+	}
	510	+	}
	511	+	a.afterDataPaceWait = func(deadline time.Time) {
	512	+	select {
	513	+	case dataPaceWait <- deadline:
	514	+	default:
	515	+	}
	516	+	}
	517	+	a.afterDataJob = func() {
	518	+	select {
	519	+	case dataJobDone <- struct{}{}:
	520	+	default:
	521	+	}
	522	+	}
469	523		a.mu.Unlock()
470	524		aPC.addAfterWrite(func() {
471	525		select {

@@ -485,9 +539,10 @@ func TestTransportFloodClassification(t *testing.T) {
485	539		}
486	540		awaitSignal(t, dataPaceStarted, "output pacing")
487	541		awaitSignal(t, dataWritten, "first paced output write")
	542	+	nextPace := awaitResult(t, dataPaceWait, "data pacing deadline")
488	543
489	544		var maxContactAge time.Duration
490		-	pendingACKWrites := 0
	545	+	awaitingACKWrite := false
491	546		lossReducedCwnd := false
492	547		latestState := uint64(0)
493	548		probed := false

@@ -495,21 +550,45 @@ func TestTransportFloodClassification(t *testing.T) {
495	550		probeWho := ""
496	551		for range 2000 {
497	552		clk.advance(25 * time.Millisecond)
498		-	for range pendingACKWrites {
	553	+	elapsed := clk.Now().Sub(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	554	+	if elapsed%opts.ResendAfter == 0 {
	555	+	// Both peers complete retransmit and health work before this
	556	+	// advance may be followed by another manual-clock step.
	557	+	for range 4 {
	558	+	awaitSignal(t, timerWork, "retransmit or health timer work")
	559	+	}
	560	+	}
	561	+	if elapsed%opts.Heartbeat == 0 {
	562	+	for range 2 {
	563	+	awaitSignal(t, timerWork, "heartbeat timer work")
	564	+	}
	565	+	}
	566	+	// A manual-clock advance is not a scheduler yield. When it fires the
	567	+	// data pacer, wait until the write reaches its next deterministic
	568	+	// deadline (or its job completes) before advancing again.
	569	+	if !nextPace.IsZero() && !nextPace.After(clk.Now()) {
	570	+	awaitSignal(t, dataWritten, "paced data write")
	571	+	select {
	572	+	case nextPace = <-dataPaceWait:
	573	+	case <-dataJobDone:
	574	+	// A completed job may be awaiting an ACK or retransmit timer;
	575	+	// neither has a data-pacer deadline to await yet.
	576	+	nextPace = time.Time{}
	577	+	}
	578	+	}
	579	+	if awaitingACKWrite {
	580	+	awaitSignal(t, ackDispatched, "cumulative ACK dispatch")
499	581		awaitSignal(t, ackWritten, "cumulative ACK write")
	582	+	awaitingACKWrite = false
500	583		}
501		-	pendingACKWrites = 0
502		-	flushAndAwait(t, forward, bPC, processed)
503		-	for {
	584	+	if flushAndAwait(t, forward, bPC, processed) > 0 {
504	585		select {
505		-	case <-ackQueued:
	586	+	case <-ackWakeAccepted:
506	587		awaitSignal(t, ackScheduled, "cumulative ACK scheduling")
507		-	pendingACKWrites++
	588	+	awaitingACKWrite = true
508	589		default:
509		-	goto ackRecordsDrained
510	590		}
511	591		}
512		-	ackRecordsDrained:
513	592		flushAndAwait(t, reverse, aPC, processed)
514	593		a.mu.Lock()
515	594		queuedOutput := len(a.outputQueue) > 0

 		}
 		defer closeFloodTransports(a, b)
 		waitForManualTimers(t, clk, 6)
+		for range 6 {
+			awaitSignal(t, clk.timerCreated, "initial transport timer creation")
+		}
 		var contacted atomic.Bool
-		processed := make(chan struct{})
+		// All fixture hooks are lossy notifications: transport read/control loops
+		// must never wait for the test goroutine to be scheduled.
+		processed := make(chan struct{}, 256)
 		primerWritten := make(chan struct{}, 1)
-		primerACKScheduled := make(chan struct{}, 1)
 		primerAcknowledged := make(chan struct{}, 1)
-		ackQueued := make(chan struct{}, 128)
+		ackWakeAccepted := make(chan struct{}, 128)
 		ackScheduled := make(chan struct{}, 128)
+		ackDispatched := make(chan struct{}, 128)
 		ackWritten := make(chan struct{}, 128)
 		aPC.replaceAfterWrite(func() {
 			select {
 			}
 		})
 		a.mu.Lock()
-		a.afterPacketProcessed = func() { processed <- struct{}{} }
+		a.afterPacketProcessed = func() {
+			select {
+			case processed <- struct{}{}:
+			default:
+			}
+		}
 		a.mu.Unlock()
 		b.mu.Lock()
 		b.afterAuthenticatedPacket = func() { contacted.Store(true) }
-		b.afterPacketProcessed = func() { processed <- struct{}{} }
+		b.afterPacketProcessed = func() {
-		b.afterACKScheduled = func() {
 			select {
-			case primerACKScheduled <- struct{}{}:
+			case processed <- struct{}{}:
 			default:
 			}
+		}
+		b.afterACKWakeAccepted = func() {
+			select {
+			case ackWakeAccepted <- struct{}{}:
+			default:
+			}
+		}
+		b.afterACKScheduled = func() {
 			select {
 			case ackScheduled <- struct{}{}:
 			default:
 			}
 		}
-		b.afterACKQueued = func() { ackQueued <- struct{}{} }
+		b.afterACKDispatched = func() {
+			select {
+			case ackDispatched <- struct{}{}:
+			default:
+			}
+		}
 		b.mu.Unlock()
 		primer := ports.MarshalOutput(ports.Output{BaseStateNum: 0, NewStateNum: 1, Data: []byte("primer")})
 		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: primer}); err != nil {
 				awaitingPrimerACK = false
 			}
 			if flushAndAwait(t, forward, bPC, processed) > 0 {
-				awaitSignal(t, primerACKScheduled, "primer ACK scheduling")
+				awaitSignal(t, ackWakeAccepted, "primer accepted ACK wake")
+				awaitSignal(t, ackScheduled, "primer ACK scheduling")
 				awaitingPrimerACK = true
 			}
 			flushAndAwait(t, reverse, aPC, processed)
 		if !primerDrained {
 			t.Fatal("clean RTT primer did not drain")
 		}
-		<-ackQueued
-		<-ackScheduled
 		a.mu.Lock()
 		initialCwnd := a.congestion.cwndBytes
 		a.mu.Unlock()
+		timerWork := make(chan struct{}, 128)
 		dataPaceStarted := make(chan struct{}, 128)
-		dataWritten := make(chan struct{}, 1)
+		dataPaceWait := make(chan time.Time, 128)
+		dataJobDone := make(chan struct{}, 128)
+		dataWritten := make(chan struct{}, 128)
+		notifyTimerWork := func() {
+			select {
+			case timerWork <- struct{}{}:
+			default:
+			}
+		}
+		for _, tr := range []*Transport{a, b} {
+			tr.mu.Lock()
+			tr.afterRetransmitTimer = notifyTimerWork
+			tr.afterHealthTimer = notifyTimerWork
+			tr.afterHeartbeatTimer = notifyTimerWork
+			tr.mu.Unlock()
+		}
 		a.mu.Lock()
-		a.beforeDataPace = func() { dataPaceStarted <- struct{}{} }
+		a.beforeDataPace = func() {
+			select {
+			case dataPaceStarted <- struct{}{}:
+			default:
+			}
+		}
+		a.afterDataPaceWait = func(deadline time.Time) {
+			select {
+			case dataPaceWait <- deadline:
+			default:
+			}
+		}
+		a.afterDataJob = func() {
+			select {
+			case dataJobDone <- struct{}{}:
+			default:
+			}
+		}
 		a.mu.Unlock()
 		aPC.addAfterWrite(func() {
 			select {
 		}
 		awaitSignal(t, dataPaceStarted, "output pacing")
 		awaitSignal(t, dataWritten, "first paced output write")
+		nextPace := awaitResult(t, dataPaceWait, "data pacing deadline")
 		var maxContactAge time.Duration
-		pendingACKWrites := 0
+		awaitingACKWrite := false
 		lossReducedCwnd := false
 		latestState := uint64(0)
 		probed := false
 		probeWho := ""
 		for range 2000 {
 			clk.advance(25 * time.Millisecond)
-			for range pendingACKWrites {
+			elapsed := clk.Now().Sub(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
+			if elapsed%opts.ResendAfter == 0 {
+				// Both peers complete retransmit and health work before this
+				// advance may be followed by another manual-clock step.
+				for range 4 {
+					awaitSignal(t, timerWork, "retransmit or health timer work")
+				}
+			}
+			if elapsed%opts.Heartbeat == 0 {
+				for range 2 {
+					awaitSignal(t, timerWork, "heartbeat timer work")
+				}
+			}
+			// A manual-clock advance is not a scheduler yield. When it fires the
+			// data pacer, wait until the write reaches its next deterministic
+			// deadline (or its job completes) before advancing again.
+			if !nextPace.IsZero() && !nextPace.After(clk.Now()) {
+				awaitSignal(t, dataWritten, "paced data write")
+				select {
+				case nextPace = <-dataPaceWait:
+				case <-dataJobDone:
+					// A completed job may be awaiting an ACK or retransmit timer;
+					// neither has a data-pacer deadline to await yet.
+					nextPace = time.Time{}
+				}
+			}
+			if awaitingACKWrite {
+				awaitSignal(t, ackDispatched, "cumulative ACK dispatch")
 				awaitSignal(t, ackWritten, "cumulative ACK write")
+				awaitingACKWrite = false
 			}
-			pendingACKWrites = 0
+			if flushAndAwait(t, forward, bPC, processed) > 0 {
-			flushAndAwait(t, forward, bPC, processed)
-			for {
 				select {
-				case <-ackQueued:
+				case <-ackWakeAccepted:
 					awaitSignal(t, ackScheduled, "cumulative ACK scheduling")
-					pendingACKWrites++
+					awaitingACKWrite = true
 				default:
-					goto ackRecordsDrained
 				}
 			}
-		ackRecordsDrained:
 			flushAndAwait(t, reverse, aPC, processed)
 			a.mu.Lock()
 			queuedOutput := len(a.outputQueue) > 0

M internal/adapters/dgram/pacer.go +7 -3


@@ -14,9 +14,10 @@ var errPacerClosed = errors.New("dgram: pacer closed")
14	14		type paceLimits func() (bytesPerSecond int, burstBytes int)
15	15
16	16		type bytePacer struct {
17		-	clk ports.Clock
18		-	tokens int64
19		-	last time.Time
	17	+	clk ports.Clock
	18	+	tokens int64
	19	+	last time.Time
	20	+	afterTimer func(time.Time)
20	21		}
21	22
22	23		func (p *bytePacer) wait(done <-chan struct{}, n int, limits paceLimits) error {

@@ -52,6 +53,9 @@ func (p *bytePacer) wait(done <-chan struct{}, n int, limits paceLimits) error {
52	53		deficit := int64(n) - p.tokens
53	54		wait := time.Duration(mulDivCeilSaturating(deficit, int64(time.Second), int64(rate)))
54	55		timer := p.clk.NewTimer(wait)
	56	+	if p.afterTimer != nil {
	57	+	p.afterTimer(now.Add(wait))
	58	+	}
55	59		select {
56	60		case <-timer.C():
57	61		timer.Stop()

 type paceLimits func() (bytesPerSecond int, burstBytes int)
 type bytePacer struct {
-	clk    ports.Clock
+	clk        ports.Clock
-	tokens int64
+	tokens     int64
-	last   time.Time
+	last       time.Time
+	afterTimer func(time.Time)
 }
 func (p *bytePacer) wait(done <-chan struct{}, n int, limits paceLimits) error {
 		deficit := int64(n) - p.tokens
 		wait := time.Duration(mulDivCeilSaturating(deficit, int64(time.Second), int64(rate)))
 		timer := p.clk.NewTimer(wait)
+		if p.afterTimer != nil {
+			p.afterTimer(now.Add(wait))
+		}
 		select {
 		case <-timer.C():
 			timer.Stop()

M internal/adapters/dgram/transport.go +49 -7


@@ -216,15 +216,24 @@ type Transport struct {
216	216		writeMu sync.Mutex
217	217		// beforeDataPace is a test synchronization hook. It is read under mu.
218	218		beforeDataPace func()
	219	+	// afterDataPaceWait observes a data pacer's next manual-clock deadline.
	220	+	// Test observers must be non-blocking.
	221	+	afterDataPaceWait func(time.Time)
	222	+	// afterDataJob observes completion of a data or retransmit send job. Test
	223	+	// observers must be non-blocking.
	224	+	afterDataJob func()
	225	+	// Timer hooks are test-only completion notifications. Observers must be
	226	+	// non-blocking so timer loops never depend on test scheduling.
	227	+	afterRetransmitTimer func()
	228	+	afterHealthTimer func()
	229	+	afterHeartbeatTimer func()
219	230		// afterPacketProcessed is a test synchronization hook called after an
220	231		// authenticated packet has been fully processed. It is read under mu.
221	232		afterPacketProcessed func()
222		-	// afterACKScheduled is a test synchronization hook called after the ACK
223		-	// scheduler handles a wake. It is read under mu.
224		-	afterACKScheduled func()
225		-	// afterACKQueued is a test synchronization hook called after an ACK is
226		-	// queued. It is read under mu.
227		-	afterACKQueued func()
	233	+	// ACK hooks are test-only and observers must be non-blocking.
	234	+	afterACKWakeAccepted func()
	235	+	afterACKScheduled func()
	236	+	afterACKDispatched func()
228	237		// outboundMu orders first transmissions and paced batches. Retransmits use
229	238		// writeMu directly because they do not establish new sequence order.
230	239		outboundMu sync.Mutex

@@ -982,6 +991,7 @@ func (t *Transport) retransmitLoop() {
982	991		case <-resend.C():
983	992		t.queueRetransmits()
984	993		resend.Reset(t.resendAfter)
	994	+	t.notifyTimerHook(func() func() { return t.afterRetransmitTimer })
985	995		case <-t.done:
986	996		return
987	997		}

@@ -998,6 +1008,7 @@ func (t *Transport) healthLoop() {
998	1008		case <-health.C():
999	1009		t.checkSilence()
1000	1010		health.Reset(t.resendAfter)
	1011	+	t.notifyTimerHook(func() func() { return t.afterHealthTimer })
1001	1012		case <-t.done:
1002	1013		return
1003	1014		}

@@ -1005,7 +1016,14 @@ func (t *Transport) healthLoop() {
1005	1016		}
1006	1017
1007	1018		func (t *Transport) dataSendLoop() {
1008		-	pacer := bytePacer{clk: t.clock}
	1019	+	pacer := bytePacer{clk: t.clock, afterTimer: func(deadline time.Time) {
	1020	+	t.mu.Lock()
	1021	+	hook := t.afterDataPaceWait
	1022	+	t.mu.Unlock()
	1023	+	if hook != nil {
	1024	+	hook(deadline)
	1025	+	}
	1026	+	}}
1009	1027		for {
1010	1028		select {
1011	1029		case job := <-t.dataSend:

@@ -1035,6 +1053,12 @@ func (t Transport) runDataSendJob(pacer bytePacer, job dataSendJob, initial bo
1035	1053		} else {
1036	1054		t.markPendingReady(job.seq, job.reliable)
1037	1055		}
	1056	+	t.mu.Lock()
	1057	+	afterDataJob := t.afterDataJob
	1058	+	t.mu.Unlock()
	1059	+	if afterDataJob != nil {
	1060	+	afterDataJob()
	1061	+	}
1038	1062		if job.done != nil {
1039	1063		if errors.Is(err, errPacerClosed) {
1040	1064		err = t.closedError()

@@ -1120,6 +1144,14 @@ func (t Transport) writeDataJob(pacer bytePacer, job dataSendJob, onFirstWrite
1120	1144		}
1121	1145
1122	1146		func (t Transport) runRetransmitBatch(pacer bytePacer, batch []retransmitRecord) {
	1147	+	defer func() {
	1148	+	t.mu.Lock()
	1149	+	hook := t.afterDataJob
	1150	+	t.mu.Unlock()
	1151	+	if hook != nil {
	1152	+	hook()
	1153	+	}
	1154	+	}()
1123	1155		lossApplied := false
1124	1156		onFirstWrite := func() {
1125	1157		if lossApplied {

@@ -1150,12 +1182,22 @@ func (t *Transport) heartbeatLoop() {
1150	1182		case <-heartbeat.C():
1151	1183		t.queueControl(recProbe, 0)
1152	1184		heartbeat.Reset(t.heartbeat)
	1185	+	t.notifyTimerHook(func() func() { return t.afterHeartbeatTimer })
1153	1186		case <-t.done:
1154	1187		return
1155	1188		}
1156	1189		}
1157	1190		}
1158	1191
	1192	+	func (t *Transport) notifyTimerHook(get func() func()) {
	1193	+	t.mu.Lock()
	1194	+	hook := get()
	1195	+	t.mu.Unlock()
	1196	+	if hook != nil {
	1197	+	hook()
	1198	+	}
	1199	+	}
	1200	+
1159	1201		func (t *Transport) queueRetransmits() {
1160	1202		// Do not mark records in-flight while the one-slot sender queue is full.
1161	1203		// There is one producer and the sender only consumes, so capacity cannot be

 	writeMu sync.Mutex
 	// beforeDataPace is a test synchronization hook. It is read under mu.
 	beforeDataPace func()
+	// afterDataPaceWait observes a data pacer's next manual-clock deadline.
+	// Test observers must be non-blocking.
+	afterDataPaceWait func(time.Time)
+	// afterDataJob observes completion of a data or retransmit send job. Test
+	// observers must be non-blocking.
+	afterDataJob func()
+	// Timer hooks are test-only completion notifications. Observers must be
+	// non-blocking so timer loops never depend on test scheduling.
+	afterRetransmitTimer func()
+	afterHealthTimer     func()
+	afterHeartbeatTimer  func()
 	// afterPacketProcessed is a test synchronization hook called after an
 	// authenticated packet has been fully processed. It is read under mu.
 	afterPacketProcessed func()
-	// afterACKScheduled is a test synchronization hook called after the ACK
+	// ACK hooks are test-only and observers must be non-blocking.
-	// scheduler handles a wake. It is read under mu.
+	afterACKWakeAccepted func()
-	afterACKScheduled func()
+	afterACKScheduled    func()
-	// afterACKQueued is a test synchronization hook called after an ACK is
+	afterACKDispatched   func()
-	// queued. It is read under mu.
-	afterACKQueued func()
 	// outboundMu orders first transmissions and paced batches. Retransmits use
 	// writeMu directly because they do not establish new sequence order.
 	outboundMu sync.Mutex
 		case <-resend.C():
 			t.queueRetransmits()
 			resend.Reset(t.resendAfter)
+			t.notifyTimerHook(func() func() { return t.afterRetransmitTimer })
 		case <-t.done:
 			return
 		}
 		case <-health.C():
 			t.checkSilence()
 			health.Reset(t.resendAfter)
+			t.notifyTimerHook(func() func() { return t.afterHealthTimer })
 		case <-t.done:
 			return
 		}
 }
 func (t *Transport) dataSendLoop() {
-	pacer := bytePacer{clk: t.clock}
+	pacer := bytePacer{clk: t.clock, afterTimer: func(deadline time.Time) {
+		t.mu.Lock()
+		hook := t.afterDataPaceWait
+		t.mu.Unlock()
+		if hook != nil {
+			hook(deadline)
+		}
+	}}
 	for {
 		select {
 		case job := <-t.dataSend:
 	} else {
 		t.markPendingReady(job.seq, job.reliable)
 	}
+	t.mu.Lock()
+	afterDataJob := t.afterDataJob
+	t.mu.Unlock()
+	if afterDataJob != nil {
+		afterDataJob()
+	}
 	if job.done != nil {
 		if errors.Is(err, errPacerClosed) {
 			err = t.closedError()
 }
 func (t *Transport) runRetransmitBatch(pacer *bytePacer, batch []retransmitRecord) {
+	defer func() {
+		t.mu.Lock()
+		hook := t.afterDataJob
+		t.mu.Unlock()
+		if hook != nil {
+			hook()
+		}
+	}()
 	lossApplied := false
 	onFirstWrite := func() {
 		if lossApplied {
 		case <-heartbeat.C():
 			t.queueControl(recProbe, 0)
 			heartbeat.Reset(t.heartbeat)
+			t.notifyTimerHook(func() func() { return t.afterHeartbeatTimer })
 		case <-t.done:
 			return
 		}
 	}
 }
+func (t *Transport) notifyTimerHook(get func() func()) {
+	t.mu.Lock()
+	hook := get()
+	t.mu.Unlock()
+	if hook != nil {
+		hook()
+	}
+}
 func (t *Transport) queueRetransmits() {
 	// Do not mark records in-flight while the one-slot sender queue is full.
 	// There is one producer and the sender only consumes, so capacity cannot be