improve(dgram): pace UDP data with congestion feedback

commit: 43a66791ca8e55b1ec503641b35f9544e115da1c
author: brice · 3mo ago · 2026-07-10 07:11 UTC
parent: 7f56b4a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +609 -369

A internal/adapters/dgram/congestion.go +68 -0

@@ -0,0 +1,68 @@
	1	+	package dgram
	2	+
	3	+	import "time"
	4	+
	5	+	const (
	6	+	initialCongestionPackets = 10
	7	+	minimumCongestionPackets = 2
	8	+	maximumCongestionBytes = 4 << 20
	9	+	initialPacingRTT = 250 * time.Millisecond
	10	+	dataBurstPackets = 2
	11	+	)
	12	+
	13	+	type congestionController struct {
	14	+	mtu int
	15	+	cwndBytes int
	16	+	minBytes int
	17	+	maxBytes int
	18	+	srtt time.Duration
	19	+	}
	20	+
	21	+	func newCongestionController(mtu int) congestionController {
	22	+	minimum := minimumCongestionPackets * mtu
	23	+	maximum := maximumCongestionBytes
	24	+	if maximum < minimum {
	25	+	maximum = minimum
	26	+	}
	27	+	initial := initialCongestionPackets * mtu
	28	+	if initial > maximum {
	29	+	initial = maximum
	30	+	}
	31	+	return congestionController{
	32	+	mtu: mtu,
	33	+	cwndBytes: initial,
	34	+	minBytes: minimum,
	35	+	maxBytes: maximum,
	36	+	srtt: initialPacingRTT,
	37	+	}
	38	+	}
	39	+
	40	+	func (c *congestionController) onRTT(srtt time.Duration) {
	41	+	if srtt > 0 {
	42	+	c.srtt = srtt
	43	+	}
	44	+	}
	45	+
	46	+	func (c *congestionController) onACK(ackedBytes int) {
	47	+	if ackedBytes <= 0 \|\| c.cwndBytes >= c.maxBytes {
	48	+	return
	49	+	}
	50	+	increase := int(int64(c.mtu) * int64(ackedBytes) / int64(c.cwndBytes))
	51	+	if increase < 1 {
	52	+	increase = 1
	53	+	}
	54	+	c.cwndBytes = min(c.cwndBytes+increase, c.maxBytes)
	55	+	}
	56	+
	57	+	func (c *congestionController) onLoss() {
	58	+	c.cwndBytes = max(c.cwndBytes/2, c.minBytes)
	59	+	}
	60	+
	61	+	func (c congestionController) bytesPerSecond() int {
	62	+	rtt := max(c.srtt, time.Millisecond)
	63	+	return int(int64(c.cwndBytes) * int64(time.Second) / int64(rtt))
	64	+	}
	65	+
	66	+	func (c congestionController) burstBytes() int {
	67	+	return min(c.cwndBytes, dataBurstPackets*c.mtu)
	68	+	}

@@ -0,0 +1,68 @@
	1	+	package dgram
	2	+
	3	+	import "time"
	4	+
	5	+	const (
	6	+	initialCongestionPackets = 10
	7	+	minimumCongestionPackets = 2
	8	+	maximumCongestionBytes = 4 << 20
	9	+	initialPacingRTT = 250 * time.Millisecond
	10	+	dataBurstPackets = 2
	11	+	)
	12	+
	13	+	type congestionController struct {
	14	+	mtu int
	15	+	cwndBytes int
	16	+	minBytes int
	17	+	maxBytes int
	18	+	srtt time.Duration
	19	+	}
	20	+
	21	+	func newCongestionController(mtu int) congestionController {
	22	+	minimum := minimumCongestionPackets * mtu
	23	+	maximum := maximumCongestionBytes
	24	+	if maximum < minimum {
	25	+	maximum = minimum
	26	+	}
	27	+	initial := initialCongestionPackets * mtu
	28	+	if initial > maximum {
	29	+	initial = maximum
	30	+	}
	31	+	return congestionController{
	32	+	mtu: mtu,
	33	+	cwndBytes: initial,
	34	+	minBytes: minimum,
	35	+	maxBytes: maximum,
	36	+	srtt: initialPacingRTT,
	37	+	}
	38	+	}
	39	+
	40	+	func (c *congestionController) onRTT(srtt time.Duration) {
	41	+	if srtt > 0 {
	42	+	c.srtt = srtt
	43	+	}
	44	+	}
	45	+
	46	+	func (c *congestionController) onACK(ackedBytes int) {
	47	+	if ackedBytes <= 0 \|\| c.cwndBytes >= c.maxBytes {
	48	+	return
	49	+	}
	50	+	increase := int(int64(c.mtu) * int64(ackedBytes) / int64(c.cwndBytes))
	51	+	if increase < 1 {
	52	+	increase = 1
	53	+	}
	54	+	c.cwndBytes = min(c.cwndBytes+increase, c.maxBytes)
	55	+	}
	56	+
	57	+	func (c *congestionController) onLoss() {
	58	+	c.cwndBytes = max(c.cwndBytes/2, c.minBytes)
	59	+	}
	60	+
	61	+	func (c congestionController) bytesPerSecond() int {
	62	+	rtt := max(c.srtt, time.Millisecond)
	63	+	return int(int64(c.cwndBytes) * int64(time.Second) / int64(rtt))
	64	+	}
	65	+
	66	+	func (c congestionController) burstBytes() int {
	67	+	return min(c.cwndBytes, dataBurstPackets*c.mtu)
	68	+	}

M internal/adapters/dgram/network_sim_test.go +140 -49


@@ -54,6 +54,7 @@ type simulatedLink struct {
54	54		queue []scheduledPacket
55	55		queueBytes int
56	56		next time.Time
	57	+	dropped int
57	58		}
58	59
59	60		func newSimulatedLink(clock interface{ Now() time.Time }, policy packetPolicy) *simulatedLink {

@@ -71,6 +72,7 @@ func (l simulatedLink) enqueue(p fakePC, peer *fakePC, b []byte) bool {
71	72		return l.deliverLocked(peer, packet{b: append([]byte(nil), b...), addr: p.addr})
72	73		}
73	74		if l.policy.MaxQueueBytes <= 0 \|\| l.queueBytes+len(b) > l.policy.MaxQueueBytes {
	75	+	l.dropped++
74	76		return false
75	77		}
76	78		due := meta.At.Add(l.policy.Delay)

@@ -107,6 +109,12 @@ func (l *simulatedLink) queuedBytes() int {
107	109		return l.queueBytes
108	110		}
109	111
	112	+	func (l *simulatedLink) droppedPackets() int {
	113	+	l.mu.Lock()
	114	+	defer l.mu.Unlock()
	115	+	return l.dropped
	116	+	}
	117	+
110	118		func (l simulatedLink) deliverLocked(peer fakePC, q packet) bool {
111	119		peer.mu.Lock()
112	120		defer peer.mu.Unlock()

@@ -281,8 +289,8 @@ func TestTransportFloodClassification(t *testing.T) {
281	289		if len(dropped) != floodRecordCount {
282	290		t.Fatalf("flood state = %+v, dropped records=%d, want exactly one fragment from each of %d records", state, len(dropped), floodRecordCount)
283	291		}
284		-	if state.packetAge != 0 \|\| state.recordAge != time.Nanosecond \|\| state.ackAge != time.Nanosecond {
285		-	t.Fatalf("flood state = %+v, want packet contact only after %d output records of %d MTUs", state, floodRecordCount, floodRecordMTUs)
	292	+	if state.packetAge > initialPacingRTT \|\| state.recordAge <= 0 \|\| state.ackAge <= 0 {
	293	+	t.Fatalf("flood state = %+v, want fresh packet contact without record or ACK progress", state)
286	294		}
287	295		if !clk.Now().After(start) {
288	296		t.Fatal("real authenticated fragments did not update contact")

@@ -291,34 +299,90 @@ func TestTransportFloodClassification(t *testing.T) {
291	299
292	300		t.Run("bounded byte queue advances and releases on fake clock", func(t *testing.T) {
293	301		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
294		-	limit := 2 * 128
	302	+	const mtu = 128
	303	+	limit := dataBurstPackets * mtu
295	304		link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: 128})
296	305		aPC, bPC := newFloodPair(link)
297		-	const mtu = 128
298		-	a, b := newFloodTransports(t, aPC, bPC, clk, mtu)
	306	+	bPC.link = nil // ACKs are unconstrained; this case isolates outbound congestion.
	307	+	opts := Options{
	308	+	Clock: clk, MTU: mtu, ResendAfter: time.Hour, Heartbeat: time.Hour,
	309	+	DegradedAfter: time.Hour, ProbeAfter: time.Hour, OfflineAfter: time.Hour, DeadAfter: time.Hour,
	310	+	}
	311	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts)
	312	+	if err != nil {
	313	+	t.Fatal(err)
	314	+	}
	315	+	b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, opts)
	316	+	if err != nil {
	317	+	_ = a.Close()
	318	+	t.Fatal(err)
	319	+	}
299	320		defer closeFloodTransports(a, b)
300	321		contacted := make(chan struct{}, 1)
301	322		b.mu.Lock()
302		-	b.afterAuthenticatedPacket = func() { contacted <- struct{}{} }
	323	+	b.afterAuthenticatedPacket = func() {
	324	+	select {
	325	+	case contacted <- struct{}{}:
	326	+	default:
	327	+	}
	328	+	}
303	329		b.mu.Unlock()
304		-	a.mu.Lock()
305		-	a.rto = time.Hour // Keep the fake-clock queue case about bandwidth, not retries.
306		-	a.mu.Unlock()
307		-	sendFloodOutputs(t, a, floodRecordCount, mtu)
	330	+	maxFragmentData := mtu - pdgram.HeaderSize - a.codec.Overhead()
	331	+	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 4*maxFragmentData)}); err != nil {
	332	+	t.Fatal(err)
	333	+	}
	334	+	eventually(t, time.Second, func() bool { return link.queuedBytes() == limit })
	335	+	clk.advance(initialPacingRTT)
	336	+	time.Sleep(time.Millisecond)
308	337		state := floodState(a, b, link)
309		-	if state.queuedBytes <= 0 \|\| state.queuedBytes > limit {
310		-	t.Fatalf("flood state = %+v, want queued bytes in 1..%d", state, limit)
	338	+	if state.queuedBytes != limit \|\| link.droppedPackets() == 0 {
	339	+	t.Fatalf("flood state = %+v drops=%d, want bounded queue overflow at %d bytes", state, link.droppedPackets(), limit)
311	340		}
312		-	clk.advance(10 * time.Second)
	341	+	clk.advance(time.Second)
313	342		link.flush(bPC)
314	343		awaitSignal(t, contacted, "queued authenticated fragment")
315		-	state = floodState(a, b, link)
316		-	if state.queuedBytes != 0 {
317		-	t.Fatalf("flood state = %+v, want empty queue after fake-clock release", state)
	344	+	for link.queuedBytes() > 0 {
	345	+	clk.advance(time.Second)
	346	+	link.flush(bPC)
318	347		}
319		-	if state.packetAge != 0 \|\| state.recordAge != 10time.Second \|\| state.ackAge != 10time.Second {
320		-	t.Fatalf("flood state = %+v, want delayed packet contact without record or ACK progress", state)
	348	+	if age := floodState(a, b, link).packetAge; age >= a.probeAfter {
	349	+	t.Fatalf("authenticated contact age=%v, want below probe threshold %v", age, a.probeAfter)
	350	+	}
	351	+	eventually(t, time.Second, func() bool {
	352	+	a.mu.Lock()
	353	+	defer a.mu.Unlock()
	354	+	p := a.pending[1]
	355	+	return p != nil && !p.initialInFlight && !p.last.IsZero()
	356	+	})
	357	+
	358	+	a.mu.Lock()
	359	+	beforeLoss := a.congestion.cwndBytes
	360	+	a.rto = time.Millisecond
	361	+	a.mu.Unlock()
	362	+	aPC.link = nil // The transient bottleneck clears before the paced retry.
	363	+	clk.advance(2 * time.Millisecond)
	364	+	a.queueRetransmits()
	365	+	eventually(t, time.Second, func() bool {
	366	+	a.mu.Lock()
	367	+	defer a.mu.Unlock()
	368	+	return a.congestion.cwndBytes < beforeLoss
	369	+	})
	370	+	for range 16 {
	371	+	clk.advance(time.Second)
	372	+	link.flush(bPC)
	373	+	clk.advance(maxACKDelay)
	374	+	time.Sleep(time.Millisecond)
	375	+	a.mu.Lock()
	376	+	drained := len(a.pending) == 0
	377	+	a.mu.Unlock()
	378	+	if drained {
	379	+	return
	380	+	}
321	381		}
	382	+	a.mu.Lock()
	383	+	pending := a.pending[1]
	384	+	a.mu.Unlock()
	385	+	t.Fatalf("pending records did not drain after paced retransmission: %+v pending=%+v drops=%d", floodState(a, b, link), pending, link.droppedPackets())
322	386		})
323	387
324	388		t.Run("control stays responsive while fresh output and large retransmits contend", func(t *testing.T) {

@@ -350,8 +414,6 @@ func TestTransportFloodClassification(t *testing.T) {
350	414		releaseRetransmit := make(chan struct{})
351	415		lastRetransmitted := make(chan struct{}, 1)
352	416		probeSent := make(chan struct{}, 1)
353		-	freshAtPace := make(chan struct{})
354		-	allowFreshPace := make(chan struct{})
355	417		freshReturned := make(chan error, 1)
356	418		writes := make(map[uint64]int)
357	419		var writesMu sync.Mutex

@@ -406,6 +468,9 @@ func TestTransportFloodClassification(t *testing.T) {
406	468		for range 6 { // Both transports install retransmit, health, and heartbeat timers.
407	469		awaitSignal(t, clk.timerCreated, "transport timer creation")
408	470		}
	471	+	a.mu.Lock()
	472	+	a.rto = time.Hour
	473	+	a.mu.Unlock()
409	474
410	475		// Fresh, fragmented output records become overdue while their real
411	476		// cumulative ACKs are lost. Disabling pacing intentionally forces overload

@@ -414,48 +479,60 @@ func TestTransportFloodClassification(t *testing.T) {
414	479		awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
415	480		clk.advance(maxACKDelay)
416	481		awaitSignal(t, acksDropped, "dropped cumulative ACK")
	482	+	a.mu.Lock()
	483	+	a.rto = 10 * time.Millisecond
	484	+	a.mu.Unlock()
	485	+	a.queueRetransmits()
	486	+	a.mu.Lock()
	487	+	a.rto = time.Hour
	488	+	a.mu.Unlock()
	489	+	clk.advance(initialPacingRTT)
417	490
418	491		awaitSignal(t, retransmitBlocked, "retransmit pacing barrier")
419	492		if state := floodState(a, b, nil); state.retransmits < floodRecordCount {
420	493		t.Fatalf("flood state = %+v, want all %d overdue records selected", state, floodRecordCount)
421	494		}
422		-
423		-	// Start a third large output at this same fake-clock instant. The pacing
424		-	// barrier proves its real Send reaches the shared pacing gate while the
425		-	// retransmit owns it inside WriteTo.
426		-	a.mu.Lock()
427		-	a.outputNext = time.Time{}
428		-	a.beforeDataPace = func() {
429		-	close(freshAtPace)
430		-	awaitSignal(t, allowFreshPace, "fresh pacing release")
	495	+	select {
	496	+	case <-probeSent:
	497	+	default:
431	498		}
432		-	a.mu.Unlock()
	499	+
	500	+	// A fresh record queues behind the sole data sender while control traffic
	501	+	// remains independent of congestion tokens and the blocked data write.
433	502		go func() {
434	503		freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(floodRecordCount, 128)})
435	504		}()
436		-	awaitSignal(t, freshAtPace, "fresh output pacing barrier")
437		-	a.mu.Lock()
438		-	a.beforeDataPace = nil
439		-	a.mu.Unlock()
440		-	close(allowFreshPace)
441	505		select {
442	506		case err := <-freshReturned:
443	507		t.Fatalf("fresh output completed while retransmit held pacing: %v", err)
444	508		default:
445	509		}
446	510
	511	+	clk.advance(11 * time.Millisecond)
447	512		awaitSignal(t, probeSent, "heartbeat probe during output contention")
448	513
449	514		close(releaseRetransmit)
450		-	if err := awaitResult(t, freshReturned, "fresh output completion"); err != nil {
451		-	t.Fatal(err)
	515	+	for range 1024 {
	516	+	clk.advance(initialPacingRTT)
	517	+	time.Sleep(time.Millisecond)
	518	+	select {
	519	+	case err := <-freshReturned:
	520	+	if err != nil {
	521	+	t.Fatal(err)
	522	+	}
	523	+	awaitSignal(t, lastRetransmitted, "last overdue record retransmission")
	524	+	writesMu.Lock()
	525	+	defer writesMu.Unlock()
	526	+	if writes[1] < 2 \|\| writes[floodRecordCount] < 2 \|\| writes[floodRecordCount+1] == 0 {
	527	+	t.Fatalf("flood state = %+v, writes=%v, want fresh large traffic and retransmissions across %d overdue records", floodState(a, b, nil), writes, floodRecordCount)
	528	+	}
	529	+	return
	530	+	default:
	531	+	}
452	532		}
453		-	awaitSignal(t, lastRetransmitted, "last overdue record retransmission")
454	533		writesMu.Lock()
455	534		defer writesMu.Unlock()
456		-	if writes[1] < 2 \|\| writes[floodRecordCount] < 2 \|\| writes[floodRecordCount+1] == 0 {
457		-	t.Fatalf("flood state = %+v, writes=%v, want fresh large traffic and retransmissions across %d overdue records", floodState(a, b, nil), writes, floodRecordCount)
458		-	}
	535	+	t.Fatalf("fresh output did not complete after paced retransmissions: writes=%v state=%+v", writes, floodState(a, b, nil))
459	536		})
460	537		}
461	538

@@ -510,23 +587,37 @@ func TestFloodOutputPayloadIsIncrementalStateBearingOutput(t *testing.T) {
510	587
511	588		func sendFloodOutputs(t testing.T, sender Transport, count, mtu int) {
512	589		t.Helper()
513		-	sender.dataPaceMu.Lock()
514		-	sender.dataPaceRemaining = count * floodRecordMTUs * 4
515		-	sender.dataPaceNext = sender.clock.Now().Add(time.Hour)
516		-	sender.dataPaceMu.Unlock()
517	590		for state := range count {
	591	+	if err := sender.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(state, mtu)}); err != nil {
	592	+	t.Fatal(err)
	593	+	}
	594	+	}
	595	+	clk, ok := sender.clock.(*manualClock)
	596	+	if !ok {
	597	+	return
	598	+	}
	599	+	for range count * floodRecordMTUs * 2 {
	600	+	clk.advance(initialPacingRTT)
	601	+	time.Sleep(time.Millisecond)
518	602		sender.mu.Lock()
519		-	sender.outputNext = time.Time{}
	603	+	complete := len(sender.outputQueue) == 0 && len(sender.dataSend) == 0
	604	+	for _, p := range sender.pending {
	605	+	complete = complete && !p.initialInFlight && !p.last.IsZero()
	606	+	}
520	607		sender.mu.Unlock()
521		-	if err := sender.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(state, mtu)}); err != nil {
522		-	t.Fatal(err)
	608	+	if complete {
	609	+	return
523	610		}
524	611		}
	612	+	t.Fatal("flood output pacing did not drain")
525	613		}
526	614
527	615		func newFloodTransports(t testing.T, aPC, bPC fakePC, clk manualClock, mtu int) (Transport, *Transport) {
528	616		t.Helper()
529		-	opts := Options{Clock: clk, MTU: mtu, ResendAfter: 10 * time.Millisecond, Heartbeat: time.Hour}
	617	+	opts := Options{
	618	+	Clock: clk, MTU: mtu, ResendAfter: time.Hour, Heartbeat: time.Hour,
	619	+	DegradedAfter: time.Hour, ProbeAfter: time.Hour, OfflineAfter: time.Hour, DeadAfter: time.Hour,
	620	+	}
530	621		a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts)
531	622		if err != nil {
532	623		t.Fatal(err)


@@ -54,6 +54,7 @@ type simulatedLink struct {
54		queue []scheduledPacket	54		queue []scheduledPacket
55		queueBytes int	55		queueBytes int
56		next time.Time	56		next time.Time
			57	+	dropped int
57		}	58		}
58			59
59		func newSimulatedLink(clock interface{ Now() time.Time }, policy packetPolicy) *simulatedLink {	60		func newSimulatedLink(clock interface{ Now() time.Time }, policy packetPolicy) *simulatedLink {

@@ -71,6 +72,7 @@ func (l simulatedLink) enqueue(p fakePC, peer *fakePC, b []byte) bool {
71		return l.deliverLocked(peer, packet{b: append([]byte(nil), b...), addr: p.addr})	72		return l.deliverLocked(peer, packet{b: append([]byte(nil), b...), addr: p.addr})
72		}	73		}
73		if l.policy.MaxQueueBytes <= 0 \|\| l.queueBytes+len(b) > l.policy.MaxQueueBytes {	74		if l.policy.MaxQueueBytes <= 0 \|\| l.queueBytes+len(b) > l.policy.MaxQueueBytes {
			75	+	l.dropped++
74		return false	76		return false
75		}	77		}
76		due := meta.At.Add(l.policy.Delay)	78		due := meta.At.Add(l.policy.Delay)

@@ -107,6 +109,12 @@ func (l *simulatedLink) queuedBytes() int {
107		return l.queueBytes	109		return l.queueBytes
108		}	110		}
109			111
			112	+	func (l *simulatedLink) droppedPackets() int {
			113	+	l.mu.Lock()
			114	+	defer l.mu.Unlock()
			115	+	return l.dropped
			116	+	}
			117	+
110		func (l simulatedLink) deliverLocked(peer fakePC, q packet) bool {	118		func (l simulatedLink) deliverLocked(peer fakePC, q packet) bool {
111		peer.mu.Lock()	119		peer.mu.Lock()
112		defer peer.mu.Unlock()	120		defer peer.mu.Unlock()

@@ -281,8 +289,8 @@ func TestTransportFloodClassification(t *testing.T) {
281		if len(dropped) != floodRecordCount {	289		if len(dropped) != floodRecordCount {
282		t.Fatalf("flood state = %+v, dropped records=%d, want exactly one fragment from each of %d records", state, len(dropped), floodRecordCount)	290		t.Fatalf("flood state = %+v, dropped records=%d, want exactly one fragment from each of %d records", state, len(dropped), floodRecordCount)
283		}	291		}
284	-	if state.packetAge != 0 \|\| state.recordAge != time.Nanosecond \|\| state.ackAge != time.Nanosecond {	292	+	if state.packetAge > initialPacingRTT \|\| state.recordAge <= 0 \|\| state.ackAge <= 0 {
285	-	t.Fatalf("flood state = %+v, want packet contact only after %d output records of %d MTUs", state, floodRecordCount, floodRecordMTUs)	293	+	t.Fatalf("flood state = %+v, want fresh packet contact without record or ACK progress", state)
286		}	294		}
287		if !clk.Now().After(start) {	295		if !clk.Now().After(start) {
288		t.Fatal("real authenticated fragments did not update contact")	296		t.Fatal("real authenticated fragments did not update contact")

@@ -291,34 +299,90 @@ func TestTransportFloodClassification(t *testing.T) {
291			299
292		t.Run("bounded byte queue advances and releases on fake clock", func(t *testing.T) {	300		t.Run("bounded byte queue advances and releases on fake clock", func(t *testing.T) {
293		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))	301		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
294	-	limit := 2 * 128	302	+	const mtu = 128
			303	+	limit := dataBurstPackets * mtu
295		link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: 128})	304		link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: 128})
296		aPC, bPC := newFloodPair(link)	305		aPC, bPC := newFloodPair(link)
297	-	const mtu = 128	306	+	bPC.link = nil // ACKs are unconstrained; this case isolates outbound congestion.
298	-	a, b := newFloodTransports(t, aPC, bPC, clk, mtu)	307	+	opts := Options{
			308	+	Clock: clk, MTU: mtu, ResendAfter: time.Hour, Heartbeat: time.Hour,
			309	+	DegradedAfter: time.Hour, ProbeAfter: time.Hour, OfflineAfter: time.Hour, DeadAfter: time.Hour,
			310	+	}
			311	+	a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts)
			312	+	if err != nil {
			313	+	t.Fatal(err)
			314	+	}
			315	+	b, err := NewTransportWithOptions(bPC, aPC.addr, key(), 2, 1, opts)
			316	+	if err != nil {
			317	+	_ = a.Close()
			318	+	t.Fatal(err)
			319	+	}
299		defer closeFloodTransports(a, b)	320		defer closeFloodTransports(a, b)
300		contacted := make(chan struct{}, 1)	321		contacted := make(chan struct{}, 1)
301		b.mu.Lock()	322		b.mu.Lock()
302	-	b.afterAuthenticatedPacket = func() { contacted <- struct{}{} }	323	+	b.afterAuthenticatedPacket = func() {
			324	+	select {
			325	+	case contacted <- struct{}{}:
			326	+	default:
			327	+	}
			328	+	}
303		b.mu.Unlock()	329		b.mu.Unlock()
304	-	a.mu.Lock()	330	+	maxFragmentData := mtu - pdgram.HeaderSize - a.codec.Overhead()
305	-	a.rto = time.Hour // Keep the fake-clock queue case about bandwidth, not retries.	331	+	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 4*maxFragmentData)}); err != nil {
306	-	a.mu.Unlock()	332	+	t.Fatal(err)
307	-	sendFloodOutputs(t, a, floodRecordCount, mtu)	333	+	}
			334	+	eventually(t, time.Second, func() bool { return link.queuedBytes() == limit })
			335	+	clk.advance(initialPacingRTT)
			336	+	time.Sleep(time.Millisecond)
308		state := floodState(a, b, link)	337		state := floodState(a, b, link)
309	-	if state.queuedBytes <= 0 \|\| state.queuedBytes > limit {	338	+	if state.queuedBytes != limit \|\| link.droppedPackets() == 0 {
310	-	t.Fatalf("flood state = %+v, want queued bytes in 1..%d", state, limit)	339	+	t.Fatalf("flood state = %+v drops=%d, want bounded queue overflow at %d bytes", state, link.droppedPackets(), limit)
311		}	340		}
312	-	clk.advance(10 * time.Second)	341	+	clk.advance(time.Second)
313		link.flush(bPC)	342		link.flush(bPC)
314		awaitSignal(t, contacted, "queued authenticated fragment")	343		awaitSignal(t, contacted, "queued authenticated fragment")
315	-	state = floodState(a, b, link)	344	+	for link.queuedBytes() > 0 {
316	-	if state.queuedBytes != 0 {	345	+	clk.advance(time.Second)
317	-	t.Fatalf("flood state = %+v, want empty queue after fake-clock release", state)	346	+	link.flush(bPC)
318		}	347		}
319	-	if state.packetAge != 0 \|\| state.recordAge != 10time.Second \|\| state.ackAge != 10time.Second {	348	+	if age := floodState(a, b, link).packetAge; age >= a.probeAfter {
320	-	t.Fatalf("flood state = %+v, want delayed packet contact without record or ACK progress", state)	349	+	t.Fatalf("authenticated contact age=%v, want below probe threshold %v", age, a.probeAfter)
			350	+	}
			351	+	eventually(t, time.Second, func() bool {
			352	+	a.mu.Lock()
			353	+	defer a.mu.Unlock()
			354	+	p := a.pending[1]
			355	+	return p != nil && !p.initialInFlight && !p.last.IsZero()
			356	+	})
			357	+
			358	+	a.mu.Lock()
			359	+	beforeLoss := a.congestion.cwndBytes
			360	+	a.rto = time.Millisecond
			361	+	a.mu.Unlock()
			362	+	aPC.link = nil // The transient bottleneck clears before the paced retry.
			363	+	clk.advance(2 * time.Millisecond)
			364	+	a.queueRetransmits()
			365	+	eventually(t, time.Second, func() bool {
			366	+	a.mu.Lock()
			367	+	defer a.mu.Unlock()
			368	+	return a.congestion.cwndBytes < beforeLoss
			369	+	})
			370	+	for range 16 {
			371	+	clk.advance(time.Second)
			372	+	link.flush(bPC)
			373	+	clk.advance(maxACKDelay)
			374	+	time.Sleep(time.Millisecond)
			375	+	a.mu.Lock()
			376	+	drained := len(a.pending) == 0
			377	+	a.mu.Unlock()
			378	+	if drained {
			379	+	return
			380	+	}
321		}	381		}
			382	+	a.mu.Lock()
			383	+	pending := a.pending[1]
			384	+	a.mu.Unlock()
			385	+	t.Fatalf("pending records did not drain after paced retransmission: %+v pending=%+v drops=%d", floodState(a, b, link), pending, link.droppedPackets())
322		})	386		})
323			387
324		t.Run("control stays responsive while fresh output and large retransmits contend", func(t *testing.T) {	388		t.Run("control stays responsive while fresh output and large retransmits contend", func(t *testing.T) {

@@ -350,8 +414,6 @@ func TestTransportFloodClassification(t *testing.T) {
350		releaseRetransmit := make(chan struct{})	414		releaseRetransmit := make(chan struct{})
351		lastRetransmitted := make(chan struct{}, 1)	415		lastRetransmitted := make(chan struct{}, 1)
352		probeSent := make(chan struct{}, 1)	416		probeSent := make(chan struct{}, 1)
353	-	freshAtPace := make(chan struct{})
354	-	allowFreshPace := make(chan struct{})
355		freshReturned := make(chan error, 1)	417		freshReturned := make(chan error, 1)
356		writes := make(map[uint64]int)	418		writes := make(map[uint64]int)
357		var writesMu sync.Mutex	419		var writesMu sync.Mutex

@@ -406,6 +468,9 @@ func TestTransportFloodClassification(t *testing.T) {
406		for range 6 { // Both transports install retransmit, health, and heartbeat timers.	468		for range 6 { // Both transports install retransmit, health, and heartbeat timers.
407		awaitSignal(t, clk.timerCreated, "transport timer creation")	469		awaitSignal(t, clk.timerCreated, "transport timer creation")
408		}	470		}
			471	+	a.mu.Lock()
			472	+	a.rto = time.Hour
			473	+	a.mu.Unlock()
409			474
410		// Fresh, fragmented output records become overdue while their real	475		// Fresh, fragmented output records become overdue while their real
411		// cumulative ACKs are lost. Disabling pacing intentionally forces overload	476		// cumulative ACKs are lost. Disabling pacing intentionally forces overload

@@ -414,48 +479,60 @@ func TestTransportFloodClassification(t *testing.T) {
414		awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")	479		awaitSignal(t, clk.timerCreated, "ACK coalescing timer creation")
415		clk.advance(maxACKDelay)	480		clk.advance(maxACKDelay)
416		awaitSignal(t, acksDropped, "dropped cumulative ACK")	481		awaitSignal(t, acksDropped, "dropped cumulative ACK")
			482	+	a.mu.Lock()
			483	+	a.rto = 10 * time.Millisecond
			484	+	a.mu.Unlock()
			485	+	a.queueRetransmits()
			486	+	a.mu.Lock()
			487	+	a.rto = time.Hour
			488	+	a.mu.Unlock()
			489	+	clk.advance(initialPacingRTT)
417			490
418		awaitSignal(t, retransmitBlocked, "retransmit pacing barrier")	491		awaitSignal(t, retransmitBlocked, "retransmit pacing barrier")
419		if state := floodState(a, b, nil); state.retransmits < floodRecordCount {	492		if state := floodState(a, b, nil); state.retransmits < floodRecordCount {
420		t.Fatalf("flood state = %+v, want all %d overdue records selected", state, floodRecordCount)	493		t.Fatalf("flood state = %+v, want all %d overdue records selected", state, floodRecordCount)
421		}	494		}
422	-		495	+	select {
423	-	// Start a third large output at this same fake-clock instant. The pacing	496	+	case <-probeSent:
424	-	// barrier proves its real Send reaches the shared pacing gate while the	497	+	default:
425	-	// retransmit owns it inside WriteTo.
426	-	a.mu.Lock()
427	-	a.outputNext = time.Time{}
428	-	a.beforeDataPace = func() {
429	-	close(freshAtPace)
430	-	awaitSignal(t, allowFreshPace, "fresh pacing release")
431		}	498		}
432	-	a.mu.Unlock()	499	+
			500	+	// A fresh record queues behind the sole data sender while control traffic
			501	+	// remains independent of congestion tokens and the blocked data write.
433		go func() {	502		go func() {
434		freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(floodRecordCount, 128)})	503		freshReturned <- a.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(floodRecordCount, 128)})
435		}()	504		}()
436	-	awaitSignal(t, freshAtPace, "fresh output pacing barrier")
437	-	a.mu.Lock()
438	-	a.beforeDataPace = nil
439	-	a.mu.Unlock()
440	-	close(allowFreshPace)
441		select {	505		select {
442		case err := <-freshReturned:	506		case err := <-freshReturned:
443		t.Fatalf("fresh output completed while retransmit held pacing: %v", err)	507		t.Fatalf("fresh output completed while retransmit held pacing: %v", err)
444		default:	508		default:
445		}	509		}
446			510
			511	+	clk.advance(11 * time.Millisecond)
447		awaitSignal(t, probeSent, "heartbeat probe during output contention")	512		awaitSignal(t, probeSent, "heartbeat probe during output contention")
448			513
449		close(releaseRetransmit)	514		close(releaseRetransmit)
450	-	if err := awaitResult(t, freshReturned, "fresh output completion"); err != nil {	515	+	for range 1024 {
451	-	t.Fatal(err)	516	+	clk.advance(initialPacingRTT)
			517	+	time.Sleep(time.Millisecond)
			518	+	select {
			519	+	case err := <-freshReturned:
			520	+	if err != nil {
			521	+	t.Fatal(err)
			522	+	}
			523	+	awaitSignal(t, lastRetransmitted, "last overdue record retransmission")
			524	+	writesMu.Lock()
			525	+	defer writesMu.Unlock()
			526	+	if writes[1] < 2 \|\| writes[floodRecordCount] < 2 \|\| writes[floodRecordCount+1] == 0 {
			527	+	t.Fatalf("flood state = %+v, writes=%v, want fresh large traffic and retransmissions across %d overdue records", floodState(a, b, nil), writes, floodRecordCount)
			528	+	}
			529	+	return
			530	+	default:
			531	+	}
452		}	532		}
453	-	awaitSignal(t, lastRetransmitted, "last overdue record retransmission")
454		writesMu.Lock()	533		writesMu.Lock()
455		defer writesMu.Unlock()	534		defer writesMu.Unlock()
456	-	if writes[1] < 2 \|\| writes[floodRecordCount] < 2 \|\| writes[floodRecordCount+1] == 0 {	535	+	t.Fatalf("fresh output did not complete after paced retransmissions: writes=%v state=%+v", writes, floodState(a, b, nil))
457	-	t.Fatalf("flood state = %+v, writes=%v, want fresh large traffic and retransmissions across %d overdue records", floodState(a, b, nil), writes, floodRecordCount)
458	-	}
459		})	536		})
460		}	537		}
461			538

@@ -510,23 +587,37 @@ func TestFloodOutputPayloadIsIncrementalStateBearingOutput(t *testing.T) {
510			587
511		func sendFloodOutputs(t testing.T, sender Transport, count, mtu int) {	588		func sendFloodOutputs(t testing.T, sender Transport, count, mtu int) {
512		t.Helper()	589		t.Helper()
513	-	sender.dataPaceMu.Lock()
514	-	sender.dataPaceRemaining = count * floodRecordMTUs * 4
515	-	sender.dataPaceNext = sender.clock.Now().Add(time.Hour)
516	-	sender.dataPaceMu.Unlock()
517		for state := range count {	590		for state := range count {
			591	+	if err := sender.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(state, mtu)}); err != nil {
			592	+	t.Fatal(err)
			593	+	}
			594	+	}
			595	+	clk, ok := sender.clock.(*manualClock)
			596	+	if !ok {
			597	+	return
			598	+	}
			599	+	for range count * floodRecordMTUs * 2 {
			600	+	clk.advance(initialPacingRTT)
			601	+	time.Sleep(time.Millisecond)
518		sender.mu.Lock()	602		sender.mu.Lock()
519	-	sender.outputNext = time.Time{}	603	+	complete := len(sender.outputQueue) == 0 && len(sender.dataSend) == 0
			604	+	for _, p := range sender.pending {
			605	+	complete = complete && !p.initialInFlight && !p.last.IsZero()
			606	+	}
520		sender.mu.Unlock()	607		sender.mu.Unlock()
521	-	if err := sender.Send(ports.Frame{Type: ports.MsgOutput, Payload: floodOutputPayload(state, mtu)}); err != nil {	608	+	if complete {
522	-	t.Fatal(err)	609	+	return
523		}	610		}
524		}	611		}
			612	+	t.Fatal("flood output pacing did not drain")
525		}	613		}
526			614
527		func newFloodTransports(t testing.T, aPC, bPC fakePC, clk manualClock, mtu int) (Transport, *Transport) {	615		func newFloodTransports(t testing.T, aPC, bPC fakePC, clk manualClock, mtu int) (Transport, *Transport) {
528		t.Helper()	616		t.Helper()
529	-	opts := Options{Clock: clk, MTU: mtu, ResendAfter: 10 * time.Millisecond, Heartbeat: time.Hour}	617	+	opts := Options{
			618	+	Clock: clk, MTU: mtu, ResendAfter: time.Hour, Heartbeat: time.Hour,
			619	+	DegradedAfter: time.Hour, ProbeAfter: time.Hour, OfflineAfter: time.Hour, DeadAfter: time.Hour,
			620	+	}
530		a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts)	621		a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, opts)
531		if err != nil {	622		if err != nil {
532		t.Fatal(err)	623		t.Fatal(err)

M internal/adapters/dgram/output_pacer.go +11 -73

@@ -1,18 +1,16 @@
1	1		package dgram
2	2
3		-	import (
4		-	"time"
	3	+	import "github.com/bnema/vev/internal/ports"
5	4
6		-	"github.com/bnema/vev/internal/ports"
7		-	)
8		-
9		-	type queuedSend struct {
	5	+	type dataSendJob struct {
10	6		seq uint64
11	7		reliable bool
12	8		frame ports.Frame
13	9		done chan error
14	10		}
15	11
	12	+	type queuedSend = dataSendJob
	13	+
16	14		func (t *Transport) outputPaceLoop() {
17	15		for {
18	16		t.outboundMu.Lock()

@@ -33,45 +31,14 @@ func (t *Transport) outputPaceLoop() {
33	31		t.outboundMu.Unlock()
34	32		return
35	33		}
36		-	now := t.clock.Now()
37		-	wait := t.outputNext.Sub(now)
38		-	if wait > 0 {
39		-	wake := t.outputWake
40		-	timer := t.clock.NewTimer(wait)
41		-	t.mu.Unlock()
42		-	t.outboundMu.Unlock()
43		-	select {
44		-	case <-timer.C():
45		-	case <-wake:
46		-	case <-t.done:
47		-	timer.Stop()
48		-	return
49		-	}
50		-	timer.Stop()
51		-	continue
52		-	}
53		-	limit := 0
54		-	budget := t.mtu
55		-	for limit < len(t.outputQueue) && limit < defaultOutputPaceBatch {
56		-	cost := len(t.outputQueue[limit].frame.Payload) + dataRecordHeaderSize
57		-	if limit > 0 && cost > budget {
58		-	break
59		-	}
60		-	limit++
61		-	budget -= min(cost, budget)
62		-	}
63		-	batch := append([]queuedSend(nil), t.outputQueue[:limit]...)
64		-	copy(t.outputQueue, t.outputQueue[limit:])
65		-	t.outputQueue = t.outputQueue[:len(t.outputQueue)-limit]
66		-	t.outputNext = now.Add(t.outputPaceDelayLocked())
	34	+	q := t.outputQueue[0]
	35	+	copy(t.outputQueue, t.outputQueue[1:])
	36	+	t.outputQueue = t.outputQueue[:len(t.outputQueue)-1]
67	37		t.mu.Unlock()
68		-	for i, q := range batch {
69		-	if err := t.sendQueuedData(q); err != nil {
70		-	t.removeQueuedPending(batch[i+1:], err)
71		-	t.closeWithError(err)
72		-	t.outboundMu.Unlock()
73		-	return
74		-	}
	38	+	if err := t.queueDataJob(dataSendJob(q)); err != nil {
	39	+	t.removeQueuedPending([]queuedSend{q}, err)
	40	+	t.outboundMu.Unlock()
	41	+	return
75	42		}
76	43		t.outboundMu.Unlock()
77	44		}

@@ -85,39 +52,10 @@ func (t *Transport) notifyOutputPacerLocked() {
85	52		t.outputWake = make(chan struct{})
86	53		}
87	54
88		-	func (t *Transport) outputPaceDelayLocked() time.Duration {
89		-	if t.srtt <= 0 {
90		-	return defaultOutputPaceMinDelay
91		-	}
92		-	d := t.srtt / 2
93		-	if d < defaultOutputPaceMinCadence {
94		-	d = defaultOutputPaceMinCadence
95		-	}
96		-	if d > defaultOutputPaceMaxDelay {
97		-	d = defaultOutputPaceMaxDelay
98		-	}
99		-	return d
100		-	}
101		-
102	55		func shouldPaceOutput(f ports.Frame) bool {
103	56		return f.Type == ports.MsgOutput
104	57		}
105	58
106		-	func (t *Transport) sendQueuedData(q queuedSend) error {
107		-	t.markPendingSent(q.seq, q.reliable)
108		-	err := t.sendData(q.seq, q.reliable, q.frame)
109		-	if err != nil {
110		-	t.removePending(q.seq, q.reliable)
111		-	} else {
112		-	t.markPendingReady(q.seq, q.reliable)
113		-	}
114		-	if q.done != nil {
115		-	q.done <- err
116		-	close(q.done)
117		-	}
118		-	return err
119		-	}
120		-
121	59		func (t *Transport) markPendingSent(seq uint64, reliable bool) {
122	60		if !reliable {
123	61		return

@@ -1,18 +1,16 @@
1		package dgram	1		package dgram
2			2
3	-	import (	3	+	import "github.com/bnema/vev/internal/ports"
4	-	"time"
5			4
6	-	"github.com/bnema/vev/internal/ports"	5	+	type dataSendJob struct {
7	-	)
8	-
9	-	type queuedSend struct {
10		seq uint64	6		seq uint64
11		reliable bool	7		reliable bool
12		frame ports.Frame	8		frame ports.Frame
13		done chan error	9		done chan error
14		}	10		}
15			11
			12	+	type queuedSend = dataSendJob
			13	+
16		func (t *Transport) outputPaceLoop() {	14		func (t *Transport) outputPaceLoop() {
17		for {	15		for {
18		t.outboundMu.Lock()	16		t.outboundMu.Lock()

@@ -33,45 +31,14 @@ func (t *Transport) outputPaceLoop() {
33		t.outboundMu.Unlock()	31		t.outboundMu.Unlock()
34		return	32		return
35		}	33		}
36	-	now := t.clock.Now()	34	+	q := t.outputQueue[0]
37	-	wait := t.outputNext.Sub(now)	35	+	copy(t.outputQueue, t.outputQueue[1:])
38	-	if wait > 0 {	36	+	t.outputQueue = t.outputQueue[:len(t.outputQueue)-1]
39	-	wake := t.outputWake
40	-	timer := t.clock.NewTimer(wait)
41	-	t.mu.Unlock()
42	-	t.outboundMu.Unlock()
43	-	select {
44	-	case <-timer.C():
45	-	case <-wake:
46	-	case <-t.done:
47	-	timer.Stop()
48	-	return
49	-	}
50	-	timer.Stop()
51	-	continue
52	-	}
53	-	limit := 0
54	-	budget := t.mtu
55	-	for limit < len(t.outputQueue) && limit < defaultOutputPaceBatch {
56	-	cost := len(t.outputQueue[limit].frame.Payload) + dataRecordHeaderSize
57	-	if limit > 0 && cost > budget {
58	-	break
59	-	}
60	-	limit++
61	-	budget -= min(cost, budget)
62	-	}
63	-	batch := append([]queuedSend(nil), t.outputQueue[:limit]...)
64	-	copy(t.outputQueue, t.outputQueue[limit:])
65	-	t.outputQueue = t.outputQueue[:len(t.outputQueue)-limit]
66	-	t.outputNext = now.Add(t.outputPaceDelayLocked())
67		t.mu.Unlock()	37		t.mu.Unlock()
68	-	for i, q := range batch {	38	+	if err := t.queueDataJob(dataSendJob(q)); err != nil {
69	-	if err := t.sendQueuedData(q); err != nil {	39	+	t.removeQueuedPending([]queuedSend{q}, err)
70	-	t.removeQueuedPending(batch[i+1:], err)	40	+	t.outboundMu.Unlock()
71	-	t.closeWithError(err)	41	+	return
72	-	t.outboundMu.Unlock()
73	-	return
74	-	}
75		}	42		}
76		t.outboundMu.Unlock()	43		t.outboundMu.Unlock()
77		}	44		}

@@ -85,39 +52,10 @@ func (t *Transport) notifyOutputPacerLocked() {
85		t.outputWake = make(chan struct{})	52		t.outputWake = make(chan struct{})
86		}	53		}
87			54
88	-	func (t *Transport) outputPaceDelayLocked() time.Duration {
89	-	if t.srtt <= 0 {
90	-	return defaultOutputPaceMinDelay
91	-	}
92	-	d := t.srtt / 2
93	-	if d < defaultOutputPaceMinCadence {
94	-	d = defaultOutputPaceMinCadence
95	-	}
96	-	if d > defaultOutputPaceMaxDelay {
97	-	d = defaultOutputPaceMaxDelay
98	-	}
99	-	return d
100	-	}
101	-
102		func shouldPaceOutput(f ports.Frame) bool {	55		func shouldPaceOutput(f ports.Frame) bool {
103		return f.Type == ports.MsgOutput	56		return f.Type == ports.MsgOutput
104		}	57		}
105			58
106	-	func (t *Transport) sendQueuedData(q queuedSend) error {
107	-	t.markPendingSent(q.seq, q.reliable)
108	-	err := t.sendData(q.seq, q.reliable, q.frame)
109	-	if err != nil {
110	-	t.removePending(q.seq, q.reliable)
111	-	} else {
112	-	t.markPendingReady(q.seq, q.reliable)
113	-	}
114	-	if q.done != nil {
115	-	q.done <- err
116	-	close(q.done)
117	-	}
118	-	return err
119	-	}
120	-
121		func (t *Transport) markPendingSent(seq uint64, reliable bool) {	59		func (t *Transport) markPendingSent(seq uint64, reliable bool) {
122		if !reliable {	60		if !reliable {
123		return	61		return

A internal/adapters/dgram/pacer.go +60 -0

@@ -0,0 +1,60 @@
	1	+	package dgram
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"time"
	6	+
	7	+	"github.com/bnema/vev/internal/ports"
	8	+	)
	9	+
	10	+	var errPacerClosed = errors.New("dgram: pacer closed")
	11	+
	12	+	type paceLimits func() (bytesPerSecond int, burstBytes int)
	13	+
	14	+	type bytePacer struct {
	15	+	clk ports.Clock
	16	+	tokens int64
	17	+	last time.Time
	18	+	}
	19	+
	20	+	func (p *bytePacer) wait(done <-chan struct{}, n int, limits paceLimits) error {
	21	+	if n <= 0 {
	22	+	return nil
	23	+	}
	24	+	for {
	25	+	now := p.clk.Now()
	26	+	rate, burst := limits()
	27	+	if rate < 1 {
	28	+	rate = 1
	29	+	}
	30	+	if burst < n {
	31	+	burst = n
	32	+	}
	33	+	if p.last.IsZero() {
	34	+	p.last = now
	35	+	p.tokens = int64(burst)
	36	+	} else {
	37	+	elapsed := now.Sub(p.last)
	38	+	if elapsed > 0 {
	39	+	p.tokens += int64(elapsed) * int64(rate) / int64(time.Second)
	40	+	p.last = now
	41	+	}
	42	+	p.tokens = min(p.tokens, int64(burst))
	43	+	}
	44	+	if p.tokens >= int64(n) {
	45	+	p.tokens -= int64(n)
	46	+	return nil
	47	+	}
	48	+
	49	+	deficit := int64(n) - p.tokens
	50	+	wait := time.Duration((deficit*int64(time.Second) + int64(rate) - 1) / int64(rate))
	51	+	timer := p.clk.NewTimer(wait)
	52	+	select {
	53	+	case <-timer.C():
	54	+	timer.Stop()
	55	+	case <-done:
	56	+	timer.Stop()
	57	+	return errPacerClosed
	58	+	}
	59	+	}
	60	+	}

@@ -0,0 +1,60 @@
	1	+	package dgram
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"time"
	6	+
	7	+	"github.com/bnema/vev/internal/ports"
	8	+	)
	9	+
	10	+	var errPacerClosed = errors.New("dgram: pacer closed")
	11	+
	12	+	type paceLimits func() (bytesPerSecond int, burstBytes int)
	13	+
	14	+	type bytePacer struct {
	15	+	clk ports.Clock
	16	+	tokens int64
	17	+	last time.Time
	18	+	}
	19	+
	20	+	func (p *bytePacer) wait(done <-chan struct{}, n int, limits paceLimits) error {
	21	+	if n <= 0 {
	22	+	return nil
	23	+	}
	24	+	for {
	25	+	now := p.clk.Now()
	26	+	rate, burst := limits()
	27	+	if rate < 1 {
	28	+	rate = 1
	29	+	}
	30	+	if burst < n {
	31	+	burst = n
	32	+	}
	33	+	if p.last.IsZero() {
	34	+	p.last = now
	35	+	p.tokens = int64(burst)
	36	+	} else {
	37	+	elapsed := now.Sub(p.last)
	38	+	if elapsed > 0 {
	39	+	p.tokens += int64(elapsed) * int64(rate) / int64(time.Second)
	40	+	p.last = now
	41	+	}
	42	+	p.tokens = min(p.tokens, int64(burst))
	43	+	}
	44	+	if p.tokens >= int64(n) {
	45	+	p.tokens -= int64(n)
	46	+	return nil
	47	+	}
	48	+
	49	+	deficit := int64(n) - p.tokens
	50	+	wait := time.Duration((deficit*int64(time.Second) + int64(rate) - 1) / int64(rate))
	51	+	timer := p.clk.NewTimer(wait)
	52	+	select {
	53	+	case <-timer.C():
	54	+	timer.Stop()
	55	+	case <-done:
	56	+	timer.Stop()
	57	+	return errPacerClosed
	58	+	}
	59	+	}
	60	+	}

M internal/adapters/dgram/transport.go +162 -135


@@ -20,28 +20,23 @@ const (
20	20		recProbe byte = 3
21	21		recPong byte = 4
22	22
23		-	defaultMTU = pdgram.DefaultMTU
24		-	defaultResend = 250 * time.Millisecond
25		-	defaultMaxResendAfter = 2 * time.Second
26		-	defaultMaxResendPerTick = 64
27		-	defaultWriteTimeout = 250 * time.Millisecond
28		-	defaultHeartbeat = 3 * time.Second
29		-	defaultDegraded = 10 * time.Second
30		-	defaultProbe = 20 * time.Second
31		-	defaultOffline = 30 * time.Second
32		-	defaultDead = 60 * time.Second
33		-	defaultMaxPending = 1024
34		-	defaultMaxPendingWait = 50 * time.Millisecond
35		-	defaultMaxRecvBuf = 1024
36		-	defaultOutputPaceMinDelay = 8 * time.Millisecond
37		-	defaultOutputPaceMinCadence = 20 * time.Millisecond
38		-	defaultOutputPaceMaxDelay = 250 * time.Millisecond
39		-	defaultOutputPaceBatch = 2
40		-	defaultPacketPaceBudget = 8
41		-	dataRecordHeaderSize = 12
42		-	maxRecvBuffer = defaultMaxRecvBuf
43		-	linkEventBufferSize = 16
44		-	probeReplyBufferSize = controlQueueSize // legacy test burst bound
	23	+	defaultMTU = pdgram.DefaultMTU
	24	+	defaultResend = 250 * time.Millisecond
	25	+	defaultMaxResendAfter = 2 * time.Second
	26	+	defaultMaxResendPerTick = 64
	27	+	defaultWriteTimeout = 250 * time.Millisecond
	28	+	defaultHeartbeat = 3 * time.Second
	29	+	defaultDegraded = 10 * time.Second
	30	+	defaultProbe = 20 * time.Second
	31	+	defaultOffline = 30 * time.Second
	32	+	defaultDead = 60 * time.Second
	33	+	defaultMaxPending = 1024
	34	+	defaultMaxPendingWait = 50 * time.Millisecond
	35	+	defaultMaxRecvBuf = 1024
	36	+	dataRecordHeaderSize = 12
	37	+	maxRecvBuffer = defaultMaxRecvBuf
	38	+	linkEventBufferSize = 16
	39	+	probeReplyBufferSize = controlQueueSize // legacy test burst bound
45	40		)
46	41
47	42		var (

@@ -188,6 +183,7 @@ type Transport struct {
188	183		maxPendingWait time.Duration
189	184		maxRecvBuffer int
190	185		clock ports.Clock
	186	+	congestion congestionController
191	187		linkState ports.LinkState
192	188		linkStateGeneration uint64
193	189		linkEvents chan ports.LinkEvent

@@ -205,7 +201,7 @@ type Transport struct {
205	201		hopGeneration uint64
206	202		outputQueue []queuedSend
207	203		outputWake chan struct{}
208		-	outputNext time.Time
	204	+	dataSend chan dataSendJob
209	205		retransmitWork chan []retransmitRecord
210	206		writeDeadlines *writeDeadlineState
211	207

@@ -215,10 +211,7 @@ type Transport struct {
215	211		in chan ports.Frame
216	212		done chan struct{}
217	213
218		-	writeMu sync.Mutex
219		-	dataPaceMu sync.Mutex
220		-	dataPaceRemaining int
221		-	dataPaceNext time.Time
	214	+	writeMu sync.Mutex
222	215		// beforeDataPace is a test synchronization hook. It is read under mu.
223	216		beforeDataPace func()
224	217		// outboundMu orders first transmissions and paced batches. Retransmits use

@@ -239,6 +232,7 @@ type pending struct {
239	232		initialInFlight bool
240	233		attempts int
241	234		retransmitted bool
	235	+	wireBytes int
242	236		}
243	237
244	238		type retransmitRecord struct {

@@ -338,6 +332,7 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
338	332		maxPendingWait: opts.MaxPendingWait,
339	333		maxRecvBuffer: opts.MaxRecvBuffer,
340	334		clock: opts.Clock,
	335	+	congestion: newCongestionController(opts.MTU),
341	336		linkState: ports.LinkStateConnected,
342	337		linkEvents: make(chan ports.LinkEvent, linkEventBufferSize),
343	338		probeWait: make(map[uint64]chan struct{}),

@@ -350,6 +345,7 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
350	345		t.ackWake = make(chan struct{}, 1)
351	346		t.ackSend = make(chan uint64, 1)
352	347		t.outputWake = make(chan struct{})
	348	+	t.dataSend = make(chan dataSendJob, opts.MaxPending)
353	349		t.writeDeadlines = newWriteDeadlineState(pc)
354	350		// One queued batch bounds retransmit work while its sole sender is paced.
355	351		t.retransmitWork = make(chan []retransmitRecord, 1)

@@ -362,7 +358,7 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
362	358		go t.retransmitLoop()
363	359		go t.heartbeatLoop()
364	360		go t.healthLoop()
365		-	go t.retransmitSenderLoop()
	361	+	go t.dataSendLoop()
366	362		go t.deliveryLoop()
367	363		go t.controlLoop()
368	364		go t.ackSendLoop()

@@ -399,60 +395,47 @@ func (t *Transport) send(f ports.Frame, async bool) error {
399	395		if len(t.pending) == 0 {
400	396		t.health.pendingStarted(now)
401	397		}
402		-	t.pending[seq] = &pending{frame: f, enqueued: now}
	398	+	t.pending[seq] = &pending{frame: f, enqueued: now, wireBytes: dataRecordHeaderSize + len(f.Payload)}
403	399		}
404		-	if shouldPaceOutput(f) {
405		-	now := t.clock.Now()
406		-	if len(t.outputQueue) > 0 \|\| (!t.outputNext.IsZero() && now.Before(t.outputNext)) {
407		-	wasEmpty := len(t.outputQueue) == 0
408		-	var done chan error
409		-	if !async {
410		-	done = make(chan error, 1)
411		-	}
412		-	t.outputQueue = append(t.outputQueue, queuedSend{seq: seq, reliable: reliable, frame: f, done: done})
413		-	if wasEmpty {
414		-	t.notifyOutputPacerLocked()
415		-	}
416		-	t.mu.Unlock()
417		-	if done == nil {
418		-	return nil
419		-	}
420		-	t.outboundMu.Unlock()
421		-	unlockOutbound = false
422		-	return t.waitQueuedSend(done)
423		-	}
424		-	t.outputNext = now.Add(t.outputPaceDelayLocked())
425		-	t.mu.Unlock()
426		-	t.markPendingSent(seq, reliable)
427		-	if err := t.sendData(seq, reliable, f); err != nil {
428		-	t.removePending(seq, reliable)
429		-	return err
430		-	}
431		-	t.markPendingReady(seq, reliable)
432		-	return nil
	400	+	var done chan error
	401	+	if !async {
	402	+	done = make(chan error, 1)
433	403		}
434		-	if len(t.outputQueue) > 0 {
435		-	var done chan error
436		-	if !async {
437		-	done = make(chan error, 1)
	404	+	job := dataSendJob{seq: seq, reliable: reliable, frame: f, done: done}
	405	+	if shouldPaceOutput(f) \|\| len(t.outputQueue) > 0 {
	406	+	wasEmpty := len(t.outputQueue) == 0
	407	+	t.outputQueue = append(t.outputQueue, queuedSend(job))
	408	+	if wasEmpty {
	409	+	t.notifyOutputPacerLocked()
438	410		}
439		-	t.outputQueue = append(t.outputQueue, queuedSend{seq: seq, reliable: reliable, frame: f, done: done})
440	411		t.mu.Unlock()
441		-	if done == nil {
442		-	return nil
443		-	}
444	412		t.outboundMu.Unlock()
445	413		unlockOutbound = false
	414	+	if async {
	415	+	return nil
	416	+	}
446	417		return t.waitQueuedSend(done)
447	418		}
448	419		t.mu.Unlock()
449		-	t.markPendingSent(seq, reliable)
450		-	if err := t.sendData(seq, reliable, f); err != nil {
	420	+	if err := t.queueDataJob(job); err != nil {
451	421		t.removePending(seq, reliable)
452	422		return err
453	423		}
454		-	t.markPendingReady(seq, reliable)
455		-	return nil
	424	+	t.outboundMu.Unlock()
	425	+	unlockOutbound = false
	426	+	if async {
	427	+	return nil
	428	+	}
	429	+	return t.waitQueuedSend(done)
	430	+	}
	431	+
	432	+	func (t *Transport) queueDataJob(job dataSendJob) error {
	433	+	select {
	434	+	case t.dataSend <- job:
	435	+	return nil
	436	+	case <-t.done:
	437	+	return t.closedError()
	438	+	}
456	439		}
457	440
458	441		func (t *Transport) waitQueuedSend(done <-chan error) error {

@@ -653,40 +636,18 @@ func (t *Transport) unregisterProbe(id uint64) {
653	636		t.mu.Unlock()
654	637		}
655	638
656		-	func (t *Transport) sendData(seq uint64, reliable bool, f ports.Frame) error {
657		-	p := encodeData(seq, reliable, f)
658		-	return t.sendPayload(p, true, seq)
659		-	}
660	639		func (t *Transport) sendAck(seq uint64) {
661	640		var b [9]byte
662	641		b[0] = recAck
663	642		binary.BigEndian.PutUint64(b[1:], seq)
664		-	_ = t.sendPayload(b[:], false, 0)
	643	+	_ = t.sendPayload(b[:])
665	644		}
666		-	func (t *Transport) sendPayload(p []byte, paced bool, pendingSeq uint64) error {
667		-	if paced {
668		-	t.mu.Lock()
669		-	beforeDataPace := t.beforeDataPace
670		-	t.mu.Unlock()
671		-	if beforeDataPace != nil {
672		-	beforeDataPace()
673		-	}
674		-	t.dataPaceMu.Lock()
675		-	defer t.dataPaceMu.Unlock()
676		-	}
	645	+	func (t *Transport) sendPayload(p []byte) error {
677	646		frags, err := pdgram.FragmentPayload(t.nextCounter(), p, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
678	647		if err != nil {
679	648		return err
680	649		}
681		-	for i, f := range frags {
682		-	if pendingSeq != 0 && !t.pendingExists(pendingSeq) {
683		-	return nil
684		-	}
685		-	if paced {
686		-	if err := t.takeDataPaceSlot(); err != nil {
687		-	return err
688		-	}
689		-	}
	650	+	for _, f := range frags {
690	651		raw, err := pdgram.MarshalFragment(f)
691	652		if err != nil {
692	653		return err

@@ -706,9 +667,6 @@ func (t *Transport) sendPayload(p []byte, paced bool, pendingSeq uint64) error {
706	667		if peer == nil {
707	668		return errors.New("dgram: no peer")
708	669		}
709		-	if pendingSeq != 0 && i == len(frags)-1 {
710		-	t.markPendingFinalWrite(pendingSeq)
711		-	}
712	670		if err := t.writePacket(pkt, peer); err != nil {
713	671		return err
714	672		}

@@ -723,30 +681,6 @@ func (t *Transport) pendingExists(seq uint64) bool {
723	681		return ok
724	682		}
725	683
726		-	func (t *Transport) takeDataPaceSlot() error {
727		-	now := t.clock.Now()
728		-	if t.dataPaceRemaining > 0 && now.Before(t.dataPaceNext) {
729		-	t.dataPaceRemaining--
730		-	return nil
731		-	}
732		-	if t.dataPaceNext.After(now) {
733		-	timer := t.clock.NewTimer(t.dataPaceNext.Sub(now))
734		-	select {
735		-	case <-timer.C():
736		-	case <-t.done:
737		-	timer.Stop()
738		-	return t.closedError()
739		-	}
740		-	timer.Stop()
741		-	}
742		-	t.mu.Lock()
743		-	delay := t.outputPaceDelayLocked()
744		-	t.mu.Unlock()
745		-	t.dataPaceRemaining = defaultPacketPaceBudget - 1
746		-	t.dataPaceNext = t.clock.Now().Add(delay)
747		-	return nil
748		-	}
749		-
750	684		func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
751	685		t.writeMu.Lock()
752	686		defer t.writeMu.Unlock()

@@ -878,6 +812,7 @@ func (t *Transport) handleRecord(p []byte) {
878	812		seq := binary.BigEndian.Uint64(p[1:])
879	813		t.mu.Lock()
880	814		acked := false
	815	+	ackedBytes := 0
881	816		for pendingSeq, p := range t.pending {
882	817		if pendingSeq > seq {
883	818		continue

@@ -885,10 +820,14 @@ func (t *Transport) handleRecord(p []byte) {
885	820		if pendingSeq == seq && p != nil && !p.retransmitted {
886	821		t.updateRTTLocked(t.clock.Now().Sub(p.first))
887	822		}
	823	+	if p != nil {
	824	+	ackedBytes += p.wireBytes
	825	+	}
888	826		delete(t.pending, pendingSeq)
889	827		acked = true
890	828		}
891	829		if acked {
	830	+	t.congestion.onACK(ackedBytes)
892	831		now := t.clock.Now()
893	832		t.health.ackProgress(now)
894	833		if len(t.pending) == 0 {

@@ -1042,20 +981,15 @@ func (t *Transport) healthLoop() {
1042	981		}
1043	982		}
1044	983
1045		-	func (t *Transport) retransmitSenderLoop() {
	984	+	func (t *Transport) dataSendLoop() {
	985	+	pacer := bytePacer{clk: t.clock}
1046	986		for {
1047	987		select {
	988	+	case job := <-t.dataSend:
	989	+	t.runDataSendJob(&pacer, job, true)
1048	990		case batch := <-t.retransmitWork:
1049	991		for _, r := range batch {
1050		-	if err := t.sendData(r.seq, true, r.f); err == nil {
1051		-	t.markPendingReady(r.seq, true)
1052		-	} else {
1053		-	t.mu.Lock()
1054		-	if p := t.pending[r.seq]; p != nil {
1055		-	p.initialInFlight = false
1056		-	}
1057		-	t.mu.Unlock()
1058		-	}
	992	+	t.runDataSendJob(&pacer, dataSendJob{seq: r.seq, reliable: true, frame: r.f}, false)
1059	993		}
1060	994		case <-t.done:
1061	995		return

@@ -1063,6 +997,88 @@ func (t *Transport) retransmitSenderLoop() {
1063	997		}
1064	998		}
1065	999
	1000	+	func (t Transport) runDataSendJob(pacer bytePacer, job dataSendJob, initial bool) {
	1001	+	if initial {
	1002	+	t.markPendingSent(job.seq, job.reliable)
	1003	+	}
	1004	+	err := t.writeDataJob(pacer, job)
	1005	+	if err != nil {
	1006	+	if initial {
	1007	+	t.removePending(job.seq, job.reliable)
	1008	+	} else {
	1009	+	t.markPendingReady(job.seq, job.reliable)
	1010	+	}
	1011	+	if !errors.Is(err, errPacerClosed) {
	1012	+	t.closeWithError(err)
	1013	+	}
	1014	+	} else {
	1015	+	t.markPendingReady(job.seq, job.reliable)
	1016	+	}
	1017	+	if job.done != nil {
	1018	+	if errors.Is(err, errPacerClosed) {
	1019	+	err = t.closedError()
	1020	+	}
	1021	+	job.done <- err
	1022	+	close(job.done)
	1023	+	}
	1024	+	}
	1025	+
	1026	+	func (t Transport) writeDataJob(pacer bytePacer, job dataSendJob) error {
	1027	+	payload := encodeData(job.seq, job.reliable, job.frame)
	1028	+	frags, err := pdgram.FragmentPayload(t.nextCounter(), payload, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
	1029	+	if err != nil {
	1030	+	return err
	1031	+	}
	1032	+	t.mu.Lock()
	1033	+	beforeDataPace := t.beforeDataPace
	1034	+	t.mu.Unlock()
	1035	+	if beforeDataPace != nil {
	1036	+	beforeDataPace()
	1037	+	}
	1038	+	limits := func() (int, int) {
	1039	+	t.mu.Lock()
	1040	+	defer t.mu.Unlock()
	1041	+	return t.congestion.bytesPerSecond(), t.congestion.burstBytes()
	1042	+	}
	1043	+	for i, frag := range frags {
	1044	+	if job.reliable && !t.pendingExists(job.seq) {
	1045	+	return nil
	1046	+	}
	1047	+	raw, err := pdgram.MarshalFragment(frag)
	1048	+	if err != nil {
	1049	+	return err
	1050	+	}
	1051	+	pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
	1052	+	if err := pacer.wait(t.done, len(pkt), limits); err != nil {
	1053	+	return err
	1054	+	}
	1055	+	if job.reliable && !t.pendingExists(job.seq) {
	1056	+	return nil
	1057	+	}
	1058	+	t.mu.Lock()
	1059	+	peer := t.peer
	1060	+	closed := t.closed
	1061	+	closeErr := t.closeErr
	1062	+	t.mu.Unlock()
	1063	+	if closed {
	1064	+	if closeErr != nil {
	1065	+	return closeErr
	1066	+	}
	1067	+	return net.ErrClosed
	1068	+	}
	1069	+	if peer == nil {
	1070	+	return errors.New("dgram: no peer")
	1071	+	}
	1072	+	if i == len(frags)-1 {
	1073	+	t.markPendingFinalWrite(job.seq)
	1074	+	}
	1075	+	if err := t.writePacket(pkt, peer); err != nil {
	1076	+	return err
	1077	+	}
	1078	+	}
	1079	+	return nil
	1080	+	}
	1081	+
1066	1082		func (t *Transport) heartbeatLoop() {
1067	1083		heartbeat := t.clock.NewTimer(t.heartbeat)
1068	1084		defer heartbeat.Stop()

@@ -1125,6 +1141,7 @@ func (t *Transport) selectRetransmitsLocked(now time.Time) ([]retransmitRecord,
1125	1141		p.initialInFlight = true
1126	1142		p.attempts++
1127	1143		p.retransmitted = true
	1144	+	t.congestion.onLoss()
1128	1145		t.retransmits++
1129	1146		emitDiagnostic = emitDiagnostic \|\| t.retransmits%64 == 0
1130	1147		resend = append(resend, retransmitRecord{seq: seq, f: p.frame})

@@ -1133,15 +1150,24 @@ func (t *Transport) selectRetransmitsLocked(now time.Time) ([]retransmitRecord,
1133	1150		return resend, emitDiagnostic
1134	1151		}
1135	1152
1136		-	// resendPending remains a synchronous test hook for retransmission selection
1137		-	// unit tests. Production timer paths use queueRetransmits and the sole sender.
	1153	+	// resendPending remains a test hook for retransmission selection. Constructed
	1154	+	// transports transfer its batch to the sole data sender; small unit fixtures
	1155	+	// without transport goroutines execute the selected writes inline.
1138	1156		func (t *Transport) resendPending() {
1139	1157		now := t.clock.Now()
1140	1158		t.mu.Lock()
1141	1159		resend, _ := t.selectRetransmitsLocked(now)
1142	1160		t.mu.Unlock()
	1161	+	if t.retransmitWork != nil {
	1162	+	select {
	1163	+	case t.retransmitWork <- resend:
	1164	+	case <-t.done:
	1165	+	}
	1166	+	return
	1167	+	}
	1168	+	pacer := bytePacer{clk: t.clock}
1143	1169		for _, r := range resend {
1144		-	if err := t.sendData(r.seq, true, r.f); err == nil {
	1170	+	if err := t.writeDataJob(&pacer, dataSendJob{seq: r.seq, reliable: true, frame: r.f}); err == nil {
1145	1171		t.markPendingReady(r.seq, true)
1146	1172		} else {
1147	1173		t.mu.Lock()

@@ -1194,6 +1220,7 @@ func (t *Transport) updateRTTLocked(sample time.Duration) {
1194	1220		rto = t.maxResendAfter
1195	1221		}
1196	1222		t.rto = rto
	1223	+	t.congestion.onRTT(t.srtt)
1197	1224		}
1198	1225
1199	1226		func (t *Transport) hopPacketConnOnce(generation uint64) {

 	recProbe byte = 3
 	recPong  byte = 4
-	defaultMTU                  = pdgram.DefaultMTU
+	defaultMTU              = pdgram.DefaultMTU
-	defaultResend               = 250 * time.Millisecond
+	defaultResend           = 250 * time.Millisecond
-	defaultMaxResendAfter       = 2 * time.Second
+	defaultMaxResendAfter   = 2 * time.Second
-	defaultMaxResendPerTick     = 64
+	defaultMaxResendPerTick = 64
-	defaultWriteTimeout         = 250 * time.Millisecond
+	defaultWriteTimeout     = 250 * time.Millisecond
-	defaultHeartbeat            = 3 * time.Second
+	defaultHeartbeat        = 3 * time.Second
-	defaultDegraded             = 10 * time.Second
+	defaultDegraded         = 10 * time.Second
-	defaultProbe                = 20 * time.Second
+	defaultProbe            = 20 * time.Second
-	defaultOffline              = 30 * time.Second
+	defaultOffline          = 30 * time.Second
-	defaultDead                 = 60 * time.Second
+	defaultDead             = 60 * time.Second
-	defaultMaxPending           = 1024
+	defaultMaxPending       = 1024
-	defaultMaxPendingWait       = 50 * time.Millisecond
+	defaultMaxPendingWait   = 50 * time.Millisecond
-	defaultMaxRecvBuf           = 1024
+	defaultMaxRecvBuf       = 1024
-	defaultOutputPaceMinDelay   = 8 * time.Millisecond
+	dataRecordHeaderSize    = 12
-	defaultOutputPaceMinCadence = 20 * time.Millisecond
+	maxRecvBuffer           = defaultMaxRecvBuf
-	defaultOutputPaceMaxDelay   = 250 * time.Millisecond
+	linkEventBufferSize     = 16
-	defaultOutputPaceBatch      = 2
+	probeReplyBufferSize    = controlQueueSize // legacy test burst bound
-	defaultPacketPaceBudget     = 8
-	dataRecordHeaderSize        = 12
-	maxRecvBuffer               = defaultMaxRecvBuf
-	linkEventBufferSize         = 16
-	probeReplyBufferSize        = controlQueueSize // legacy test burst bound
 )
 var (
 	maxPendingWait       time.Duration
 	maxRecvBuffer        int
 	clock                ports.Clock
+	congestion           congestionController
 	linkState            ports.LinkState
 	linkStateGeneration  uint64
 	linkEvents           chan ports.LinkEvent
 	hopGeneration        uint64
 	outputQueue          []queuedSend
 	outputWake           chan struct{}
-	outputNext           time.Time
+	dataSend             chan dataSendJob
 	retransmitWork       chan []retransmitRecord
 	writeDeadlines       *writeDeadlineState
 	in     chan ports.Frame
 	done   chan struct{}
-	writeMu           sync.Mutex
+	writeMu sync.Mutex
-	dataPaceMu        sync.Mutex
-	dataPaceRemaining int
-	dataPaceNext      time.Time
 	// beforeDataPace is a test synchronization hook. It is read under mu.
 	beforeDataPace func()
 	// outboundMu orders first transmissions and paced batches. Retransmits use
 	initialInFlight bool
 	attempts        int
 	retransmitted   bool
+	wireBytes       int
 }
 type retransmitRecord struct {
 		maxPendingWait:          opts.MaxPendingWait,
 		maxRecvBuffer:           opts.MaxRecvBuffer,
 		clock:                   opts.Clock,
+		congestion:              newCongestionController(opts.MTU),
 		linkState:               ports.LinkStateConnected,
 		linkEvents:              make(chan ports.LinkEvent, linkEventBufferSize),
 		probeWait:               make(map[uint64]chan struct{}),
 	t.ackWake = make(chan struct{}, 1)
 	t.ackSend = make(chan uint64, 1)
 	t.outputWake = make(chan struct{})
+	t.dataSend = make(chan dataSendJob, opts.MaxPending)
 	t.writeDeadlines = newWriteDeadlineState(pc)
 	// One queued batch bounds retransmit work while its sole sender is paced.
 	t.retransmitWork = make(chan []retransmitRecord, 1)
 	go t.retransmitLoop()
 	go t.heartbeatLoop()
 	go t.healthLoop()
-	go t.retransmitSenderLoop()
+	go t.dataSendLoop()
 	go t.deliveryLoop()
 	go t.controlLoop()
 	go t.ackSendLoop()
 		if len(t.pending) == 0 {
 			t.health.pendingStarted(now)
 		}
-		t.pending[seq] = &pending{frame: f, enqueued: now}
+		t.pending[seq] = &pending{frame: f, enqueued: now, wireBytes: dataRecordHeaderSize + len(f.Payload)}
 	}
-	if shouldPaceOutput(f) {
+	var done chan error
-		now := t.clock.Now()
+	if !async {
-		if len(t.outputQueue) > 0 || (!t.outputNext.IsZero() && now.Before(t.outputNext)) {
+		done = make(chan error, 1)
-			wasEmpty := len(t.outputQueue) == 0
-			var done chan error
-			if !async {
-				done = make(chan error, 1)
-			}
-			t.outputQueue = append(t.outputQueue, queuedSend{seq: seq, reliable: reliable, frame: f, done: done})
-			if wasEmpty {
-				t.notifyOutputPacerLocked()
-			}
-			t.mu.Unlock()
-			if done == nil {
-				return nil
-			}
-			t.outboundMu.Unlock()
-			unlockOutbound = false
-			return t.waitQueuedSend(done)
-		}
-		t.outputNext = now.Add(t.outputPaceDelayLocked())
-		t.mu.Unlock()
-		t.markPendingSent(seq, reliable)
-		if err := t.sendData(seq, reliable, f); err != nil {
-			t.removePending(seq, reliable)
-			return err
-		}
-		t.markPendingReady(seq, reliable)
-		return nil
 	}
-	if len(t.outputQueue) > 0 {
+	job := dataSendJob{seq: seq, reliable: reliable, frame: f, done: done}
-		var done chan error
+	if shouldPaceOutput(f) || len(t.outputQueue) > 0 {
-		if !async {
+		wasEmpty := len(t.outputQueue) == 0
-			done = make(chan error, 1)
+		t.outputQueue = append(t.outputQueue, queuedSend(job))
+		if wasEmpty {
+			t.notifyOutputPacerLocked()
 		}
-		t.outputQueue = append(t.outputQueue, queuedSend{seq: seq, reliable: reliable, frame: f, done: done})
 		t.mu.Unlock()
-		if done == nil {
-			return nil
-		}
 		t.outboundMu.Unlock()
 		unlockOutbound = false
+		if async {
+			return nil
+		}
 		return t.waitQueuedSend(done)
 	}
 	t.mu.Unlock()
-	t.markPendingSent(seq, reliable)
+	if err := t.queueDataJob(job); err != nil {
-	if err := t.sendData(seq, reliable, f); err != nil {
 		t.removePending(seq, reliable)
 		return err
 	}
-	t.markPendingReady(seq, reliable)
+	t.outboundMu.Unlock()
-	return nil
+	unlockOutbound = false
+	if async {
+		return nil
+	}
+	return t.waitQueuedSend(done)
+}
+func (t *Transport) queueDataJob(job dataSendJob) error {
+	select {
+	case t.dataSend <- job:
+		return nil
+	case <-t.done:
+		return t.closedError()
+	}
 }
 func (t *Transport) waitQueuedSend(done <-chan error) error {
 	t.mu.Unlock()
 }
-func (t *Transport) sendData(seq uint64, reliable bool, f ports.Frame) error {
-	p := encodeData(seq, reliable, f)
-	return t.sendPayload(p, true, seq)
-}
 func (t *Transport) sendAck(seq uint64) {
 	var b [9]byte
 	b[0] = recAck
 	binary.BigEndian.PutUint64(b[1:], seq)
-	_ = t.sendPayload(b[:], false, 0)
+	_ = t.sendPayload(b[:])
 }
-func (t *Transport) sendPayload(p []byte, paced bool, pendingSeq uint64) error {
+func (t *Transport) sendPayload(p []byte) error {
-	if paced {
-		t.mu.Lock()
-		beforeDataPace := t.beforeDataPace
-		t.mu.Unlock()
-		if beforeDataPace != nil {
-			beforeDataPace()
-		}
-		t.dataPaceMu.Lock()
-		defer t.dataPaceMu.Unlock()
-	}
 	frags, err := pdgram.FragmentPayload(t.nextCounter(), p, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
 	if err != nil {
 		return err
 	}
-	for i, f := range frags {
+	for _, f := range frags {
-		if pendingSeq != 0 && !t.pendingExists(pendingSeq) {
-			return nil
-		}
-		if paced {
-			if err := t.takeDataPaceSlot(); err != nil {
-				return err
-			}
-		}
 		raw, err := pdgram.MarshalFragment(f)
 		if err != nil {
 			return err
 		if peer == nil {
 			return errors.New("dgram: no peer")
 		}
-		if pendingSeq != 0 && i == len(frags)-1 {
-			t.markPendingFinalWrite(pendingSeq)
-		}
 		if err := t.writePacket(pkt, peer); err != nil {
 			return err
 		}
 	return ok
 }
-func (t *Transport) takeDataPaceSlot() error {
-	now := t.clock.Now()
-	if t.dataPaceRemaining > 0 && now.Before(t.dataPaceNext) {
-		t.dataPaceRemaining--
-		return nil
-	}
-	if t.dataPaceNext.After(now) {
-		timer := t.clock.NewTimer(t.dataPaceNext.Sub(now))
-		select {
-		case <-timer.C():
-		case <-t.done:
-			timer.Stop()
-			return t.closedError()
-		}
-		timer.Stop()
-	}
-	t.mu.Lock()
-	delay := t.outputPaceDelayLocked()
-	t.mu.Unlock()
-	t.dataPaceRemaining = defaultPacketPaceBudget - 1
-	t.dataPaceNext = t.clock.Now().Add(delay)
-	return nil
-}
 func (t *Transport) writePacket(pkt []byte, peer net.Addr) error {
 	t.writeMu.Lock()
 	defer t.writeMu.Unlock()
 		seq := binary.BigEndian.Uint64(p[1:])
 		t.mu.Lock()
 		acked := false
+		ackedBytes := 0
 		for pendingSeq, p := range t.pending {
 			if pendingSeq > seq {
 				continue
 			if pendingSeq == seq && p != nil && !p.retransmitted {
 				t.updateRTTLocked(t.clock.Now().Sub(p.first))
 			}
+			if p != nil {
+				ackedBytes += p.wireBytes
+			}
 			delete(t.pending, pendingSeq)
 			acked = true
 		}
 		if acked {
+			t.congestion.onACK(ackedBytes)
 			now := t.clock.Now()
 			t.health.ackProgress(now)
 			if len(t.pending) == 0 {
 	}
 }
-func (t *Transport) retransmitSenderLoop() {
+func (t *Transport) dataSendLoop() {
+	pacer := bytePacer{clk: t.clock}
 	for {
 		select {
+		case job := <-t.dataSend:
+			t.runDataSendJob(&pacer, job, true)
 		case batch := <-t.retransmitWork:
 			for _, r := range batch {
-				if err := t.sendData(r.seq, true, r.f); err == nil {
+				t.runDataSendJob(&pacer, dataSendJob{seq: r.seq, reliable: true, frame: r.f}, false)
-					t.markPendingReady(r.seq, true)
-				} else {
-					t.mu.Lock()
-					if p := t.pending[r.seq]; p != nil {
-						p.initialInFlight = false
-					}
-					t.mu.Unlock()
-				}
 			}
 		case <-t.done:
 			return
 	}
 }
+func (t *Transport) runDataSendJob(pacer *bytePacer, job dataSendJob, initial bool) {
+	if initial {
+		t.markPendingSent(job.seq, job.reliable)
+	}
+	err := t.writeDataJob(pacer, job)
+	if err != nil {
+		if initial {
+			t.removePending(job.seq, job.reliable)
+		} else {
+			t.markPendingReady(job.seq, job.reliable)
+		}
+		if !errors.Is(err, errPacerClosed) {
+			t.closeWithError(err)
+		}
+	} else {
+		t.markPendingReady(job.seq, job.reliable)
+	}
+	if job.done != nil {
+		if errors.Is(err, errPacerClosed) {
+			err = t.closedError()
+		}
+		job.done <- err
+		close(job.done)
+	}
+}
+func (t *Transport) writeDataJob(pacer *bytePacer, job dataSendJob) error {
+	payload := encodeData(job.seq, job.reliable, job.frame)
+	frags, err := pdgram.FragmentPayload(t.nextCounter(), payload, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
+	if err != nil {
+		return err
+	}
+	t.mu.Lock()
+	beforeDataPace := t.beforeDataPace
+	t.mu.Unlock()
+	if beforeDataPace != nil {
+		beforeDataPace()
+	}
+	limits := func() (int, int) {
+		t.mu.Lock()
+		defer t.mu.Unlock()
+		return t.congestion.bytesPerSecond(), t.congestion.burstBytes()
+	}
+	for i, frag := range frags {
+		if job.reliable && !t.pendingExists(job.seq) {
+			return nil
+		}
+		raw, err := pdgram.MarshalFragment(frag)
+		if err != nil {
+			return err
+		}
+		pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
+		if err := pacer.wait(t.done, len(pkt), limits); err != nil {
+			return err
+		}
+		if job.reliable && !t.pendingExists(job.seq) {
+			return nil
+		}
+		t.mu.Lock()
+		peer := t.peer
+		closed := t.closed
+		closeErr := t.closeErr
+		t.mu.Unlock()
+		if closed {
+			if closeErr != nil {
+				return closeErr
+			}
+			return net.ErrClosed
+		}
+		if peer == nil {
+			return errors.New("dgram: no peer")
+		}
+		if i == len(frags)-1 {
+			t.markPendingFinalWrite(job.seq)
+		}
+		if err := t.writePacket(pkt, peer); err != nil {
+			return err
+		}
+	}
+	return nil
+}
 func (t *Transport) heartbeatLoop() {
 	heartbeat := t.clock.NewTimer(t.heartbeat)
 	defer heartbeat.Stop()
 		p.initialInFlight = true
 		p.attempts++
 		p.retransmitted = true
+		t.congestion.onLoss()
 		t.retransmits++
 		emitDiagnostic = emitDiagnostic || t.retransmits%64 == 0
 		resend = append(resend, retransmitRecord{seq: seq, f: p.frame})
 	return resend, emitDiagnostic
 }
-// resendPending remains a synchronous test hook for retransmission selection
+// resendPending remains a test hook for retransmission selection. Constructed
-// unit tests. Production timer paths use queueRetransmits and the sole sender.
+// transports transfer its batch to the sole data sender; small unit fixtures
+// without transport goroutines execute the selected writes inline.
 func (t *Transport) resendPending() {
 	now := t.clock.Now()
 	t.mu.Lock()
 	resend, _ := t.selectRetransmitsLocked(now)
 	t.mu.Unlock()
+	if t.retransmitWork != nil {
+		select {
+		case t.retransmitWork <- resend:
+		case <-t.done:
+		}
+		return
+	}
+	pacer := bytePacer{clk: t.clock}
 	for _, r := range resend {
-		if err := t.sendData(r.seq, true, r.f); err == nil {
+		if err := t.writeDataJob(&pacer, dataSendJob{seq: r.seq, reliable: true, frame: r.f}); err == nil {
 			t.markPendingReady(r.seq, true)
 		} else {
 			t.mu.Lock()
 		rto = t.maxResendAfter
 	}
 	t.rto = rto
+	t.congestion.onRTT(t.srtt)
 }
 func (t *Transport) hopPacketConnOnce(generation uint64) {

M internal/adapters/dgram/transport_test.go +168 -112


@@ -213,6 +213,148 @@ func waitForManualTimers(t testing.T, clk manualClock, n int) {
213	213		t.Fatalf("manual timers=%d, want at least %d", active, n)
214	214		}
215	215
	216	+	func TestCongestionControllerBoundsAndIntegerGrowth(t *testing.T) {
	217	+	const mtu = 1200
	218	+	c := newCongestionController(mtu)
	219	+	if got, want := c.cwndBytes, initialCongestionPackets*mtu; got != want {
	220	+	t.Fatalf("initial cwnd=%d, want %d", got, want)
	221	+	}
	222	+	if got, want := c.burstBytes(), dataBurstPackets*mtu; got != want {
	223	+	t.Fatalf("initial burst=%d, want %d", got, want)
	224	+	}
	225	+	if got, want := c.bytesPerSecond(), initialCongestionPacketsmtuint(time.Second/initialPacingRTT); got != want {
	226	+	t.Fatalf("initial pacing rate=%d, want %d", got, want)
	227	+	}
	228	+
	229	+	c.onLoss()
	230	+	if got, want := c.cwndBytes, initialCongestionPackets*mtu/2; got != want {
	231	+	t.Fatalf("cwnd after first loss=%d, want %d", got, want)
	232	+	}
	233	+	c.cwndBytes = 3 * mtu
	234	+	c.onLoss()
	235	+	if got, want := c.cwndBytes, minimumCongestionPackets*mtu; got != want {
	236	+	t.Fatalf("minimum cwnd=%d, want %d", got, want)
	237	+	}
	238	+
	239	+	c.cwndBytes = 10 * mtu
	240	+	c.onACK(1)
	241	+	if got, want := c.cwndBytes, 10*mtu+1; got != want {
	242	+	t.Fatalf("minimum additive increase cwnd=%d, want %d", got, want)
	243	+	}
	244	+	c.onACK(10 * mtu)
	245	+	if got, want := c.cwndBytes, 10mtu+1+(mtu10mtu)/(10mtu+1); got != want {
	246	+	t.Fatalf("integer additive increase cwnd=%d, want %d", got, want)
	247	+	}
	248	+	c.cwndBytes = maximumCongestionBytes
	249	+	c.onACK(maximumCongestionBytes)
	250	+	if got := c.cwndBytes; got != maximumCongestionBytes {
	251	+	t.Fatalf("maximum cwnd=%d, want %d", got, maximumCongestionBytes)
	252	+	}
	253	+
	254	+	c.onRTT(500 * time.Microsecond)
	255	+	if got, want := c.bytesPerSecond(), maximumCongestionBytes*1000; got != want {
	256	+	t.Fatalf("sub-millisecond pacing rate=%d, want %d", got, want)
	257	+	}
	258	+	}
	259	+
	260	+	func TestBytePacerUsesBurstThenFakeClockRate(t *testing.T) {
	261	+	const mtu = 1200
	262	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	263	+	c := newCongestionController(mtu)
	264	+	p := bytePacer{clk: clk}
	265	+	done := make(chan struct{})
	266	+	limits := func() (int, int) { return c.bytesPerSecond(), c.burstBytes() }
	267	+
	268	+	for range dataBurstPackets {
	269	+	if err := p.wait(done, mtu, limits); err != nil {
	270	+	t.Fatal(err)
	271	+	}
	272	+	}
	273	+	wait := make(chan error, 1)
	274	+	go func() { wait <- p.wait(done, mtu, limits) }()
	275	+	awaitSignal(t, clk.timerCreated, "byte pacer timer")
	276	+	clk.advance(24 * time.Millisecond)
	277	+	select {
	278	+	case err := <-wait:
	279	+	t.Fatalf("third MTU released early: %v", err)
	280	+	default:
	281	+	}
	282	+	clk.advance(time.Millisecond)
	283	+	if err := awaitResult(t, wait, "third paced MTU"); err != nil {
	284	+	t.Fatal(err)
	285	+	}
	286	+
	287	+	c.onLoss()
	288	+	next := make(chan error, 1)
	289	+	go func() { next <- p.wait(done, mtu, limits) }()
	290	+	awaitSignal(t, clk.timerCreated, "reduced-cwnd byte pacer timer")
	291	+	clk.advance(49 * time.Millisecond)
	292	+	select {
	293	+	case err := <-next:
	294	+	t.Fatalf("reduced cwnd did not lengthen wait: %v", err)
	295	+	default:
	296	+	}
	297	+	clk.advance(time.Millisecond)
	298	+	if err := awaitResult(t, next, "reduced-cwnd paced MTU"); err != nil {
	299	+	t.Fatal(err)
	300	+	}
	301	+	}
	302	+
	303	+	func TestBytePacerCancellation(t *testing.T) {
	304	+	const mtu = 1200
	305	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	306	+	p := bytePacer{clk: clk}
	307	+	done := make(chan struct{})
	308	+	limits := func() (int, int) { return mtu, mtu }
	309	+	if err := p.wait(done, mtu, limits); err != nil {
	310	+	t.Fatal(err)
	311	+	}
	312	+	wait := make(chan error, 1)
	313	+	go func() { wait <- p.wait(done, mtu, limits) }()
	314	+	awaitSignal(t, clk.timerCreated, "cancelled byte pacer timer")
	315	+	close(done)
	316	+	if err := awaitResult(t, wait, "byte pacer cancellation"); !errors.Is(err, errPacerClosed) {
	317	+	t.Fatalf("cancellation error=%v, want %v", err, errPacerClosed)
	318	+	}
	319	+	}
	320	+
	321	+	func BenchmarkCongestionControllerACK(b *testing.B) {
	322	+	c := newCongestionController(defaultMTU)
	323	+	b.ReportAllocs()
	324	+	for b.Loop() {
	325	+	c.onACK(defaultMTU)
	326	+	if c.cwndBytes == c.maxBytes {
	327	+	c = newCongestionController(defaultMTU)
	328	+	}
	329	+	}
	330	+	}
	331	+
	332	+	func TestSendAsyncTransfersFragmentedRecordToBytePacedSender(t *testing.T) {
	333	+	const mtu = 128
	334	+	aPC, bPC := newPair()
	335	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	336	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
	337	+	Clock: clk, MTU: mtu, ResendAfter: time.Hour, Heartbeat: time.Hour,
	338	+	})
	339	+	if err != nil {
	340	+	t.Fatal(err)
	341	+	}
	342	+	defer func() { _ = a.Close() }()
	343	+
	344	+	maxFragmentData := mtu - pdgram.HeaderSize - a.codec.Overhead()
	345	+	frame := ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 2*maxFragmentData)}
	346	+	if err := a.SendAsync(frame); err != nil {
	347	+	t.Fatal(err)
	348	+	}
	349	+	eventuallyPacketCount(t, bPC, dataBurstPackets)
	350	+	if got := len(bPC.in); got != dataBurstPackets {
	351	+	t.Fatalf("immediate datagrams=%d, want %d MTU burst", got, dataBurstPackets)
	352	+	}
	353	+	awaitSignal(t, clk.timerCreated, "shared data sender pacing timer")
	354	+	clk.advance(initialPacingRTT)
	355	+	eventuallyPacketCount(t, bPC, 3)
	356	+	}
	357	+
216	358		type deadlineCapturePC struct {
217	359		addr net.Addr
218	360		deadline atomic.Value

@@ -242,7 +384,7 @@ func (p *deadlineCapturePC) SetWriteDeadline(t time.Time) error {
242	384		return nil
243	385		}
244	386
245		-	func TestOutputFirstSendIsPacedAndQueuedOutputsAreBatched(t *testing.T) {
	387	+	func TestSmallOutputsShareInitialByteBurst(t *testing.T) {
246	388		aPC, bPC := newPair()
247	389		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
248	390		a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})

@@ -256,19 +398,10 @@ func TestOutputFirstSendIsPacedAndQueuedOutputsAreBatched(t *testing.T) {
256	398		t.Fatal(err)
257	399		}
258	400		}
259		-	if got := len(bPC.in); got != 1 {
260		-	t.Fatalf("immediate datagrams=%d, want only the first output sent before pacing timer", got)
261		-	}
262		-	waitForManualTimers(t, clk, 4)
263		-	clk.advance(defaultOutputPaceMinDelay)
264	401		eventuallyPacketCount(t, bPC, 3)
265	402		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}); err != nil {
266	403		t.Fatal(err)
267	404		}
268		-	if got := len(bPC.in); got != 3 {
269		-	t.Fatalf("datagrams before second pace tick=%d, want 3", got)
270		-	}
271		-	clk.advance(defaultOutputPaceMinDelay)
272	405		eventuallyPacketCount(t, bPC, 4)
273	406		}
274	407

@@ -293,11 +426,11 @@ func TestOutputPacingSendsAtMostOneOversizedFramePerTick(t *testing.T) {
293	426		waitForManualTimers(t, clk, 4)
294	427		previousPackets := len(aPC.in)
295	428		for tick := 0; tick < 128; tick++ {
296		-	clk.advance(defaultOutputPaceMaxDelay)
	429	+	clk.advance(initialPacingRTT)
297	430		time.Sleep(time.Millisecond)
298	431		packets := len(aPC.in)
299		-	if delta := packets - previousPackets; delta > defaultPacketPaceBudget {
300		-	t.Fatalf("tick %d emitted %d packets, budget %d", tick, delta, defaultPacketPaceBudget)
	432	+	if delta := packets - previousPackets; delta > initialCongestionPackets {
	433	+	t.Fatalf("tick %d emitted %d packets, cwnd packets %d", tick, delta, initialCongestionPackets)
301	434		}
302	435		previousPackets = packets
303	436		a.mu.Lock()

@@ -323,7 +456,7 @@ func TestPacedInitialSendTimestampsFinalFragmentCompletion(t *testing.T) {
323	456		defer func() { _ = a.Close() }()
324	457		large := ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 24*a.mtu)}
325	458		done := make(chan error, 1)
326		-	go func() { done <- a.SendAsync(large) }()
	459	+	go func() { done <- a.Send(large) }()
327	460		start := clk.Now()
328	461		for tick := 0; tick < 128; tick++ {
329	462		a.mu.Lock()

@@ -333,7 +466,7 @@ func TestPacedInitialSendTimestampsFinalFragmentCompletion(t *testing.T) {
333	466		t.Fatal("pending last timestamp set before final fragment")
334	467		}
335	468		a.mu.Unlock()
336		-	clk.advance(defaultOutputPaceMaxDelay)
	469	+	clk.advance(initialPacingRTT)
337	470		time.Sleep(time.Millisecond)
338	471		select {
339	472		case err := <-done:

@@ -343,7 +476,7 @@ func TestPacedInitialSendTimestampsFinalFragmentCompletion(t *testing.T) {
343	476		a.mu.Lock()
344	477		completed := a.pending[1].last
345	478		a.mu.Unlock()
346		-	if elapsed := completed.Sub(start); elapsed <= defaultOutputPaceMaxDelay {
	479	+	if elapsed := completed.Sub(start); elapsed <= initialPacingRTT {
347	480		t.Fatalf("large send completed too early after %v", elapsed)
348	481		}
349	482		before := len(aPC.in)

@@ -396,7 +529,7 @@ func TestOwnedSynchronousMaxSideEffectCompletesThroughConcurrentPacedWork(t *tes
396	529		go func() { done <- a.SendSynchronous(sideEffect) }()
397	530
398	531		for tick := 0; tick < 512; tick++ {
399		-	clk.advance(defaultOutputPaceMaxDelay)
	532	+	clk.advance(initialPacingRTT)
400	533		time.Sleep(time.Millisecond)
401	534		select {
402	535		case err := <-done:

@@ -448,31 +581,6 @@ func TestResendSkipsOutputDuringFirstWireAttempt(t *testing.T) {
448	581		}
449	582		}
450	583
451		-	func TestOutputPaceDelayUsesSRTTHalfClamped(t *testing.T) {
452		-	tr, _ := newResendTestTransport(t, 10)
453		-	tests := []struct {
454		-	name string
455		-	srtt time.Duration
456		-	want time.Duration
457		-	}{
458		-	{name: "high rtt clamps to max", srtt: time.Second, want: defaultOutputPaceMaxDelay},
459		-	{name: "low rtt clamps to min cadence", srtt: 30 * time.Millisecond, want: defaultOutputPaceMinCadence},
460		-	{name: "mid rtt uses half", srtt: 80 * time.Millisecond, want: 40 * time.Millisecond},
461		-	{name: "initial rtt uses min delay", want: defaultOutputPaceMinDelay},
462		-	}
463		-	for _, tt := range tests {
464		-	t.Run(tt.name, func(t *testing.T) {
465		-	tr.mu.Lock()
466		-	tr.srtt = tt.srtt
467		-	got := tr.outputPaceDelayLocked()
468		-	tr.mu.Unlock()
469		-	if got != tt.want {
470		-	t.Fatalf("output pace delay=%v, want %v", got, tt.want)
471		-	}
472		-	})
473		-	}
474		-	}
475		-
476	584		func TestInputWaitsForDequeuedOutputBatch(t *testing.T) {
477	585		aPC, bPC := newPair()
478	586		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))

@@ -482,29 +590,12 @@ func TestInputWaitsForDequeuedOutputBatch(t *testing.T) {
482	590		}
483	591		defer func() { _ = a.Close() }()
484	592
	593	+	a.writeMu.Lock()
485	594		for _, payload := range []string{"first", "queued"} {
486	595		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte(payload)}); err != nil {
487	596		t.Fatal(err)
488	597		}
489	598		}
490		-	waitForManualTimers(t, clk, 4)
491		-	a.writeMu.Lock()
492		-	clk.advance(defaultOutputPaceMinDelay)
493		-	deadline := time.Now().Add(time.Second)
494		-	for {
495		-	a.mu.Lock()
496		-	dequeued := len(a.outputQueue) == 0
497		-	a.mu.Unlock()
498		-	if dequeued {
499		-	break
500		-	}
501		-	if time.Now().After(deadline) {
502		-	a.writeMu.Unlock()
503		-	t.Fatal("pacer did not dequeue output batch")
504		-	}
505		-	time.Sleep(time.Millisecond)
506		-	}
507		-
508	599		done := make(chan error, 1)
509	600		go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}) }()
510	601		select {

@@ -536,29 +627,13 @@ func TestPacedControlQueueFailureClosesTransport(t *testing.T) {
536	627		}
537	628		defer func() { _ = a.Close() }()
538	629
539		-	for i := range 3 {
540		-	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
541		-	t.Fatal(err)
542		-	}
543		-	}
544		-	if got := len(bPC.in); got != 1 {
545		-	t.Fatalf("immediate output datagrams=%d, want 1", got)
546		-	}
547	630		if err := bPC.Close(); err != nil {
548	631		t.Fatal(err)
549	632		}
550		-
551	633		inputDone := make(chan error, 1)
552	634		go func() { inputDone <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}) }()
553		-	eventually(t, time.Second, func() bool {
554		-	a.mu.Lock()
555		-	defer a.mu.Unlock()
556		-	return len(a.outputQueue) == 2
557		-	})
558		-	waitForManualTimers(t, clk, 4)
559		-	clk.advance(defaultOutputPaceMinDelay)
560	635		if err := <-inputDone; err == nil {
561		-	t.Fatal("queued control send succeeded after paced wire failure")
	636	+	t.Fatal("data send succeeded after paced wire failure")
562	637		}
563	638		select {
564	639		case <-a.done:

@@ -602,10 +677,7 @@ func TestCloseUnblocksQueuedSynchronousFrames(t *testing.T) {
602	677		if err != nil {
603	678		t.Fatal(err)
604	679		}
605		-	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("prime")}); err != nil {
606		-	t.Fatal(err)
607		-	}
608		-	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued")}); err != nil {
	680	+	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 3*a.mtu)}); err != nil {
609	681		t.Fatal(err)
610	682		}
611	683		done := make(chan error, 1)

@@ -613,7 +685,7 @@ func TestCloseUnblocksQueuedSynchronousFrames(t *testing.T) {
613	685		eventually(t, time.Second, func() bool {
614	686		a.mu.Lock()
615	687		defer a.mu.Unlock()
616		-	return len(a.outputQueue) == 2
	688	+	return len(a.pending) == 2
617	689		})
618	690		_ = a.Close()
619	691		if err := <-done; err == nil {

@@ -632,16 +704,12 @@ func TestOutputPaceFailureClosesTransportWithWriteError(t *testing.T) {
632	704		}
633	705		defer func() { _ = a.Close() }()
634	706
635		-	for i := range 2 {
636		-	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
637		-	t.Fatal(err)
638		-	}
639		-	}
640		-	waitForManualTimers(t, clk, 4)
641	707		if err := bPC.Close(); err != nil {
642	708		t.Fatal(err)
643	709		}
644		-	clk.advance(defaultOutputPaceMinDelay)
	710	+	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("fails")}); err != nil {
	711	+	t.Fatal(err)
	712	+	}
645	713
646	714		select {
647	715		case <-a.done:

@@ -668,16 +736,9 @@ func TestInputAndControlRemainOrderedInBoundedPacedQueue(t *testing.T) {
668	736		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued")}); err != nil {
669	737		t.Fatal(err)
670	738		}
671		-	if got := len(bPC.in); got != 1 {
672		-	t.Fatalf("immediate output datagrams=%d, want 1", got)
673		-	}
	739	+	eventuallyPacketCount(t, bPC, 2)
674	740		inputDone := make(chan error, 1)
675	741		go func() { inputDone <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}) }()
676		-	if got := len(bPC.in); got != 1 {
677		-	t.Fatalf("datagrams after input=%d, want prerequisites and input still paced", got)
678		-	}
679		-	waitForManualTimers(t, clk, 4)
680		-	clk.advance(defaultOutputPaceMinDelay)
681	742		eventuallyPacketCount(t, bPC, 3)
682	743		if err := <-inputDone; err != nil {
683	744		t.Fatal(err)

@@ -685,20 +746,8 @@ func TestInputAndControlRemainOrderedInBoundedPacedQueue(t *testing.T) {
685	746		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued-before-control")}); err != nil {
686	747		t.Fatal(err)
687	748		}
688		-	if got := len(bPC.in); got != 3 {
689		-	t.Fatalf("datagrams before control=%d, want queued output still paced", got)
690		-	}
691	749		controlDone := make(chan error, 1)
692	750		go func() { controlDone <- a.Send(ports.Frame{Type: ports.MsgPing}) }()
693		-	eventually(t, time.Second, func() bool {
694		-	a.mu.Lock()
695		-	defer a.mu.Unlock()
696		-	return len(a.outputQueue) == 2
697		-	})
698		-	if got := len(bPC.in); got != 3 {
699		-	t.Fatalf("datagrams after control=%d, want ordered paced queue", got)
700		-	}
701		-	clk.advance(defaultOutputPaceMinDelay)
702	751		eventuallyPacketCount(t, bPC, 5)
703	752		if err := <-controlDone; err != nil {
704	753		t.Fatal(err)

@@ -1179,6 +1228,11 @@ func TestImmediateLoopbackAckSamplesStampedFinalWrite(t *testing.T) {
1179	1228		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("state")}); err != nil {
1180	1229		t.Fatal(err)
1181	1230		}
	1231	+	eventually(t, time.Second, func() bool {
	1232	+	a.mu.Lock()
	1233	+	defer a.mu.Unlock()
	1234	+	return a.srtt > 0
	1235	+	})
1182	1236		a.mu.Lock()
1183	1237		srtt, rto := a.srtt, a.rto
1184	1238		a.mu.Unlock()

@@ -1293,6 +1347,8 @@ func newResendTestTransport(t testing.T, maxResendPerTick int) (Transport, *at
1293	1347		maxResendPerTick: maxResendPerTick,
1294	1348		rto: 100 * time.Millisecond,
1295	1349		clock: realClock{},
	1350	+	congestion: newCongestionController(pdgram.DefaultMTU),
	1351	+	done: make(chan struct{}),
1296	1352		}, &writes
1297	1353		}
1298	1354

@@ -1326,11 +1382,11 @@ func TestOversizedRetransmitHonorsPacketBudget(t *testing.T) {
1326	1382
1327	1383		previous := int32(0)
1328	1384		for tick := 0; tick < 128; tick++ {
1329		-	clk.advance(defaultOutputPaceMaxDelay)
	1385	+	clk.advance(initialPacingRTT)
1330	1386		time.Sleep(time.Millisecond)
1331	1387		current := writes.Load()
1332		-	if delta := current - previous; delta > defaultPacketPaceBudget {
1333		-	t.Fatalf("tick %d retransmitted %d packets, budget %d", tick, delta, defaultPacketPaceBudget)
	1388	+	if delta := current - previous; delta > dataBurstPackets {
	1389	+	t.Fatalf("tick %d retransmitted %d packets, burst %d", tick, delta, dataBurstPackets)
1334	1390		}
1335	1391		previous = current
1336	1392		select {

 	t.Fatalf("manual timers=%d, want at least %d", active, n)
 }
+func TestCongestionControllerBoundsAndIntegerGrowth(t *testing.T) {
+	const mtu = 1200
+	c := newCongestionController(mtu)
+	if got, want := c.cwndBytes, initialCongestionPackets*mtu; got != want {
+		t.Fatalf("initial cwnd=%d, want %d", got, want)
+	}
+	if got, want := c.burstBytes(), dataBurstPackets*mtu; got != want {
+		t.Fatalf("initial burst=%d, want %d", got, want)
+	}
+	if got, want := c.bytesPerSecond(), initialCongestionPackets*mtu*int(time.Second/initialPacingRTT); got != want {
+		t.Fatalf("initial pacing rate=%d, want %d", got, want)
+	}
+	c.onLoss()
+	if got, want := c.cwndBytes, initialCongestionPackets*mtu/2; got != want {
+		t.Fatalf("cwnd after first loss=%d, want %d", got, want)
+	}
+	c.cwndBytes = 3 * mtu
+	c.onLoss()
+	if got, want := c.cwndBytes, minimumCongestionPackets*mtu; got != want {
+		t.Fatalf("minimum cwnd=%d, want %d", got, want)
+	}
+	c.cwndBytes = 10 * mtu
+	c.onACK(1)
+	if got, want := c.cwndBytes, 10*mtu+1; got != want {
+		t.Fatalf("minimum additive increase cwnd=%d, want %d", got, want)
+	}
+	c.onACK(10 * mtu)
+	if got, want := c.cwndBytes, 10*mtu+1+(mtu*10*mtu)/(10*mtu+1); got != want {
+		t.Fatalf("integer additive increase cwnd=%d, want %d", got, want)
+	}
+	c.cwndBytes = maximumCongestionBytes
+	c.onACK(maximumCongestionBytes)
+	if got := c.cwndBytes; got != maximumCongestionBytes {
+		t.Fatalf("maximum cwnd=%d, want %d", got, maximumCongestionBytes)
+	}
+	c.onRTT(500 * time.Microsecond)
+	if got, want := c.bytesPerSecond(), maximumCongestionBytes*1000; got != want {
+		t.Fatalf("sub-millisecond pacing rate=%d, want %d", got, want)
+	}
+}
+func TestBytePacerUsesBurstThenFakeClockRate(t *testing.T) {
+	const mtu = 1200
+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
+	c := newCongestionController(mtu)
+	p := bytePacer{clk: clk}
+	done := make(chan struct{})
+	limits := func() (int, int) { return c.bytesPerSecond(), c.burstBytes() }
+	for range dataBurstPackets {
+		if err := p.wait(done, mtu, limits); err != nil {
+			t.Fatal(err)
+		}
+	}
+	wait := make(chan error, 1)
+	go func() { wait <- p.wait(done, mtu, limits) }()
+	awaitSignal(t, clk.timerCreated, "byte pacer timer")
+	clk.advance(24 * time.Millisecond)
+	select {
+	case err := <-wait:
+		t.Fatalf("third MTU released early: %v", err)
+	default:
+	}
+	clk.advance(time.Millisecond)
+	if err := awaitResult(t, wait, "third paced MTU"); err != nil {
+		t.Fatal(err)
+	}
+	c.onLoss()
+	next := make(chan error, 1)
+	go func() { next <- p.wait(done, mtu, limits) }()
+	awaitSignal(t, clk.timerCreated, "reduced-cwnd byte pacer timer")
+	clk.advance(49 * time.Millisecond)
+	select {
+	case err := <-next:
+		t.Fatalf("reduced cwnd did not lengthen wait: %v", err)
+	default:
+	}
+	clk.advance(time.Millisecond)
+	if err := awaitResult(t, next, "reduced-cwnd paced MTU"); err != nil {
+		t.Fatal(err)
+	}
+}
+func TestBytePacerCancellation(t *testing.T) {
+	const mtu = 1200
+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
+	p := bytePacer{clk: clk}
+	done := make(chan struct{})
+	limits := func() (int, int) { return mtu, mtu }
+	if err := p.wait(done, mtu, limits); err != nil {
+		t.Fatal(err)
+	}
+	wait := make(chan error, 1)
+	go func() { wait <- p.wait(done, mtu, limits) }()
+	awaitSignal(t, clk.timerCreated, "cancelled byte pacer timer")
+	close(done)
+	if err := awaitResult(t, wait, "byte pacer cancellation"); !errors.Is(err, errPacerClosed) {
+		t.Fatalf("cancellation error=%v, want %v", err, errPacerClosed)
+	}
+}
+func BenchmarkCongestionControllerACK(b *testing.B) {
+	c := newCongestionController(defaultMTU)
+	b.ReportAllocs()
+	for b.Loop() {
+		c.onACK(defaultMTU)
+		if c.cwndBytes == c.maxBytes {
+			c = newCongestionController(defaultMTU)
+		}
+	}
+}
+func TestSendAsyncTransfersFragmentedRecordToBytePacedSender(t *testing.T) {
+	const mtu = 128
+	aPC, bPC := newPair()
+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
+		Clock: clk, MTU: mtu, ResendAfter: time.Hour, Heartbeat: time.Hour,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = a.Close() }()
+	maxFragmentData := mtu - pdgram.HeaderSize - a.codec.Overhead()
+	frame := ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 2*maxFragmentData)}
+	if err := a.SendAsync(frame); err != nil {
+		t.Fatal(err)
+	}
+	eventuallyPacketCount(t, bPC, dataBurstPackets)
+	if got := len(bPC.in); got != dataBurstPackets {
+		t.Fatalf("immediate datagrams=%d, want %d MTU burst", got, dataBurstPackets)
+	}
+	awaitSignal(t, clk.timerCreated, "shared data sender pacing timer")
+	clk.advance(initialPacingRTT)
+	eventuallyPacketCount(t, bPC, 3)
+}
 type deadlineCapturePC struct {
 	addr     net.Addr
 	deadline atomic.Value
 	return nil
 }
-func TestOutputFirstSendIsPacedAndQueuedOutputsAreBatched(t *testing.T) {
+func TestSmallOutputsShareInitialByteBurst(t *testing.T) {
 	aPC, bPC := newPair()
 	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
 	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
 			t.Fatal(err)
 		}
 	}
-	if got := len(bPC.in); got != 1 {
-		t.Fatalf("immediate datagrams=%d, want only the first output sent before pacing timer", got)
-	}
-	waitForManualTimers(t, clk, 4)
-	clk.advance(defaultOutputPaceMinDelay)
 	eventuallyPacketCount(t, bPC, 3)
 	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}); err != nil {
 		t.Fatal(err)
 	}
-	if got := len(bPC.in); got != 3 {
-		t.Fatalf("datagrams before second pace tick=%d, want 3", got)
-	}
-	clk.advance(defaultOutputPaceMinDelay)
 	eventuallyPacketCount(t, bPC, 4)
 }
 	waitForManualTimers(t, clk, 4)
 	previousPackets := len(aPC.in)
 	for tick := 0; tick < 128; tick++ {
-		clk.advance(defaultOutputPaceMaxDelay)
+		clk.advance(initialPacingRTT)
 		time.Sleep(time.Millisecond)
 		packets := len(aPC.in)
-		if delta := packets - previousPackets; delta > defaultPacketPaceBudget {
+		if delta := packets - previousPackets; delta > initialCongestionPackets {
-			t.Fatalf("tick %d emitted %d packets, budget %d", tick, delta, defaultPacketPaceBudget)
+			t.Fatalf("tick %d emitted %d packets, cwnd packets %d", tick, delta, initialCongestionPackets)
 		}
 		previousPackets = packets
 		a.mu.Lock()
 	defer func() { _ = a.Close() }()
 	large := ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 24*a.mtu)}
 	done := make(chan error, 1)
-	go func() { done <- a.SendAsync(large) }()
+	go func() { done <- a.Send(large) }()
 	start := clk.Now()
 	for tick := 0; tick < 128; tick++ {
 		a.mu.Lock()
 			t.Fatal("pending last timestamp set before final fragment")
 		}
 		a.mu.Unlock()
-		clk.advance(defaultOutputPaceMaxDelay)
+		clk.advance(initialPacingRTT)
 		time.Sleep(time.Millisecond)
 		select {
 		case err := <-done:
 			a.mu.Lock()
 			completed := a.pending[1].last
 			a.mu.Unlock()
-			if elapsed := completed.Sub(start); elapsed <= defaultOutputPaceMaxDelay {
+			if elapsed := completed.Sub(start); elapsed <= initialPacingRTT {
 				t.Fatalf("large send completed too early after %v", elapsed)
 			}
 			before := len(aPC.in)
 	go func() { done <- a.SendSynchronous(sideEffect) }()
 	for tick := 0; tick < 512; tick++ {
-		clk.advance(defaultOutputPaceMaxDelay)
+		clk.advance(initialPacingRTT)
 		time.Sleep(time.Millisecond)
 		select {
 		case err := <-done:
 	}
 }
-func TestOutputPaceDelayUsesSRTTHalfClamped(t *testing.T) {
-	tr, _ := newResendTestTransport(t, 10)
-	tests := []struct {
-		name string
-		srtt time.Duration
-		want time.Duration
-	}{
-		{name: "high rtt clamps to max", srtt: time.Second, want: defaultOutputPaceMaxDelay},
-		{name: "low rtt clamps to min cadence", srtt: 30 * time.Millisecond, want: defaultOutputPaceMinCadence},
-		{name: "mid rtt uses half", srtt: 80 * time.Millisecond, want: 40 * time.Millisecond},
-		{name: "initial rtt uses min delay", want: defaultOutputPaceMinDelay},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			tr.mu.Lock()
-			tr.srtt = tt.srtt
-			got := tr.outputPaceDelayLocked()
-			tr.mu.Unlock()
-			if got != tt.want {
-				t.Fatalf("output pace delay=%v, want %v", got, tt.want)
-			}
-		})
-	}
-}
 func TestInputWaitsForDequeuedOutputBatch(t *testing.T) {
 	aPC, bPC := newPair()
 	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
 	}
 	defer func() { _ = a.Close() }()
+	a.writeMu.Lock()
 	for _, payload := range []string{"first", "queued"} {
 		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte(payload)}); err != nil {
 			t.Fatal(err)
 		}
 	}
-	waitForManualTimers(t, clk, 4)
-	a.writeMu.Lock()
-	clk.advance(defaultOutputPaceMinDelay)
-	deadline := time.Now().Add(time.Second)
-	for {
-		a.mu.Lock()
-		dequeued := len(a.outputQueue) == 0
-		a.mu.Unlock()
-		if dequeued {
-			break
-		}
-		if time.Now().After(deadline) {
-			a.writeMu.Unlock()
-			t.Fatal("pacer did not dequeue output batch")
-		}
-		time.Sleep(time.Millisecond)
-	}
 	done := make(chan error, 1)
 	go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}) }()
 	select {
 	}
 	defer func() { _ = a.Close() }()
-	for i := range 3 {
-		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
-			t.Fatal(err)
-		}
-	}
-	if got := len(bPC.in); got != 1 {
-		t.Fatalf("immediate output datagrams=%d, want 1", got)
-	}
 	if err := bPC.Close(); err != nil {
 		t.Fatal(err)
 	}
 	inputDone := make(chan error, 1)
 	go func() { inputDone <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}) }()
-	eventually(t, time.Second, func() bool {
-		a.mu.Lock()
-		defer a.mu.Unlock()
-		return len(a.outputQueue) == 2
-	})
-	waitForManualTimers(t, clk, 4)
-	clk.advance(defaultOutputPaceMinDelay)
 	if err := <-inputDone; err == nil {
-		t.Fatal("queued control send succeeded after paced wire failure")
+		t.Fatal("data send succeeded after paced wire failure")
 	}
 	select {
 	case <-a.done:
 			if err != nil {
 				t.Fatal(err)
 			}
-			if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("prime")}); err != nil {
+			if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 3*a.mtu)}); err != nil {
-				t.Fatal(err)
-			}
-			if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued")}); err != nil {
 				t.Fatal(err)
 			}
 			done := make(chan error, 1)
 			eventually(t, time.Second, func() bool {
 				a.mu.Lock()
 				defer a.mu.Unlock()
-				return len(a.outputQueue) == 2
+				return len(a.pending) == 2
 			})
 			_ = a.Close()
 			if err := <-done; err == nil {
 	}
 	defer func() { _ = a.Close() }()
-	for i := range 2 {
-		if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
-			t.Fatal(err)
-		}
-	}
-	waitForManualTimers(t, clk, 4)
 	if err := bPC.Close(); err != nil {
 		t.Fatal(err)
 	}
-	clk.advance(defaultOutputPaceMinDelay)
+	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("fails")}); err != nil {
+		t.Fatal(err)
+	}
 	select {
 	case <-a.done:
 	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued")}); err != nil {
 		t.Fatal(err)
 	}
-	if got := len(bPC.in); got != 1 {
+	eventuallyPacketCount(t, bPC, 2)
-		t.Fatalf("immediate output datagrams=%d, want 1", got)
-	}
 	inputDone := make(chan error, 1)
 	go func() { inputDone <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}) }()
-	if got := len(bPC.in); got != 1 {
-		t.Fatalf("datagrams after input=%d, want prerequisites and input still paced", got)
-	}
-	waitForManualTimers(t, clk, 4)
-	clk.advance(defaultOutputPaceMinDelay)
 	eventuallyPacketCount(t, bPC, 3)
 	if err := <-inputDone; err != nil {
 		t.Fatal(err)
 	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued-before-control")}); err != nil {
 		t.Fatal(err)
 	}
-	if got := len(bPC.in); got != 3 {
-		t.Fatalf("datagrams before control=%d, want queued output still paced", got)
-	}
 	controlDone := make(chan error, 1)
 	go func() { controlDone <- a.Send(ports.Frame{Type: ports.MsgPing}) }()
-	eventually(t, time.Second, func() bool {
-		a.mu.Lock()
-		defer a.mu.Unlock()
-		return len(a.outputQueue) == 2
-	})
-	if got := len(bPC.in); got != 3 {
-		t.Fatalf("datagrams after control=%d, want ordered paced queue", got)
-	}
-	clk.advance(defaultOutputPaceMinDelay)
 	eventuallyPacketCount(t, bPC, 5)
 	if err := <-controlDone; err != nil {
 		t.Fatal(err)
 	if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("state")}); err != nil {
 		t.Fatal(err)
 	}
+	eventually(t, time.Second, func() bool {
+		a.mu.Lock()
+		defer a.mu.Unlock()
+		return a.srtt > 0
+	})
 	a.mu.Lock()
 	srtt, rto := a.srtt, a.rto
 	a.mu.Unlock()
 		maxResendPerTick: maxResendPerTick,
 		rto:              100 * time.Millisecond,
 		clock:            realClock{},
+		congestion:       newCongestionController(pdgram.DefaultMTU),
+		done:             make(chan struct{}),
 	}, &writes
 }
 	previous := int32(0)
 	for tick := 0; tick < 128; tick++ {
-		clk.advance(defaultOutputPaceMaxDelay)
+		clk.advance(initialPacingRTT)
 		time.Sleep(time.Millisecond)
 		current := writes.Load()
-		if delta := current - previous; delta > defaultPacketPaceBudget {
+		if delta := current - previous; delta > dataBurstPackets {
-			t.Fatalf("tick %d retransmitted %d packets, budget %d", tick, delta, defaultPacketPaceBudget)
+			t.Fatalf("tick %d retransmitted %d packets, burst %d", tick, delta, dataBurstPackets)
 		}
 		previous = current
 		select {