From 9e3477e617da84737924a4985cd9ba6680f3e217 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sat, 15 Aug 2026 15:58:48 +0700 Subject: [PATCH] feat(client): capture kernel transmit timestamps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On Linux, best-effort upgrade the socket's SO_TIMESTAMPING flags to TX_SOFTWARE | OPT_ID | OPT_TSONLY after a successful Open and before the first Echo probe. Promote the #103 error-queue foundation to production: MSG_ERRQUEUE is drained opportunistically with a fixed per-call budget at several natural choke points (after a probe send, before processing a reply, before evaluating timeouts), and each automatic OPT_ID timestamp is correlated directly with wire_seq — safe because Linux rolls back its speculative sk_tskey increment on every send failure, wire_seq and the kernel counter start together at every Open, and a socket can never be reopened for a second session. The observed kernel TX wall timestamp is retained on PendingProbe (surviving pending -> timed-out retention) as dormant metadata only; RTT, OWD, and IPDV are unchanged and continue to use the userspace send timestamp exclusively. Capability failure is fully non-fatal and RX timestamping is unaffected either way. Non-Linux and non-ancillary builds are unchanged. --- crates/irtt-client/AGENTS.md | 108 ++++++ crates/irtt-client/src/async_client.rs | 33 +- crates/irtt-client/src/async_client/tests.rs | 1 + crates/irtt-client/src/client.rs | 36 +- crates/irtt-client/src/managed/tests.rs | 1 + crates/irtt-client/src/probe.rs | 24 +- crates/irtt-client/src/receive.rs | 35 ++ crates/irtt-client/src/receive/linux.rs | 149 +++++++- .../src/receive/linux/error_queue.rs | 287 ++++++++++++--- crates/irtt-client/src/session/machine.rs | 343 ++++++++++++++++++ 10 files changed, 948 insertions(+), 69 deletions(-) create mode 100644 crates/irtt-client/AGENTS.md diff --git a/crates/irtt-client/AGENTS.md b/crates/irtt-client/AGENTS.md new file mode 100644 index 0000000..c96ba5b --- /dev/null +++ b/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` +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. diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index 22220e6..dc4bf46 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/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, prepared_open: Option, prepared_probe: Option, + /// 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, ClientError> { + self.drain_tx_timestamps()?; self.machine.poll_timeouts_at(now) } @@ -305,6 +310,7 @@ impl AsyncClient { now: Instant, limit: usize, ) -> Result { + 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, diff --git a/crates/irtt-client/src/async_client/tests.rs b/crates/irtt-client/src/async_client/tests.rs index b9d5620..91cc044 100644 --- a/crates/irtt-client/src/async_client/tests.rs +++ b/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, }); } diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 064f271..b6e421f 100644 --- a/crates/irtt-client/src/client.rs +++ b/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, applied_traffic_class: Option, + /// 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, 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) { #[cfg(test)] if self.test_hooks.fail_dscp_restore.replace(false) { diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index 284d0c2..78eb75b 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/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, } } diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index 951fb1e..a1f842a 100644 --- a/crates/irtt-client/src/probe.rs +++ b/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, } #[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 { 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, } } diff --git a/crates/irtt-client/src/receive.rs b/crates/irtt-client/src/receive.rs index d65fc6e..d9d7e00 100644 --- a/crates/irtt-client/src/receive.rs +++ b/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(_socket: &S) -> bool { + false +} + +#[cfg(all(target_os = "linux", feature = "ancillary"))] +pub(crate) fn try_enable_tx_timestamping(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( + _socket: &S, + _on_timestamp: impl FnMut(u32, SystemTime), +) -> io::Result<()> { + Ok(()) +} + +#[cfg(all(target_os = "linux", feature = "ancillary"))] +pub(crate) fn drain_tx_timestamps( + 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; diff --git a/crates/irtt-client/src/receive/linux.rs b/crates/irtt-client/src/receive/linux.rs index ed1eb6c..cdefb6b 100644 --- a/crates/irtt-client/src/receive/linux.rs +++ b/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(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( + 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); + } } diff --git a/crates/irtt-client/src/receive/linux/error_queue.rs b/crates/irtt-client/src/receive/linux/error_queue.rs index c8e1ed9..58e280d 100644 --- a/crates/irtt-client/src/receive/linux/error_queue.rs +++ b/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) -> ErrorQueueRecord { let mut extended_error: Option<(u32, u8, u32, u32)> = None; let mut timestamp: Option = None; @@ -114,20 +120,21 @@ pub(crate) fn classify(cmsgs: impl Iterator) -> 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`. } diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 7fd2882..a2f9368 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/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); + } } -- 2.54.0