feat(server): capture kernel receive timestamps

crates/irtt-server/src/socket_io/ancillary.rs

@@ -1,7 +1,21 @@
-//! Destination-address ancillary data on Linux, macOS and FreeBSD.
+//! Ancillary receive data on Linux, macOS and FreeBSD.
 //!
-//! Three platforms, two mechanisms per address family, all reached through
-//! `nix`'s safe wrappers:
+//! Two independent kinds of metadata travel with a received datagram here, and
+//! they are not equally important:
+//!
+//! - the request's **local destination address**, which a wildcard listener
+//!   must have to answer from the endpoint the client contacted. Losing it
+//!   makes the request unanswerable, so the datagram is dropped.
+//! - the kernel's **software receive timestamp**, an accuracy enhancement that
+//!   Linux listeners request opportunistically. Losing it costs nothing but
+//!   precision, so the datagram is served with no timestamp.
+//!
+//! Both are collected from one `recvmsg`, and neither is looked for at a fixed
+//! position: the kernel may return them in either order, so the whole usable
+//! control sequence is walked and each kind recorded where it appears.
+//!
+//! Three platforms, two destination mechanisms per address family, all reached
+//! through `nix`'s safe wrappers:
 //!
 //! | | receive | send |
 //! |---|---|---|
@@ -17,6 +31,11 @@
 //! receive of which only two mean anything on send. Each direction is built
 //! explicitly for that reason.
 //!
+//! On Linux the kernel receive timestamp arrives as `SCM_TIMESTAMPNS`, enabled
+//! by `SO_TIMESTAMPNS`. It is a software timestamp taken by the kernel's
+//! realtime clock as the datagram is received — not a NIC hardware timestamp,
+//! not a monotonic reading, and nothing to do with transmission.
+//!
 //! Both directions run nonblocking through Tokio readiness, following the same
 //! `readable()`/`try_io` shape the client's ancillary receive path uses. One
 //! socket, one task, no blocking call and no spawned thread.
@@ -25,23 +44,63 @@ use std::{
     io::{self, IoSlice, IoSliceMut},
     net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV6},
     os::fd::AsRawFd,
+    time::SystemTime,
 };
 
 use nix::{
     libc,
     sys::socket::{
-        recvmsg, sendmsg, setsockopt, sockopt, ControlMessage, ControlMessageOwned, MsgFlags,
-        RecvMsg, SockaddrStorage,
+        cmsg_space, recvmsg, sendmsg, setsockopt, sockopt, ControlMessage, ControlMessageOwned,
+        MsgFlags, RecvMsg, SockaddrStorage,
     },
 };
 use tokio::{io::Interest, net::UdpSocket};
 
+#[cfg(target_os = "linux")]
+use std::time::{Duration, UNIX_EPOCH};
+
+#[cfg(target_os = "linux")]
+use nix::sys::time::TimeSpec;
+
 use super::{ReceivedDatagram, ReplySource};
 
-/// Room for the destination control message. The largest of them is an
-/// `in6_pktinfo`, and this leaves headroom for a second message the kernel may
-/// add, so `MSG_CTRUNC` reflects a real problem rather than a tight buffer.
-const CONTROL_LEN: usize = 128;
+/// Room for the IPv4 destination control message, whose payload is the
+/// platform's own: Linux reports an `in_pktinfo`, the BSDs a bare `in_addr`.
+#[cfg(target_os = "linux")]
+const DESTINATION_V4_LEN: usize = cmsg_space::<libc::in_pktinfo>();
+#[cfg(any(target_os = "macos", target_os = "freebsd"))]
+const DESTINATION_V4_LEN: usize = cmsg_space::<libc::in_addr>();
+
+/// Room for the IPv6 destination control message, which is an `in6_pktinfo`
+/// everywhere here.
+const DESTINATION_V6_LEN: usize = cmsg_space::<libc::in6_pktinfo>();
+
+/// Room for the kernel receive timestamp, which only Linux is asked for.
+#[cfg(target_os = "linux")]
+const TIMESTAMP_LEN: usize = cmsg_space::<libc::timespec>();
+#[cfg(not(target_os = "linux"))]
+const TIMESTAMP_LEN: usize = 0;
+
+/// Capacity for every control message this server asks the kernel for, derived
+/// from the payload types themselves rather than from a round number believed
+/// to be large enough.
+///
+/// Both address families are counted although one socket only ever receives
+/// one of them: a few bytes of headroom is cheaper than a per-family buffer
+/// type, and a dual-stack listener's family is a runtime fact. Undersizing this
+/// would not fail loudly — it would set `MSG_CTRUNC` and turn valid wildcard
+/// requests into silent drops — so the sum is stated in full and proved by a
+/// receive test rather than by inspection.
+const CONTROL_LEN: usize = DESTINATION_V4_LEN + DESTINATION_V6_LEN + TIMESTAMP_LEN;
+
+/// Whichever family a socket turns out to be, its destination message and the
+/// timestamp must fit together. Checked at compile time on every target this
+/// module builds for, so a future control message that outgrows the buffer
+/// fails the build rather than the wildcard listeners that depend on it.
+const _: () = {
+    assert!(CONTROL_LEN >= DESTINATION_V4_LEN + TIMESTAMP_LEN);
+    assert!(CONTROL_LEN >= DESTINATION_V6_LEN + TIMESTAMP_LEN);
+};
 
 pub(super) fn configure_destination_metadata(socket: &UdpSocket, is_ipv4: bool) -> io::Result<()> {
     if is_ipv4 {
@@ -63,20 +122,59 @@ fn configure_ipv4_destination(socket: &UdpSocket) -> io::Result<()> {
     setsockopt(socket, sockopt::Ipv4RecvDstAddr, &true).map_err(io::Error::from)
 }
 
+/// Asks the kernel for a software receive timestamp on each datagram, and
+/// reports whether it agreed.
+///
+/// Deliberately infallible for the caller. Unlike destination metadata, this is
+/// an accuracy enhancement and not a correctness requirement: a listener that
+/// does not get it answers every request exactly as it otherwise would, just
+/// without the kernel's view of when each one arrived.
+#[cfg(target_os = "linux")]
+pub(super) fn try_configure_kernel_rx_timestamp(socket: &UdpSocket) -> bool {
+    setsockopt(socket, sockopt::ReceiveTimestampns, &true).is_ok()
+}
+
 pub(super) async fn receive_with_destination(
     socket: &UdpSocket,
     buffer: &mut [u8],
+) -> io::Result<Option<ReceivedDatagram>> {
+    receive_ancillary(socket, buffer, true).await
+}
+
+/// The Linux explicit-address receive: no destination metadata is needed or
+/// asked for, and the datagram is taken through `recvmsg` only so an enabled
+/// `SCM_TIMESTAMPNS` can be read off it.
+#[cfg(target_os = "linux")]
+pub(super) async fn receive_bound(
+    socket: &UdpSocket,
+    buffer: &mut [u8],
+) -> io::Result<Option<ReceivedDatagram>> {
+    receive_ancillary(socket, buffer, false).await
+}
+
+/// `require_destination` is the listener's own wildcard-ness. It decides both
+/// what this receive must recover and how severe a truncated control buffer is.
+async fn receive_ancillary(
+    socket: &UdpSocket,
+    buffer: &mut [u8],
+    require_destination: bool,
 ) -> io::Result<Option<ReceivedDatagram>> {
     loop {
         socket.readable().await?;
-        match socket.try_io(Interest::READABLE, || receive_once(socket, buffer)) {
+        match socket.try_io(Interest::READABLE, || {
+            receive_once(socket, buffer, require_destination)
+        }) {
             Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue,
             result => return result,
         }
     }
 }
 
-fn receive_once(socket: &UdpSocket, buffer: &mut [u8]) -> io::Result<Option<ReceivedDatagram>> {
+fn receive_once(
+    socket: &UdpSocket,
+    buffer: &mut [u8],
+    require_destination: bool,
+) -> io::Result<Option<ReceivedDatagram>> {
     let mut control = ControlBuffer::new();
     let mut iov = [IoSliceMut::new(buffer)];
     let message = recvmsg::<SockaddrStorage>(
@@ -86,7 +184,7 @@ fn receive_once(socket: &UdpSocket, buffer: &mut [u8]) -> io::Result<Option<Rece
         MsgFlags::MSG_DONTWAIT,
     )?;
 
-    Ok(received_datagram(&message))
+    Ok(received_datagram(&message, require_destination))
 }
 
 /// Turns one received message into a datagram the runtime may serve, or `None`
@@ -97,21 +195,57 @@ fn receive_once(socket: &UdpSocket, buffer: &mut [u8]) -> io::Result<Option<Rece
 /// no destination arrived with the datagram, that promise cannot be kept for
 /// this request — so it is dropped here, before the core sees it and before any
 /// session's receive, rate or lifetime state could move.
-fn received_datagram(message: &RecvMsg<'_, '_, SockaddrStorage>) -> Option<ReceivedDatagram> {
-    if message
-        .flags
-        .intersects(MsgFlags::MSG_TRUNC | MsgFlags::MSG_CTRUNC)
-    {
+///
+/// A kernel receive timestamp is never part of that judgement. Absent,
+/// truncated away or structurally unrepresentable, it simply does not appear on
+/// the datagram.
+fn received_datagram(
+    message: &RecvMsg<'_, '_, SockaddrStorage>,
+    require_destination: bool,
+) -> Option<ReceivedDatagram> {
+    if is_unusable(message.flags, require_destination) {
         return None;
     }
 
+    let metadata = receive_metadata(message);
     Some(ReceivedDatagram {
         len: message.bytes,
         peer: peer_endpoint(message.address.as_ref()?)?,
-        reply_source: reply_source(message)?,
+        reply_source: resolved_reply_source(metadata.reply_source, require_destination)?,
+        kernel_rx_timestamp: metadata.kernel_rx_timestamp,
     })
 }
 
+/// Whether the kernel's own report of this receive rules the datagram out.
+///
+/// A truncated payload always does: what arrived is not what was sent. A
+/// truncated *control* buffer only does where the destination was required,
+/// because then the metadata that may have been cut is the metadata the reply
+/// cannot be built without. For a listener that asked only for a timestamp,
+/// the same flag means at worst that the timestamp was lost — a reason to serve
+/// the request without one, not to discard a client's packet.
+fn is_unusable(flags: MsgFlags, require_destination: bool) -> bool {
+    flags.contains(MsgFlags::MSG_TRUNC)
+        || (require_destination && flags.contains(MsgFlags::MSG_CTRUNC))
+}
+
+/// Where this datagram's reply must leave from, given what the control data
+/// yielded.
+///
+/// An explicit-address listener has nothing to recover: it sends from the one
+/// address it is bound to. A wildcard listener has everything to recover, and a
+/// request whose destination did not arrive is one it cannot answer.
+fn resolved_reply_source(
+    recovered: Option<ReplySource>,
+    require_destination: bool,
+) -> Option<ReplySource> {
+    if require_destination {
+        recovered
+    } else {
+        Some(ReplySource::Bound)
+    }
+}
+
 /// The request's full source endpoint, keeping the IPv6 zone the core compares
 /// endpoints by.
 fn peer_endpoint(address: &SockaddrStorage) -> Option<SocketAddr> {
@@ -123,34 +257,84 @@ fn peer_endpoint(address: &SockaddrStorage) -> Option<SocketAddr> {
         .map(|peer| SocketAddr::from(*peer))
 }
 
-fn reply_source(message: &RecvMsg<'_, '_, SockaddrStorage>) -> Option<ReplySource> {
-    for control_message in message.cmsgs().ok()? {
+/// What one receive's control data carried, kind by kind.
+///
+/// Independent fields rather than an early return, because the kernel is free
+/// to order control messages as it likes: stopping at the destination would
+/// discard a timestamp that followed it, and stopping at the timestamp would
+/// discard the destination a wildcard reply depends on.
+#[derive(Default)]
+struct ReceiveMetadata {
+    reply_source: Option<ReplySource>,
+    /// Only Linux is ever asked for one, so on the BSDs this stays absent and
+    /// the datagram simply carries no kernel view of its arrival.
+    kernel_rx_timestamp: Option<SystemTime>,
+}
+
+fn receive_metadata(message: &RecvMsg<'_, '_, SockaddrStorage>) -> ReceiveMetadata {
+    let mut metadata = ReceiveMetadata::default();
+    // `cmsgs` refuses to walk a control buffer the kernel had to truncate
+    // (`MSG_CTRUNC`), because what is there cannot be trusted to be complete
+    // messages. Whether that is fatal was already decided by `is_unusable`;
+    // here it just means no metadata was recovered.
+    let Ok(control_messages) = message.cmsgs() else {
+        return metadata;
+    };
+
+    for control_message in control_messages {
         match control_message {
             // Linux reports the header destination in `ipi_addr`; `ipi_spec_dst`
             // is the route's local address, which is the same for the unicast
             // traffic a server sees and is not what the client addressed.
             #[cfg(target_os = "linux")]
             ControlMessageOwned::Ipv4PacketInfo(info) => {
-                return ipv4_reply_source(ipv4_addr(info.ipi_addr));
+                metadata.reply_source = ipv4_reply_source(ipv4_addr(info.ipi_addr));
             }
             #[cfg(any(target_os = "macos", target_os = "freebsd"))]
             ControlMessageOwned::Ipv4RecvDstAddr(address) => {
-                return ipv4_reply_source(ipv4_addr(address));
+                metadata.reply_source = ipv4_reply_source(ipv4_addr(address));
             }
             ControlMessageOwned::Ipv6PacketInfo(info) => {
-                let address = Ipv6Addr::from(info.ipi6_addr.s6_addr);
-                if address.is_unspecified() {
-                    return None;
-                }
-                return Some(ReplySource::V6 {
-                    address,
-                    interface_index: info.ipi6_ifindex,
-                });
+                metadata.reply_source = ipv6_reply_source(info);
+            }
+            #[cfg(target_os = "linux")]
+            ControlMessageOwned::ScmTimestampns(timestamp) => {
+                metadata.kernel_rx_timestamp = system_time_from_timespec(timestamp);
             }
             _ => {}
         }
     }
-    None
+    metadata
+}
+
+/// An unspecified destination is no destination: nothing can be sent from `::`,
+/// so the datagram is dropped rather than answered from a source the kernel
+/// would pick instead.
+fn ipv6_reply_source(info: libc::in6_pktinfo) -> Option<ReplySource> {
+    let address = Ipv6Addr::from(info.ipi6_addr.s6_addr);
+    (!address.is_unspecified()).then_some(ReplySource::V6 {
+        address,
+        interface_index: info.ipi6_ifindex,
+    })
+}
+
+/// A kernel `SCM_TIMESTAMPNS` reading as a wall-clock instant, or `None` where
+/// it is not one.
+///
+/// Purely structural: a negative or out-of-range `timespec`, or one past what
+/// `SystemTime` can represent, describes no instant and is reported as absent
+/// metadata. Whether a representable reading is *plausible* — close enough to
+/// the server's own userspace sample to be worth measuring against — is a
+/// question for whoever eventually measures with it, and is deliberately not
+/// asked here.
+#[cfg(target_os = "linux")]
+fn system_time_from_timespec(timespec: TimeSpec) -> Option<SystemTime> {
+    let seconds = u64::try_from(timespec.tv_sec()).ok()?;
+    let nanos = u32::try_from(timespec.tv_nsec()).ok()?;
+    if nanos >= 1_000_000_000 {
+        return None;
+    }
+    UNIX_EPOCH.checked_add(Duration::new(seconds, nanos))
 }
 
 /// An unspecified destination is no destination: nothing can be sent from
@@ -373,4 +557,113 @@ mod tests {
             );
         }
     }
+
+    /// What the two listener kinds do with the metadata they did or did not
+    /// get, which is the whole distinction this module rests on: a destination
+    /// is correctness-critical for one of them and irrelevant to the other.
+    #[test]
+    fn only_a_listener_that_needs_a_destination_is_stopped_by_a_missing_one() {
+        let recovered = ReplySource::V4 {
+            address: Ipv4Addr::LOCALHOST,
+        };
+
+        assert_eq!(
+            resolved_reply_source(Some(recovered), true),
+            Some(recovered),
+            "a wildcard listener replies from the destination it recovered"
+        );
+        assert_eq!(
+            resolved_reply_source(None, true),
+            None,
+            "and cannot answer a request whose destination did not arrive"
+        );
+        assert_eq!(
+            resolved_reply_source(None, false),
+            Some(ReplySource::Bound),
+            "an explicit listener never needed one"
+        );
+        assert_eq!(
+            resolved_reply_source(Some(recovered), false),
+            Some(ReplySource::Bound),
+            "and sends from its bind even if one turned up"
+        );
+    }
+
+    /// Which truncation is fatal to whom. A cut payload always is; cut control
+    /// data is fatal only where it may have taken the destination with it.
+    /// Turning a lost *timestamp* into a lost packet would trade a real client
+    /// request for optional precision.
+    #[test]
+    fn control_truncation_only_condemns_a_datagram_whose_destination_it_may_have_taken() {
+        for (flags, require_destination, unusable, why) in [
+            (MsgFlags::empty(), true, false, "a clean wildcard receive"),
+            (MsgFlags::empty(), false, false, "a clean explicit receive"),
+            (
+                MsgFlags::MSG_TRUNC,
+                false,
+                true,
+                "a cut payload is not what the client sent",
+            ),
+            (
+                MsgFlags::MSG_TRUNC,
+                true,
+                true,
+                "and no less so for a wildcard listener",
+            ),
+            (
+                MsgFlags::MSG_CTRUNC,
+                true,
+                true,
+                "cut control data may have taken the destination",
+            ),
+            (
+                MsgFlags::MSG_CTRUNC,
+                false,
+                false,
+                "but for an explicit listener it can only have cost a timestamp",
+            ),
+            (
+                MsgFlags::MSG_TRUNC | MsgFlags::MSG_CTRUNC,
+                false,
+                true,
+                "the payload rule still applies",
+            ),
+        ] {
+            assert_eq!(is_unusable(flags, require_destination), unusable, "{why}");
+        }
+    }
+
+    /// A representable kernel reading becomes an instant; anything else becomes
+    /// absent metadata. Note what is *not* rejected here: a reading far from
+    /// now is structurally fine, and judging it belongs to whoever measures
+    /// with it.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_kernel_timestamp_converts_only_where_it_describes_an_instant() {
+        assert_eq!(
+            system_time_from_timespec(TimeSpec::new(1, 2))
+                .unwrap()
+                .duration_since(UNIX_EPOCH)
+                .unwrap(),
+            Duration::new(1, 2)
+        );
+        assert_eq!(
+            system_time_from_timespec(TimeSpec::new(0, 0)),
+            Some(UNIX_EPOCH),
+            "the epoch itself is representable"
+        );
+
+        assert_eq!(system_time_from_timespec(TimeSpec::new(-1, 0)), None);
+        assert_eq!(system_time_from_timespec(TimeSpec::new(0, -1)), None);
+        assert_eq!(
+            system_time_from_timespec(TimeSpec::new(0, 1_000_000_000)),
+            None
+        );
+        assert_eq!(
+            system_time_from_timespec(TimeSpec::new(i64::MIN, i64::MIN)),
+            None,
+            "and the extremes convert rather than panic"
+        );
+        let _ = system_time_from_timespec(TimeSpec::new(i64::MAX, 999_999_999));
+    }
 }