feat(tcp): receiver-side RTT estimator for DRS (autotune step 3 / R1)

Third phase of the Tier-3 buffer auto-tuning plan
(docs/refactor/tcp_buffer_autotuning.md §4 R1 / §10 step 3): add the
receiver-side RTT estimator that receive-buffer Dynamic Right-Sizing
(Track R, still to land in step 4) needs for its per-RTT measurement
cadence.

Why a new estimator: the RFC 6298 sender SRTT only gets samples when
our own data is acked, so on a pure download (we send only ACKs) it
stays None and DRS could never fire. Linux solves this with a separate
rcv_rtt_est fed by tcp_rcv_rtt_measure_ts: an inbound data segment whose
TSecr echoes a timestamp we sent one round trip ago yields
rtt = now - TSecr, valid even for a pure receiver.

- New RcvRttState (state/tcp__state__rcv_rtt.py): rtt_ms (alpha = 1/8
  EWMA, None until first sample) + last_tsecr for the within-RTT dedup.
  observe() seeds on the first sample, folds via (7*old + sample)//8
  after, and no-ops on a repeated TSecr so a burst within one round trip
  contributes a single sample (Linux dedups on rcv_rtt_last_tsecr).
- Instantiated as self._rcv_rtt in TcpSession.__init__.
- Hook in tcp__session__ack.py phase 5: on a new-data segment with
  send_ts and a truthy TSecr, sample (now_ms - TSecr) & 0xFFFFFFFF (the
  same formula the phase-3 sender sample uses) and fold it in.
  Timestamps-gated for the first pass; the no-TS tcp_rcv_rtt_measure
  fallback stays deferred (plan §7).

No consumer yet — step 4 (R2-R4) reads _rcv_rtt.rtt_ms to gate the DRS
grow cadence. No behaviour change.

Tests-first: test__tcp__state__rcv_rtt.py (unit — defaults, first-sample
seed, EWMA fold, TSecr dedup) + test__tcp__session__rcv_rtt.py
(integration — RX-path seed/fold/dedup, no-TS leaves it unset). RED
pre-impl: ModuleNotFoundError + AttributeError (_rcv_rtt absent) → 8
green post-impl. Plan §4 / §10 step 3 marked done.

Reference: RFC 7323 §4 (RTTM via TSecr).
Reference: Linux tcp_rcv_rtt_measure_ts / rcv_rtt_est.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: ccie18643

docs/refactor/tcp_buffer_autotuning.md

@@ -166,6 +166,22 @@ once per receiver-RTT, when the app has just drained the rx buffer:
 
 ### R1 — receiver-side RTT estimator (the prerequisite, medium)
 
+**DONE.** `RcvRttState` (`state/tcp__state__rcv_rtt.py`, `@dataclass(slots=True)`):
+`rtt_ms: int | None` (α=1/8 EWMA, `None` until first sample) + `last_tsecr`
+for the within-RTT dedup; `observe(sample_ms, tsecr)` seeds on the first
+sample, folds via `(7*old + sample)//8` after, and no-ops on a repeated
+TSecr. Instantiated `self._rcv_rtt` in `TcpSession.__init__`. Hook: phase 5
+of `tcp__session__ack.py`, on a new-data segment with `send_ts` and a
+truthy `tcp__tsecr`, sample `(stack.timer.now_ms - tcp__tsecr) & 0xFFFF_FFFF`
+— the same formula the phase-3 *sender* sample uses, but fired on inbound
+data so a pure receiver (ACK-only) still gets an RTT. Timestamps-gated
+(the no-TS `tcp_rcv_rtt_measure` fallback stays deferred, §7). Tests:
+`test__tcp__state__rcv_rtt.py` (unit — EWMA/dedup) +
+`test__tcp__session__rcv_rtt.py` (integration — RX-path seed/fold/dedup +
+no-TS leaves it unset). The DRS reader (R3/R4) consumes `_rcv_rtt.rtt_ms`.
+
+Original plan text:
+
 **Why it's needed:** DRS gates its once-per-RTT cadence on a *receiver*
 RTT. On a bulk download PyTCP only sends bare ACKs, so `_rto_state.srtt_ms`
 never gets a sample (no acked data of ours) — it stays `None` and DRS
@@ -493,7 +509,9 @@ CLAUDE.md "Linux as tiebreaker" precedence.
    fully observable once step 5 lands the small default. Tests:
    `test__tcp__session__sndbuf_autotune.py`.
 3. **R1** (receiver RTT estimator) — the DRS prerequisite; self-contained,
-   unit-testable in isolation.
+   unit-testable in isolation. **DONE** — `RcvRttState` +
+   `tcp__session__ack.py` phase-5 TSecr hook; timestamps-gated. Tests:
+   `test__tcp__state__rcv_rtt.py`, `test__tcp__session__rcv_rtt.py`.
 4. **R2 + R3 + R4** (DRS struct, trigger, grow policy) — the large piece;
    lands on top of R1.
 5. **The small-default switch** (§2 / §7) — flip the unset `SO_SNDBUF`

packages/pytcp/pytcp/protocols/tcp/session/tcp__session.py

@@ -62,6 +62,7 @@
 from pytcp.protocols.tcp.state.tcp__state__keepalive import KeepaliveState
 from pytcp.protocols.tcp.state.tcp__state__persist import PersistState
 from pytcp.protocols.tcp.state.tcp__state__rack_tlp import RackTlpState
+from pytcp.protocols.tcp.state.tcp__state__rcv_rtt import RcvRttState
 from pytcp.protocols.tcp.state.tcp__state__recv_seq import RecvSeqState
 from pytcp.protocols.tcp.state.tcp__state__rtt_sample import RttSampleState
 from pytcp.protocols.tcp.state.tcp__state__send_seq import SendSeqState
@@ -339,6 +340,10 @@ def __init__(
         # §5.7 idle-baseline 'last_send_time_ms'. See
         # 'state/tcp__state__rtt_sample.py'.
         self._rtt: RttSampleState = RttSampleState()
+        # RFC 7323 §4 receiver-side RTT estimator (Tier-3 Track R DRS
+        # cadence). Fed from the inbound TSecr echo, so it produces an
+        # RTT even on a pure receiver. See 'state/tcp__state__rcv_rtt.py'.
+        self._rcv_rtt: RcvRttState = RcvRttState()
         self._retransmit_count: int = 0
         # RFC 6298 §5.7 second-clause SYN-retransmit counter.
         # Decoupled from '_retransmit_count' (which

packages/pytcp/pytcp/protocols/tcp/session/tcp__session__ack.py

@@ -715,6 +715,19 @@ def _phase5_consume_segment_and_postprocess(self, packet_rx_md: TcpMetadata) ->
         if packet_rx_md.tcp__data and overlap_prefix < len(packet_rx_md.tcp__data):
             new_data = packet_rx_md.tcp__data[overlap_prefix:]
             session._enqueue_rx_buffer(new_data)
+            # RFC 7323 §4 / Linux 'tcp_rcv_rtt_measure_ts': sample a
+            # RECEIVER-side RTT from the echoed TSecr on this new-data
+            # segment. Unlike the phase-3 sender sample (which needs our
+            # data to be acked), this fires on a pure download where we
+            # only send ACKs — the DRS cadence gate (Tier-3 Track R)
+            # needs it. Deduped on TSecr inside 'observe' so a burst
+            # within one RTT yields a single sample. A truthy 'tcp__tsecr'
+            # covers both the None and the no-echo 0 cases.
+            if session._ts.send_ts and packet_rx_md.tcp__tsecr:
+                session._rcv_rtt.observe(
+                    sample_ms=(stack.timer.now_ms - packet_rx_md.tcp__tsecr) & 0xFFFF_FFFF,
+                    tsecr=packet_rx_md.tcp__tsecr,
+                )
             __debug__ and log(
                 "tcp-ss",
                 f"[{session}] - Enqueued {len(new_data)} bytes starting at "

packages/pytcp/pytcp/protocols/tcp/state/tcp__state__rcv_rtt.py

@@ -0,0 +1,77 @@
+################################################################################
+##                                                                            ##
+##   PyTCP - Python TCP/IP stack                                              ##
+##   Copyright (C) 2020-present Sebastian Majewski                            ##
+##                                                                            ##
+##   This program is free software: you can redistribute it and/or modify     ##
+##   it under the terms of the GNU General Public License as published by     ##
+##   the Free Software Foundation, either version 3 of the License, or        ##
+##   (at your option) any later version.                                      ##
+##                                                                            ##
+##   This program is distributed in the hope that it will be useful,          ##
+##   but WITHOUT ANY WARRANTY; without even the implied warranty of           ##
+##   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the             ##
+##   GNU General Public License for more details.                             ##
+##                                                                            ##
+##   You should have received a copy of the GNU General Public License        ##
+##   along with this program. If not, see <https://www.gnu.org/licenses/>.    ##
+##                                                                            ##
+##   Author's email: ccie18643@gmail.com                                      ##
+##   Github repository: https://github.com/ccie18643/PyTCP                    ##
+##                                                                            ##
+################################################################################
+
+
+"""
+This module contains the per-session receiver-side RTT estimator state
+container. Unlike the RFC 6298 sender SRTT (which needs our data to be
+acked), this is fed from the RFC 7323 TSecr echo on inbound data
+segments, so it produces an RTT measurement even on a pure receiver
+that only sends ACKs — the round-trip cadence the Tier-3 receive-buffer
+Dynamic Right-Sizing (Track R) gate depends on (Linux 'rcv_rtt_est' /
+'tcp_rcv_rtt_measure_ts').
+
+pytcp/protocols/tcp/state/tcp__state__rcv_rtt.py
+
+ver 3.0.8
+"""
+
+from dataclasses import dataclass
+
+
+@dataclass(slots=True)
+class RcvRttState:
+    """
+    Per-session receiver-side RTT estimator. Owned by 'TcpSession';
+    written on the RX/ACK thread from the inbound-data TSecr echo and
+    read by the DRS measurement (app thread). Timestamps-gated: only
+    fed while bilateral RFC 7323 TSopt is active.
+    """
+
+    # Smoothed receiver RTT (ms), 'None' until the first sample —
+    # DRS treats 'None' as "no cadence yet" and skips its adjust.
+    rtt_ms: int | None = None
+
+    # The most recently sampled TSecr. A segment echoing the same
+    # TSecr is a within-RTT duplicate and contributes no new sample,
+    # so a burst does not overweight the estimate (Linux dedups on
+    # 'rcv_rtt_last_tsecr').
+    last_tsecr: int | None = None
+
+    def observe(self, *, sample_ms: int, tsecr: int) -> None:
+        """
+        Fold a receiver RTT sample ('now_ms - TSecr') into the
+        smoothed estimate. Deduplicated on 'tsecr' so at most one
+        sample lands per round trip. The first sample seeds the
+        estimate directly; subsequent samples fold via the
+        alpha = 1/8 EWMA '(7 * old + sample) // 8' (the RFC 6298 SRTT
+        smoothing, matching Linux 'tcp_rcv_rtt_update' win_dep=0).
+        """
+
+        if tsecr == self.last_tsecr:
+            return
+        self.last_tsecr = tsecr
+        if self.rtt_ms is None:
+            self.rtt_ms = sample_ms
+        else:
+            self.rtt_ms = (7 * self.rtt_ms + sample_ms) // 8

packages/pytcp/pytcp/tests/integration/protocols/tcp/test__tcp__session__rcv_rtt.py

@@ -0,0 +1,178 @@
+################################################################################
+##                                                                            ##
+##   PyTCP - Python TCP/IP stack                                              ##
+##   Copyright (C) 2020-present Sebastian Majewski                            ##
+##                                                                            ##
+##   This program is free software: you can redistribute it and/or modify     ##
+##   it under the terms of the GNU General Public License as published by     ##
+##   the Free Software Foundation, either version 3 of the License, or        ##
+##   (at your option) any later version.                                      ##
+##                                                                            ##
+##   This program is distributed in the hope that it will be useful,          ##
+##   but WITHOUT ANY WARRANTY; without even the implied warranty of           ##
+##   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the             ##
+##   GNU General Public License for more details.                             ##
+##                                                                            ##
+##   You should have received a copy of the GNU General Public License        ##
+##   along with this program. If not, see <https://www.gnu.org/licenses/>.    ##
+##                                                                            ##
+##   Author's email: ccie18643@gmail.com                                      ##
+##   Github repository: https://github.com/ccie18643/PyTCP                    ##
+##                                                                            ##
+################################################################################
+
+
+# pylint: disable=protected-access
+# pyright: reportPrivateUsage=false
+
+
+"""
+Integration tests for the receiver-side RTT estimator (Tier-3 Track R,
+R1). On an inbound new-data segment carrying a valid RFC 7323 TSecr, the
+session folds 'now_ms - TSecr' into '_rcv_rtt' — the RTT measurement DRS
+needs even on a pure receiver (which only sends ACKs, so the sender SRTT
+never gets a sample). Timestamps-gated for the first pass.
+
+pytcp/tests/integration/protocols/tcp/test__tcp__session__rcv_rtt.py
+
+ver 3.0.8
+"""
+
+from pytcp.protocols.tcp.session import TcpSession
+from pytcp.tests.lib.tcp_segment_factory import build_tcp4
+from pytcp.tests.lib.tcp_testcase import TcpTestCase
+
+_LOCAL_PORT = 12345
+_REMOTE_PORT = 80
+_PEER_ISS = 5000
+_LOCAL_ISS = 1000
+
+
+class TestTcpSessionRcvRtt(TcpTestCase):
+    """
+    The receiver-side RTT estimator RX-path tests.
+    """
+
+    def _establish_with_tsopt(self) -> TcpSession:
+        """
+        Drive an active-open handshake with bilateral timestamps and
+        advance the virtual clock past the sampled RTTs so 'now_ms -
+        TSecr' stays positive.
+        """
+
+        session = self._drive_handshake_to_established(
+            iss=_LOCAL_ISS,
+            peer_iss=_PEER_ISS,
+            peer_tsval=7000,
+            peer_tsecr=0,
+        )
+        assert session._ts.send_ts, "Setup: bilateral TSopt negotiation must succeed."
+        self._advance(ms=200)
+        return session
+
+    def _peer_data(self, *, seq: int, tsval: int, tsecr: int, payload: bytes) -> bytes:
+        """
+        Build an inbound data segment carrying the supplied TSval /
+        TSecr and payload.
+        """
+
+        return build_tcp4(
+            sport=_REMOTE_PORT,
+            dport=_LOCAL_PORT,
+            seq=seq,
+            ack=_LOCAL_ISS + 1,
+            flags=("ACK",),
+            win=64240,
+            tsval=tsval,
+            tsecr=tsecr,
+            payload=payload,
+        )
+
+    def test__rcv_rtt__first_data_segment_seeds_estimate(self) -> None:
+        """
+        Ensure an inbound new-data segment whose TSecr was sent one
+        round trip ago seeds the receiver RTT estimate with
+        'now_ms - TSecr'.
+
+        Reference: RFC 7323 §4 (RTTM via TSecr).
+        """
+
+        session = self._establish_with_tsopt()
+        now = self._timer.now_ms
+        self._drive_rx(frame=self._peer_data(seq=_PEER_ISS + 1, tsval=7001, tsecr=now - 40, payload=b"X" * 100))
+
+        self.assertEqual(
+            session._rcv_rtt.rtt_ms,
+            40,
+            msg="The first data segment must seed the receiver RTT with now_ms - TSecr.",
+        )
+
+    def test__rcv_rtt__second_segment_folds_ewma(self) -> None:
+        """
+        Ensure a second new-data segment with a fresh TSecr folds into
+        the receiver RTT via the alpha = 1/8 EWMA.
+
+        Reference: RFC 7323 §4 (RTTM via TSecr).
+        """
+
+        session = self._establish_with_tsopt()
+        now = self._timer.now_ms
+        self._drive_rx(frame=self._peer_data(seq=_PEER_ISS + 1, tsval=7001, tsecr=now - 40, payload=b"X" * 100))
+        self._drive_rx(frame=self._peer_data(seq=_PEER_ISS + 101, tsval=7002, tsecr=now - 80, payload=b"Y" * 100))
+
+        # (7 * 40 + 80) // 8 == 45.
+        self.assertEqual(
+            session._rcv_rtt.rtt_ms,
+            45,
+            msg="A second fresh-TSecr sample must fold via the 1/8 EWMA.",
+        )
+
+    def test__rcv_rtt__no_timestamps_leaves_estimate_unset(self) -> None:
+        """
+        Ensure a connection without bilateral timestamps never
+        populates the receiver RTT estimate — the first pass is
+        timestamps-gated.
+
+        Reference: RFC 7323 §4 (RTTM via TSecr).
+        """
+
+        session = self._drive_handshake_to_established(iss=_LOCAL_ISS, peer_iss=_PEER_ISS)
+        assert not session._ts.send_ts, "Setup: no bilateral TSopt without peer TSval."
+        self._advance(ms=200)
+        self._drive_rx(
+            frame=build_tcp4(
+                sport=_REMOTE_PORT,
+                dport=_LOCAL_PORT,
+                seq=_PEER_ISS + 1,
+                ack=_LOCAL_ISS + 1,
+                flags=("ACK",),
+                win=64240,
+                payload=b"X" * 100,
+            )
+        )
+
+        self.assertIsNone(
+            session._rcv_rtt.rtt_ms,
+            msg="Without timestamps the receiver RTT estimate must stay unset.",
+        )
+
+    def test__rcv_rtt__duplicate_tsecr_yields_single_sample(self) -> None:
+        """
+        Ensure two new-data segments echoing the same TSecr contribute
+        only one RTT sample, so a within-RTT burst does not overweight
+        the estimate.
+
+        Reference: RFC 7323 §4 (RTTM via TSecr).
+        """
+
+        session = self._establish_with_tsopt()
+        now = self._timer.now_ms
+        self._drive_rx(frame=self._peer_data(seq=_PEER_ISS + 1, tsval=7001, tsecr=now - 40, payload=b"X" * 100))
+        # Same TSecr, fresh data — the second sample must be ignored.
+        self._drive_rx(frame=self._peer_data(seq=_PEER_ISS + 101, tsval=7002, tsecr=now - 40, payload=b"Y" * 100))
+
+        self.assertEqual(
+            session._rcv_rtt.rtt_ms,
+            40,
+            msg="A repeated TSecr must not contribute a second sample.",
+        )