Merge pull request #102 from Lochnair/fix/server-kernel-rx-wall

fix(server): use kernel receive time for wall timestamps

crates/irtt-server/AGENTS.md

@@ -177,17 +177,74 @@ the mirror image of destination metadata, and the contrast is the point.
   on an explicit listener, or one that describes no representable instant, all
   mean the datagram is served with `kernel_rx_timestamp: None`. Only the
   destination rules a datagram out, and only for the listener that needs one.
-- **Nothing consumes it yet.** It stops at the runtime boundary. Do not thread it
-  into `ServerCore`, add a timestamp argument or variant to `handle_datagram`,
-  stash it in a "next receive" side channel, or wrap the core's clock. How a
-  kernel-observed arrival time should enter the core is an open design question,
-  to be settled deliberately and not as a side effect of some other change.
 - **Do not map wall to monotonic.** It is a realtime reading with no monotonic
   counterpart, and `ClockSample` is a paired userspace instant. Synthesizing the
   missing half would invent precision the kernel never reported.
-- **Do not judge plausibility here.** Conversion is structural only. Whether a
+- **Do not judge plausibility in the socket layer.** Conversion there is
-  representable reading is close enough to the server's own sample to measure
+  structural only. Whether a representable reading is close enough to the
-  against belongs with whoever measures.
+  server's own sample to measure against belongs with whoever measures, which is
+  the core.
+
+### How it reaches the core, and what it may change
+
+There are two receive entry points and one implementation behind them.
+
+- **`ServerCore::handle_datagram(peer, packet)` is the public one and takes no
+  timing.** A caller hands over bytes and an endpoint; every instant the reply
+  reports is one the core sampled itself. This is exactly the behavior it had
+  before the kernel timestamp was consumed, and it is what a direct or manual
+  `ServerCore` user gets.
+- **The Tokio runtime uses a crate-private entry point** that additionally takes
+  the optional kernel-observed wall receive time, passed with the datagram it
+  describes. That is the whole mechanism: an explicit per-datagram argument. Do
+  **not** replace it with a "next receive timestamp" setter, a pending-metadata
+  field, thread-local or global state, or a `ClockSource` wrapper that returns
+  transport metadata on the next sample — those hide a per-datagram input behind
+  ambient mutable state. Do not make the runtime path a second public API, and
+  do not pass `ReplySource` or the whole `ReceivedDatagram` into the core: it
+  needs a peer, bytes and an arrival observation, not socket addressing.
+- **`ClockSample` is unchanged and still one paired userspace instant.** The
+  preferred receive wall value is carried beside it on the classified echo, never
+  inside it. A `ClockSample` whose `wall_ns` came from the kernel and whose
+  `mono_ns` came from userspace would claim two physical instants were one
+  sample.
+
+The acceptance rule is the client's, in principle and in its bound, and lives in
+the core as `preferred_receive_wall_ns`. A reading is used only when it converts
+into the wire's signed-nanosecond vocabulary, is not later than the core's own
+paired receive sample, and lags it by no more than `MAX_KERNEL_RX_LAG` — one
+second, inclusive, and a sanity guard rather than an expected wakeup latency.
+Anything else falls back to the userspace wall reading. **A rejected reading is
+never a rejected datagram**: no error reply, no dropped packet, no session state
+moved. Keep the two implementations separate from the client's; a shared helper
+is not worth a crate dependency for a dozen lines of timing policy.
+
+What the observation may change is one field:
+
+- **Measurement policy**: `recv_wall` under `StampAt::Receive` and `StampAt::Both`
+  on a clock that reports wall time. Nothing else.
+- **Lifecycle policy**: always `received_at.mono_ns`, the core's own userspace
+  monotonic reading. Idle expiry, rate allowance and the maximum-duration
+  deadline never see a kernel value, including for the rate-limited echo that
+  produces no reply and for the final echo that carries the close.
+- **Midpoint**: always the paired userspace receive and send samples. Moving only
+  the wall receive endpoint would improve apparent upstream delay by exactly what
+  it cost apparent downstream delay, and leave `midpoint_wall` and
+  `midpoint_mono` describing different midpoints — with no residency interval
+  recoverable from a midpoint to show for it.
+- **`StampAt::Send`, `StampAt::None` and `Clock::Monotonic`** emit no arrival wall
+  field, so an observation cannot reach them. An open or a close ignores it too.
+- **Ordering still holds.** The reported arrival wall value is held back to the
+  send reading in its own right, because it may not have come from the receive
+  sample `ClockSample::not_after` orders.
+
+One consequence is intended: under `Clock::Wall` with `StampAt::Both`, the
+server-processing interval a client computes now runs from the kernel's arrival
+reading rather than the server's wakeup, so it includes the kernel-to-userspace
+portion of the server's residency and the adjusted round-trip shrinks to match.
+Under `Clock::Both` the monotonic pair still yields the unchanged userspace
+interval. Do not force the two to agree, and in particular do not estimate a
+kernel monotonic instant to do it.
 - Linux only, and receive only. No macOS, FreeBSD or Windows implementation, and
   no transmit timestamping, without a reason and agreement.
 - Control-buffer capacity is derived from the cmsg payload types and asserted at
@@ -283,7 +340,9 @@ effective parameters must be safe for the server to run.
   clock source and is stable for its life.
 - A reply's receive instant is held back to its send instant where the wall
   clock stepped backwards between the two readings, so the required ordering
-  holds. That correction is per reply and stateless on purpose: latching a
+  holds — and so is a reported arrival wall value that came from the kernel
+  instead, which needs the rule applied in its own right because it was not part
+  of the pair. That correction is per reply and stateless on purpose: latching a
   pre-step wall value across packets would keep reporting a time the host has
   corrected away, which is the smoothing the specification forbids, and
   anchoring the wall clock to the monotonic origin would drift and never pick up

crates/irtt-server/src/clock.rs

@@ -8,9 +8,18 @@
 //!
 //! This is the clock counterpart of [`TokenSource`](crate::token::TokenSource)
 //! and nothing more. It is **not** a runtime abstraction, not a transport
-//! boundary and not a pluggable product API: it is crate-private, the public
+//! boundary and not a pluggable product API: it is crate-private and the public
-//! constructor still takes only a [`ServerConfig`](crate::ServerConfig), and
+//! constructor still takes only a [`ServerConfig`](crate::ServerConfig).
-//! callers never pass timestamps into the core.
+//!
+//! The public [`ServerCore::handle_datagram`](crate::ServerCore::handle_datagram)
+//! entry point takes no timestamp: the core samples every instant it reports.
+//! The Tokio runtime uses a crate-private entry point that may additionally hand
+//! over a kernel-observed wall receive time for one datagram, which this module
+//! contributes only the [`wall_ns_of`] conversion for. Whether such an
+//! observation is usable, and which reply field it may reach, is measurement
+//! policy and lives in the core — this stays the module that reads clocks, not
+//! one that models transport metadata. A [`ClockSample`] remains one paired
+//! userspace instant and never carries a kernel reading.
 
 use std::{
     fmt,
@@ -123,6 +132,28 @@ fn wall_ns() -> i64 {
     }
 }
 
+/// An observed instant as nanoseconds since the Unix epoch, or `None` when it
+/// does not fit the wire's signed-nanosecond field.
+///
+/// This is the conversion for an instant the server did **not** read from its
+/// own [`ClockSource`] — today, a kernel receive timestamp the transport
+/// observed. It reports unrepresentability rather than saturating, because a
+/// clamped instant is indistinguishable from a real one at the boundary and
+/// would then be compared against a genuine sample as though it were plausible.
+/// A local reading has no such problem and keeps [`saturating_ns`].
+///
+/// The negation cannot fail for any [`SystemTime`] a host can hold: it is
+/// reachable only from a value already inside `i64`, and only `i64::MIN` has no
+/// negation. It is checked anyway so that no path here can panic.
+pub(crate) fn wall_ns_of(at: SystemTime) -> Option<i64> {
+    match at.duration_since(UNIX_EPOCH) {
+        Ok(since_epoch) => i64::try_from(since_epoch.as_nanos()).ok(),
+        Err(before_epoch) => i64::try_from(before_epoch.duration().as_nanos())
+            .ok()
+            .and_then(i64::checked_neg),
+    }
+}
+
 /// A duration as nanoseconds, saturating rather than panicking.
 ///
 /// This is the crate's one [`Duration`] → wire-nanoseconds conversion, used for
@@ -215,4 +246,25 @@ mod tests {
         assert_eq!(saturating_ns(Duration::from_nanos(1_500)), 1_500);
         assert_eq!(saturating_ns(Duration::MAX), i64::MAX);
     }
+
+    #[test]
+    fn an_observed_instant_converts_or_reports_that_it_cannot() {
+        assert_eq!(
+            wall_ns_of(UNIX_EPOCH + Duration::from_nanos(1_500)),
+            Some(1_500)
+        );
+        assert_eq!(wall_ns_of(UNIX_EPOCH), Some(0));
+
+        // Before the epoch is negative, not a wrap and not a rejection.
+        assert_eq!(
+            wall_ns_of(UNIX_EPOCH - Duration::from_nanos(1_500)),
+            Some(-1_500)
+        );
+
+        // Beyond the field, in either direction, is `None` rather than a
+        // clamped value that would then read as an ordinary instant.
+        let beyond = Duration::from_nanos(u64::MAX) + Duration::from_secs(1);
+        assert_eq!(wall_ns_of(UNIX_EPOCH + beyond), None);
+        assert_eq!(wall_ns_of(UNIX_EPOCH - beyond), None);
+    }
 }

crates/irtt-server/src/core.rs

@@ -1,4 +1,8 @@
-use std::{collections::HashMap, net::SocketAddr};
+use std::{
+    collections::HashMap,
+    net::SocketAddr,
+    time::{Duration, SystemTime},
+};
 
 use irtt_proto::{
     decode_request, echo_packet_len, encode_echo_reply, encode_open_reply, verify_packet_hmac,
@@ -7,7 +11,7 @@ use irtt_proto::{
 };
 
 use crate::{
-    clock::{saturating_ns, ClockSample, ClockSource, SystemClock},
+    clock::{saturating_ns, wall_ns_of, ClockSample, ClockSource, SystemClock},
     config::ServerConfig,
     error::ServerError,
     fill::{echo_payload_len, negotiate_server_fill},
@@ -154,6 +158,12 @@ impl ServerCore {
     /// data or consulting the session table, and no reply behavior may reveal
     /// which stage rejected it.
     ///
+    /// Every instant this reports is one the core sampled itself: a caller hands
+    /// over bytes and an endpoint, not timings. The Tokio runtime, which does
+    /// have transport metadata, reaches the same implementation through a
+    /// crate-private entry point that may add a kernel-observed arrival time;
+    /// this path behaves exactly as it does with none.
+    ///
     /// # Errors
     ///
     /// Returns [`ServerError`] only for an internal failure — the random source
@@ -164,6 +174,49 @@ impl ServerCore {
         &mut self,
         peer: SocketAddr,
         packet: &[u8],
+    ) -> Result<Option<OutboundDatagram>, ServerError> {
+        self.process(peer, packet, None)
+    }
+
+    /// Handles one received datagram together with what the transport observed
+    /// about its arrival.
+    ///
+    /// This is the entry point the Tokio runtime uses, and the only difference
+    /// from [`handle_datagram`](Self::handle_datagram) is `kernel_rx_timestamp`:
+    /// a wall-clock arrival instant the kernel recorded for this datagram, where
+    /// the listener could obtain one. Both paths run the same implementation,
+    /// and `None` reproduces the public path exactly.
+    ///
+    /// The observation is measurement input and nothing else. It may become the
+    /// receive wall value of an echo reply that negotiated one, and it can never
+    /// reach lifecycle, rate or deadline policy, the monotonic domain, or a
+    /// midpoint — those stay on the core's own paired userspace sample. An
+    /// implausible or unrepresentable value is not a reason to reject the
+    /// datagram; it simply is not used. See
+    /// [`preferred_receive_wall_ns`].
+    ///
+    /// It is crate-private on purpose: transport metadata reaches the core by
+    /// being passed with the datagram it describes, and there is no second
+    /// public API for it.
+    ///
+    /// # Errors
+    ///
+    /// As [`handle_datagram`](Self::handle_datagram).
+    pub(crate) fn handle_received_datagram(
+        &mut self,
+        peer: SocketAddr,
+        packet: &[u8],
+        kernel_rx_timestamp: Option<SystemTime>,
+    ) -> Result<Option<OutboundDatagram>, ServerError> {
+        self.process(peer, packet, kernel_rx_timestamp)
+    }
+
+    /// The one implementation behind both receive entry points.
+    fn process(
+        &mut self,
+        peer: SocketAddr,
+        packet: &[u8],
+        kernel_rx_timestamp: Option<SystemTime>,
     ) -> Result<Option<OutboundDatagram>, ServerError> {
         let Ok(request) = decode_request(packet) else {
             return Ok(None);
@@ -176,7 +229,7 @@ impl ServerCore {
         // protocol state, and it is a strictly better trade than looking a
         // session up before authentication to discover whether it even wanted
         // timestamps.
-        let kind = self.classify(&request.kind, packet.len());
+        let kind = self.classify(&request.kind, packet.len(), kernel_rx_timestamp);
 
         // The HMAC flag must agree with this server's configuration in both
         // directions of mismatch, and the MAC itself is verified separately.
@@ -248,10 +301,17 @@ impl ServerCore {
 
     /// Pairs a structurally decoded request with the clock sample an echo
     /// needs, leaving every other kind untouched.
+    ///
+    /// An echo also settles here which wall instant its reply may report as the
+    /// arrival time, because that is the moment both candidates are in hand.
+    /// `kernel_rx_timestamp` is ignored for every other kind: an open or a close
+    /// emits no receive timestamp, so having observed one changes nothing about
+    /// it.
     fn classify<'a>(
         &mut self,
         kind: &DecodedRequestKind<'a>,
         datagram_len: usize,
+        kernel_rx_timestamp: Option<SystemTime>,
     ) -> ClassifiedKind<'a> {
         match *kind {
             DecodedRequestKind::Open { no_test, params } => {
@@ -262,12 +322,16 @@ impl ServerCore {
             // read no further than its length.
             DecodedRequestKind::Echo {
                 token, sequence, ..
-            } => ClassifiedKind::Echo(EchoRequest {
+            } => {
+                let received_at = self.clock.sample();
+                ClassifiedKind::Echo(EchoRequest {
                     token,
                     sequence,
                     datagram_len,
-                received_at: self.clock.sample(),
+                    received_at,
-            }),
+                    receive_wall_ns: preferred_receive_wall_ns(kernel_rx_timestamp, received_at),
+                })
+            }
         }
     }
 
@@ -377,7 +441,12 @@ impl ServerCore {
             sequence: request.sequence,
             recv_count: params.received_stats.has_count().then_some(next.count),
             recv_window: params.received_stats.has_window().then_some(next.window),
-            timestamps: timestamp_fields(params, request.received_at, sent_at),
+            timestamps: timestamp_fields(
+                params,
+                request.received_at,
+                sent_at,
+                request.receive_wall_ns,
+            ),
             // The session's negotiated fill, generated fresh for this reply and
             // sized from the layout rather than from `params.length`. A `none`
             // session contributes no bytes and the encoder's zero-fill stands,
@@ -621,12 +690,67 @@ enum ClassifiedKind<'a> {
 /// fields is carried forward. `datagram_len` is the length the datagram
 /// actually arrived with, which the configured maximum judges; it is not
 /// required to match the negotiated length, and it never sizes the reply.
+///
+/// The two timing fields are deliberately separate and are not interchangeable.
+/// `received_at` is the core's own paired userspace instant and is the sole
+/// authority for everything about this request that is a same-host question:
+/// idle expiry, rate allowance, the maximum-duration deadline, and the
+/// monotonic timestamps a reply may carry. `receive_wall_ns` is only the
+/// wall-clock arrival value a reply may *report*, which is the kernel's reading
+/// where the transport observed a usable one. Keeping it out of `ClockSample`
+/// is the point: a kernel wall reading has no monotonic counterpart, and pairing
+/// it with `received_at.mono_ns` would claim two different physical instants
+/// were one sample.
 #[derive(Debug, Clone, Copy)]
 struct EchoRequest {
     token: u64,
     sequence: u32,
     datagram_len: usize,
     received_at: ClockSample,
+    receive_wall_ns: i64,
+}
+
+/// Largest lag behind the core's own receive sample that a kernel receive
+/// timestamp may show and still be used as a measurement endpoint.
+///
+/// This is a sanity guard rather than an expected kernel-to-userspace wakeup
+/// latency, which is orders of magnitude smaller. A reading this far behind the
+/// server's own sample most likely reflects a realtime clock discontinuity or
+/// otherwise unusable metadata, and falling back to a value this server knows it
+/// read is better than publishing a bad cross-host timing endpoint. The client
+/// applies the same bound to its own receive metadata.
+pub(crate) const MAX_KERNEL_RX_LAG: Duration = Duration::from_secs(1);
+
+/// The wall instant an echo reply should report as the request's arrival time.
+///
+/// A kernel receive timestamp is taken before the socket queue and the runtime's
+/// wakeup, so preferring it removes that delay from the upstream one-way delay a
+/// client computes. It is used only when it is plausible *for this datagram*: it
+/// must convert into the wire's signed-nanosecond vocabulary, it cannot be later
+/// than the sample that observed the same datagram, and it may not lag that
+/// sample by more than [`MAX_KERNEL_RX_LAG`]. The bound is inclusive.
+///
+/// Anything else falls back to the userspace wall reading. A failed check is
+/// never a reason to drop the datagram, refuse a reply or disturb session state:
+/// implausible metadata means the server measures as it did before the kernel
+/// offered any.
+///
+/// The comparison goes through `i128` because the difference of two `i64`
+/// nanosecond values can exceed `i64`.
+pub(crate) fn preferred_receive_wall_ns(
+    kernel_rx_timestamp: Option<SystemTime>,
+    received_at: ClockSample,
+) -> i64 {
+    let userspace_wall_ns = received_at.wall_ns;
+    let Some(kernel_wall_ns) = kernel_rx_timestamp.and_then(wall_ns_of) else {
+        return userspace_wall_ns;
+    };
+    let lag_ns = i128::from(userspace_wall_ns) - i128::from(kernel_wall_ns);
+    if (0..=i128::from(saturating_ns(MAX_KERNEL_RX_LAG))).contains(&lag_ns) {
+        kernel_wall_ns
+    } else {
+        userspace_wall_ns
+    }
 }
 
 /// Builds the timestamp fields a session negotiated, and only those.
@@ -650,20 +774,65 @@ struct EchoRequest {
 /// receive instant must not be later than its send instant and the wall clock
 /// can be stepped backwards between the two readings. See
 /// [`ClockSample::not_after`]; the correction is confined to this one reply.
-fn timestamp_fields(params: &Params, received: ClockSample, sent: ClockSample) -> TimestampFields {
+///
+/// `receive_wall_ns` is the wall instant a reported arrival time should use —
+/// [`preferred_receive_wall_ns`], so the kernel's reading where there was a
+/// usable one and the userspace sample's own otherwise. It reaches exactly two
+/// fields, `recv_wall` under [`StampAt::Receive`] and under [`StampAt::Both`],
+/// and it is held to the send reading by the same rule the paired sample is.
+/// Everything else keeps the paired userspace samples:
+///
+/// - **Monotonic never moves.** A kernel wall reading has no monotonic
+///   counterpart, and inventing one would be precision the kernel never
+///   reported. So under [`Clock::Both`] a `recv_wall`/`recv_mono` pair may
+///   describe two different physical instants. That is a deliberate wire
+///   policy — each field is honest about the domain it names, and nothing
+///   computes across the two — and it is why the core does not model them as
+///   one sample.
+/// - **A midpoint is untouched.** It is the mean of the receive and send
+///   instants, so substituting only the wall receive endpoint would move
+///   `midpoint_wall` earlier by half the kernel-to-userspace delay: apparent
+///   upstream delay would improve by exactly as much as apparent downstream
+///   delay worsened, `midpoint_wall` and `midpoint_mono` would stop describing
+///   the same midpoint, and no server residency interval is recoverable from a
+///   midpoint to make up for it.
+/// - **[`StampAt::Send`] and [`StampAt::None`] report no arrival time at all**,
+///   so an observation cannot reach them.
+///
+/// One consequence is intended and worth stating: under [`Clock::Wall`] with
+/// [`StampAt::Both`], the interval a client computes from the two reported
+/// instants now spans from the kernel's arrival reading rather than from the
+/// server's wakeup, so it includes the kernel-to-userspace portion of the
+/// server's residency. Under [`Clock::Both`] the monotonic pair still yields the
+/// unchanged userspace interval.
+fn timestamp_fields(
+    params: &Params,
+    received: ClockSample,
+    sent: ClockSample,
+    receive_wall_ns: i64,
+) -> TimestampFields {
     let received = received.not_after(sent);
+    // The reported arrival wall value obeys the same ordering rule, and needs
+    // its own application of it: it may not have come from `received`, so a
+    // backward wall step between the kernel's reading and the send sample would
+    // otherwise leave the reply claiming it was received after it was sent.
+    let receive_wall_ns = receive_wall_ns.min(sent.wall_ns);
     let per_clock = |sample: ClockSample| {
         (
             params.clock.has_wall().then_some(sample.wall_ns),
             params.clock.has_mono().then_some(sample.mono_ns),
         )
     };
+    let arrival = (
+        params.clock.has_wall().then_some(receive_wall_ns),
+        params.clock.has_mono().then_some(received.mono_ns),
+    );
     let absent = (None, None);
     let (recv, midpoint, send) = match params.stamp_at {
         StampAt::None => (absent, absent, absent),
-        StampAt::Receive => (per_clock(received), absent, absent),
+        StampAt::Receive => (arrival, absent, absent),
         StampAt::Send => (absent, absent, per_clock(sent)),
-        StampAt::Both => (per_clock(received), absent, per_clock(sent)),
+        StampAt::Both => (arrival, absent, per_clock(sent)),
         StampAt::Midpoint => (absent, per_clock(received.midpoint(sent)), absent),
     };
     TimestampFields {

crates/irtt-server/src/runtime.rs

@@ -229,14 +229,21 @@ impl Server {
                         len,
                         peer,
                         reply_source,
-                        // Observed by the transport and deliberately unused.
+                        // Handed to the core with the datagram it describes, as
-                        // The core samples its own receive instant, and this
+                        // an explicit per-datagram input. The core samples its
-                        // is where a kernel-observed arrival time stops until a
+                        // own receive instant regardless and decides whether
-                        // follow-up change decides how it should enter the core
+                        // this one is usable; nothing here judges it, and there
-                        // — feeding it in here would change protocol timing.
+                        // is no "next receive" state between the two.
-                        kernel_rx_timestamp: _kernel_rx_timestamp,
+                        kernel_rx_timestamp,
                     } = received;
-                    if let Some(reply) = self.core.handle_datagram(peer, &self.recv_buffer[..len])? {
+                    // `reply_source` stays on this side of the boundary: which
+                    // of the host's addresses a request arrived on is transport
+                    // state, not session policy.
+                    if let Some(reply) = self.core.handle_received_datagram(
+                        peer,
+                        &self.recv_buffer[..len],
+                        kernel_rx_timestamp,
+                    )? {
                         if !self.prepare_reply_traffic_class(reply.traffic_class()) {
                             continue;
                         }

crates/irtt-server/src/tests/kernel_rx.rs

@@ -0,0 +1,668 @@
+//! Consuming a kernel-observed receive timestamp.
+//!
+//! A Linux listener reports the kernel's arrival time for each datagram, and an
+//! echo reply may report it as the wall-clock instant the request was received.
+//! These tests pin what that is allowed to change and — mostly — what it is
+//! not: only `recv_wall`, only under [`StampAt::Receive`] and [`StampAt::Both`],
+//! only on a clock that reports wall time, and never a monotonic value, a
+//! midpoint, or any lifecycle, rate or deadline decision.
+//!
+//! Whether the transport actually captures a timestamp is settled by the socket
+//! layer's own tests; nothing here recertifies `SCM_TIMESTAMPNS`. These drive
+//! the core with fabricated values, so they run identically on every target.
+
+use std::{
+    net::SocketAddr,
+    time::{Duration, SystemTime, UNIX_EPOCH},
+};
+
+use irtt_proto::{Clock, Params, ReceivedStats, StampAt, TimestampFields};
+
+use super::support::{
+    close_request, core_with_sources, echo_at, echo_params, echo_request,
+    expect_closing_echo_reply, expect_echo_reply, expect_normal_open_reply, manual_core,
+    open_negotiated, open_request, peer, sample, ManualClock, ScriptedClock, ScriptedTokens,
+};
+use crate::{
+    clock::ClockSample,
+    core::{preferred_receive_wall_ns, OutboundDatagram, ServerCore, MAX_KERNEL_RX_LAG},
+    ServerConfig,
+};
+
+const TOKEN_A: u64 = 0x0102_0304_0506_0708;
+const MS: i64 = 1_000_000;
+const SECOND: i64 = 1_000 * MS;
+
+/// The lifecycle instant the session-creating open consumes.
+fn open_sample() -> ClockSample {
+    sample(900_000, 9_000)
+}
+
+/// The core's own receive reading for the echo every matrix test sends.
+fn received() -> ClockSample {
+    sample(1_000_000, 10_000)
+}
+
+/// The core's send reading for the same echo.
+fn sent() -> ClockSample {
+    sample(1_200_000, 10_300)
+}
+
+/// A kernel arrival reading 100 µs before [`received`], which is the ordinary
+/// shape: earlier than the server's own sample and well inside the bound.
+fn kernel_rx() -> SystemTime {
+    at_epoch_ns(900_000)
+}
+
+/// The [`SystemTime`] `ns` nanoseconds from the Unix epoch, before it when
+/// negative.
+fn at_epoch_ns(ns: i64) -> SystemTime {
+    let magnitude = Duration::from_nanos(ns.unsigned_abs());
+    if ns.is_negative() {
+        UNIX_EPOCH - magnitude
+    } else {
+        UNIX_EPOCH + magnitude
+    }
+}
+
+/// Sends one echo carrying `kernel_rx_timestamp` through the runtime's entry
+/// point and returns the reply's timestamp fields.
+fn echo_timestamps(
+    stamp_at: StampAt,
+    clock: Clock,
+    kernel_rx_timestamp: Option<SystemTime>,
+) -> TimestampFields {
+    let mut core = core_with_sources(
+        ServerConfig::default(),
+        ScriptedTokens::new([TOKEN_A]),
+        ScriptedClock::new([open_sample(), received(), sent()]),
+    );
+    let params = echo_params(ReceivedStats::None, stamp_at, clock, /* length */ 0);
+    let (token, negotiated) = open_negotiated(&mut core, peer(), &params, None);
+
+    let packet = core
+        .handle_received_datagram(
+            peer(),
+            &echo_request(token, 0, &negotiated, &[], None),
+            kernel_rx_timestamp,
+        )
+        .unwrap()
+        .unwrap_or_else(|| panic!("{stamp_at:?} on {clock:?} must be answered"));
+    expect_echo_reply(&packet, &negotiated, None).timestamps
+}
+
+#[test]
+fn the_preferred_receive_wall_is_the_kernel_reading_only_when_it_is_plausible() {
+    let userspace = received();
+    let userspace_wall = userspace.wall_ns;
+    let bound_ns = i64::try_from(MAX_KERNEL_RX_LAG.as_nanos()).unwrap();
+
+    // No observation at all: the server's own reading, which is the whole of
+    // the pre-existing behavior and what every non-Linux target sees.
+    assert_eq!(preferred_receive_wall_ns(None, userspace), userspace_wall);
+
+    // Earlier, and inside the bound.
+    assert_eq!(
+        preferred_receive_wall_ns(Some(kernel_rx()), userspace),
+        900_000
+    );
+
+    // Equal is accepted: the two clocks may simply have read the same
+    // nanosecond.
+    assert_eq!(
+        preferred_receive_wall_ns(Some(at_epoch_ns(userspace_wall)), userspace),
+        userspace_wall
+    );
+
+    // Later than the sample that observed the same datagram is impossible, so
+    // it is metadata this server will not measure against. This is the shape a
+    // backward wall step between the kernel's reading and the core's sample
+    // produces.
+    assert_eq!(
+        preferred_receive_wall_ns(Some(at_epoch_ns(userspace_wall + 1)), userspace),
+        userspace_wall
+    );
+
+    // The lag bound is inclusive.
+    assert_eq!(
+        preferred_receive_wall_ns(Some(at_epoch_ns(userspace_wall - bound_ns)), userspace),
+        userspace_wall - bound_ns
+    );
+    assert_eq!(
+        preferred_receive_wall_ns(Some(at_epoch_ns(userspace_wall - bound_ns - 1)), userspace),
+        userspace_wall
+    );
+
+    // A pre-epoch reading is representable as a negative value and is judged
+    // by exactly the same rule, so against a host clock set anywhere near the
+    // present it simply lags far too much. Nothing here special-cases the sign.
+    let present = sample(1_700_000_000 * SECOND, 10_000);
+    assert_eq!(
+        preferred_receive_wall_ns(Some(UNIX_EPOCH - Duration::from_nanos(1)), present),
+        present.wall_ns
+    );
+    assert_eq!(
+        preferred_receive_wall_ns(Some(at_epoch_ns(present.wall_ns - bound_ns)), present),
+        present.wall_ns - bound_ns,
+        "and a realistic wall clock accepts a reading at the bound like any other"
+    );
+}
+
+#[test]
+fn an_unrepresentable_kernel_reading_falls_back_without_panicking() {
+    // Far outside the wire's signed-nanosecond field in both directions, and
+    // judged against a userspace sample at the field's own limits, so nothing
+    // here can be rescued by a lucky comparison. A conversion that saturated
+    // instead of reporting failure would turn one of these into an ordinary
+    // instant.
+    let beyond = Duration::from_nanos(u64::MAX) + Duration::from_secs(1);
+
+    for userspace_wall in [0, i64::MAX, i64::MIN] {
+        let userspace = sample(userspace_wall, 10_000);
+        for kernel in [UNIX_EPOCH + beyond, UNIX_EPOCH - beyond] {
+            assert_eq!(
+                preferred_receive_wall_ns(Some(kernel), userspace),
+                userspace_wall,
+                "an unrepresentable reading is never a measurement endpoint"
+            );
+        }
+    }
+}
+
+#[test]
+fn only_the_receive_wall_field_can_come_from_the_kernel() {
+    // The whole StampAt x Clock matrix, each asserted as one struct so that a
+    // field appearing where it should not is as loud as a wrong instant. Every
+    // expectation below is what the same session emits with no kernel metadata,
+    // except `recv_wall` under Receive and Both on a wall-reporting clock.
+    for (stamp_at, clock, expected) in [
+        // A: the kernel reading is the reported arrival time.
+        (
+            StampAt::Receive,
+            Clock::Wall,
+            TimestampFields {
+                recv_wall: Some(900_000),
+                ..TimestampFields::default()
+            },
+        ),
+        // B: and the monotonic half of the same arrival stays the core's own.
+        // The two fields deliberately name different physical instants.
+        (
+            StampAt::Receive,
+            Clock::Both,
+            TimestampFields {
+                recv_wall: Some(900_000),
+                recv_mono: Some(10_000),
+                ..TimestampFields::default()
+            },
+        ),
+        // C: nothing about the departure side moves, in either domain.
+        (
+            StampAt::Both,
+            Clock::Both,
+            TimestampFields {
+                recv_wall: Some(900_000),
+                recv_mono: Some(10_000),
+                send_wall: Some(1_200_000),
+                send_mono: Some(10_300),
+                ..TimestampFields::default()
+            },
+        ),
+        // D: wall only. The interval between these two now includes the
+        // kernel-to-userspace part of the server's residency, which is the one
+        // intended behavior change.
+        (
+            StampAt::Both,
+            Clock::Wall,
+            TimestampFields {
+                recv_wall: Some(900_000),
+                send_wall: Some(1_200_000),
+                ..TimestampFields::default()
+            },
+        ),
+        // E: no wall field exists, so there is nothing to improve.
+        (
+            StampAt::Both,
+            Clock::Monotonic,
+            TimestampFields {
+                recv_mono: Some(10_000),
+                send_mono: Some(10_300),
+                ..TimestampFields::default()
+            },
+        ),
+        (
+            StampAt::Receive,
+            Clock::Monotonic,
+            TimestampFields {
+                recv_mono: Some(10_000),
+                ..TimestampFields::default()
+            },
+        ),
+        // F: a reply reporting only its departure never learns of an arrival
+        // observation.
+        (
+            StampAt::Send,
+            Clock::Wall,
+            TimestampFields {
+                send_wall: Some(1_200_000),
+                ..TimestampFields::default()
+            },
+        ),
+        (
+            StampAt::Send,
+            Clock::Both,
+            TimestampFields {
+                send_wall: Some(1_200_000),
+                send_mono: Some(10_300),
+                ..TimestampFields::default()
+            },
+        ),
+        // G: no timestamp at all.
+        (StampAt::None, Clock::Both, TimestampFields::default()),
+        // H and I: a midpoint is the mean of the core's own paired samples and
+        // stays exactly that. Substituting only the wall arrival endpoint would
+        // pull `midpoint_wall` to 1_050_000 — half the kernel-to-userspace
+        // delay early — improving apparent upstream delay by precisely as much
+        // as it worsened apparent downstream delay, and leaving the two
+        // midpoint fields describing different midpoints.
+        (
+            StampAt::Midpoint,
+            Clock::Wall,
+            TimestampFields {
+                midpoint_wall: Some(1_100_000),
+                ..TimestampFields::default()
+            },
+        ),
+        (
+            StampAt::Midpoint,
+            Clock::Both,
+            TimestampFields {
+                midpoint_wall: Some(1_100_000),
+                midpoint_mono: Some(10_150),
+                ..TimestampFields::default()
+            },
+        ),
+        (
+            StampAt::Midpoint,
+            Clock::Monotonic,
+            TimestampFields {
+                midpoint_mono: Some(10_150),
+                ..TimestampFields::default()
+            },
+        ),
+    ] {
+        assert_eq!(
+            echo_timestamps(stamp_at, clock, Some(kernel_rx())),
+            expected,
+            "{stamp_at:?} on {clock:?} with a plausible kernel reading"
+        );
+    }
+}
+
+#[test]
+fn a_datagram_with_no_kernel_reading_emits_what_it_always_did() {
+    // The same matrix without metadata, which is every non-Linux target and
+    // every Linux listener whose kernel offered nothing. `recv_wall` is the
+    // core's own 1_000_000 throughout, and the two paths are otherwise the same
+    // implementation.
+    for (stamp_at, clock, expected) in [
+        (
+            StampAt::Receive,
+            Clock::Wall,
+            TimestampFields {
+                recv_wall: Some(1_000_000),
+                ..TimestampFields::default()
+            },
+        ),
+        (
+            StampAt::Both,
+            Clock::Both,
+            TimestampFields {
+                recv_wall: Some(1_000_000),
+                recv_mono: Some(10_000),
+                send_wall: Some(1_200_000),
+                send_mono: Some(10_300),
+                ..TimestampFields::default()
+            },
+        ),
+        (
+            StampAt::Midpoint,
+            Clock::Both,
+            TimestampFields {
+                midpoint_wall: Some(1_100_000),
+                midpoint_mono: Some(10_150),
+                ..TimestampFields::default()
+            },
+        ),
+    ] {
+        assert_eq!(
+            echo_timestamps(stamp_at, clock, None),
+            expected,
+            "{stamp_at:?} on {clock:?} without kernel metadata"
+        );
+    }
+}
+
+#[test]
+fn an_implausible_kernel_reading_falls_back_to_the_servers_own_sample() {
+    // Case A of the wall-step analysis: the wall clock stepped *backwards*
+    // between the kernel's reading and the core's sample, so the kernel value
+    // is later than a sample that observed the same datagram. Case B: it
+    // stepped forwards so far that the reading lags beyond the bound. Neither
+    // is a reason to drop the datagram or to answer differently in any other
+    // respect.
+    let userspace_wall = received().wall_ns;
+    for kernel in [
+        at_epoch_ns(userspace_wall + 1),
+        at_epoch_ns(userspace_wall - i64::try_from(MAX_KERNEL_RX_LAG.as_nanos()).unwrap() - 1),
+    ] {
+        assert_eq!(
+            echo_timestamps(StampAt::Both, Clock::Both, Some(kernel)),
+            TimestampFields {
+                recv_wall: Some(1_000_000),
+                recv_mono: Some(10_000),
+                send_wall: Some(1_200_000),
+                send_mono: Some(10_300),
+                ..TimestampFields::default()
+            },
+        );
+    }
+}
+
+#[test]
+fn a_backward_wall_step_after_receive_never_reports_receive_after_send() {
+    // The kernel's reading and the send sample no longer come from one paired
+    // source, so the ordering rule has to reach the reported arrival value on
+    // its own: an accepted kernel reading of 900_000, against a send sample the
+    // wall clock stepped back to 850_000, must still settle on the send
+    // reading.
+    let mut core = core_with_sources(
+        ServerConfig::default(),
+        ScriptedTokens::new([TOKEN_A]),
+        ScriptedClock::new([
+            open_sample(),
+            sample(1_000_000, 10_000),
+            sample(850_000, 10_300),
+        ]),
+    );
+    let params = echo_params(ReceivedStats::None, StampAt::Both, Clock::Both, 0);
+    let (token, negotiated) = open_negotiated(&mut core, peer(), &params, None);
+
+    let packet = core
+        .handle_received_datagram(
+            peer(),
+            &echo_request(token, 0, &negotiated, &[], None),
+            Some(at_epoch_ns(900_000)),
+        )
+        .unwrap()
+        .expect("a backward wall step must not stop the echo being answered");
+    let timestamps = expect_echo_reply(&packet, &negotiated, None).timestamps;
+
+    assert_eq!(
+        timestamps,
+        TimestampFields {
+            recv_wall: Some(850_000),
+            recv_mono: Some(10_000),
+            send_wall: Some(850_000),
+            send_mono: Some(10_300),
+            ..TimestampFields::default()
+        },
+        "the reported arrival wall value is held back to the send reading"
+    );
+    assert!(timestamps.recv_wall <= timestamps.send_wall);
+    // The monotonic pair is ordered by its own source and owes nothing to what
+    // the wall clock did.
+    assert_eq!(timestamps.recv_mono, Some(10_000));
+    assert!(timestamps.recv_mono <= timestamps.send_mono);
+}
+
+#[test]
+fn the_public_core_path_emits_exactly_the_userspace_receive_wall() {
+    // A direct `ServerCore` user has no transport metadata and must see the
+    // behavior it always did, from the same implementation the runtime path
+    // uses.
+    let mut core = core_with_sources(
+        ServerConfig::default(),
+        ScriptedTokens::new([TOKEN_A]),
+        ScriptedClock::new([open_sample(), received(), sent()]),
+    );
+    let params = echo_params(ReceivedStats::None, StampAt::Both, Clock::Both, 0);
+    let (token, negotiated) = open_negotiated(&mut core, peer(), &params, None);
+
+    let packet = core
+        .handle_datagram(peer(), &echo_request(token, 0, &negotiated, &[], None))
+        .unwrap()
+        .expect("the public path must answer an ordinary echo");
+
+    assert_eq!(
+        expect_echo_reply(&packet, &negotiated, None).timestamps,
+        TimestampFields {
+            recv_wall: Some(1_000_000),
+            recv_mono: Some(10_000),
+            send_wall: Some(1_200_000),
+            send_mono: Some(10_300),
+            ..TimestampFields::default()
+        },
+    );
+}
+
+/// One echo's outcome, in the terms a lifecycle test cares about: whether it was
+/// answered at all, what it counted, and how much session state the server still
+/// holds afterwards.
+#[derive(Debug, PartialEq, Eq)]
+struct Outcome {
+    answered: bool,
+    closing: bool,
+    recv_count: Option<u32>,
+    recv_window: Option<u64>,
+    session_count: usize,
+    recv_wall: Option<i64>,
+}
+
+fn outcome_of(
+    core: &ServerCore,
+    reply: Option<OutboundDatagram>,
+    params: &Params,
+    closing: bool,
+) -> Outcome {
+    let Some(reply) = reply else {
+        return Outcome {
+            answered: false,
+            closing: false,
+            recv_count: None,
+            recv_window: None,
+            session_count: core.session_count(),
+            recv_wall: None,
+        };
+    };
+    let decoded = if closing {
+        expect_closing_echo_reply(&reply, params, None)
+    } else {
+        expect_echo_reply(&reply, params, None)
+    };
+    Outcome {
+        answered: true,
+        closing,
+        recv_count: decoded.recv_count,
+        recv_window: decoded.recv_window,
+        session_count: core.session_count(),
+        recv_wall: decoded.timestamps.recv_wall,
+    }
+}
+
+/// Sends one echo as it would arrive at `at_ns`, with the kernel having observed
+/// it 1 µs earlier when `with_kernel_rx` is set.
+fn echo_at_with_kernel_rx(
+    core: &mut ServerCore,
+    clock: &ManualClock,
+    at_ns: i64,
+    token: u64,
+    sequence: u32,
+    params: &Params,
+    with_kernel_rx: bool,
+) -> Option<OutboundDatagram> {
+    if !with_kernel_rx {
+        return echo_at(core, clock, at_ns, token, sequence, params, None);
+    }
+    clock.set(at_ns);
+    core.handle_received_datagram(
+        peer(),
+        &echo_request(token, sequence, params, &[], None),
+        Some(at_epoch_ns(at_ns - 1_000)),
+    )
+    .expect("a rejected or rate-limited echo is not an internal error")
+}
+
+#[test]
+fn kernel_metadata_changes_no_lifecycle_rate_or_deadline_decision() {
+    // Two identical runs over one scripted arrival schedule, one with a
+    // plausible kernel reading on every echo and one with none. The schedule
+    // crosses a rate-limit burst, a refill, an idle refresh that only a
+    // rate-limited request can have produced, and the maximum-duration
+    // boundary — and the two runs must agree on every one of those outcomes.
+    //
+    // The session reports its receive wall, so the one permitted difference is
+    // visible rather than merely assumed absent.
+    let params = Params {
+        interval_ns: 100 * MS,
+        ..echo_params(ReceivedStats::Both, StampAt::Receive, Clock::Wall, 0)
+    };
+    let config = ServerConfig::default()
+        .with_min_send_interval(Duration::from_millis(100))
+        .with_burst_allowance(2)
+        .with_idle_timeout(Duration::from_secs(2))
+        .with_max_test_duration(Duration::from_secs(1));
+
+    let run = |with_kernel_rx: bool| {
+        let (mut core, clock, token, negotiated) = manual_core(
+            config.clone(),
+            &params,
+            ScriptedTokens::new([TOKEN_A]),
+            None,
+        );
+        let mut outcomes = Vec::new();
+        let mut send = |core: &mut ServerCore, at_ns: i64, sequence: u32, closing: bool| {
+            let reply = echo_at_with_kernel_rx(
+                core,
+                &clock,
+                at_ns,
+                token,
+                sequence,
+                &negotiated,
+                with_kernel_rx,
+            );
+            outcomes.push(outcome_of(core, reply, &negotiated, closing));
+        };
+
+        // A burst of three against an allowance of two: the third is
+        // rate-limited, answers nothing and advances no statistic, but does
+        // refresh the idle deadline.
+        send(&mut core, 10 * MS, 0, false);
+        send(&mut core, 10 * MS, 1, false);
+        send(&mut core, 10 * MS, 2, false);
+        // Past the refill, and out of order, so the window transition is
+        // exercised too.
+        send(&mut core, 200 * MS, 4, false);
+        send(&mut core, 400 * MS, 3, false);
+        // The idle timeout is 2 s and the last activity was at 400 ms, so this
+        // request at 2.3 s finds the session alive only because activity — not
+        // the open — is what the deadline runs from.
+        send(&mut core, 2_300 * MS, 5, false);
+        // The maximum-duration deadline is the first served echo plus one
+        // second plus the two-second grace: 10 ms + 3 s. This crosses it and
+        // carries the close, releasing the session.
+        send(&mut core, 3_100 * MS, 6, true);
+        // And every later request is an unknown token.
+        send(&mut core, 3_200 * MS, 7, false);
+        assert_eq!(core.session_count(), 0);
+        outcomes
+    };
+
+    let without = run(false);
+    let with = run(true);
+
+    assert_eq!(
+        without.len(),
+        8,
+        "the schedule must actually have run to the end"
+    );
+    for (index, (without, with)) in without.iter().zip(&with).enumerate() {
+        assert_eq!(
+            (
+                with.answered,
+                with.closing,
+                with.recv_count,
+                with.recv_window,
+                with.session_count
+            ),
+            (
+                without.answered,
+                without.closing,
+                without.recv_count,
+                without.recv_window,
+                without.session_count
+            ),
+            "echo {index}: kernel metadata may not change any same-host decision"
+        );
+    }
+
+    // The permitted difference, and proof the runs were not simply identical:
+    // every answered reply reported the kernel's reading rather than the
+    // server's own, which is 1 µs later.
+    let reported: Vec<_> = with
+        .iter()
+        .filter_map(|outcome| outcome.recv_wall)
+        .collect();
+    let userspace: Vec<_> = without
+        .iter()
+        .filter_map(|outcome| outcome.recv_wall)
+        .collect();
+    assert_eq!(reported.len(), 6, "six of the eight echoes were answered");
+    assert_eq!(
+        reported,
+        userspace
+            .iter()
+            .map(|wall| wall - 1_000)
+            .collect::<Vec<_>>(),
+        "each answered reply reported the kernel's arrival reading"
+    );
+}
+
+#[test]
+fn an_open_or_close_ignores_kernel_metadata_entirely() {
+    // Neither emits a receive timestamp, and neither may take a different
+    // lifecycle decision for having been observed. The clock script is what
+    // pins that: an open consumes exactly its own lifecycle sample whatever the
+    // transport reported.
+    let clock = ScriptedClock::new([open_sample(), sample(1_000_000, 10_000)]);
+    let mut core = core_with_sources(
+        ServerConfig::default(),
+        ScriptedTokens::new([TOKEN_A]),
+        clock.clone(),
+    );
+    let params = echo_params(ReceivedStats::None, StampAt::Both, Clock::Both, 0);
+
+    let packet = core
+        .handle_received_datagram(peer(), &open_request(&params, None), Some(kernel_rx()))
+        .unwrap()
+        .expect("an open must be answered");
+    let reply = expect_normal_open_reply(&packet, None);
+    assert_eq!(
+        clock.remaining(),
+        1,
+        "an open consumes one lifecycle sample"
+    );
+    assert_eq!(core.session_count(), 1);
+
+    let closed: SocketAddr = peer();
+    assert!(core
+        .handle_received_datagram(closed, &close_request(reply.token, None), Some(kernel_rx()))
+        .unwrap()
+        .is_none());
+    assert_eq!(core.session_count(), 0, "the close released its session");
+    assert_eq!(
+        clock.remaining(),
+        0,
+        "and a close consumes one of its own, whatever the transport observed"
+    );
+}

crates/irtt-server/src/tests/mod.rs

@@ -16,6 +16,7 @@ mod close;
 mod echo;
 mod fill;
 mod hmac;
+mod kernel_rx;
 mod lifecycle;
 mod negotiation;
 mod no_test;