Merge pull request #104 from Lochnair/feat/client-kernel-tx-capture

feat(client): capture kernel transmit timestamps

crates/irtt-client/AGENTS.md

@@ -0,0 +1,108 @@
+# `irtt-client` guidance
+
+Repository-wide guidance in the root `AGENTS.md` also applies here.
+
+## Linux kernel TX timestamp capture
+
+On Linux, with the `ancillary` feature, the client best-effort upgrades its
+socket from RX-only `SO_TIMESTAMPING` to RX+TX immediately after a successful
+Open, before the first Echo probe can be sent:
+
+```
+socket created
+    -> RX timestamping configured (RX_SOFTWARE | SOFTWARE)
+    -> Open attempts/retries (untimestamped)
+    -> successful Open
+    -> best-effort enable TX timestamping (+ TX_SOFTWARE | OPT_ID | OPT_TSONLY)
+    -> first Echo probe
+```
+
+`TX_SOFTWARE` asks the kernel for a software send timestamp; `OPT_ID` asks it
+to tag each notification with an automatically incrementing per-socket ID;
+`OPT_TSONLY` means the notification never needs to carry the original
+payload, so a zero-length receive buffer is enough. The upgrade is entirely
+best-effort: if it fails, the client keeps operating on the original RX-only
+configuration (`try_enable_tx_timestamping` restores it explicitly) and
+timing falls back to userspace `sent_at` as it always has. TX capability
+failure is never a public error and is never logged from the library.
+
+### Correlation invariant: kernel ID == wire sequence
+
+The client uses the kernel's automatic `OPT_ID` directly as the probe's wire
+sequence number — no separate correlation map. This is safe because:
+
+- **A failed/nonblocking send never consumes an ID.** Linux increments the
+  per-socket ID counter (`sk->sk_tskey`) speculatively while building a
+  datagram and rolls the increment back (`atomic_dec`) on every error path
+  out of `__ip_append_data`/`__ip6_append_data`
+  (`net/ipv4/ip_output.c`, `net/ipv6/ip6_output.c`), guarded by a local
+  `hold_tskey` flag. A `WouldBlock` `try_send` therefore can never leave a
+  gap in, or otherwise advance, the ID the kernel will assign to the next
+  *successful* submission. (`include/net/sock.h`'s `_sock_tx_timestamp`
+  performs the same increment for the non-corked path.) Client-side, the
+  existing send transaction only calls `commit_probe_sent` — which is the
+  only thing that advances `next_wire_seq` — after a successful socket send;
+  a `WouldBlock` result never reaches it.
+- **The two counters start together and advance together.** `wire_seq`
+  starts at 0 on every successful Open (`SessionMachine::commit_open` builds
+  a fresh `ActiveSession` with `next_wire_seq: 0`) and advances by exactly 1
+  only via `commit_probe_sent`, which only runs after a confirmed successful
+  send — the same condition under which the kernel's counter advances by 1.
+  TX timestamping is enabled only after that same successful Open and before
+  any probe is sent, so both counters' zero points line up exactly.
+- **One socket, one session, no reopen.** `SessionMachine`'s state machine
+  only reaches `Open` from `Connected`, and `commit_open`/`commit_local_close`
+  never route back to `Connected`. A `Client`/`AsyncClient` cannot reopen a
+  second session on the same socket, so there is no second epoch of wire
+  sequence numbers to collide with a stale kernel ID.
+- **The one extra send (Close) is harmless.** `Close` sends one datagram
+  after TX timestamping is already enabled, generating one TX timestamp
+  record with no corresponding `wire_seq` (`Close` does not advance
+  `next_wire_seq`). Because the session is terminal immediately after,
+  nothing will ever look up that ID; `record_kernel_tx_timestamp`'s
+  "unknown ID -> ignore" rule discards it safely.
+
+## MSG_ERRQUEUE drains
+
+`receive::linux::error_queue` decodes one `MSG_ERRQUEUE` record at a time
+(`try_recv_error_queue_record`) and classifies it: a genuine send-timestamp
+completion (`TxTimestamp`), a timestamp-origin record that is not a usable
+`SND` completion (`MalformedOrUnsupportedTimestamp` — never treated as
+fatal, since this client only ever requests `TX_SOFTWARE`/`SND`, so anything
+else on that origin is unexpected metadata, not a network error), a genuine
+non-timestamp socket/network error (`SocketError`), or nothing (`Ignored`).
+Origin (`SO_EE_ORIGIN_TIMESTAMPING` or not) is the only thing that decides
+between the last two categories.
+
+`receive::linux::drain_tx_timestamps` reads up to
+`MAX_TX_TIMESTAMP_RECORDS_PER_DRAIN` (32) records per call: nonblocking, one
+record at a time, no heap allocation, stopping early once the queue is
+empty. It is a starvation guard, not a queue-capacity promise — one TX
+timestamp is generated per successful probe, and drains happen at several
+natural choke points, so a modest constant comfortably covers realistic
+backlog. `Client`/`AsyncClient` call it after a successful probe send,
+before processing an inbound reply, and before evaluating timeouts; it never
+waits for a timestamp; an absent timestamp after all of those opportunities
+simply stays absent. A genuine `SocketError` record stops the drain and is
+surfaced through the ordinary `ClientError::Socket` path.
+
+The Tokio adapter uses the same raw nonblocking `recvmsg(MSG_ERRQUEUE |
+MSG_DONTWAIT)` rather than `try_io`/`poll_recv_ready`: error-queue readiness
+is not ordinary read readiness, and driving it through the Tokio reactor
+risks clearing readiness state that the normal Echo receive path depends on.
+There is no background task or second `AsyncFd` around the Tokio-owned
+socket.
+
+## Retained state
+
+`PendingProbe` carries an optional `kernel_tx_timestamp: Option<SystemTime>`
+alongside the authoritative userspace `sent_at`. It survives pending ->
+timed-out retention (`SessionMachine::record_kernel_tx_timestamp` updates
+whichever of `PendingMap`/`TimedOutMap` still holds the probe) and is
+otherwise bounded by the same limits those maps already enforce — no new
+unbounded state. First valid observation wins on duplicates. **It is
+currently dormant metadata only**: no RTT, OWD, or IPDV computation reads
+it. `sent_at` remains the sole input to all of those, exactly as before this
+capability existed. A later change may select it as a preferred send
+timestamp; until then, treat any code path that reads
+`kernel_tx_timestamp` for a measurement as a bug.

crates/irtt-client/src/async_client.rs

@@ -24,7 +24,7 @@ use crate::{
     config::ClientConfig,
     error::ClientError,
     event::{ClientEvent, OpenOutcome},
-    receive::try_recv_tokio_datagram,
+    receive::{drain_tx_timestamps, try_enable_tx_timestamping, try_recv_tokio_datagram},
     session::machine::{
         recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, PreparedOpenRequest,
         PreparedProbe, SessionMachine, TimeoutBatch, MAX_OPEN_PACKET_SIZE,
@@ -222,6 +222,9 @@ pub struct AsyncClient {
     applied_traffic_class: Option<u8>,
     prepared_open: Option<PreparedOpenRequest>,
     prepared_probe: Option<PreparedProbe>,
+    /// See `Client::tx_timestamping_enabled` in `client.rs` for the
+    /// invariant this tracks; it is the same adapter-owned capability flag.
+    tx_timestamping_enabled: bool,
     #[cfg(test)]
     test_hooks: AsyncClientTestHooks,
 }
@@ -250,6 +253,7 @@ impl AsyncClient {
             applied_traffic_class: None,
             prepared_open: Some(prepared_open),
             prepared_probe: None,
+            tx_timestamping_enabled: false,
             #[cfg(test)]
             test_hooks: AsyncClientTestHooks::default(),
         })
@@ -297,6 +301,7 @@ impl AsyncClient {
 
     /// Poll protocol timeouts using the caller's monotonic timestamp.
     pub fn poll_timeouts_at(&mut self, now: Instant) -> Result<Vec<ClientEvent>, ClientError> {
+        self.drain_tx_timestamps()?;
         self.machine.poll_timeouts_at(now)
     }
 
@@ -305,6 +310,7 @@ impl AsyncClient {
         now: Instant,
         limit: usize,
     ) -> Result<TimeoutBatch, ClientError> {
+        self.drain_tx_timestamps()?;
         self.machine.poll_timeouts_bounded_at(now, limit)
     }
 
@@ -566,6 +572,9 @@ impl AsyncClient {
             if let Err(error) = validate_datagram_length(expected_bytes, bytes) {
                 return Poll::Ready(Err(error));
             }
+            if let Err(error) = self.drain_tx_timestamps() {
+                return Poll::Ready(Err(error));
+            }
             events.push(echo_sent_event(
                 self.remote,
                 sent,
@@ -606,6 +615,14 @@ impl AsyncClient {
                 Err(error) => return Poll::Ready(Err(ClientError::Socket(error))),
             };
 
+            // One final bounded drain before processing the reply, so a TX
+            // timestamp that raced the reply is still associated with its
+            // probe before that probe is looked up and possibly removed
+            // below.
+            if let Err(error) = self.drain_tx_timestamps() {
+                return Poll::Ready(Err(error));
+            }
+
             let events = match self.machine.process_received_echo_packet(
                 &self.recv_buffer[..datagram.len],
                 datagram.received_at,
@@ -884,9 +901,23 @@ impl AsyncClient {
         self.schedule = prepared.schedule;
         self.applied_traffic_class = prepared.negotiated_traffic_class;
         self.prepared_open = None;
+        self.tx_timestamping_enabled = try_enable_tx_timestamping(&self.socket);
         outcome
     }
 
+    /// See `Client::drain_tx_timestamps` in `client.rs`: same bounded,
+    /// nonblocking, best-effort drain, adapted to the Tokio socket.
+    fn drain_tx_timestamps(&mut self) -> Result<(), ClientError> {
+        if !self.tx_timestamping_enabled {
+            return Ok(());
+        }
+        let machine = &mut self.machine;
+        drain_tx_timestamps(&self.socket, |id, timestamp| {
+            machine.record_kernel_tx_timestamp(id, timestamp);
+        })
+        .map_err(ClientError::Socket)
+    }
+
     fn poll_open_cleanup(
         &self,
         state: &mut AsyncOpenState,

crates/irtt-client/src/async_client/tests.rs

@@ -741,6 +741,7 @@ fn public_timeout_polling_remains_exhaustive() {
                     wall: SystemTime::UNIX_EPOCH,
                 },
                 timeout_at: now - Duration::from_secs(1),
+                kernel_tx_timestamp: None,
             });
         }
 

crates/irtt-client/src/client.rs

@@ -14,7 +14,7 @@ use crate::{
     config::{ClientConfig, RecvBudget},
     error::ClientError,
     event::{ClientEvent, OpenOutcome},
-    receive::recv_datagram,
+    receive::{drain_tx_timestamps, recv_datagram, try_enable_tx_timestamping},
     session::machine::{
         recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, ProbeSent,
         SessionMachine, MAX_OPEN_PACKET_SIZE,
@@ -104,6 +104,13 @@ pub struct Client {
     remote: SocketAddr,
     recv_buffer: Vec<u8>,
     applied_traffic_class: Option<u8>,
+    /// Whether the socket's `SO_TIMESTAMPING` flags carry `TX_SOFTWARE` +
+    /// `OPT_ID` + `OPT_TSONLY`. Always `false` off Linux, without the
+    /// `ancillary` feature, or before the best-effort post-Open upgrade;
+    /// tracked here (adapter-owned capability, not protocol state) so
+    /// unnecessary error-queue drains are skipped once it is known there is
+    /// nothing to drain.
+    tx_timestamping_enabled: bool,
     #[cfg(test)]
     test_hooks: ClientTestHooks,
     #[cfg(test)]
@@ -133,6 +140,7 @@ impl Client {
             remote,
             recv_buffer: vec![0_u8; recv_buffer_size(false, None)?],
             applied_traffic_class: None,
+            tx_timestamping_enabled: false,
             #[cfg(test)]
             test_hooks: ClientTestHooks::default(),
             #[cfg(test)]
@@ -267,6 +275,11 @@ impl Client {
             Err(err) => return Err(ClientError::Socket(err)),
         };
 
+        // One final bounded drain before processing the reply, so a TX
+        // timestamp that raced the reply is still associated with its probe
+        // before that probe is looked up and possibly removed below.
+        self.drain_tx_timestamps()?;
+
         let events = self.runtime.process_received_echo_packet(
             &self.recv_buffer[..datagram.len],
             datagram.received_at,
@@ -313,6 +326,7 @@ impl Client {
     ///
     /// `now` is monotonic time only; wall-clock time is not used for timeout expiry.
     pub fn poll_timeouts_at(&mut self, now: Instant) -> Result<Vec<ClientEvent>, ClientError> {
+        self.drain_tx_timestamps()?;
         self.runtime.poll_timeouts_at(now)
     }
 
@@ -430,6 +444,7 @@ impl Client {
         #[cfg(test)]
         let bytes = reported_bytes.unwrap_or(bytes);
         validate_datagram_length(expected_bytes, bytes)?;
+        self.drain_tx_timestamps()?;
 
         events.push(echo_sent_event(
             remote,
@@ -610,6 +625,7 @@ impl Client {
             let outcome = self.runtime.commit_open(machine);
             self.schedule = schedule;
             self.applied_traffic_class = Some(negotiated_traffic_class);
+            self.tx_timestamping_enabled = try_enable_tx_timestamping(&self.socket);
             Ok(outcome)
         } else {
             debug_assert!(schedule.is_none());
@@ -619,10 +635,28 @@ impl Client {
             let outcome = self.runtime.commit_open(machine);
             self.schedule = None;
             self.applied_traffic_class = None;
+            self.tx_timestamping_enabled = try_enable_tx_timestamping(&self.socket);
             Ok(outcome)
         }
     }
 
+    /// Best-effort, bounded, nonblocking drain of the socket's
+    /// `MSG_ERRQUEUE`. A no-op unless [`Self::apply_prepared_open`]
+    /// successfully upgraded the socket to TX timestamping. Any TX
+    /// timestamp found is handed to the session machine to attach to its
+    /// matching probe; a genuine socket/network error found on the queue is
+    /// surfaced through the normal socket-error path.
+    fn drain_tx_timestamps(&mut self) -> Result<(), ClientError> {
+        if !self.tx_timestamping_enabled {
+            return Ok(());
+        }
+        let runtime = &mut self.runtime;
+        drain_tx_timestamps(&self.socket, |id, timestamp| {
+            runtime.record_kernel_tx_timestamp(id, timestamp);
+        })
+        .map_err(ClientError::Socket)
+    }
+
     fn restore_dscp_best_effort(&self, previous_traffic_class: Option<u8>) {
         #[cfg(test)]
         if self.test_hooks.fail_dscp_restore.replace(false) {

crates/irtt-client/src/managed/tests.rs

@@ -354,6 +354,7 @@ fn timeout_probe(seq: u32, sent_at: Instant, timeout_at: Instant) -> PendingProb
             wall: SystemTime::UNIX_EPOCH,
         },
         timeout_at,
+        kernel_tx_timestamp: None,
     }
 }
 

crates/irtt-client/src/probe.rs

@@ -1,6 +1,6 @@
 use std::{
     collections::{HashMap, HashSet, TryReserveError, VecDeque},
-    time::Instant,
+    time::{Instant, SystemTime},
 };
 
 use crate::{error::ClientError, timing::ClientTimestamp};
@@ -10,6 +10,12 @@ pub(crate) struct PendingProbe {
     pub wire_seq: u32,
     pub sent_at: ClientTimestamp,
     pub timeout_at: Instant,
+    /// Observed Linux kernel TX_SOFTWARE wall timestamp for this probe's
+    /// send, when the socket has TX timestamping enabled and a matching
+    /// `MSG_ERRQUEUE` record has been drained. Dormant metadata only: no
+    /// measurement in this change reads it. `sent_at` remains authoritative
+    /// for RTT, timeout, and current one-way delay.
+    pub kernel_tx_timestamp: Option<SystemTime>,
 }
 
 #[derive(Debug)]
@@ -158,6 +164,14 @@ impl PendingMap {
         self.map.len()
     }
 
+    /// Mutable access to a still-pending probe by wire sequence, without
+    /// disturbing its position in the timeout order. Used to attach an
+    /// observed kernel TX timestamp to a probe that has not yet completed or
+    /// timed out.
+    pub fn get_mut(&mut self, wire_seq: u32) -> Option<&mut PendingProbe> {
+        self.map.get_mut(&wire_seq).map(|entry| &mut entry.probe)
+    }
+
     #[cfg(any(feature = "tokio", test))]
     pub fn next_timeout_deadline(&self) -> Option<Instant> {
         self.first
@@ -266,6 +280,13 @@ impl TimedOutMap {
         removed
     }
 
+    /// Mutable access to a still-retained timed-out probe by wire sequence,
+    /// without disturbing eviction order. Used to attach an observed kernel
+    /// TX timestamp that arrives after the probe has already timed out.
+    pub fn get_mut(&mut self, wire_seq: u32) -> Option<&mut PendingProbe> {
+        self.map.get_mut(&wire_seq)
+    }
+
     pub fn clear(&mut self) {
         self.map.clear();
         self.insertion_order.clear();
@@ -367,6 +388,7 @@ mod tests {
             wire_seq: seq,
             sent_at: ts(timeout_at - Duration::from_secs(1)),
             timeout_at,
+            kernel_tx_timestamp: None,
         }
     }
 

crates/irtt-client/src/receive.rs

@@ -1,6 +1,7 @@
 use std::{
     io,
     net::{SocketAddr, UdpSocket},
+    time::SystemTime,
 };
 
 use crate::{metadata::ReceiveMeta, timing::ClientTimestamp};
@@ -77,6 +78,40 @@ pub(crate) fn configure_receive_metadata(socket: &UdpSocket, remote: SocketAddr)
     linux::configure_receive_metadata(socket, remote)
 }
 
+/// Best-effort upgrade a socket from RX-only to RX+TX kernel software
+/// timestamping. Returns whether the upgrade succeeded; a caller should
+/// track this to skip later drains entirely rather than perform harmless
+/// no-op ones. Always returns `false` off Linux or without the `ancillary`
+/// feature, so callers never need their own platform `cfg`.
+#[cfg(not(all(target_os = "linux", feature = "ancillary")))]
+pub(crate) fn try_enable_tx_timestamping<S>(_socket: &S) -> bool {
+    false
+}
+
+#[cfg(all(target_os = "linux", feature = "ancillary"))]
+pub(crate) fn try_enable_tx_timestamping<S: std::os::fd::AsFd>(socket: &S) -> bool {
+    linux::try_enable_tx_timestamping(socket)
+}
+
+/// Bounded, nonblocking drain of `socket`'s `MSG_ERRQUEUE`, reporting each
+/// usable TX timestamp completion to `on_timestamp`. Always a no-op off
+/// Linux or without the `ancillary` feature.
+#[cfg(not(all(target_os = "linux", feature = "ancillary")))]
+pub(crate) fn drain_tx_timestamps<S>(
+    _socket: &S,
+    _on_timestamp: impl FnMut(u32, SystemTime),
+) -> io::Result<()> {
+    Ok(())
+}
+
+#[cfg(all(target_os = "linux", feature = "ancillary"))]
+pub(crate) fn drain_tx_timestamps<S: std::os::fd::AsRawFd>(
+    socket: &S,
+    on_timestamp: impl FnMut(u32, SystemTime),
+) -> io::Result<()> {
+    linux::drain_tx_timestamps(socket, on_timestamp)
+}
+
 #[cfg(test)]
 mod tests {
     use std::net::UdpSocket;

crates/irtt-client/src/receive/linux.rs

@@ -1,25 +1,30 @@
-//! Linux ancillary receive metadata.
+//! Linux ancillary receive and transmit metadata.
 //!
 //! Receive kernel timing uses `SO_TIMESTAMPING` with `RX_SOFTWARE` +
 //! `SOFTWARE` (reported via `SCM_TIMESTAMPING`), not the older
 //! `SO_TIMESTAMPNS`/`SCM_TIMESTAMPNS`. Linux documents fallback quirks when
 //! `SO_TIMESTAMP[NS]` and `SO_TIMESTAMPING` are both enabled on one socket,
-//! so this establishes one unambiguous software-timestamp facility before
-//! TX-side timing is introduced in a follow-up change. `ReceiveMeta`'s
+//! so this establishes one unambiguous software-timestamp facility. `ReceiveMeta`'s
 //! semantics — an optional observed kernel receive wall time, used for
 //! downstream one-way delay only when plausible relative to the userspace
-//! receive sample — are unchanged by this.
+//! receive sample — are unchanged by any of this module's TX-side additions.
 //!
-//! `SOF_TIMESTAMPING_TX_SOFTWARE` is never requested on the production
-//! socket in [`configure_receive_metadata`]. TX flag construction,
-//! `MSG_ERRQUEUE` parsing, and extended-error classification exist as
-//! test-only primitives in [`error_queue`] for a follow-up change to build
-//! on; production does not generate or drain TX timestamp records yet.
+//! After a successful Open, the client adapter best-effort upgrades the
+//! socket from [`RX_TIMESTAMPING_FLAGS`] to [`error_queue::TX_TIMESTAMPING_FLAGS`]
+//! via [`try_enable_tx_timestamping`]. When that succeeds, the automatic
+//! `SOF_TIMESTAMPING_OPT_ID` counter the kernel assigns to each successfully
+//! submitted datagram is used directly as the probe's wire sequence number
+//! (see the client crate's `AGENTS.md` for the correlation invariant this
+//! relies on). [`drain_tx_timestamps`] performs a small bounded, nonblocking
+//! read of `MSG_ERRQUEUE` so the adapter can opportunistically collect those
+//! timestamps without ever waiting for one. The observed kernel TX wall time
+//! is retained as metadata only; it is not consumed by any measurement in
+//! this change.
 
 use std::{
     io::{self, IoSliceMut},
     net::{SocketAddr, UdpSocket},
-    os::fd::{AsRawFd, RawFd},
+    os::fd::{AsFd, AsRawFd, RawFd},
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
@@ -33,10 +38,74 @@ use nix::sys::{
 
 use crate::{metadata::ReceiveMeta, receive::ReceivedDatagram, timing::ClientTimestamp};
 
-#[cfg(test)]
-mod error_queue;
+pub(crate) mod error_queue;
 
-/// `SO_TIMESTAMPING` flags requested on the production receive socket.
+use error_queue::ErrorQueueRecord;
+
+/// Starvation guard on one opportunistic `MSG_ERRQUEUE` drain.
+///
+/// Each successfully submitted, TX-timestamped datagram generates at most
+/// one record, and drains happen at several natural choke points (after a
+/// probe send, after a normal receive, before timeout expiry), so this is
+/// not a queue-capacity promise: it just bounds the work one drain call can
+/// do. A handful of in-flight probes' worth of backlog is the realistic
+/// case between choke points, so a small constant comfortably covers it
+/// without any unbounded loop.
+pub(crate) const MAX_TX_TIMESTAMP_RECORDS_PER_DRAIN: usize = 32;
+
+/// Best-effort upgrade from RX-only to RX+TX `SO_TIMESTAMPING` flags.
+///
+/// Returns `true` when the upgrade succeeded and the caller may now expect
+/// `MSG_ERRQUEUE` TX timestamp records for subsequent sends. On failure, the
+/// original [`RX_TIMESTAMPING_FLAGS`] are best-effort restored so a failed
+/// upgrade cannot leave the socket without the RX timestamping this client
+/// already depends on; that restore's own failure is not observable here
+/// (both leave the caller without TX capability, which this return value
+/// already reports).
+pub(crate) fn try_enable_tx_timestamping<S: AsFd>(socket: &S) -> bool {
+    if setsockopt(
+        socket,
+        sockopt::Timestamping,
+        &error_queue::TX_TIMESTAMPING_FLAGS,
+    )
+    .is_ok()
+    {
+        return true;
+    }
+    let _ = setsockopt(socket, sockopt::Timestamping, &RX_TIMESTAMPING_FLAGS);
+    false
+}
+
+/// Bounded, nonblocking drain of `fd`'s `MSG_ERRQUEUE`.
+///
+/// Reads and classifies up to [`MAX_TX_TIMESTAMP_RECORDS_PER_DRAIN`] records,
+/// one at a time with no intermediate allocation. Each usable TX timestamp
+/// completion is reported to `on_timestamp` immediately. Malformed or
+/// unsupported timestamp notifications and unrelated records are silently
+/// dropped: timestamp metadata is optional and never fails a probe. A
+/// genuine non-timestamp socket/network error stops the drain and is
+/// returned to the caller, which decides how to surface it.
+pub(crate) fn drain_tx_timestamps<S: AsRawFd>(
+    socket: &S,
+    mut on_timestamp: impl FnMut(u32, SystemTime),
+) -> io::Result<()> {
+    let fd = socket.as_raw_fd();
+    for _ in 0..MAX_TX_TIMESTAMP_RECORDS_PER_DRAIN {
+        match error_queue::try_recv_error_queue_record(fd)? {
+            None => return Ok(()),
+            Some(ErrorQueueRecord::TxTimestamp { id, timestamp }) => on_timestamp(id, timestamp),
+            Some(ErrorQueueRecord::MalformedOrUnsupportedTimestamp | ErrorQueueRecord::Ignored) => {
+            }
+            Some(ErrorQueueRecord::SocketError { errno, .. }) => {
+                return Err(io::Error::from_raw_os_error(errno));
+            }
+        }
+    }
+    Ok(())
+}
+
+/// `SO_TIMESTAMPING` flags requested on every production socket at
+/// connect time, before any TX upgrade is attempted.
 ///
 /// `SOF_TIMESTAMPING_RX_SOFTWARE` asks the kernel to generate a software
 /// receive timestamp for each datagram; `SOF_TIMESTAMPING_SOFTWARE` asks it
@@ -44,7 +113,8 @@ mod error_queue;
 /// the older `SO_TIMESTAMP`/`SO_TIMESTAMPNS` on the same socket has
 /// documented fallback quirks, so the client uses `SO_TIMESTAMPING`
 /// exclusively for receive timing. Hardware and TX-side flags are
-/// deliberately excluded: this socket is receive-only.
+/// deliberately excluded here: a fresh socket is receive-only until (and
+/// unless) [`try_enable_tx_timestamping`] later upgrades it.
 const RX_TIMESTAMPING_FLAGS: TimestampingFlag = TimestampingFlag::SOF_TIMESTAMPING_RX_SOFTWARE
     .union(TimestampingFlag::SOF_TIMESTAMPING_SOFTWARE);
 
@@ -430,4 +500,55 @@ mod tests {
             io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported
         )
     }
+
+    #[test]
+    fn try_enable_tx_timestamping_then_drain_reports_ids_through_the_adapter_api() {
+        let (socket, _peer) = connected_ipv4_loopback_pair();
+        // Mirrors production: RX-only first, then a best-effort upgrade.
+        configure_receive_metadata(&socket, socket.peer_addr().unwrap()).unwrap();
+
+        if !super::try_enable_tx_timestamping(&socket) {
+            eprintln!(
+                "skipping adapter-level TX timestamping smoke test: kernel/container denied \
+                 SO_TIMESTAMPING TX flags"
+            );
+            return;
+        }
+
+        if let Err(error) = socket.send(b"one") {
+            eprintln!("skipping adapter-level TX timestamping smoke test: send failed: {error}");
+            return;
+        }
+
+        let deadline = std::time::Instant::now() + Duration::from_secs(2);
+        let mut observed = None;
+        while observed.is_none() && std::time::Instant::now() < deadline {
+            super::drain_tx_timestamps(&socket, |id, timestamp| {
+                observed = Some((id, timestamp));
+            })
+            .unwrap();
+            std::thread::yield_now();
+        }
+
+        let Some((id, timestamp)) = observed else {
+            eprintln!(
+                "skipping adapter-level TX timestamping smoke test: no TX timestamp record \
+                 arrived within the poll deadline"
+            );
+            return;
+        };
+        assert_eq!(id, 0);
+        assert!(timestamp.duration_since(UNIX_EPOCH).unwrap() > Duration::ZERO);
+    }
+
+    #[test]
+    fn drain_tx_timestamps_is_a_bounded_nonblocking_no_op_on_an_untouched_socket() {
+        let (socket, _peer) = connected_ipv4_loopback_pair();
+        configure_receive_metadata(&socket, socket.peer_addr().unwrap()).unwrap();
+
+        let mut calls = 0_usize;
+        super::drain_tx_timestamps(&socket, |_, _| calls += 1).unwrap();
+
+        assert_eq!(calls, 0);
+    }
 }

crates/irtt-client/src/receive/linux/error_queue.rs

@@ -1,11 +1,12 @@
-//! Safe `MSG_ERRQUEUE` primitives for the future client TX timestamp work
-//! (see the parent module's TX-foundation note).
+//! Safe `MSG_ERRQUEUE` primitives backing the client's TX timestamp capture.
 //!
-//! Test-only for this PR: production sockets never enable
-//! `SOF_TIMESTAMPING_TX_SOFTWARE` and never drain the error queue, so
-//! nothing here runs outside `#[cfg(test)]`. It exists so the follow-up PR
-//! that wires TX timestamps into probes does not need to rediscover flag
-//! construction, error-queue parsing, or extended-error classification.
+//! A production socket that successfully upgraded to
+//! [`TX_TIMESTAMPING_FLAGS`] receives one extended-error record per
+//! successfully submitted timestamped datagram on `MSG_ERRQUEUE`.
+//! [`try_recv_error_queue_record`] performs one nonblocking read of that
+//! queue and [`classify`] turns its ancillary data into an
+//! [`ErrorQueueRecord`] without ever panicking or requiring the caller to
+//! understand cmsg layout.
 
 use std::{
     io::{self, IoSliceMut},
@@ -22,13 +23,11 @@ use nix::{
 
 use super::system_time_from_timespec;
 
-/// Desired `SO_TIMESTAMPING` configuration for a future TX-timestamped
-/// socket. Combines this PR's RX flags with the TX-side flags the next PR
-/// needs: `TX_SOFTWARE` to generate a send timestamp, `OPT_ID` for automatic
-/// per-datagram correlation, and `OPT_TSONLY` so the notification does not
-/// need to retain or copy the original payload.
-///
-/// Not applied to any production socket in this PR.
+/// `SO_TIMESTAMPING` configuration for a TX-timestamped socket. Combines the
+/// module's RX flags with the TX-side flags TX capture needs: `TX_SOFTWARE`
+/// to generate a send timestamp, `OPT_ID` for automatic per-datagram
+/// correlation, and `OPT_TSONLY` so the notification does not need to
+/// retain or copy the original payload.
 pub(crate) const TX_TIMESTAMPING_FLAGS: TimestampingFlag =
     TimestampingFlag::SOF_TIMESTAMPING_RX_SOFTWARE
         .union(TimestampingFlag::SOF_TIMESTAMPING_SOFTWARE)
@@ -70,14 +69,18 @@ pub(crate) enum ErrorQueueRecord {
     /// `SO_EE_ORIGIN_TIMESTAMPING`, its kind was `SCM_TSTAMP_SND`, and a
     /// usable software timestamp accompanied it.
     TxTimestamp { id: u32, timestamp: SystemTime },
-    /// A timestamp-origin extended error was present, but no usable
-    /// timestamp accompanied it: either the `SCM_TIMESTAMPING` cmsg was
-    /// missing, or its value failed to convert to a `SystemTime`. An
-    /// accuracy failure, not a socket failure.
-    MalformedTimestamp,
-    /// An extended error that is not a timestamp completion: a real
-    /// network, local, or ICMP error surfaced on the error queue. Must not
-    /// be misclassified as malformed timing metadata.
+    /// The extended error's origin was `SO_EE_ORIGIN_TIMESTAMPING` (i.e. this
+    /// is unambiguously a timestamp facility record, never a real
+    /// socket/network error), but it was not a usable send-timestamp
+    /// completion: the completion kind was not `SCM_TSTAMP_SND` (this client
+    /// requests only `TX_SOFTWARE`, so only `SND` is expected, but an
+    /// unrequested kind must not be mistaken for a socket failure), the
+    /// `SCM_TIMESTAMPING` cmsg was missing, or its value failed to convert
+    /// to a `SystemTime`. An accuracy failure, not a socket failure.
+    MalformedOrUnsupportedTimestamp,
+    /// An extended error whose origin was not `SO_EE_ORIGIN_TIMESTAMPING`: a
+    /// real network, local, or ICMP error surfaced on the error queue. Must
+    /// not be misclassified as malformed timing metadata.
     SocketError { errno: i32, origin: u8 },
     /// No extended error was present in this record's control data (e.g.
     /// only unrelated/unknown cmsgs).
@@ -87,10 +90,13 @@ pub(crate) enum ErrorQueueRecord {
 /// Classify a decoded set of cmsgs from one `MSG_ERRQUEUE` datagram.
 ///
 /// Collects the extended error and any accompanying timestamp
-/// independently before classifying, so cmsg order never matters. A
-/// timestamp is only ever attached to a timestamp-origin extended error;
-/// an ID is never reported without a timestamp, and a timestamp is never
-/// accepted without a valid timestamp-origin extended error.
+/// independently before classifying, so cmsg order never matters. Origin is
+/// the sole discriminator between a timestamp facility record and a real
+/// socket error: any `SO_EE_ORIGIN_TIMESTAMPING` record is timing metadata,
+/// however unexpected its `errno`/`ee_info`, and only ever becomes
+/// [`ErrorQueueRecord::MalformedOrUnsupportedTimestamp`] on the way to
+/// `None` — never [`ErrorQueueRecord::SocketError`]. A timestamp is only
+/// ever attached to a timestamp-origin record reporting `SCM_TSTAMP_SND`.
 pub(crate) fn classify(cmsgs: impl Iterator<Item = ControlMessageOwned>) -> ErrorQueueRecord {
     let mut extended_error: Option<(u32, u8, u32, u32)> = None;
     let mut timestamp: Option<Timestamps> = None;
@@ -114,20 +120,21 @@ pub(crate) fn classify(cmsgs: impl Iterator<Item = ControlMessageOwned>) -> Erro
         return ErrorQueueRecord::Ignored;
     };
 
-    let is_timestamp_origin = origin == nix::libc::SO_EE_ORIGIN_TIMESTAMPING
-        && errno == nix::libc::ENOMSG as u32
-        && info == SCM_TSTAMP_SND;
-
-    if !is_timestamp_origin {
+    if origin != nix::libc::SO_EE_ORIGIN_TIMESTAMPING {
         return ErrorQueueRecord::SocketError {
             errno: errno as i32,
             origin,
         };
     }
 
+    let is_send_completion = errno == nix::libc::ENOMSG as u32 && info == SCM_TSTAMP_SND;
+    if !is_send_completion {
+        return ErrorQueueRecord::MalformedOrUnsupportedTimestamp;
+    }
+
     match timestamp.and_then(|observed| system_time_from_timespec(observed.system)) {
         Some(timestamp) => ErrorQueueRecord::TxTimestamp { id, timestamp },
-        None => ErrorQueueRecord::MalformedTimestamp,
+        None => ErrorQueueRecord::MalformedOrUnsupportedTimestamp,
     }
 }
 
@@ -248,7 +255,7 @@ mod tests {
 
         assert_eq!(
             classify(cmsgs.into_iter()),
-            ErrorQueueRecord::MalformedTimestamp
+            ErrorQueueRecord::MalformedOrUnsupportedTimestamp
         );
     }
 
@@ -261,7 +268,7 @@ mod tests {
 
         assert_eq!(
             classify(cmsgs.into_iter()),
-            ErrorQueueRecord::MalformedTimestamp
+            ErrorQueueRecord::MalformedOrUnsupportedTimestamp
         );
     }
 
@@ -287,10 +294,12 @@ mod tests {
     }
 
     #[test]
-    fn classifies_timestamping_origin_with_wrong_completion_kind_as_socket_error() {
+    fn classifies_timestamping_origin_with_wrong_completion_kind_as_malformed() {
         // Same origin byte as a send-timestamp completion, but a different
-        // `ee_info` (e.g. SCM_TSTAMP_SCHED/ACK) must not be mistaken for
-        // SCM_TSTAMP_SND.
+        // `ee_info` (e.g. SCM_TSTAMP_SCHED/ACK, which this client never
+        // requests). A timestamping-origin record must never become a
+        // fatal SocketError merely because it reports a completion kind we
+        // did not ask for; it stays timing metadata.
         let cmsgs = vec![ControlMessageOwned::Ipv4RecvErr(
             extended_error(
                 nix::libc::ENOMSG as u32,
@@ -303,10 +312,29 @@ mod tests {
 
         assert_eq!(
             classify(cmsgs.into_iter()),
-            ErrorQueueRecord::SocketError {
-                errno: nix::libc::ENOMSG,
-                origin: nix::libc::SO_EE_ORIGIN_TIMESTAMPING,
-            }
+            ErrorQueueRecord::MalformedOrUnsupportedTimestamp
+        );
+    }
+
+    #[test]
+    fn classifies_timestamping_origin_with_unexpected_errno_as_malformed() {
+        // Origin is the sole discriminator: even an origin==TIMESTAMPING
+        // record with a surprising errno (the kernel always reports ENOMSG
+        // for this origin, but classification must not assume that) is
+        // still timing metadata, never a real socket error.
+        let cmsgs = vec![ControlMessageOwned::Ipv4RecvErr(
+            extended_error(
+                nix::libc::ECONNREFUSED as u32,
+                nix::libc::SO_EE_ORIGIN_TIMESTAMPING,
+                SCM_TSTAMP_SND,
+                0,
+            ),
+            None,
+        )];
+
+        assert_eq!(
+            classify(cmsgs.into_iter()),
+            ErrorQueueRecord::MalformedOrUnsupportedTimestamp
         );
     }
 
@@ -436,15 +464,170 @@ mod tests {
         );
     }
 
-    // NOTE: a deterministic test proving that a failed nonblocking `sendmsg`
-    // does not consume an automatic OPT_ID is not included here. Forcing a
-    // reliable, non-flaky `WouldBlock` on a connected loopback UDP socket
-    // (e.g. via socket-buffer saturation) has no deterministic OS-level
-    // mechanism available in this environment, and the repository's testing
-    // policy explicitly forbids faking coverage with sleep/saturation-based
-    // flakiness. This remains an open assumption for the follow-up PR that
-    // wires automatic ID correlation into production: it must verify, by
-    // construction or by a reliable test double, that a `WouldBlock` send
-    // never advances the ID the kernel will assign to the next
-    // successfully transmitted timestamped datagram.
+    #[test]
+    fn rx_only_sends_before_tx_enable_do_not_consume_the_later_tx_id_namespace() {
+        let socket = UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap();
+        let peer = UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap();
+        socket.connect(peer.local_addr().unwrap()).unwrap();
+        setsockopt(
+            &socket,
+            sockopt::Timestamping,
+            &super::super::RX_TIMESTAMPING_FLAGS,
+        )
+        .unwrap();
+
+        // Ordinary pre-enable sends: TX_SOFTWARE is not yet requested, so
+        // none of these may leave anything on the error queue for the
+        // later-enabled OPT_ID counter to have skipped past.
+        for _ in 0..3 {
+            if let Err(error) = socket.send(b"pre-enable") {
+                eprintln!("skipping pre-enable send/RX-only smoke test: {error}");
+                return;
+            }
+        }
+        assert_eq!(
+            try_recv_error_queue_record(socket.as_raw_fd()).unwrap(),
+            None,
+            "RX-only sends must never populate MSG_ERRQUEUE"
+        );
+
+        // This mirrors production: RX-only is configured first (e.g. at
+        // connect time), and only later, after a successful Open, is the
+        // socket upgraded to TX_TIMESTAMPING_FLAGS.
+        if setsockopt(&socket, sockopt::Timestamping, &TX_TIMESTAMPING_FLAGS).is_err() {
+            eprintln!(
+                "skipping pre-enable send/RX-only smoke test: kernel/container denied \
+                 SO_TIMESTAMPING TX flags"
+            );
+            return;
+        }
+
+        if let Err(error) = socket.send(b"first-timestamped") {
+            eprintln!(
+                "skipping pre-enable send/RX-only smoke test: post-upgrade send failed: {error}"
+            );
+            return;
+        }
+        let Some(record) = poll_for_record(socket.as_raw_fd(), Duration::from_secs(2)) else {
+            eprintln!(
+                "skipping pre-enable send/RX-only smoke test: no TX timestamp record was \
+                 delivered within the poll deadline"
+            );
+            return;
+        };
+        assert!(
+            matches!(record, ErrorQueueRecord::TxTimestamp { id: 0, .. }),
+            "expected the first send after upgrade to report id 0 (pre-enable sends must not \
+             have consumed part of the namespace), got {record:?}"
+        );
+    }
+
+    #[test]
+    fn normal_rx_scm_timestamping_still_works_after_tx_flags_are_enabled() {
+        let Some((socket, peer)) = tx_timestamped_loopback_socket() else {
+            return;
+        };
+        peer.send_to(b"reply", socket.local_addr().unwrap())
+            .unwrap();
+
+        let mut buf = [0_u8; 16];
+        let Ok(datagram) = super::super::recv_datagram(&socket, &mut buf) else {
+            eprintln!("skipping post-TX-enable RX smoke test: receive failed");
+            return;
+        };
+        assert_eq!(&buf[..datagram.len], b"reply");
+        let Some(timestamp) = datagram.meta.kernel_rx_timestamp else {
+            eprintln!(
+                "skipping post-TX-enable RX smoke test: kernel did not provide SCM_TIMESTAMPING"
+            );
+            return;
+        };
+        assert!(timestamp.duration_since(std::time::UNIX_EPOCH).unwrap() > Duration::ZERO);
+    }
+
+    #[test]
+    fn tx_timestamp_and_normal_reply_streams_do_not_interfere() {
+        let Some((socket, peer)) = tx_timestamped_loopback_socket() else {
+            return;
+        };
+        let fd = socket.as_raw_fd();
+
+        if let Err(error) = socket.send(b"ping") {
+            eprintln!("skipping TX/RX separation smoke test: ping send failed: {error}");
+            return;
+        }
+        let mut incoming = [0_u8; 16];
+        let Ok((len, from)) = peer.recv_from(&mut incoming) else {
+            eprintln!("skipping TX/RX separation smoke test: peer never observed ping");
+            return;
+        };
+        assert_eq!(&incoming[..len], b"ping");
+        peer.send_to(b"pong", from).unwrap();
+
+        // Reading the normal payload must not require, wait for, or be
+        // disturbed by draining MSG_ERRQUEUE first.
+        let mut reply_buf = [0_u8; 16];
+        let Ok(datagram) = super::super::recv_datagram(&socket, &mut reply_buf) else {
+            eprintln!("skipping TX/RX separation smoke test: normal receive failed");
+            return;
+        };
+        assert_eq!(&reply_buf[..datagram.len], b"pong");
+
+        // The TX timestamp for "ping" is still independently available on
+        // the error queue, unconsumed by the normal receive above.
+        let Some(record) = poll_for_record(fd, Duration::from_secs(2)) else {
+            eprintln!("skipping TX/RX separation smoke test: no TX timestamp record arrived");
+            return;
+        };
+        assert!(matches!(
+            record,
+            ErrorQueueRecord::TxTimestamp { id: 0, .. }
+        ));
+    }
+
+    #[test]
+    fn tx_timestamp_correlation_does_not_depend_on_original_payload() {
+        let Some((socket, _peer)) = tx_timestamped_loopback_socket() else {
+            return;
+        };
+
+        // OPT_TSONLY notifications never carry the original payload; a
+        // larger payload must classify identically to the zero-length one
+        // used elsewhere, since correlation is by `id` alone.
+        let payload = vec![0x5A_u8; 512];
+        if let Err(error) = socket.send(&payload) {
+            eprintln!("skipping OPT_TSONLY payload-independence smoke test: send failed: {error}");
+            return;
+        }
+        let Some(record) = poll_for_record(socket.as_raw_fd(), Duration::from_secs(2)) else {
+            eprintln!(
+                "skipping OPT_TSONLY payload-independence smoke test: no TX timestamp record \
+                 arrived"
+            );
+            return;
+        };
+        assert!(matches!(
+            record,
+            ErrorQueueRecord::TxTimestamp { id: 0, .. }
+        ));
+    }
+
+    // NOTE: a live test proving that a failed nonblocking `sendmsg` does not
+    // consume an automatic OPT_ID is not included here. Forcing a reliable,
+    // non-flaky `WouldBlock` on a connected loopback UDP socket (e.g. via
+    // socket-buffer saturation) has no deterministic OS-level mechanism
+    // available in this environment, and the repository's testing policy
+    // explicitly forbids faking coverage with sleep/saturation-based
+    // flakiness. That safety is instead established from Linux kernel
+    // source: `_sock_tx_timestamp` (include/net/sock.h) and
+    // `__ip_append_data`/`__ip6_append_data` (net/ipv4/ip_output.c,
+    // net/ipv6/ip6_output.c) increment `sk->sk_tskey` speculatively before
+    // a datagram is fully built and explicitly `atomic_dec` it back out on
+    // every error path (tracked via a local `hold_tskey` flag), so a failed
+    // send can never leave a gap in — or otherwise advance — the ID space
+    // the kernel hands out to successfully submitted datagrams. See the
+    // client crate's `AGENTS.md` for the full citation. The client-side
+    // half of this invariant (a `WouldBlock` `try_send` never advances the
+    // local wire-sequence counter used as the correlation ID) is covered by
+    // a deterministic test in the adapter module that injects `WouldBlock`.
 }

crates/irtt-client/src/session/machine.rs

@@ -366,6 +366,7 @@ impl SessionMachine {
                 wire_seq: preflight.seq,
                 sent_at,
                 timeout_at,
+                kernel_tx_timestamp: None,
             },
             next_wire_seq: preflight.next_wire_seq,
             next_packets_sent: preflight.next_packets_sent,
@@ -392,6 +393,35 @@ impl SessionMachine {
         }
     }
 
+    /// Record an observed Linux kernel TX timestamp for `wire_seq`, the
+    /// automatic `SOF_TIMESTAMPING_OPT_ID` the kernel assigned when the
+    /// datagram was submitted.
+    ///
+    /// `wire_seq` doubles as that correlation ID: both counters start at
+    /// zero for a session's first timestamped send and advance by exactly
+    /// one only on a confirmed successful submission, so they stay
+    /// identical for the life of one open session (see the client crate's
+    /// `AGENTS.md` for the full invariant).
+    ///
+    /// Updates a still-pending or already-timed-out probe in place. Never
+    /// resurrects a completed, evicted, or unknown probe, and never touches
+    /// `sent_at`, timeout/loss/reply state, or any counter — this is purely
+    /// dormant metadata attachment. A probe that already has a timestamp
+    /// keeps it: first usable observation wins, so a later duplicate cannot
+    /// silently overwrite it.
+    pub(crate) fn record_kernel_tx_timestamp(&mut self, wire_seq: u32, timestamp: SystemTime) {
+        let MachineState::Open(session) = &mut self.state else {
+            return;
+        };
+        if let Some(probe) = session.pending.get_mut(wire_seq) {
+            probe.kernel_tx_timestamp.get_or_insert(timestamp);
+            return;
+        }
+        if let Some(probe) = session.timed_out.get_mut(wire_seq) {
+            probe.kernel_tx_timestamp.get_or_insert(timestamp);
+        }
+    }
+
     pub(crate) fn process_received_echo_packet(
         &mut self,
         packet: &[u8],
@@ -1053,6 +1083,7 @@ impl SessionMachine {
             wire_seq: 0,
             sent_at: now,
             timeout_at: now.mono,
+            kernel_tx_timestamp: None,
         });
         session.completed.insert(0);
     }
@@ -1789,6 +1820,7 @@ mod tests {
                 wire_seq: seq,
                 sent_at,
                 timeout_at: sent_at.mono + Duration::from_secs(1),
+                kernel_tx_timestamp: None,
             });
         }
 
@@ -1828,6 +1860,7 @@ mod tests {
             wire_seq: 0,
             sent_at: timestamp(now - Duration::from_secs(1)),
             timeout_at: now,
+            kernel_tx_timestamp: None,
         };
         let session = active_mut(&mut machine);
         session.next_wire_seq = 0;
@@ -1916,6 +1949,7 @@ mod tests {
             wire_seq: 0,
             sent_at,
             timeout_at: sent_at.mono + Duration::from_secs(1),
+            kernel_tx_timestamp: None,
         });
         session.next_wire_seq = 1;
         machine
@@ -1930,6 +1964,7 @@ mod tests {
             wire_seq: 0,
             sent_at,
             timeout_at: sent_at.mono + Duration::from_secs(1),
+            kernel_tx_timestamp: None,
         });
         session.next_wire_seq = 1;
         machine
@@ -2217,4 +2252,312 @@ mod tests {
             Some(SignedDuration::from_nanos(15_000_000))
         );
     }
+
+    // Kernel TX timestamp correlation (`record_kernel_tx_timestamp`).
+    //
+    // These probe `SessionMachine`'s private association state directly.
+    // The kernel TX timestamp is dormant metadata in this change: none of
+    // these tests assert anything about RTT/OWD/IPDV output, only about
+    // where `PendingProbe::kernel_tx_timestamp` ends up.
+
+    fn tx_ts(offset_secs: u64) -> SystemTime {
+        UNIX_EPOCH + Duration::from_secs(offset_secs)
+    }
+
+    fn send_probe(machine: &mut SessionMachine, sent_at: ClientTimestamp) -> u32 {
+        let prepared = machine.prepare_probe().unwrap().unwrap();
+        let preflight = machine.preflight_probe_commit(&prepared).unwrap();
+        let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap();
+        machine.commit_probe_sent(commit, prepared.bytes.len()).seq
+    }
+
+    #[test]
+    fn kernel_tx_timestamp_attaches_to_pending_probe() {
+        let mut machine = open_machine(4, Duration::from_secs(1));
+        let now = Instant::now();
+        let seq = send_probe(&mut machine, timestamp(now));
+
+        machine.record_kernel_tx_timestamp(seq, tx_ts(1));
+
+        assert_eq!(
+            active_mut(&mut machine)
+                .pending
+                .get_mut(seq)
+                .unwrap()
+                .kernel_tx_timestamp,
+            Some(tx_ts(1))
+        );
+    }
+
+    #[test]
+    fn kernel_tx_timestamp_for_unknown_id_does_nothing() {
+        let mut machine = open_machine(4, Duration::from_secs(1));
+        let now = Instant::now();
+        let seq = send_probe(&mut machine, timestamp(now));
+
+        // No probe was ever sent for sequence 41; recording against it must
+        // not panic, allocate a phantom entry, or affect the real probe.
+        machine.record_kernel_tx_timestamp(41, tx_ts(9));
+
+        assert!(active_mut(&mut machine)
+            .pending
+            .get_mut(seq)
+            .unwrap()
+            .kernel_tx_timestamp
+            .is_none());
+        assert_eq!(active(&machine).pending.len(), 1);
+    }
+
+    #[test]
+    fn kernel_tx_timestamps_associate_correctly_when_observed_out_of_order() {
+        let mut machine = open_machine(4, Duration::from_secs(1));
+        let now = Instant::now();
+        let seq10 = send_probe(&mut machine, timestamp(now));
+        let seq11 = send_probe(&mut machine, timestamp(now));
+        let seq12 = send_probe(&mut machine, timestamp(now));
+        assert_eq!((seq10, seq11, seq12), (0, 1, 2));
+
+        // Linux does not guarantee MSG_ERRQUEUE dequeue order matches send
+        // order; observe them out of order (12, 10, 11).
+        machine.record_kernel_tx_timestamp(seq12, tx_ts(12));
+        machine.record_kernel_tx_timestamp(seq10, tx_ts(10));
+        machine.record_kernel_tx_timestamp(seq11, tx_ts(11));
+
+        let session = active_mut(&mut machine);
+        assert_eq!(
+            session.pending.get_mut(seq10).unwrap().kernel_tx_timestamp,
+            Some(tx_ts(10))
+        );
+        assert_eq!(
+            session.pending.get_mut(seq11).unwrap().kernel_tx_timestamp,
+            Some(tx_ts(11))
+        );
+        assert_eq!(
+            session.pending.get_mut(seq12).unwrap().kernel_tx_timestamp,
+            Some(tx_ts(12))
+        );
+    }
+
+    #[test]
+    fn first_valid_kernel_tx_timestamp_wins_over_a_later_duplicate() {
+        let mut machine = open_machine(4, Duration::from_secs(1));
+        let now = Instant::now();
+        let seq = send_probe(&mut machine, timestamp(now));
+
+        machine.record_kernel_tx_timestamp(seq, tx_ts(1));
+        machine.record_kernel_tx_timestamp(seq, tx_ts(2));
+
+        assert_eq!(
+            active_mut(&mut machine)
+                .pending
+                .get_mut(seq)
+                .unwrap()
+                .kernel_tx_timestamp,
+            Some(tx_ts(1))
+        );
+    }
+
+    #[test]
+    fn probe_timeout_preserves_an_existing_kernel_tx_timestamp() {
+        let mut machine = open_machine(4, Duration::from_secs(1));
+        let now = Instant::now();
+        let seq = send_probe(&mut machine, timestamp(now));
+        machine.record_kernel_tx_timestamp(seq, tx_ts(1));
+
+        let events = machine
+            .poll_timeouts_at(now + Duration::from_secs(2))
+            .unwrap();
+        assert!(matches!(events.as_slice(), [ClientEvent::EchoLoss { .. }]));
+
+        let session = active_mut(&mut machine);
+        assert!(!session.pending.contains(seq));
+        assert_eq!(
+            session.timed_out.get_mut(seq).unwrap().kernel_tx_timestamp,
+            Some(tx_ts(1))
+        );
+    }
+
+    #[test]
+    fn kernel_tx_timestamp_arriving_after_timeout_attaches_to_timed_out_probe() {
+        let mut machine = open_machine(4, Duration::from_secs(1));
+        let now = Instant::now();
+        let seq = send_probe(&mut machine, timestamp(now));
+
+        machine
+            .poll_timeouts_at(now + Duration::from_secs(2))
+            .unwrap();
+        assert!(active(&machine).timed_out.contains(seq));
+
+        machine.record_kernel_tx_timestamp(seq, tx_ts(3));
+
+        assert_eq!(
+            active_mut(&mut machine)
+                .timed_out
+                .get_mut(seq)
+                .unwrap()
+                .kernel_tx_timestamp,
+            Some(tx_ts(3))
+        );
+    }
+
+    #[test]
+    fn measurable_late_reply_retains_the_kernel_tx_timestamp() {
+        let now = Instant::now();
+        let mut machine = machine_with_timed_out_probe(owd_sent_at(now));
+        machine.record_kernel_tx_timestamp(0, tx_ts(4));
+        assert_eq!(
+            active_mut(&mut machine)
+                .timed_out
+                .get_mut(0)
+                .unwrap()
+                .kernel_tx_timestamp,
+            Some(tx_ts(4))
+        );
+
+        // The reply is measurable (a late reply with a retained sent_at);
+        // OWD/RTT still come from sent_at only, per NO_OWD_CHANGE, but the
+        // dormant kernel timestamp must have already been retained above
+        // and this call must not disturb that.
+        let events = process_owd_reply(
+            &mut machine,
+            owd_timestamps(),
+            kernel_rx_meta(25_000_000),
+            owd_received_at(now),
+        );
+        assert!(matches!(events.as_slice(), [ClientEvent::LateReply { .. }]));
+    }
+
+    #[test]
+    fn completed_probe_ignores_a_later_kernel_tx_timestamp() {
+        let mut machine = open_machine(4, Duration::from_secs(1));
+        let now = Instant::now();
+        let seq = send_probe(&mut machine, timestamp(now));
+        let reply = EchoReply {
+            flags: 0,
+            token: active(&machine).token,
+            sequence: seq,
+            recv_count: None,
+            recv_window: None,
+            timestamps: TimestampFields::default(),
+            payload: Vec::new(),
+        };
+        machine
+            .process_echo_reply(
+                reply,
+                64,
+                ClientTimestamp {
+                    mono: now,
+                    wall: SystemTime::now(),
+                },
+                ReceiveMeta::default(),
+            )
+            .unwrap();
+        assert!(active(&machine).completed.contains(seq));
+        assert!(!active(&machine).pending.contains(seq));
+
+        // Must not resurrect the probe into either map.
+        machine.record_kernel_tx_timestamp(seq, tx_ts(5));
+
+        let session = active_mut(&mut machine);
+        assert!(session.pending.get_mut(seq).is_none());
+        assert!(session.timed_out.get_mut(seq).is_none());
+    }
+
+    #[test]
+    fn evicted_timed_out_probe_ignores_a_later_kernel_tx_timestamp() {
+        let mut machine = open_machine(1, Duration::from_secs(1));
+        let now = Instant::now();
+        let first = send_probe(&mut machine, timestamp(now));
+        machine
+            .poll_timeouts_at(now + Duration::from_secs(2))
+            .unwrap();
+        assert!(active(&machine).timed_out.contains(first));
+
+        // capacity 1: sending and timing out a second probe evicts the first
+        // from the bounded TimedOutMap.
+        let second = send_probe(&mut machine, timestamp(now + Duration::from_secs(2)));
+        machine
+            .poll_timeouts_at(now + Duration::from_secs(4))
+            .unwrap();
+        assert!(active(&machine).timed_out.contains(second));
+        assert!(!active(&machine).timed_out.contains(first));
+
+        // Recording against the evicted ID must not panic or resurrect it.
+        machine.record_kernel_tx_timestamp(first, tx_ts(6));
+        assert!(active_mut(&mut machine).timed_out.get_mut(first).is_none());
+    }
+
+    #[test]
+    fn kernel_tx_timestamp_recording_is_a_no_op_once_closed() {
+        let mut machine = open_machine(4, Duration::from_secs(1));
+        let now = Instant::now();
+        let seq = send_probe(&mut machine, timestamp(now));
+        let prepared = machine.prepare_close().unwrap();
+        let commit = prepared.commit;
+        machine.commit_local_close(commit, timestamp(now));
+        assert!(machine.is_terminal());
+
+        // Recording after close must not panic (there is no open session to
+        // index into) and must remain inert.
+        machine.record_kernel_tx_timestamp(seq, tx_ts(7));
+    }
+
+    #[test]
+    fn kernel_tx_timestamp_ids_wrap_like_wire_seq() {
+        let mut machine = open_machine(4, Duration::from_secs(1));
+        active_mut(&mut machine).next_wire_seq = u32::MAX;
+        let now = Instant::now();
+        let last = send_probe(&mut machine, timestamp(now));
+        let wrapped = send_probe(&mut machine, timestamp(now));
+        assert_eq!(last, u32::MAX);
+        assert_eq!(wrapped, 0);
+
+        machine.record_kernel_tx_timestamp(u32::MAX, tx_ts(1));
+        machine.record_kernel_tx_timestamp(0, tx_ts(2));
+
+        let session = active_mut(&mut machine);
+        assert_eq!(
+            session
+                .pending
+                .get_mut(u32::MAX)
+                .unwrap()
+                .kernel_tx_timestamp,
+            Some(tx_ts(1))
+        );
+        assert_eq!(
+            session.pending.get_mut(0).unwrap().kernel_tx_timestamp,
+            Some(tx_ts(2))
+        );
+    }
+
+    #[test]
+    fn kernel_tx_timestamp_presence_does_not_change_rtt_or_owd() {
+        let mono = Instant::now();
+        let mut without = machine_with_pending_probe(owd_sent_at(mono));
+        let mut with = machine_with_pending_probe(owd_sent_at(mono));
+        with.record_kernel_tx_timestamp(0, tx_ts(1));
+
+        let without_events = process_owd_reply(
+            &mut without,
+            owd_timestamps(),
+            kernel_rx_meta(25_000_000),
+            owd_received_at(mono),
+        );
+        let with_events = process_owd_reply(
+            &mut with,
+            owd_timestamps(),
+            kernel_rx_meta(25_000_000),
+            owd_received_at(mono),
+        );
+
+        let without_reply = match without_events.as_slice() {
+            [ClientEvent::EchoReply { rtt, one_way, .. }] => (*rtt, *one_way),
+            other => panic!("expected one EchoReply, got {other:?}"),
+        };
+        let with_reply = match with_events.as_slice() {
+            [ClientEvent::EchoReply { rtt, one_way, .. }] => (*rtt, *one_way),
+            other => panic!("expected one EchoReply, got {other:?}"),
+        };
+        assert_eq!(without_reply, with_reply);
+    }
 }