fix(dgram): harden paced retransmission loss handling

commit: 8d15e6ec7cbb3c7e486d88d76268c0fc131a57b7
author: brice · 3mo ago · 2026-07-10 10:59 UTC
parent: 4cb7b0d
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +174 -47

M internal/adapters/dgram/pacer.go +29 -2


@@ -2,6 +2,8 @@ package dgram
2	2
3	3		import (
4	4		"errors"
	5	+	"math"
	6	+	"math/bits"
5	7		"time"
6	8
7	9		"github.com/bnema/vev/internal/ports"

@@ -36,7 +38,8 @@ func (p *bytePacer) wait(done <-chan struct{}, n int, limits paceLimits) error {
36	38		} else {
37	39		elapsed := now.Sub(p.last)
38	40		if elapsed > 0 {
39		-	p.tokens += int64(elapsed) * int64(rate) / int64(time.Second)
	41	+	capacity := int64(burst) - p.tokens
	42	+	p.tokens += mulDivFloorSaturating(int64(elapsed), int64(rate), int64(time.Second), capacity)
40	43		p.last = now
41	44		}
42	45		p.tokens = min(p.tokens, int64(burst))

@@ -47,7 +50,7 @@ func (p *bytePacer) wait(done <-chan struct{}, n int, limits paceLimits) error {
47	50		}
48	51
49	52		deficit := int64(n) - p.tokens
50		-	wait := time.Duration((deficit*int64(time.Second) + int64(rate) - 1) / int64(rate))
	53	+	wait := time.Duration(mulDivCeilSaturating(deficit, int64(time.Second), int64(rate)))
51	54		timer := p.clk.NewTimer(wait)
52	55		select {
53	56		case <-timer.C():

@@ -58,3 +61,27 @@ func (p *bytePacer) wait(done <-chan struct{}, n int, limits paceLimits) error {
58	61		}
59	62		}
60	63		}
	64	+
	65	+	func mulDivFloorSaturating(a, b, divisor, limit int64) int64 {
	66	+	if a <= 0 \|\| b <= 0 \|\| limit <= 0 {
	67	+	return 0
	68	+	}
	69	+	hi, lo := bits.Mul64(uint64(a), uint64(b))
	70	+	if hi >= uint64(divisor) {
	71	+	return limit
	72	+	}
	73	+	quotient, _ := bits.Div64(hi, lo, uint64(divisor))
	74	+	return min(int64(min(quotient, uint64(math.MaxInt64))), limit)
	75	+	}
	76	+
	77	+	func mulDivCeilSaturating(a, b, divisor int64) int64 {
	78	+	hi, lo := bits.Mul64(uint64(a), uint64(b))
	79	+	if hi >= uint64(divisor) {
	80	+	return math.MaxInt64
	81	+	}
	82	+	quotient, remainder := bits.Div64(hi, lo, uint64(divisor))
	83	+	if remainder != 0 {
	84	+	quotient++
	85	+	}
	86	+	return int64(min(quotient, uint64(math.MaxInt64)))
	87	+	}


@@ -2,6 +2,8 @@ package dgram
2			2
3		import (	3		import (
4		"errors"	4		"errors"
			5	+	"math"
			6	+	"math/bits"
5		"time"	7		"time"
6			8
7		"github.com/bnema/vev/internal/ports"	9		"github.com/bnema/vev/internal/ports"

@@ -36,7 +38,8 @@ func (p *bytePacer) wait(done <-chan struct{}, n int, limits paceLimits) error {
36		} else {	38		} else {
37		elapsed := now.Sub(p.last)	39		elapsed := now.Sub(p.last)
38		if elapsed > 0 {	40		if elapsed > 0 {
39	-	p.tokens += int64(elapsed) * int64(rate) / int64(time.Second)	41	+	capacity := int64(burst) - p.tokens
			42	+	p.tokens += mulDivFloorSaturating(int64(elapsed), int64(rate), int64(time.Second), capacity)
40		p.last = now	43		p.last = now
41		}	44		}
42		p.tokens = min(p.tokens, int64(burst))	45		p.tokens = min(p.tokens, int64(burst))

@@ -47,7 +50,7 @@ func (p *bytePacer) wait(done <-chan struct{}, n int, limits paceLimits) error {
47		}	50		}
48			51
49		deficit := int64(n) - p.tokens	52		deficit := int64(n) - p.tokens
50	-	wait := time.Duration((deficit*int64(time.Second) + int64(rate) - 1) / int64(rate))	53	+	wait := time.Duration(mulDivCeilSaturating(deficit, int64(time.Second), int64(rate)))
51		timer := p.clk.NewTimer(wait)	54		timer := p.clk.NewTimer(wait)
52		select {	55		select {
53		case <-timer.C():	56		case <-timer.C():

@@ -58,3 +61,27 @@ func (p *bytePacer) wait(done <-chan struct{}, n int, limits paceLimits) error {
58		}	61		}
59		}	62		}
60		}	63		}
			64	+
			65	+	func mulDivFloorSaturating(a, b, divisor, limit int64) int64 {
			66	+	if a <= 0 \|\| b <= 0 \|\| limit <= 0 {
			67	+	return 0
			68	+	}
			69	+	hi, lo := bits.Mul64(uint64(a), uint64(b))
			70	+	if hi >= uint64(divisor) {
			71	+	return limit
			72	+	}
			73	+	quotient, _ := bits.Div64(hi, lo, uint64(divisor))
			74	+	return min(int64(min(quotient, uint64(math.MaxInt64))), limit)
			75	+	}
			76	+
			77	+	func mulDivCeilSaturating(a, b, divisor int64) int64 {
			78	+	hi, lo := bits.Mul64(uint64(a), uint64(b))
			79	+	if hi >= uint64(divisor) {
			80	+	return math.MaxInt64
			81	+	}
			82	+	quotient, remainder := bits.Div64(hi, lo, uint64(divisor))
			83	+	if remainder != 0 {
			84	+	quotient++
			85	+	}
			86	+	return int64(min(quotient, uint64(math.MaxInt64)))
			87	+	}

M internal/adapters/dgram/transport.go +61 -45


@@ -988,9 +988,7 @@ func (t *Transport) dataSendLoop() {
988	988		case job := <-t.dataSend:
989	989		t.runDataSendJob(&pacer, job, true)
990	990		case batch := <-t.retransmitWork:
991		-	for _, r := range batch {
992		-	t.runDataSendJob(&pacer, dataSendJob{seq: r.seq, reliable: true, frame: r.f}, false)
993		-	}
	991	+	t.runRetransmitBatch(&pacer, batch)
994	992		case <-t.done:
995	993		return
996	994		}

@@ -1001,7 +999,7 @@ func (t Transport) runDataSendJob(pacer bytePacer, job dataSendJob, initial bo
1001	999		if initial {
1002	1000		t.markPendingSent(job.seq, job.reliable)
1003	1001		}
1004		-	err := t.writeDataJob(pacer, job)
	1002	+	_, err := t.writeDataJob(pacer, job, nil)
1005	1003		if err != nil {
1006	1004		if initial {
1007	1005		t.removePending(job.seq, job.reliable)

@@ -1023,11 +1021,14 @@ func (t Transport) runDataSendJob(pacer bytePacer, job dataSendJob, initial bo
1023	1021		}
1024	1022		}
1025	1023
1026		-	func (t Transport) writeDataJob(pacer bytePacer, job dataSendJob) error {
	1024	+	func (t Transport) writeDataJob(pacer bytePacer, job dataSendJob, onFirstWrite func()) (bool, error) {
	1025	+	if job.reliable && !t.pendingExists(job.seq) {
	1026	+	return false, nil
	1027	+	}
1027	1028		payload := encodeData(job.seq, job.reliable, job.frame)
1028	1029		frags, err := pdgram.FragmentPayload(t.nextCounter(), payload, t.mtu-pdgram.HeaderSize-t.codec.Overhead())
1029	1030		if err != nil {
1030		-	return err
	1031	+	return false, err
1031	1032		}
1032	1033		t.mu.Lock()
1033	1034		beforeDataPace := t.beforeDataPace

@@ -1035,38 +1036,35 @@ func (t Transport) writeDataJob(pacer bytePacer, job dataSendJob) error {
1035	1036		if beforeDataPace != nil {
1036	1037		beforeDataPace()
1037	1038		}
1038		-	packets := make([][]byte, 0, len(frags))
1039		-	wireBytes := 0
1040		-	for _, frag := range frags {
1041		-	raw, err := pdgram.MarshalFragment(frag)
1042		-	if err != nil {
1043		-	return err
1044		-	}
1045		-	pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
1046		-	packets = append(packets, pkt)
1047		-	wireBytes += len(pkt)
1048		-	}
1049		-	if job.reliable {
1050		-	t.mu.Lock()
1051		-	if pending := t.pending[job.seq]; pending != nil {
1052		-	pending.wireBytes = wireBytes
1053		-	}
1054		-	t.mu.Unlock()
1055		-	}
1056	1039		limits := func() (int, int) {
1057	1040		t.mu.Lock()
1058	1041		defer t.mu.Unlock()
1059	1042		return t.congestion.bytesPerSecond(), t.congestion.burstBytes()
1060	1043		}
1061		-	for i, pkt := range packets {
	1044	+	emitted := false
	1045	+	wireBytes := 0
	1046	+	for i, frag := range frags {
1062	1047		if job.reliable && !t.pendingExists(job.seq) {
1063		-	return nil
	1048	+	return emitted, nil
	1049	+	}
	1050	+	raw, err := pdgram.MarshalFragment(frag)
	1051	+	if err != nil {
	1052	+	return emitted, err
	1053	+	}
	1054	+	pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
	1055	+	wireBytes += len(pkt)
	1056	+	if job.reliable {
	1057	+	t.mu.Lock()
	1058	+	if pending := t.pending[job.seq]; pending != nil && pending.wireBytes < wireBytes {
	1059	+	pending.wireBytes = wireBytes
	1060	+	}
	1061	+	t.mu.Unlock()
1064	1062		}
1065	1063		if err := pacer.wait(t.done, len(pkt), limits); err != nil {
1066		-	return err
	1064	+	return emitted, err
1067	1065		}
1068	1066		if job.reliable && !t.pendingExists(job.seq) {
1069		-	return nil
	1067	+	return emitted, nil
1070	1068		}
1071	1069		t.mu.Lock()
1072	1070		peer := t.peer

@@ -1075,21 +1073,50 @@ func (t Transport) writeDataJob(pacer bytePacer, job dataSendJob) error {
1075	1073		t.mu.Unlock()
1076	1074		if closed {
1077	1075		if closeErr != nil {
1078		-	return closeErr
	1076	+	return emitted, closeErr
1079	1077		}
1080		-	return net.ErrClosed
	1078	+	return emitted, net.ErrClosed
1081	1079		}
1082	1080		if peer == nil {
1083		-	return errors.New("dgram: no peer")
	1081	+	return emitted, errors.New("dgram: no peer")
1084	1082		}
1085	1083		if i == len(frags)-1 {
1086	1084		t.markPendingFinalWrite(job.seq)
1087	1085		}
1088	1086		if err := t.writePacket(pkt, peer); err != nil {
1089		-	return err
	1087	+	return emitted, err
	1088	+	}
	1089	+	if !emitted {
	1090	+	emitted = true
	1091	+	if onFirstWrite != nil {
	1092	+	onFirstWrite()
	1093	+	}
1090	1094		}
1091	1095		}
1092		-	return nil
	1096	+	return emitted, nil
	1097	+	}
	1098	+
	1099	+	func (t Transport) runRetransmitBatch(pacer bytePacer, batch []retransmitRecord) {
	1100	+	lossApplied := false
	1101	+	onFirstWrite := func() {
	1102	+	if lossApplied {
	1103	+	return
	1104	+	}
	1105	+	t.mu.Lock()
	1106	+	t.congestion.onLoss()
	1107	+	t.mu.Unlock()
	1108	+	lossApplied = true
	1109	+	}
	1110	+	for _, r := range batch {
	1111	+	job := dataSendJob{seq: r.seq, reliable: true, frame: r.f}
	1112	+	_, err := t.writeDataJob(pacer, job, onFirstWrite)
	1113	+	if err == nil {
	1114	+	t.markPendingReady(r.seq, true)
	1115	+	continue
	1116	+	}
	1117	+	t.markPendingReady(r.seq, true)
	1118	+	return
	1119	+	}
1093	1120		}
1094	1121
1095	1122		func (t *Transport) heartbeatLoop() {

@@ -1154,7 +1181,6 @@ func (t *Transport) selectRetransmitsLocked(now time.Time) ([]retransmitRecord,
1154	1181		p.initialInFlight = true
1155	1182		p.attempts++
1156	1183		p.retransmitted = true
1157		-	t.congestion.onLoss()
1158	1184		t.retransmits++
1159	1185		emitDiagnostic = emitDiagnostic \|\| t.retransmits%64 == 0
1160	1186		resend = append(resend, retransmitRecord{seq: seq, f: p.frame})

@@ -1179,17 +1205,7 @@ func (t *Transport) resendPending() {
1179	1205		return
1180	1206		}
1181	1207		pacer := bytePacer{clk: t.clock}
1182		-	for _, r := range resend {
1183		-	if err := t.writeDataJob(&pacer, dataSendJob{seq: r.seq, reliable: true, frame: r.f}); err == nil {
1184		-	t.markPendingReady(r.seq, true)
1185		-	} else {
1186		-	t.mu.Lock()
1187		-	if p := t.pending[r.seq]; p != nil {
1188		-	p.initialInFlight = false
1189		-	}
1190		-	t.mu.Unlock()
1191		-	}
1192		-	}
	1208	+	t.runRetransmitBatch(&pacer, resend)
1193	1209		}
1194	1210
1195	1211		func (t Transport) resendDelayLocked(p pending) time.Duration {

 		case job := <-t.dataSend:
 			t.runDataSendJob(&pacer, job, true)
 		case batch := <-t.retransmitWork:
-			for _, r := range batch {
+			t.runRetransmitBatch(&pacer, batch)
-				t.runDataSendJob(&pacer, dataSendJob{seq: r.seq, reliable: true, frame: r.f}, false)
-			}
 		case <-t.done:
 			return
 		}
 	if initial {
 		t.markPendingSent(job.seq, job.reliable)
 	}
-	err := t.writeDataJob(pacer, job)
+	_, err := t.writeDataJob(pacer, job, nil)
 	if err != nil {
 		if initial {
 			t.removePending(job.seq, job.reliable)
 	}
 }
-func (t *Transport) writeDataJob(pacer *bytePacer, job dataSendJob) error {
+func (t *Transport) writeDataJob(pacer *bytePacer, job dataSendJob, onFirstWrite func()) (bool, error) {
+	if job.reliable && !t.pendingExists(job.seq) {
+		return false, nil
+	}
 	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
+		return false, err
 	}
 	t.mu.Lock()
 	beforeDataPace := t.beforeDataPace
 	if beforeDataPace != nil {
 		beforeDataPace()
 	}
-	packets := make([][]byte, 0, len(frags))
-	wireBytes := 0
-	for _, frag := range frags {
-		raw, err := pdgram.MarshalFragment(frag)
-		if err != nil {
-			return err
-		}
-		pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
-		packets = append(packets, pkt)
-		wireBytes += len(pkt)
-	}
-	if job.reliable {
-		t.mu.Lock()
-		if pending := t.pending[job.seq]; pending != nil {
-			pending.wireBytes = wireBytes
-		}
-		t.mu.Unlock()
-	}
 	limits := func() (int, int) {
 		t.mu.Lock()
 		defer t.mu.Unlock()
 		return t.congestion.bytesPerSecond(), t.congestion.burstBytes()
 	}
-	for i, pkt := range packets {
+	emitted := false
+	wireBytes := 0
+	for i, frag := range frags {
 		if job.reliable && !t.pendingExists(job.seq) {
-			return nil
+			return emitted, nil
+		}
+		raw, err := pdgram.MarshalFragment(frag)
+		if err != nil {
+			return emitted, err
+		}
+		pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil)
+		wireBytes += len(pkt)
+		if job.reliable {
+			t.mu.Lock()
+			if pending := t.pending[job.seq]; pending != nil && pending.wireBytes < wireBytes {
+				pending.wireBytes = wireBytes
+			}
+			t.mu.Unlock()
 		}
 		if err := pacer.wait(t.done, len(pkt), limits); err != nil {
-			return err
+			return emitted, err
 		}
 		if job.reliable && !t.pendingExists(job.seq) {
-			return nil
+			return emitted, nil
 		}
 		t.mu.Lock()
 		peer := t.peer
 		t.mu.Unlock()
 		if closed {
 			if closeErr != nil {
-				return closeErr
+				return emitted, closeErr
 			}
-			return net.ErrClosed
+			return emitted, net.ErrClosed
 		}
 		if peer == nil {
-			return errors.New("dgram: no peer")
+			return emitted, 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 emitted, err
+		}
+		if !emitted {
+			emitted = true
+			if onFirstWrite != nil {
+				onFirstWrite()
+			}
 		}
 	}
-	return nil
+	return emitted, nil
+}
+func (t *Transport) runRetransmitBatch(pacer *bytePacer, batch []retransmitRecord) {
+	lossApplied := false
+	onFirstWrite := func() {
+		if lossApplied {
+			return
+		}
+		t.mu.Lock()
+		t.congestion.onLoss()
+		t.mu.Unlock()
+		lossApplied = true
+	}
+	for _, r := range batch {
+		job := dataSendJob{seq: r.seq, reliable: true, frame: r.f}
+		_, err := t.writeDataJob(pacer, job, onFirstWrite)
+		if err == nil {
+			t.markPendingReady(r.seq, true)
+			continue
+		}
+		t.markPendingReady(r.seq, true)
+		return
+	}
 }
 func (t *Transport) heartbeatLoop() {
 		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
 	}
 	pacer := bytePacer{clk: t.clock}
-	for _, r := range resend {
+	t.runRetransmitBatch(&pacer, resend)
-		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()
-			if p := t.pending[r.seq]; p != nil {
-				p.initialInFlight = false
-			}
-			t.mu.Unlock()
-		}
-	}
 }
 func (t *Transport) resendDelayLocked(p *pending) time.Duration {

M internal/adapters/dgram/transport_test.go +84 -0


@@ -300,6 +300,36 @@ func TestBytePacerUsesBurstThenFakeClockRate(t *testing.T) {
300	300		}
301	301		}
302	302
	303	+	func TestBytePacerMaximumRateRefillsAfterMultiSecondFakeClockJump(t *testing.T) {
	304	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	305	+	p := bytePacer{clk: clk}
	306	+	done := make(chan struct{})
	307	+	burst := int(^uint(0) >> 1)
	308	+	limits := func() (int, int) { return burst, burst }
	309	+
	310	+	if err := p.wait(done, burst, limits); err != nil {
	311	+	t.Fatal(err)
	312	+	}
	313	+	clk.advance(3 * time.Second)
	314	+	if err := p.wait(done, burst, limits); err != nil {
	315	+	t.Fatal(err)
	316	+	}
	317	+
	318	+	wait := make(chan error, 1)
	319	+	go func() { wait <- p.wait(done, burst, limits) }()
	320	+	awaitSignal(t, clk.timerCreated, "maximum-rate byte pacer timer")
	321	+	clk.advance(time.Second - time.Nanosecond)
	322	+	select {
	323	+	case err := <-wait:
	324	+	t.Fatalf("maximum-rate burst released early: %v", err)
	325	+	default:
	326	+	}
	327	+	clk.advance(time.Nanosecond)
	328	+	if err := awaitResult(t, wait, "maximum-rate burst"); err != nil {
	329	+	t.Fatal(err)
	330	+	}
	331	+	}
	332	+
303	333		func TestBytePacerCancellation(t *testing.T) {
304	334		const mtu = 1200
305	335		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))

@@ -1407,6 +1437,60 @@ func TestResendPendingCapsBurst(t *testing.T) {
1407	1437		}
1408	1438		}
1409	1439
	1440	+	func TestRetransmitBatchReducesCongestionWindowOnce(t *testing.T) {
	1441	+	tr, writes := newResendTestTransport(t, 10)
	1442	+	now := tr.clock.Now()
	1443	+	for seq := uint64(1); seq <= 3; seq++ {
	1444	+	tr.pending[seq] = &pending{
	1445	+	frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(seq)}},
	1446	+	last: now.Add(-time.Second),
	1447	+	}
	1448	+	}
	1449	+	initialCwnd := tr.congestion.cwndBytes
	1450	+
	1451	+	tr.resendPending()
	1452	+
	1453	+	if got := writes.Load(); got != 3 {
	1454	+	t.Fatalf("retransmit writes=%d, want 3", got)
	1455	+	}
	1456	+	if got, want := tr.congestion.cwndBytes, initialCwnd/2; got != want {
	1457	+	t.Fatalf("cwnd after one retransmit batch=%d, want %d", got, want)
	1458	+	}
	1459	+	}
	1460	+
	1461	+	func TestACKBeforeQueuedRetransmitDoesNotReduceCongestionWindow(t *testing.T) {
	1462	+	tr, writes := newResendTestTransport(t, 1)
	1463	+	now := tr.clock.Now()
	1464	+	tr.pending[1] = &pending{
	1465	+	frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("acked")},
	1466	+	first: now.Add(-time.Second),
	1467	+	last: now.Add(-time.Second),
	1468	+	wireBytes: tr.mtu,
	1469	+	}
	1470	+	initialCwnd := tr.congestion.cwndBytes
	1471	+	tr.mu.Lock()
	1472	+	batch, _ := tr.selectRetransmitsLocked(now)
	1473	+	tr.mu.Unlock()
	1474	+	if len(batch) != 1 {
	1475	+	t.Fatalf("selected retransmits=%d, want 1", len(batch))
	1476	+	}
	1477	+	tr.handleRecord(ackRecord(1))
	1478	+	cwndAfterACK := initialCwnd + max(1, tr.mtu*tr.mtu/initialCwnd)
	1479	+	counterAfterACK := tr.ctr
	1480	+
	1481	+	tr.runRetransmitBatch(&bytePacer{clk: tr.clock}, batch)
	1482	+
	1483	+	if got := tr.ctr; got != counterAfterACK {
	1484	+	t.Fatalf("packet counter after ACK=%d, want unchanged %d", got, counterAfterACK)
	1485	+	}
	1486	+	if got := writes.Load(); got != 0 {
	1487	+	t.Fatalf("writes after ACK=%d, want 0", got)
	1488	+	}
	1489	+	if got := tr.congestion.cwndBytes; got != cwndAfterACK {
	1490	+	t.Fatalf("cwnd after ACK race=%d, want ACK-only growth %d", got, cwndAfterACK)
	1491	+	}
	1492	+	}
	1493	+
1410	1494		func TestOversizedRetransmitHonorsPacketBudget(t *testing.T) {
1411	1495		tr, writes := newResendTestTransport(t, 1)
1412	1496		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))


@@ -300,6 +300,36 @@ func TestBytePacerUsesBurstThenFakeClockRate(t *testing.T) {
300	}	300		}
301	}	301		}
302		302
		303	+	func TestBytePacerMaximumRateRefillsAfterMultiSecondFakeClockJump(t *testing.T) {
		304	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
		305	+	p := bytePacer{clk: clk}
		306	+	done := make(chan struct{})
		307	+	burst := int(^uint(0) >> 1)
		308	+	limits := func() (int, int) { return burst, burst }
		309	+
		310	+	if err := p.wait(done, burst, limits); err != nil {
		311	+	t.Fatal(err)
		312	+	}
		313	+	clk.advance(3 * time.Second)
		314	+	if err := p.wait(done, burst, limits); err != nil {
		315	+	t.Fatal(err)
		316	+	}
		317	+
		318	+	wait := make(chan error, 1)
		319	+	go func() { wait <- p.wait(done, burst, limits) }()
		320	+	awaitSignal(t, clk.timerCreated, "maximum-rate byte pacer timer")
		321	+	clk.advance(time.Second - time.Nanosecond)
		322	+	select {
		323	+	case err := <-wait:
		324	+	t.Fatalf("maximum-rate burst released early: %v", err)
		325	+	default:
		326	+	}
		327	+	clk.advance(time.Nanosecond)
		328	+	if err := awaitResult(t, wait, "maximum-rate burst"); err != nil {
		329	+	t.Fatal(err)
		330	+	}
		331	+	}
		332	+
303	func TestBytePacerCancellation(t *testing.T) {	333		func TestBytePacerCancellation(t *testing.T) {
304	const mtu = 1200	334		const mtu = 1200
305	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))	335		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))

@@ -1407,6 +1437,60 @@ func TestResendPendingCapsBurst(t *testing.T) {
1407	}	1437		}
1408	}	1438		}
1409		1439
		1440	+	func TestRetransmitBatchReducesCongestionWindowOnce(t *testing.T) {
		1441	+	tr, writes := newResendTestTransport(t, 10)
		1442	+	now := tr.clock.Now()
		1443	+	for seq := uint64(1); seq <= 3; seq++ {
		1444	+	tr.pending[seq] = &pending{
		1445	+	frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(seq)}},
		1446	+	last: now.Add(-time.Second),
		1447	+	}
		1448	+	}
		1449	+	initialCwnd := tr.congestion.cwndBytes
		1450	+
		1451	+	tr.resendPending()
		1452	+
		1453	+	if got := writes.Load(); got != 3 {
		1454	+	t.Fatalf("retransmit writes=%d, want 3", got)
		1455	+	}
		1456	+	if got, want := tr.congestion.cwndBytes, initialCwnd/2; got != want {
		1457	+	t.Fatalf("cwnd after one retransmit batch=%d, want %d", got, want)
		1458	+	}
		1459	+	}
		1460	+
		1461	+	func TestACKBeforeQueuedRetransmitDoesNotReduceCongestionWindow(t *testing.T) {
		1462	+	tr, writes := newResendTestTransport(t, 1)
		1463	+	now := tr.clock.Now()
		1464	+	tr.pending[1] = &pending{
		1465	+	frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("acked")},
		1466	+	first: now.Add(-time.Second),
		1467	+	last: now.Add(-time.Second),
		1468	+	wireBytes: tr.mtu,
		1469	+	}
		1470	+	initialCwnd := tr.congestion.cwndBytes
		1471	+	tr.mu.Lock()
		1472	+	batch, _ := tr.selectRetransmitsLocked(now)
		1473	+	tr.mu.Unlock()
		1474	+	if len(batch) != 1 {
		1475	+	t.Fatalf("selected retransmits=%d, want 1", len(batch))
		1476	+	}
		1477	+	tr.handleRecord(ackRecord(1))
		1478	+	cwndAfterACK := initialCwnd + max(1, tr.mtu*tr.mtu/initialCwnd)
		1479	+	counterAfterACK := tr.ctr
		1480	+
		1481	+	tr.runRetransmitBatch(&bytePacer{clk: tr.clock}, batch)
		1482	+
		1483	+	if got := tr.ctr; got != counterAfterACK {
		1484	+	t.Fatalf("packet counter after ACK=%d, want unchanged %d", got, counterAfterACK)
		1485	+	}
		1486	+	if got := writes.Load(); got != 0 {
		1487	+	t.Fatalf("writes after ACK=%d, want 0", got)
		1488	+	}
		1489	+	if got := tr.congestion.cwndBytes; got != cwndAfterACK {
		1490	+	t.Fatalf("cwnd after ACK race=%d, want ACK-only growth %d", got, cwndAfterACK)
		1491	+	}
		1492	+	}
		1493	+
1410	func TestOversizedRetransmitHonorsPacketBudget(t *testing.T) {	1494		func TestOversizedRetransmitHonorsPacketBudget(t *testing.T) {
1411	tr, writes := newResendTestTransport(t, 1)	1495		tr, writes := newResendTestTransport(t, 1)
1412	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))	1496		clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))