feat(server): apply reply traffic class

crates/irtt-server/AGENTS.md

@@ -157,8 +157,49 @@ effective parameters must be safe for the server to run.
 - Reply payload bytes are zero, which is the initial safe fill policy: payload
   carries no protocol meaning, and a server must never emit residue from another
   request or client. Request payload bytes never reach a reply.
-- DSCP socket application and the full `ServerFill` policy are later slices. Do
-  not add placeholder session fields for them.
+- The full `ServerFill` policy is a later slice. Do not add placeholder session
+  fields for it.
+
+## Reply traffic class
+
+The core hands each reply out as an `OutboundDatagram`: the bytes plus the
+transport policy sending them requires. The runtime must not rediscover that
+policy by decoding the outgoing packet, inspecting the inbound token, reaching
+into a `Session` or keeping its own token-to-marking map — the core is where
+session policy lives.
+
+- `traffic_class` is the **raw 8-bit IPv4 TOS / IPv6 Traffic Class byte**, which
+  is what the DSCP protocol parameter carries. The six-bit codepoint is
+  `traffic_class >> 2` and the low two bits are ECN. Never shift, mask or round
+  it inside the server.
+- Open and no-test replies always request class zero, whatever the session
+  negotiates. That is observed behavior, not a simplification.
+- An echo reply requests the session's negotiated byte, and a server-close echo
+  reply keeps it: the close is an ordinary reply with a flag added, not a
+  special packet.
+- A negotiated integer outside `0..=255` stays in `Params` — returned and stored
+  exactly as requested — but is transported unmarked. This is **`irtt-rs`
+  policy**: the clean evidence records the reference host's handling of such
+  values as platform-specific and explicitly not a compatibility requirement, so
+  there is nothing to reproduce, and a malformed-but-accepted session should
+  stay usable rather than be given a marking nobody asked for. Do not add
+  negotiation-time rejection, clamping or wrapping for it.
+- The runtime applies a class before **every** send. Zero must be applied as
+  explicitly as a nonzero value: skipping the call would send an open reply, or
+  an unmarked session's reply, under whichever marking the previous reply left
+  on the shared socket. There is no restore-to-zero-after-send step, and no
+  cached "current value" to elide the call — the next reply is authoritative.
+- **Sequential send ownership is what makes a socket-wide setting race-free.**
+  One task owns the socket and no other sends from it, including while a send is
+  suspended. Do not introduce concurrent sends without replacing this with
+  per-packet control messages.
+- Applying the class and sending are both per-packet transport steps: a failure
+  of either drops that one reply and the listener keeps serving. A class that
+  could not be applied means the reply is not sent at all, because sending it
+  would put it on the wire under the previous reply's marking.
+- The option itself comes from `socket2`'s safe API, chosen by the listener's
+  bound address family rather than the peer's. The crate forbids unsafe code;
+  there is no raw `setsockopt` and no `sendmsg`/`cmsg` machinery.
 
 ## Rate, lifetime and server-initiated close
 

crates/irtt-server/Cargo.toml

@@ -14,8 +14,9 @@ categories = ["network-programming"]
 [dependencies]
 irtt-proto.workspace = true
 getrandom = "0.4.3"
+socket2 = { version = "0.6.5", features = ["all"] }
 thiserror = "2.0.20"
 tokio = { version = "1.53.1", default-features = false, features = ["rt", "net", "time", "sync", "macros"] }
 
 [dev-dependencies]
-irtt-client = { workspace = true, features = ["tokio"] }
+irtt-client = { workspace = true, features = ["ancillary", "tokio"] }

crates/irtt-server/src/core.rs

@@ -718,9 +718,10 @@ fn same_endpoint(session: SocketAddr, peer: SocketAddr) -> bool {
 /// Every other known parameter is already validated or deliberately
 /// unrestricted here. Protocol version is accepted at any value and rewritten
 /// during negotiation. Length and DSCP are accepted as decoded, including zero,
-/// negative and out-of-byte-range values; a later slice may restrict a DSCP it
-/// cannot actually apply to the socket. Unknown tags were ignored by the
-/// decoder and are not reflected in the reply.
+/// negative and out-of-byte-range values; a DSCP the socket could never carry
+/// is negotiated unchanged and transported unmarked instead of being refused
+/// (see [`transport_traffic_class`]). Unknown tags were ignored by the decoder
+/// and are not reflected in the reply.
 fn open_params_are_admissible(decoded: &DecodedParams) -> bool {
     let admissible = |present: bool, value: i64| !present || value > 0;
     admissible(decoded.presence.duration_ns, decoded.params.duration_ns)

crates/irtt-server/src/lib.rs

@@ -43,9 +43,10 @@
 //!
 //! Open handling and negotiation, session creation, normal echo processing,
 //! per-session rate limiting, idle expiry, the maximum-duration
-//! server-initiated close and Tokio UDP orchestration. DSCP application, the
-//! full server fill policy, wildcard-bind destination-address metadata and the
-//! CLI server applet remain separate slices. Echo payloads are zero-filled.
+//! server-initiated close, the negotiated per-session reply traffic class and
+//! Tokio UDP orchestration. The full server fill policy and wildcard-bind
+//! destination-address metadata remain separate slices. Echo payloads are
+//! zero-filled.
 #![forbid(unsafe_code)]
 
 mod clock;
@@ -55,6 +56,7 @@ mod error;
 mod negotiate;
 mod runtime;
 mod session;
+mod socket_options;
 mod token;
 
 #[cfg(test)]

crates/irtt-server/src/runtime.rs

@@ -2,10 +2,11 @@
 
 use std::{future::Future, io, net::SocketAddr, time::Duration};
 
+use socket2::SockRef;
 use thiserror::Error;
 use tokio::{net::UdpSocket, time::MissedTickBehavior};
 
-use crate::{ServerConfig, ServerCore, ServerError};
+use crate::{socket_options::set_reply_traffic_class, ServerConfig, ServerCore, ServerError};
 
 /// Fixed transport receive capacity.
 ///
@@ -21,6 +22,14 @@ const MAINTENANCE_INTERVAL: Duration = Duration::from_secs(1);
 /// process it, and sends the optional reply from the same socket. No
 /// per-datagram tasks or locks are involved.
 ///
+/// **That sequential ownership is what makes the reply traffic class correct.**
+/// The class is a socket-wide setting, and every reply explicitly applies its
+/// own — zero included — immediately before its send, so the reply about to go
+/// out is always the one that set it. Nothing else sends from this socket, not
+/// even while a send is suspended, so no reply can inherit another session's
+/// marking. Adding a concurrent sender would break that and require per-packet
+/// control messages instead.
+///
 /// Bind an explicit local address when reply source-address identity matters.
 /// Wildcard binds rely on the kernel's source-address choice; destination
 /// packet metadata and per-packet source selection are not yet implemented for
@@ -30,6 +39,11 @@ pub struct Server {
     socket: UdpSocket,
     core: ServerCore,
     recv_buffer: Vec<u8>,
+    /// The listener's own address family, which selects the IPv4 TOS or IPv6
+    /// Traffic Class option. `None` only if the bound address could not be
+    /// queried at construction, in which case the send path asks again rather
+    /// than guessing a family.
+    listener_is_ipv4: Option<bool>,
 }
 
 impl Server {
@@ -45,13 +59,30 @@ impl Server {
     /// Wraps an already prepared Tokio UDP socket in a server.
     #[must_use]
     pub fn from_socket(socket: UdpSocket, config: ServerConfig) -> Self {
+        let listener_is_ipv4 = socket.local_addr().ok().map(|addr| addr.is_ipv4());
         Self {
             socket,
             core: ServerCore::new(config),
             recv_buffer: vec![0; RECEIVE_BUFFER_LEN],
+            listener_is_ipv4,
         }
     }
 
+    /// Applies one reply's raw traffic class to the listener socket.
+    ///
+    /// This runs before *every* send, and zero is applied as deliberately as
+    /// any other value: a listener serves many sessions from one socket, so
+    /// skipping the call for an unmarked reply — an open reply, or a session
+    /// that negotiated nothing — would send it under whichever marking the
+    /// previous reply left behind.
+    fn apply_reply_traffic_class(&self, traffic_class: u8) -> io::Result<()> {
+        let is_ipv4 = match self.listener_is_ipv4 {
+            Some(is_ipv4) => is_ipv4,
+            None => self.socket.local_addr()?.is_ipv4(),
+        };
+        set_reply_traffic_class(SockRef::from(&self.socket), is_ipv4, traffic_class)
+    }
+
     /// Returns the local endpoint selected for the listener.
     pub fn local_addr(&self) -> Result<SocketAddr, ServerRuntimeError> {
         self.socket
@@ -64,8 +95,13 @@ impl Server {
     /// Graceful shutdown sends no session-close packets and leaves no hidden
     /// receive task behind, even while a reply send is waiting for socket
     /// writability. Per-packet send failures and short sends drop only that
-    /// reply; receive failures and internal core failures terminate the loop
-    /// with an error.
+    /// reply, and so does a failure to apply that reply's traffic class —
+    /// preparing one packet is not a reason to stop serving every other
+    /// session, and the next reply makes its own independent attempt. Receive
+    /// failures and internal core failures terminate the loop with an error.
+    ///
+    /// A reply whose class could not be applied is never sent anyway: sending it
+    /// would put it on the wire under the previous reply's marking.
     pub async fn run<F>(&mut self, shutdown: F) -> Result<(), ServerRuntimeError>
     where
         F: Future<Output = ()>,
@@ -90,6 +126,9 @@ impl Server {
                     }
 
                     if let Some(reply) = self.core.handle_datagram(peer, &self.recv_buffer[..len])? {
+                        if self.apply_reply_traffic_class(reply.traffic_class()).is_err() {
+                            continue;
+                        }
                         let send = self.socket.send_to(reply.bytes(), peer);
                         tokio::pin!(send);
                         let sent = loop {

crates/irtt-server/src/socket_options.rs

@@ -0,0 +1,120 @@
+//! Applying a reply's traffic class to the listener socket.
+//!
+//! The value is the raw IPv4 TOS / IPv6 Traffic Class byte an
+//! [`OutboundDatagram`](crate::OutboundDatagram) asks for, and which option
+//! carries it is decided by the listener's own address family — one `Server`
+//! owns one bound socket, so the peer a reply happens to be going to never
+//! selects the option.
+//!
+//! Everything here is `socket2`'s safe API; the crate forbids unsafe code and
+//! there is no raw `setsockopt` or per-packet control message anywhere in the
+//! server.
+
+use std::io;
+
+use socket2::SockRef;
+
+/// Applies `traffic_class` to a socket bound in the given family.
+///
+/// `is_ipv4` is the *listener's* family, not the peer's.
+pub(crate) fn set_reply_traffic_class(
+    socket: SockRef<'_>,
+    is_ipv4: bool,
+    traffic_class: u8,
+) -> io::Result<()> {
+    let traffic_class = u32::from(traffic_class);
+    if is_ipv4 {
+        set_ipv4_traffic_class(socket, traffic_class)
+    } else {
+        set_ipv6_traffic_class(socket, traffic_class)
+    }
+}
+
+#[cfg(not(any(
+    target_os = "fuchsia",
+    target_os = "redox",
+    target_os = "solaris",
+    target_os = "illumos",
+    target_os = "haiku",
+)))]
+fn set_ipv4_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
+    socket.set_tos_v4(traffic_class)
+}
+
+#[cfg(any(
+    target_os = "fuchsia",
+    target_os = "redox",
+    target_os = "solaris",
+    target_os = "illumos",
+    target_os = "haiku",
+))]
+fn set_ipv4_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
+    unsupported_traffic_class(traffic_class, "IPv4 TOS socket options")
+}
+
+#[cfg(any(
+    target_os = "android",
+    target_os = "dragonfly",
+    target_os = "freebsd",
+    target_os = "fuchsia",
+    target_os = "linux",
+    target_os = "macos",
+    target_os = "netbsd",
+    target_os = "openbsd",
+    target_os = "cygwin",
+))]
+fn set_ipv6_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
+    socket.set_tclass_v6(traffic_class)
+}
+
+#[cfg(not(any(
+    target_os = "android",
+    target_os = "dragonfly",
+    target_os = "freebsd",
+    target_os = "fuchsia",
+    target_os = "linux",
+    target_os = "macos",
+    target_os = "netbsd",
+    target_os = "openbsd",
+    target_os = "cygwin",
+)))]
+fn set_ipv6_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
+    unsupported_traffic_class(traffic_class, "IPv6 Traffic Class socket options")
+}
+
+/// A target where the safe API cannot express this option.
+///
+/// Requesting *unmarked* still succeeds, because a platform that can never
+/// apply a nonzero class also never left one on the socket for the next reply to
+/// inherit — the invariant the explicit zero exists to protect. Asking for a
+/// real marking reports [`io::ErrorKind::Unsupported`] rather than pretending:
+/// the runtime drops that one reply and stays alive, and no reply ever goes out
+/// carrying a class the server did not intend.
+#[cfg(any(
+    target_os = "fuchsia",
+    target_os = "redox",
+    target_os = "solaris",
+    target_os = "illumos",
+    target_os = "haiku",
+    not(any(
+        target_os = "android",
+        target_os = "dragonfly",
+        target_os = "freebsd",
+        target_os = "fuchsia",
+        target_os = "linux",
+        target_os = "macos",
+        target_os = "netbsd",
+        target_os = "openbsd",
+        target_os = "cygwin",
+    )),
+))]
+fn unsupported_traffic_class(traffic_class: u32, option: &'static str) -> io::Result<()> {
+    if traffic_class == 0 {
+        Ok(())
+    } else {
+        Err(io::Error::new(
+            io::ErrorKind::Unsupported,
+            format!("{option} are unsupported on this target"),
+        ))
+    }
+}

crates/irtt-server/tests/traffic_class.rs

@@ -0,0 +1,225 @@
+//! What the server's replies actually carry on the wire.
+//!
+//! Everything else about reply marking is decided deterministically in the
+//! core, but "the core asked for 0xb8" is not the same claim as "the datagram
+//! left the host with TOS 0xb8". These tests close that gap with real sockets:
+//! a real [`AsyncClient`] against a real [`Server`], reading the traffic class
+//! the kernel reports for each reply it receives.
+//!
+//! Linux only, because the receive-side ancillary metadata the client exposes
+//! is. The production marking path is platform-independent and is compiled and
+//! exercised everywhere; only this observation of it is not.
+
+#![cfg(target_os = "linux")]
+
+use std::{io, net::SocketAddr, time::Duration};
+
+use irtt_client::{AsyncClient, ClientConfig, ClientEvent, OpenOutcome, PacketMeta, SocketConfig};
+use irtt_proto::{Clock, ReceivedStats, StampAt};
+use irtt_server::{Server, ServerConfig, ServerRuntimeError};
+use tokio::{sync::oneshot, task::JoinHandle, time::timeout};
+
+/// DSCP codepoints, as the client's configuration takes them, paired with the
+/// raw TOS / Traffic Class byte each one occupies the upper six bits of. The
+/// server negotiates and applies the raw byte; the shift is the client's
+/// user-facing convention.
+const EF: (u8, u8) = (46, 0xb8);
+const CS1: (u8, u8) = (8, 0x20);
+const UNMARKED: (u8, u8) = (0, 0x00);
+
+#[tokio::test(flavor = "current_thread")]
+async fn an_echo_reply_leaves_the_host_carrying_the_negotiated_traffic_class() {
+    timeout(Duration::from_secs(5), async {
+        let (server, addr) = bound_server("127.0.0.1:0".parse().unwrap())
+            .await
+            .expect("IPv4 loopback must be available");
+        let running = spawn(server);
+
+        let mut session = opened_session(addr, EF.0).await;
+        let meta = session.echo_packet_meta().await;
+        assert_eq!(meta.traffic_class, Some(EF.1), "the raw negotiated byte");
+        assert_eq!(meta.dscp, Some(EF.0), "its upper six bits");
+        assert_eq!(meta.ecn, Some(0), "and its low two");
+
+        running.stop().await;
+    })
+    .await
+    .expect("the wire traffic-class test exceeded its bounded runtime");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn one_listener_serving_many_sessions_leaks_no_marking_between_them() {
+    // The invariant that justifies applying the class before *every* send. One
+    // socket carries all three sessions, so a reply that did not set its own
+    // class would go out under whichever one the previous reply left behind.
+    // The order below returns to a marked session after another has sent, and
+    // ends on the unmarked one, which is the case a "skip the call for zero"
+    // optimization would break.
+    timeout(Duration::from_secs(10), async {
+        let (server, addr) = bound_server("127.0.0.1:0".parse().unwrap())
+            .await
+            .expect("IPv4 loopback must be available");
+        let running = spawn(server);
+
+        const MARKED: usize = 0;
+        const LIGHTLY_MARKED: usize = 1;
+        const UNMARKED_SESSION: usize = 2;
+        let mut sessions = [
+            opened_session(addr, EF.0).await,
+            opened_session(addr, CS1.0).await,
+            opened_session(addr, UNMARKED.0).await,
+        ];
+
+        for (label, session, expected) in [
+            ("first marked echo", MARKED, EF.1),
+            ("a differently marked session", LIGHTLY_MARKED, CS1.1),
+            ("back to the first session", MARKED, EF.1),
+            ("an unmarked session last", UNMARKED_SESSION, UNMARKED.1),
+        ] {
+            let meta = sessions[session].echo_packet_meta().await;
+            assert_eq!(
+                meta.traffic_class,
+                Some(expected),
+                "{label}: each reply carries its own session's marking"
+            );
+        }
+
+        running.stop().await;
+    })
+    .await
+    .expect("the cross-session leakage test exceeded its bounded runtime");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn an_ipv6_echo_reply_carries_the_negotiated_traffic_class() {
+    // One marked echo is enough here: the cross-session matrix above already
+    // pins the policy, and what this adds is the IPv6 Traffic Class option
+    // rather than the IPv4 TOS one.
+    timeout(Duration::from_secs(5), async {
+        let Some((server, addr)) = bound_server("[::1]:0".parse().unwrap()).await else {
+            return;
+        };
+        let running = spawn(server);
+
+        let mut session = opened_session(addr, EF.0).await;
+        let meta = session.echo_packet_meta().await;
+        assert_eq!(meta.traffic_class, Some(EF.1));
+
+        running.stop().await;
+    })
+    .await
+    .expect("the IPv6 wire traffic-class test exceeded its bounded runtime");
+}
+
+/// Binds a listener, or reports the family as unavailable on this host.
+///
+/// Only an address family this host does not have is skipped; any other bind
+/// failure is a real failure.
+async fn bound_server(addr: SocketAddr) -> Option<(Server, SocketAddr)> {
+    let server = match Server::bind(addr, unthrottled()).await {
+        Ok(server) => server,
+        Err(ServerRuntimeError::Bind { source, .. })
+            if matches!(
+                source.kind(),
+                io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported
+            ) =>
+        {
+            return None;
+        }
+        Err(error) => panic!("unexpected loopback bind failure: {error}"),
+    };
+    let addr = server.local_addr().unwrap();
+    Some((server, addr))
+}
+
+/// A server that rate-limits nothing, so the cadence these tests drive replies
+/// at cannot turn a marking assertion into a rate-limiting one.
+fn unthrottled() -> ServerConfig {
+    ServerConfig::default().with_min_send_interval(Duration::ZERO)
+}
+
+/// A running listener and the handle that stops it.
+struct Running {
+    shutdown: oneshot::Sender<()>,
+    task: JoinHandle<Result<(), ServerRuntimeError>>,
+}
+
+impl Running {
+    async fn stop(self) {
+        self.shutdown.send(()).unwrap();
+        self.task.await.unwrap().unwrap();
+    }
+}
+
+fn spawn(mut server: Server) -> Running {
+    let (shutdown, shutdown_rx) = oneshot::channel();
+    let task = tokio::spawn(async move {
+        server
+            .run(async {
+                let _ = shutdown_rx.await;
+            })
+            .await
+    });
+    Running { shutdown, task }
+}
+
+/// One client with an open session, and the sequence number its next echo will
+/// carry — which the caller cannot infer once several sessions are interleaved.
+struct OpenSession {
+    client: AsyncClient,
+    next_sequence: u32,
+}
+
+impl OpenSession {
+    /// Sends one echo and returns the receive metadata of its reply.
+    async fn echo_packet_meta(&mut self) -> PacketMeta {
+        let sequence = self.next_sequence;
+        self.next_sequence += 1;
+        self.client.send_probe().await.unwrap();
+        loop {
+            for event in self.client.recv().await.unwrap() {
+                if let ClientEvent::EchoReply {
+                    seq, packet_meta, ..
+                } = event
+                {
+                    assert_eq!(seq, sequence);
+                    return packet_meta;
+                }
+            }
+        }
+    }
+}
+
+/// A client with an open session requesting `dscp` as a codepoint.
+async fn opened_session(server_addr: SocketAddr, dscp: u8) -> OpenSession {
+    let config = ClientConfig {
+        server_addr: server_addr.to_string(),
+        duration: Some(Duration::from_secs(30)),
+        interval: Duration::from_millis(10),
+        length: 64,
+        dscp,
+        received_stats: ReceivedStats::Both,
+        stamp_at: StampAt::Both,
+        clock: Clock::Both,
+        open_timeouts: vec![Duration::from_millis(500)],
+        socket_config: SocketConfig {
+            recv_timeout: Some(Duration::from_millis(500)),
+            ..SocketConfig::default()
+        },
+        probe_timeout: Duration::from_millis(500),
+        ..ClientConfig::default()
+    };
+    let mut client = AsyncClient::connect(config).await.unwrap();
+    match client.open().await.unwrap() {
+        OpenOutcome::Started { negotiated, .. } => assert_eq!(
+            negotiated.params.dscp,
+            i64::from(dscp) << 2,
+            "the server negotiates the raw byte the codepoint asks for"
+        ),
+        OpenOutcome::NoTestCompleted { .. } => panic!("a normal client unexpectedly ran no-test"),
+    }
+    OpenSession {
+        client,
+        next_sequence: 0,
+    }
+}