improve(dgram): add cumulative ACK RTT resend

commit: cfccd91258fae113d35cfd3fae42cf853709c78a
author: brice · 3mo ago · 2026-07-09 16:23 UTC
parent: a95d4db
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +163 -16

M internal/adapters/dgram/transport.go +73 -14


@@ -144,6 +144,9 @@ type Transport struct {
144	144		resendAfter time.Duration
145	145		maxResendAfter time.Duration
146	146		maxResendPerTick int
	147	+	srtt time.Duration
	148	+	rttvar time.Duration
	149	+	rto time.Duration
147	150		writeTimeout time.Duration
148	151		degradedAfter time.Duration
149	152		probeAfter time.Duration

@@ -176,10 +179,11 @@ type Transport struct {
176	179		}
177	180
178	181		type pending struct {
179		-	frame ports.Frame
180		-	first time.Time
181		-	last time.Time
182		-	attempts int
	182	+	frame ports.Frame
	183	+	first time.Time
	184	+	last time.Time
	185	+	attempts int
	186	+	retransmitted bool
183	187		}
184	188
185	189		func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {

@@ -209,6 +213,7 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
209	213		resendAfter: opts.ResendAfter,
210	214		maxResendAfter: opts.MaxResendAfter,
211	215		maxResendPerTick: opts.MaxResendPerTick,
	216	+	rto: opts.ResendAfter,
212	217		writeTimeout: opts.WriteTimeout,
213	218		degradedAfter: opts.DegradedAfter,
214	219		probeAfter: opts.ProbeAfter,

@@ -499,8 +504,18 @@ func (t *Transport) handleRecord(p []byte) {
499	504		}
500	505		seq := binary.BigEndian.Uint64(p[1:])
501	506		t.mu.Lock()
502		-	if _, ok := t.pending[seq]; ok {
503		-	delete(t.pending, seq)
	507	+	acked := false
	508	+	for pendingSeq, p := range t.pending {
	509	+	if pendingSeq > seq {
	510	+	continue
	511	+	}
	512	+	if pendingSeq == seq && p != nil && !p.retransmitted {
	513	+	t.updateRTTLocked(t.clock.Now().Sub(p.first))
	514	+	}
	515	+	delete(t.pending, pendingSeq)
	516	+	acked = true
	517	+	}
	518	+	if acked {
504	519		t.notifySendWaitersLocked()
505	520		}
506	521		t.mu.Unlock()

@@ -530,9 +545,9 @@ func (t *Transport) handleRecord(p []byte) {
530	545		return
531	546		}
532	547		if reliable {
533		-	ack, queued := t.enqueueReliable(seq, f)
	548	+	ackSeq, ack, queued := t.enqueueReliable(seq, f)
534	549		if ack {
535		-	t.sendAck(seq)
	550	+	t.sendAck(ackSeq)
536	551		}
537	552		if queued {
538	553		t.signalDelivery()

@@ -543,20 +558,31 @@ func (t *Transport) handleRecord(p []byte) {
543	558		}
544	559		}
545	560
546		-	func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) (ack bool, queued bool) {
	561	+	func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) (ackSeq uint64, ack bool, queued bool) {
547	562		t.deliverMu.Lock()
548	563		defer t.deliverMu.Unlock()
549	564		if seq < t.nextRecvSeq {
550		-	return true, false
	565	+	return t.highestContiguousRecvLocked(), true, false
551	566		}
552	567		if _, exists := t.recvBuf[seq]; exists {
553		-	return true, false
	568	+	return t.highestContiguousRecvLocked(), true, false
554	569		}
555	570		if len(t.recvBuf) >= t.maxRecvBuffer && seq != t.nextRecvSeq {
556		-	return false, false
	571	+	return 0, false, false
557	572		}
558	573		t.recvBuf[seq] = f
559		-	return true, true
	574	+	return t.highestContiguousRecvLocked(), true, true
	575	+	}
	576	+
	577	+	func (t *Transport) highestContiguousRecvLocked() uint64 {
	578	+	seq := t.nextRecvSeq
	579	+	for {
	580	+	if _, ok := t.recvBuf[seq]; !ok {
	581	+	break
	582	+	}
	583	+	seq++
	584	+	}
	585	+	return seq - 1
560	586		}
561	587
562	588		func (t *Transport) signalDelivery() {

@@ -662,6 +688,7 @@ func (t *Transport) resendPending() {
662	688		}
663	689		p.last = now
664	690		p.attempts++
	691	+	p.retransmitted = true
665	692		resend = append(resend, struct {
666	693		seq uint64
667	694		f ports.Frame

@@ -675,7 +702,10 @@ func (t *Transport) resendPending() {
675	702		}
676	703
677	704		func (t Transport) resendDelayLocked(p pending) time.Duration {
678		-	d := t.resendAfter
	705	+	d := t.rto
	706	+	if d <= 0 {
	707	+	d = t.resendAfter
	708	+	}
679	709		for i := 0; i < p.attempts && d < t.maxResendAfter; i++ {
680	710		d *= 2
681	711		if d > t.maxResendAfter {

@@ -685,6 +715,35 @@ func (t Transport) resendDelayLocked(p pending) time.Duration {
685	715		return d
686	716		}
687	717
	718	+	func (t *Transport) updateRTTLocked(sample time.Duration) {
	719	+	if sample <= 0 {
	720	+	return
	721	+	}
	722	+	if t.srtt <= 0 {
	723	+	t.srtt = sample
	724	+	t.rttvar = sample / 2
	725	+	} else {
	726	+	delta := t.srtt - sample
	727	+	if delta < 0 {
	728	+	delta = -delta
	729	+	}
	730	+	t.rttvar = (3*t.rttvar + delta) / 4
	731	+	t.srtt = (7*t.srtt + sample) / 8
	732	+	}
	733	+	variation := 4 * t.rttvar
	734	+	if variation < time.Millisecond {
	735	+	variation = time.Millisecond
	736	+	}
	737	+	rto := t.srtt + variation
	738	+	if rto < t.resendAfter {
	739	+	rto = t.resendAfter
	740	+	}
	741	+	if rto > t.maxResendAfter {
	742	+	rto = t.maxResendAfter
	743	+	}
	744	+	t.rto = rto
	745	+	}
	746	+
688	747		func (t *Transport) hopPacketConnOnce() {
689	748		t.mu.Lock()
690	749		if t.rebind == nil \|\| t.hoppedOffline \|\| t.closed {

 	resendAfter      time.Duration
 	maxResendAfter   time.Duration
 	maxResendPerTick int
+	srtt             time.Duration
+	rttvar           time.Duration
+	rto              time.Duration
 	writeTimeout     time.Duration
 	degradedAfter    time.Duration
 	probeAfter       time.Duration
 }
 type pending struct {
-	frame    ports.Frame
+	frame         ports.Frame
-	first    time.Time
+	first         time.Time
-	last     time.Time
+	last          time.Time
-	attempts int
+	attempts      int
+	retransmitted bool
 }
 func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {
 		resendAfter:      opts.ResendAfter,
 		maxResendAfter:   opts.MaxResendAfter,
 		maxResendPerTick: opts.MaxResendPerTick,
+		rto:              opts.ResendAfter,
 		writeTimeout:     opts.WriteTimeout,
 		degradedAfter:    opts.DegradedAfter,
 		probeAfter:       opts.ProbeAfter,
 		}
 		seq := binary.BigEndian.Uint64(p[1:])
 		t.mu.Lock()
-		if _, ok := t.pending[seq]; ok {
+		acked := false
-			delete(t.pending, seq)
+		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))
+			}
+			delete(t.pending, pendingSeq)
+			acked = true
+		}
+		if acked {
 			t.notifySendWaitersLocked()
 		}
 		t.mu.Unlock()
 			return
 		}
 		if reliable {
-			ack, queued := t.enqueueReliable(seq, f)
+			ackSeq, ack, queued := t.enqueueReliable(seq, f)
 			if ack {
-				t.sendAck(seq)
+				t.sendAck(ackSeq)
 			}
 			if queued {
 				t.signalDelivery()
 	}
 }
-func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) (ack bool, queued bool) {
+func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) (ackSeq uint64, ack bool, queued bool) {
 	t.deliverMu.Lock()
 	defer t.deliverMu.Unlock()
 	if seq < t.nextRecvSeq {
-		return true, false
+		return t.highestContiguousRecvLocked(), true, false
 	}
 	if _, exists := t.recvBuf[seq]; exists {
-		return true, false
+		return t.highestContiguousRecvLocked(), true, false
 	}
 	if len(t.recvBuf) >= t.maxRecvBuffer && seq != t.nextRecvSeq {
-		return false, false
+		return 0, false, false
 	}
 	t.recvBuf[seq] = f
-	return true, true
+	return t.highestContiguousRecvLocked(), true, true
+}
+func (t *Transport) highestContiguousRecvLocked() uint64 {
+	seq := t.nextRecvSeq
+	for {
+		if _, ok := t.recvBuf[seq]; !ok {
+			break
+		}
+		seq++
+	}
+	return seq - 1
 }
 func (t *Transport) signalDelivery() {
 		}
 		p.last = now
 		p.attempts++
+		p.retransmitted = true
 		resend = append(resend, struct {
 			seq uint64
 			f   ports.Frame
 }
 func (t *Transport) resendDelayLocked(p *pending) time.Duration {
-	d := t.resendAfter
+	d := t.rto
+	if d <= 0 {
+		d = t.resendAfter
+	}
 	for i := 0; i < p.attempts && d < t.maxResendAfter; i++ {
 		d *= 2
 		if d > t.maxResendAfter {
 	return d
 }
+func (t *Transport) updateRTTLocked(sample time.Duration) {
+	if sample <= 0 {
+		return
+	}
+	if t.srtt <= 0 {
+		t.srtt = sample
+		t.rttvar = sample / 2
+	} else {
+		delta := t.srtt - sample
+		if delta < 0 {
+			delta = -delta
+		}
+		t.rttvar = (3*t.rttvar + delta) / 4
+		t.srtt = (7*t.srtt + sample) / 8
+	}
+	variation := 4 * t.rttvar
+	if variation < time.Millisecond {
+		variation = time.Millisecond
+	}
+	rto := t.srtt + variation
+	if rto < t.resendAfter {
+		rto = t.resendAfter
+	}
+	if rto > t.maxResendAfter {
+		rto = t.maxResendAfter
+	}
+	t.rto = rto
+}
 func (t *Transport) hopPacketConnOnce() {
 	t.mu.Lock()
 	if t.rebind == nil || t.hoppedOffline || t.closed {

M internal/adapters/dgram/transport_test.go +90 -2


@@ -511,11 +511,11 @@ func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) {
511	511
512	512		func TestReliableReceiveBufferDoesNotDropAckedFutureFrame(t *testing.T) {
513	513		tr := &Transport{maxRecvBuffer: 1, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
514		-	ack, queued := tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("future")})
	514	+	_, ack, queued := tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("future")})
515	515		if !ack \|\| !queued {
516	516		t.Fatalf("future enqueue ack=%v queued=%v, want true/true", ack, queued)
517	517		}
518		-	ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")})
	518	+	_, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")})
519	519		if !ack \|\| !queued {
520	520		t.Fatalf("next enqueue ack=%v queued=%v, want true/true", ack, queued)
521	521		}

@@ -624,6 +624,7 @@ func newResendTestTransport(t testing.T, maxResendPerTick int) (Transport, *at
624	624		resendAfter: 100 * time.Millisecond,
625	625		maxResendAfter: 500 * time.Millisecond,
626	626		maxResendPerTick: maxResendPerTick,
	627	+	rto: 100 * time.Millisecond,
627	628		clock: realClock{},
628	629		}, &writes
629	630		}

@@ -701,6 +702,93 @@ func TestRecvUnblocksWhenPeerDeadAfter(t *testing.T) {
701	702		}
702	703		}
703	704
	705	+	func TestCumulativeAckReleasesAllEarlierPendingFrames(t *testing.T) {
	706	+	tr, _ := newResendTestTransport(t, 10)
	707	+	now := tr.clock.Now()
	708	+	for seq := uint64(1); seq <= 3; seq++ {
	709	+	tr.pending[seq] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(seq)}}, first: now, last: now}
	710	+	}
	711	+
	712	+	tr.handleRecord(ackRecord(3))
	713	+
	714	+	if got := len(tr.pending); got != 0 {
	715	+	t.Fatalf("pending after cumulative ack=%d, want 0", got)
	716	+	}
	717	+	}
	718	+
	719	+	func TestReliableReceiverAcksOnlyHighestContiguousSequence(t *testing.T) {
	720	+	tr := &Transport{maxRecvBuffer: maxRecvBuffer, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
	721	+
	722	+	ackSeq, ack, queued := tr.enqueueReliable(3, ports.Frame{Type: ports.MsgOutput, Payload: []byte("third")})
	723	+	if !ack \|\| !queued {
	724	+	t.Fatalf("out-of-order enqueue ack=%v queued=%v, want true/true", ack, queued)
	725	+	}
	726	+	if got, want := ackSeq, uint64(0); got != want {
	727	+	t.Fatalf("out-of-order ack=%d, want highest contiguous %d", got, want)
	728	+	}
	729	+
	730	+	ackSeq, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("first")})
	731	+	if !ack \|\| !queued {
	732	+	t.Fatalf("first enqueue ack=%v queued=%v, want true/true", ack, queued)
	733	+	}
	734	+	if got, want := ackSeq, uint64(1); got != want {
	735	+	t.Fatalf("first contiguous ack=%d, want %d", got, want)
	736	+	}
	737	+
	738	+	ackSeq, ack, queued = tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("second")})
	739	+	if !ack \|\| !queued {
	740	+	t.Fatalf("gap fill enqueue ack=%v queued=%v, want true/true", ack, queued)
	741	+	}
	742	+	if got, want := ackSeq, uint64(3); got != want {
	743	+	t.Fatalf("filled gap ack=%d, want %d", got, want)
	744	+	}
	745	+	}
	746	+
	747	+	func TestRTTEstimatorUpdatesRTOFromAckSample(t *testing.T) {
	748	+	clk := fixedClock{now: time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)}
	749	+	tr, writes := newResendTestTransport(t, 10)
	750	+	tr.clock = clk
	751	+	tr.resendAfter = 100 * time.Millisecond
	752	+	tr.maxResendAfter = time.Second
	753	+	tr.rto = tr.resendAfter
	754	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: clk.now.Add(-200 * time.Millisecond), last: clk.now.Add(-200 * time.Millisecond)}
	755	+
	756	+	tr.handleRecord(ackRecord(1))
	757	+
	758	+	if tr.rto <= tr.resendAfter {
	759	+	t.Fatalf("rto=%v, want above initial resend after RTT sample", tr.rto)
	760	+	}
	761	+	tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: clk.now, last: clk.now.Add(-(tr.rto - time.Millisecond))}
	762	+	tr.resendPending()
	763	+	if got := writes.Load(); got != 0 {
	764	+	t.Fatalf("writes before rto=%d, want 0", got)
	765	+	}
	766	+	tr.pending[2].last = clk.now.Add(-tr.rto)
	767	+	tr.resendPending()
	768	+	if got := writes.Load(); got != 1 {
	769	+	t.Fatalf("writes at rto=%d, want 1", got)
	770	+	}
	771	+	}
	772	+
	773	+	func TestRTTEstimatorIgnoresRetransmittedFramesAfterRecovery(t *testing.T) {
	774	+	clk := fixedClock{now: time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)}
	775	+	tr, _ := newResendTestTransport(t, 10)
	776	+	tr.clock = clk
	777	+	tr.resendAfter = 100 * time.Millisecond
	778	+	tr.maxResendAfter = time.Second
	779	+	tr.rto = tr.resendAfter
	780	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: clk.now.Add(-200 * time.Millisecond), last: clk.now.Add(-tr.resendAfter)}
	781	+	tr.resendPending()
	782	+	tr.linkState = ports.LinkStateOffline
	783	+
	784	+	tr.setLinkState(ports.LinkStateConnected, nil)
	785	+	tr.handleRecord(ackRecord(1))
	786	+
	787	+	if tr.rto != tr.resendAfter \|\| tr.srtt != 0 \|\| tr.rttvar != 0 {
	788	+	t.Fatalf("rtt sampled retransmitted frame after recovery: rto=%v srtt=%v rttvar=%v", tr.rto, tr.srtt, tr.rttvar)
	789	+	}
	790	+	}
	791	+
704	792		func TestPendingReliableQueueBackpressuresUntilAck(t *testing.T) {
705	793		aPC, bPC := newPair()
706	794		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }


@@ -511,11 +511,11 @@ func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) {
511			511
512		func TestReliableReceiveBufferDoesNotDropAckedFutureFrame(t *testing.T) {	512		func TestReliableReceiveBufferDoesNotDropAckedFutureFrame(t *testing.T) {
513		tr := &Transport{maxRecvBuffer: 1, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}	513		tr := &Transport{maxRecvBuffer: 1, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
514	-	ack, queued := tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("future")})	514	+	_, ack, queued := tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("future")})
515		if !ack \|\| !queued {	515		if !ack \|\| !queued {
516		t.Fatalf("future enqueue ack=%v queued=%v, want true/true", ack, queued)	516		t.Fatalf("future enqueue ack=%v queued=%v, want true/true", ack, queued)
517		}	517		}
518	-	ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")})	518	+	_, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")})
519		if !ack \|\| !queued {	519		if !ack \|\| !queued {
520		t.Fatalf("next enqueue ack=%v queued=%v, want true/true", ack, queued)	520		t.Fatalf("next enqueue ack=%v queued=%v, want true/true", ack, queued)
521		}	521		}

@@ -624,6 +624,7 @@ func newResendTestTransport(t testing.T, maxResendPerTick int) (Transport, *at
624		resendAfter: 100 * time.Millisecond,	624		resendAfter: 100 * time.Millisecond,
625		maxResendAfter: 500 * time.Millisecond,	625		maxResendAfter: 500 * time.Millisecond,
626		maxResendPerTick: maxResendPerTick,	626		maxResendPerTick: maxResendPerTick,
			627	+	rto: 100 * time.Millisecond,
627		clock: realClock{},	628		clock: realClock{},
628		}, &writes	629		}, &writes
629		}	630		}

@@ -701,6 +702,93 @@ func TestRecvUnblocksWhenPeerDeadAfter(t *testing.T) {
701		}	702		}
702		}	703		}
703			704
			705	+	func TestCumulativeAckReleasesAllEarlierPendingFrames(t *testing.T) {
			706	+	tr, _ := newResendTestTransport(t, 10)
			707	+	now := tr.clock.Now()
			708	+	for seq := uint64(1); seq <= 3; seq++ {
			709	+	tr.pending[seq] = &pending{frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(seq)}}, first: now, last: now}
			710	+	}
			711	+
			712	+	tr.handleRecord(ackRecord(3))
			713	+
			714	+	if got := len(tr.pending); got != 0 {
			715	+	t.Fatalf("pending after cumulative ack=%d, want 0", got)
			716	+	}
			717	+	}
			718	+
			719	+	func TestReliableReceiverAcksOnlyHighestContiguousSequence(t *testing.T) {
			720	+	tr := &Transport{maxRecvBuffer: maxRecvBuffer, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
			721	+
			722	+	ackSeq, ack, queued := tr.enqueueReliable(3, ports.Frame{Type: ports.MsgOutput, Payload: []byte("third")})
			723	+	if !ack \|\| !queued {
			724	+	t.Fatalf("out-of-order enqueue ack=%v queued=%v, want true/true", ack, queued)
			725	+	}
			726	+	if got, want := ackSeq, uint64(0); got != want {
			727	+	t.Fatalf("out-of-order ack=%d, want highest contiguous %d", got, want)
			728	+	}
			729	+
			730	+	ackSeq, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("first")})
			731	+	if !ack \|\| !queued {
			732	+	t.Fatalf("first enqueue ack=%v queued=%v, want true/true", ack, queued)
			733	+	}
			734	+	if got, want := ackSeq, uint64(1); got != want {
			735	+	t.Fatalf("first contiguous ack=%d, want %d", got, want)
			736	+	}
			737	+
			738	+	ackSeq, ack, queued = tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("second")})
			739	+	if !ack \|\| !queued {
			740	+	t.Fatalf("gap fill enqueue ack=%v queued=%v, want true/true", ack, queued)
			741	+	}
			742	+	if got, want := ackSeq, uint64(3); got != want {
			743	+	t.Fatalf("filled gap ack=%d, want %d", got, want)
			744	+	}
			745	+	}
			746	+
			747	+	func TestRTTEstimatorUpdatesRTOFromAckSample(t *testing.T) {
			748	+	clk := fixedClock{now: time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)}
			749	+	tr, writes := newResendTestTransport(t, 10)
			750	+	tr.clock = clk
			751	+	tr.resendAfter = 100 * time.Millisecond
			752	+	tr.maxResendAfter = time.Second
			753	+	tr.rto = tr.resendAfter
			754	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: clk.now.Add(-200 * time.Millisecond), last: clk.now.Add(-200 * time.Millisecond)}
			755	+
			756	+	tr.handleRecord(ackRecord(1))
			757	+
			758	+	if tr.rto <= tr.resendAfter {
			759	+	t.Fatalf("rto=%v, want above initial resend after RTT sample", tr.rto)
			760	+	}
			761	+	tr.pending[2] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: clk.now, last: clk.now.Add(-(tr.rto - time.Millisecond))}
			762	+	tr.resendPending()
			763	+	if got := writes.Load(); got != 0 {
			764	+	t.Fatalf("writes before rto=%d, want 0", got)
			765	+	}
			766	+	tr.pending[2].last = clk.now.Add(-tr.rto)
			767	+	tr.resendPending()
			768	+	if got := writes.Load(); got != 1 {
			769	+	t.Fatalf("writes at rto=%d, want 1", got)
			770	+	}
			771	+	}
			772	+
			773	+	func TestRTTEstimatorIgnoresRetransmittedFramesAfterRecovery(t *testing.T) {
			774	+	clk := fixedClock{now: time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)}
			775	+	tr, _ := newResendTestTransport(t, 10)
			776	+	tr.clock = clk
			777	+	tr.resendAfter = 100 * time.Millisecond
			778	+	tr.maxResendAfter = time.Second
			779	+	tr.rto = tr.resendAfter
			780	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: clk.now.Add(-200 * time.Millisecond), last: clk.now.Add(-tr.resendAfter)}
			781	+	tr.resendPending()
			782	+	tr.linkState = ports.LinkStateOffline
			783	+
			784	+	tr.setLinkState(ports.LinkStateConnected, nil)
			785	+	tr.handleRecord(ackRecord(1))
			786	+
			787	+	if tr.rto != tr.resendAfter \|\| tr.srtt != 0 \|\| tr.rttvar != 0 {
			788	+	t.Fatalf("rtt sampled retransmitted frame after recovery: rto=%v srtt=%v rttvar=%v", tr.rto, tr.srtt, tr.rttvar)
			789	+	}
			790	+	}
			791	+
704		func TestPendingReliableQueueBackpressuresUntilAck(t *testing.T) {	792		func TestPendingReliableQueueBackpressuresUntilAck(t *testing.T) {
705		aPC, bPC := newPair()	793		aPC, bPC := newPair()
706		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }	794		aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }