feat(tcp): add RFC 6298 RTO as a LossRecovery implementation (#169)

Provide a concrete packet-loss recovery algorithm for the LossRecovery
interface added in #168: the RFC 6298 round-trip-time estimator and single
retransmission timer.

RTO is a pure, reactive state machine. It derives RTT estimates and
retransmission decisions solely from the segments observed through the
LossRecovery hooks and the monotonic time handed in at each boundary, so it
holds no clock and allocates nothing (issue #140). It tracks a shadow of the
send sequence space (snd.UNA/snd.NXT) purely from observed segments, which is
how it manages the timer without reaching into the ControlBlock and how it
distinguishes retransmissions for Karn's algorithm.

Covered:
  - §2.2/§2.3 SRTT/RTTVAR/RTO smoothing (integer-shift form)
  - §3 Karn's algorithm: one sample in flight, never sample a retransmit
  - §5.1-§5.3 timer arm/restart/stop as data is sent and acknowledged
  - §5.4-§5.6 timeout response: exponential backoff + go-back-N retransmit
  - §5.7 backoff collapse on a valid RTT sample
  - RTO clamped to [rtoMin, rtoMax]

RTT introspection (SmoothedRTT, CurrentRTO, Running) lives on the concrete type,
not the interface, per the LossRecovery design. Install with new(RTO) on
ConnConfig.LossRecovery; the connection calls Reset on open so the zero value is
ready to use.

Generated with LLM assistance.

Signed-off-by: Marvin Drees <marvin.drees@9elements.com>

tcp/rto.go

@@ -0,0 +1,215 @@
+package tcp
+
+import "time"
+
+// RFC 6298 retransmission-timeout (RTO) parameters. The algorithm keeps a
+// single retransmission timer per connection (RFC 6298 §5): the timer is
+// (re)started whenever new data is acknowledged while data remains in flight,
+// stopped when all data is acknowledged, and on expiry the oldest unacknowledged
+// segment is retransmitted and the RTO is doubled (exponential backoff, §5.5).
+const (
+	// rtoInitial is the RTO used before the first RTT measurement (RFC 6298 §2.1).
+	rtoInitial = time.Second
+	// rtoMin clamps the lower bound of the RTO. RFC 6298 §2.4 recommends a
+	// minimum of 1s, but that is punishing on the low-latency links lneto
+	// targets; like Linux we use a smaller floor so recovery on LAN/embedded
+	// links is timely.
+	rtoMin = 200 * time.Millisecond
+	// rtoMax clamps the upper bound across exponential backoff (RFC 6298 §5.5
+	// permits a maximum of at least 60s).
+	rtoMax = 60 * time.Second
+
+	// rttGainShift (alpha = 1/8) and rttvarGainShift (beta = 1/4) are the
+	// smoothing gains of RFC 6298 §2.3, applied as integer shifts.
+	rttGainShift    = 3 // alpha = 1/8
+	rttvarGainShift = 2 // beta  = 1/4
+	// rttvarK is the RTTVAR multiplier in RTO = SRTT + K*RTTVAR (RFC 6298 §2.3).
+	rttvarK = 4
+	// backoffMax caps the exponential-backoff doublings so RTO arithmetic cannot
+	// overflow and a wedged connection keeps probing at rtoMax.
+	backoffMax = 12
+)
+
+// RTO implements the RFC 6298 round-trip-time estimator and the single
+// retransmission timer as a [LossRecovery]. Construct it with new(RTO) and hand
+// it to [ConnConfig.LossRecovery]; the connection calls [RTO.Reset] on open, so
+// the zero value is ready to use.
+//
+// RTO is a pure, reactive state machine: it observes the segments a connection
+// sends and receives (via the LossRecovery hooks) and the monotonic time handed
+// in at each hook, and from those alone derives RTT estimates and retransmission
+// decisions. It holds no clock and allocates nothing, which keeps it
+// deterministic for unit testing (see issue #140).
+//
+// RTO tracks its own shadow of the send sequence space purely from the segments
+// it observes: [RTO.PostTx] advances the highest sequence sent and [RTO.PreRx]
+// advances the highest sequence acknowledged. This is what lets it manage the
+// timer (RFC 6298 §5.2/§5.3) without reaching into the tcp state machine, and it
+// is also how retransmissions are distinguished for Karn's algorithm — a segment
+// whose sequence space is not beyond the shadow snd.NXT is a retransmission and
+// is never RTT-sampled.
+type RTO struct {
+	srtt    time.Duration // smoothed round-trip time (SRTT).
+	rttvar  time.Duration // round-trip-time variation (RTTVAR).
+	rto     time.Duration // current retransmission timeout.
+	haveRTT bool          // false until the first RTT sample is taken.
+
+	// Shadow of the send sequence space, derived from observed segments.
+	haveSeq bool  // false until the first data segment is observed.
+	sndUNA  Value // highest acknowledged sequence number seen on the wire.
+	sndNXT  Value // one past the highest sequence number sent.
+
+	// RTT sampling state (Karn's algorithm, RFC 6298 §3): at most one segment is
+	// timed at a time and retransmitted segments are never sampled.
+	timing   bool
+	timedSeq Value // ACK at or beyond this value completes the sample.
+	timedAt  int64 // send time (monotonic ns) of the timed segment.
+
+	// Retransmission timer state.
+	running  bool
+	deadline int64 // time (monotonic ns) at which the timer expires.
+	backoff  uint8 // consecutive timeouts, for exponential backoff.
+}
+
+var _ LossRecovery = (*RTO)(nil)
+
+// Reset returns the estimator to its pre-connection state with the initial RTO.
+// It implements [LossRecovery] and is called when the connection opens or aborts
+// so the estimator can be reused across connection reuse.
+func (r *RTO) Reset() { *r = RTO{rto: rtoInitial} }
+
+// SmoothedRTT returns the current smoothed round-trip time (SRTT), or zero
+// before the first RTT measurement. It is concrete-type introspection and is
+// intentionally not part of [LossRecovery].
+func (r *RTO) SmoothedRTT() time.Duration { return r.srtt }
+
+// CurrentRTO returns the timeout currently in effect, clamped to [rtoMin, rtoMax].
+func (r *RTO) CurrentRTO() time.Duration {
+	rto := r.rto
+	if rto < rtoMin {
+		rto = rtoMin
+	} else if rto > rtoMax {
+		rto = rtoMax
+	}
+	return rto
+}
+
+// Running reports whether the retransmission timer is currently armed.
+func (r *RTO) Running() bool { return r.running }
+
+// NextDeadline returns the monotonic-nanosecond instant at which the timer
+// expires, or 0 when it is not armed. It implements [LossRecovery].
+func (r *RTO) NextDeadline() int64 {
+	if !r.running {
+		return 0
+	}
+	return r.deadline
+}
+
+// PreRx samples the RTT and manages the retransmission timer from a received
+// segment (RFC 6298 §5.2/§5.3). It implements [LossRecovery] and always keeps
+// the segment (the estimator never drops traffic).
+func (r *RTO) PreRx(incoming Segment, now int64) RxDirective {
+	if !r.haveSeq || !incoming.Flags.HasAny(FlagACK) {
+		return RxDirective{Keep: true}
+	}
+	ack := incoming.ACK
+	if r.timing && !ack.LessThan(r.timedSeq) {
+		// ACK covers the timed segment: take the RTT sample (§4). A valid
+		// measurement collapses the backoff (§5.7).
+		r.updateRTT(time.Duration(now - r.timedAt))
+		r.timing = false
+		r.backoff = 0
+	}
+	if r.sndUNA.LessThan(ack) && !r.sndNXT.LessThan(ack) {
+		// ACK advances snd.UNA and does not exceed what we have sent.
+		r.sndUNA = ack
+	}
+	if r.sndUNA == r.sndNXT {
+		r.running = false // §5.3: all outstanding data acknowledged.
+	} else {
+		// §5.3: new (but not all) data acknowledged — restart the timer.
+		r.running = true
+		r.deadline = now + int64(r.CurrentRTO())
+	}
+	return RxDirective{Keep: true}
+}
+
+// PreTx reports whether the retransmission timer has expired and, if so, applies
+// the RFC 6298 §5.4–§5.6 timeout response — discard the outstanding RTT sample
+// (Karn), back the RTO off exponentially and restart the timer — returning a
+// directive that asks the connection to retransmit from snd.UNA (go-back-N). It
+// implements [LossRecovery].
+func (r *RTO) PreTx(now int64) TxDirective {
+	if !r.running || now < r.deadline || r.sndUNA == r.sndNXT {
+		return TxDirective{}
+	}
+	r.timing = false // §5.4: do not sample a retransmitted segment.
+	if r.backoff < backoffMax {
+		r.backoff++
+		r.rto = min(r.CurrentRTO()*2, rtoMax) // §5.5: RTO = RTO * 2.
+	}
+	r.running = true
+	r.deadline = now + int64(r.CurrentRTO())
+	return TxDirective{RetransmitAll: true}
+}
+
+// PostTx records an emitted segment: it advances the shadow send sequence,
+// begins timing newly transmitted data (RFC 6298 §3) and arms the timer (§5.1).
+// Segments that do not extend the send sequence are retransmissions and are
+// never RTT-sampled (Karn's algorithm). Control-only segments (no data) are
+// ignored. It implements [LossRecovery].
+func (r *RTO) PostTx(outgoing Segment, now int64) {
+	if outgoing.DATALEN == 0 {
+		return // only data segments are timed / arm the RTO.
+	}
+	segStart := outgoing.SEQ
+	segEnd := segStart + Value(outgoing.LEN())
+	if !r.haveSeq {
+		r.haveSeq = true
+		r.sndUNA = segStart
+		r.sndNXT = segStart
+	}
+	if !r.sndNXT.LessThan(segEnd) {
+		// Segment does not extend the send sequence: it is a retransmission.
+		// Discard any outstanding RTT sample per Karn's algorithm. The timer was
+		// already (re)armed by PreTx on the timeout that triggered this resend.
+		r.timing = false
+		return
+	}
+	r.sndNXT = segEnd
+	if !r.timing {
+		r.timing = true
+		r.timedSeq = segEnd
+		r.timedAt = now
+	}
+	if !r.running {
+		r.running = true
+		r.deadline = now + int64(r.CurrentRTO())
+	}
+}
+
+// updateRTT folds a round-trip measurement into SRTT/RTTVAR/RTO using the
+// integer-shift form of RFC 6298 §2.2/§2.3.
+func (r *RTO) updateRTT(sample time.Duration) {
+	if sample <= 0 {
+		return
+	}
+	if !r.haveRTT {
+		// First measurement (RFC 6298 §2.2).
+		r.srtt = sample
+		r.rttvar = sample / 2
+		r.haveRTT = true
+	} else {
+		// Subsequent measurements (RFC 6298 §2.3):
+		//   RTTVAR = (1-beta)*RTTVAR + beta*|SRTT-R|
+		//   SRTT   = (1-alpha)*SRTT  + alpha*R
+		diff := r.srtt - sample
+		if diff < 0 {
+			diff = -diff
+		}
+		r.rttvar += (diff - r.rttvar) >> rttvarGainShift
+		r.srtt += (sample - r.srtt) >> rttGainShift
+	}
+	r.rto = r.srtt + rttvarK*r.rttvar
+}

tcp/rto_test.go

@@ -0,0 +1,206 @@
+package tcp
+
+import (
+	"testing"
+	"time"
+)
+
+const rtoMs = int64(time.Millisecond)
+
+// rtoDataSeg builds a data segment of datalen octets starting at seq.
+func rtoDataSeg(seq uint32, datalen int) Segment {
+	return Segment{SEQ: Value(seq), DATALEN: Size(datalen), Flags: FlagPSH | FlagACK}
+}
+
+// rtoAckSeg builds a bare ACK acknowledging up to ack.
+func rtoAckSeg(ack uint32) Segment {
+	return Segment{ACK: Value(ack), Flags: FlagACK}
+}
+
+func newRTO() *RTO {
+	var r RTO
+	r.Reset()
+	return &r
+}
+
+func TestRTO_Reset(t *testing.T) {
+	var r RTO
+	r.Reset()
+	if r.rto != rtoInitial {
+		t.Errorf("initial rto=%v, want %v", r.rto, rtoInitial)
+	}
+	if r.CurrentRTO() != rtoInitial {
+		t.Errorf("CurrentRTO=%v, want %v", r.CurrentRTO(), rtoInitial)
+	}
+	if r.haveRTT {
+		t.Error("haveRTT should be false before first sample")
+	}
+	if r.Running() || r.NextDeadline() != 0 {
+		t.Error("timer must be disarmed after Reset")
+	}
+}
+
+// TestRTO_ArmOnSendSampleOnAck sends data, verifies the timer arms, then acks it
+// and verifies an RTT sample is taken and the timer stops once all data is acked.
+func TestRTO_ArmOnSendSampleOnAck(t *testing.T) {
+	r := newRTO()
+	const iss = uint32(1000)
+
+	r.PostTx(rtoDataSeg(iss, 100), 0)
+	if !r.Running() {
+		t.Fatal("timer must arm after sending data")
+	}
+	if r.NextDeadline() != int64(rtoInitial) {
+		t.Errorf("deadline=%d, want %d", r.NextDeadline(), int64(rtoInitial))
+	}
+
+	// ACK arrives one RTT (40ms) later covering all sent data.
+	dir := r.PreRx(rtoAckSeg(iss+100), 40*rtoMs)
+	if !dir.Keep {
+		t.Error("PreRx must keep the segment")
+	}
+	if r.Running() {
+		t.Error("timer must stop once all data is acknowledged")
+	}
+	if r.SmoothedRTT() != 40*time.Millisecond {
+		t.Errorf("srtt=%v, want 40ms", r.SmoothedRTT())
+	}
+}
+
+// TestRTO_RetransmitOnTimeout verifies PreTx directs a go-back-N retransmit once
+// the deadline passes with data outstanding, and backs the RTO off.
+func TestRTO_RetransmitOnTimeout(t *testing.T) {
+	r := newRTO()
+	const iss = uint32(1000)
+	r.PostTx(rtoDataSeg(iss, 100), 0)
+
+	if r.PreTx(int64(rtoInitial) - 1).RetransmitAll {
+		t.Fatal("must not retransmit before the deadline")
+	}
+	dir := r.PreTx(int64(rtoInitial))
+	if !dir.RetransmitAll {
+		t.Fatal("RTO must fire at the deadline with data outstanding")
+	}
+	if r.CurrentRTO() != 2*rtoInitial {
+		t.Errorf("rto=%v after one backoff, want %v", r.CurrentRTO(), 2*rtoInitial)
+	}
+	// The connection resends from snd.UNA; PostTx sees a retransmission.
+	r.PostTx(rtoDataSeg(iss, 100), int64(rtoInitial))
+	if r.timing {
+		t.Error("retransmitted segment must not be RTT-sampled (Karn)")
+	}
+}
+
+// TestRTO_KarnNoSampleOnRetransmittedAck verifies that after a retransmission the
+// ACK does not produce an RTT sample (Karn's algorithm).
+func TestRTO_KarnNoSampleOnRetransmittedAck(t *testing.T) {
+	r := newRTO()
+	const iss = uint32(1000)
+	r.PostTx(rtoDataSeg(iss, 100), 0)
+	// Timeout and retransmit.
+	r.PreTx(int64(rtoInitial))
+	r.PostTx(rtoDataSeg(iss, 100), int64(rtoInitial))
+	// ACK now arrives; no sample should be taken since timing was discarded.
+	r.PreRx(rtoAckSeg(iss+100), int64(rtoInitial)+10*rtoMs)
+	if r.haveRTT {
+		t.Error("no RTT sample should exist after a retransmission (Karn)")
+	}
+}
+
+// TestRTO_TimerRestartsWhilePartiallyAcked verifies the timer restarts (not
+// stops) when an ACK advances UNA but data remains in flight (RFC 6298 §5.3).
+func TestRTO_TimerRestartsWhilePartiallyAcked(t *testing.T) {
+	r := newRTO()
+	const iss = uint32(1000)
+	r.PostTx(rtoDataSeg(iss, 100), 0)
+	r.PostTx(rtoDataSeg(iss+100, 100), 0) // 200 octets outstanding, iss..iss+200.
+
+	dir := r.PreRx(rtoAckSeg(iss+100), 40*rtoMs) // acks first 100 only.
+	if !r.Running() {
+		t.Fatal("timer must remain armed while data is still in flight")
+	}
+	if r.NextDeadline() != 40*rtoMs+int64(r.CurrentRTO()) {
+		t.Errorf("deadline=%d, want %d", r.NextDeadline(), 40*rtoMs+int64(r.CurrentRTO()))
+	}
+	if !dir.Keep {
+		t.Error("PreRx must keep the segment")
+	}
+}
+
+// TestRTO_NoArmWithoutData verifies control-only segments neither arm the timer
+// nor start an RTT sample.
+func TestRTO_NoArmWithoutData(t *testing.T) {
+	r := newRTO()
+	r.PostTx(Segment{SEQ: 1000, Flags: FlagACK}, 0) // pure ACK, DATALEN==0.
+	if r.Running() || r.timing {
+		t.Error("pure control segment must not arm the timer or start a sample")
+	}
+}
+
+// TestRTO_BackoffCollapsesOnValidSample verifies a valid RTT measurement
+// collapses the exponential backoff counter (RFC 6298 §5.7).
+func TestRTO_BackoffCollapsesOnValidSample(t *testing.T) {
+	r := newRTO()
+	const iss = uint32(1000)
+	r.PostTx(rtoDataSeg(iss, 100), 0)
+	r.PreTx(int64(rtoInitial))                        // one timeout: backoff=1.
+	r.PostTx(rtoDataSeg(iss, 100), int64(rtoInitial)) // retransmit (no sample).
+	if r.backoff != 1 {
+		t.Fatalf("backoff=%d, want 1 after a timeout", r.backoff)
+	}
+	// New data sent and freshly sampled, then acked.
+	r.PostTx(rtoDataSeg(iss+100, 100), int64(rtoInitial)+rtoMs)
+	r.PreRx(rtoAckSeg(iss+200), int64(rtoInitial)+30*rtoMs)
+	if r.backoff != 0 {
+		t.Errorf("backoff=%d, want 0 after a valid RTT sample", r.backoff)
+	}
+}
+
+// TestRTO_Clamped verifies CurrentRTO is clamped to [rtoMin, rtoMax].
+func TestRTO_Clamped(t *testing.T) {
+	var r RTO
+	r.Reset()
+	r.rto = time.Nanosecond
+	if got := r.CurrentRTO(); got != rtoMin {
+		t.Errorf("CurrentRTO=%v, want floor %v", got, rtoMin)
+	}
+	r.rto = time.Hour
+	if got := r.CurrentRTO(); got != rtoMax {
+		t.Errorf("CurrentRTO=%v, want ceiling %v", got, rtoMax)
+	}
+}
+
+// TestRTO_UpdateRTTFirstSample verifies the first-measurement initialization of
+// SRTT/RTTVAR (RFC 6298 §2.2).
+func TestRTO_UpdateRTTFirstSample(t *testing.T) {
+	var r RTO
+	r.Reset()
+	r.updateRTT(100 * time.Millisecond)
+	if r.srtt != 100*time.Millisecond {
+		t.Errorf("srtt=%v, want 100ms", r.srtt)
+	}
+	if r.rttvar != 50*time.Millisecond {
+		t.Errorf("rttvar=%v, want 50ms", r.rttvar)
+	}
+	// RTO = SRTT + K*RTTVAR = 100 + 4*50 = 300ms.
+	if r.rto != 300*time.Millisecond {
+		t.Errorf("rto=%v, want 300ms", r.rto)
+	}
+}
+
+// TestRTO_ImplementsLossRecovery exercises RTO through the [LossRecovery]
+// interface: sending data arms a deadline and a full ACK disarms it.
+func TestRTO_ImplementsLossRecovery(t *testing.T) {
+	var lr LossRecovery = newRTO()
+	lr.Reset()
+	lr.PostTx(rtoDataSeg(1000, 100), 0)
+	if lr.NextDeadline() == 0 {
+		t.Error("expected an armed deadline after sending data")
+	}
+	if !lr.PreRx(rtoAckSeg(1100), 10*rtoMs).Keep {
+		t.Error("PreRx must keep")
+	}
+	if lr.NextDeadline() != 0 {
+		t.Error("expected disarmed timer after full ack")
+	}
+}