Merge pull request #84 from Lochnair/feat/server-reply-traffic-class

feat(server): apply reply traffic class

AGENTS.md

@@ -57,7 +57,7 @@ Tokio must remain optional for `irtt-client`; default builds must remain runtime
 
 ### Server architecture (`irtt-server`)
 
-`irtt-server` is a first-class reusable server crate. It contains the deterministic OPEN/ECHO/session core — packet admission, authentication policy, open negotiation, the bounded session table, open reply construction, echo processing with its per-session receive state and timestamps, per-session rate limiting, session lifetime with idle expiry, the maximum-duration server-initiated close, and client-initiated close — plus a reusable Tokio UDP `Server` that runs one sequential `ServerCore` per listener with caller-controlled shutdown and scheduled idle maintenance. Logical expiry remains exact on authenticated, structurally valid requests, while the runtime also reclaims expired sessions once per second when no traffic arrives. DSCP application and the full server fill policy remain unimplemented; see `crates/irtt-server/AGENTS.md`. The rest of this section describes the implemented boundary and the parts not yet written.
+`irtt-server` is a first-class reusable server crate. It contains the deterministic OPEN/ECHO/session core — packet admission, authentication policy, open negotiation, the bounded session table, open reply construction, echo processing with its per-session receive state and timestamps, per-session rate limiting, session lifetime with idle expiry, the maximum-duration server-initiated close, and client-initiated close — plus a reusable Tokio UDP `Server` that runs one sequential `ServerCore` per listener with caller-controlled shutdown and scheduled idle maintenance. Logical expiry remains exact on authenticated, structurally valid requests, while the runtime also reclaims expired sessions once per second when no traffic arrives. Each reply carries the raw traffic class it must be sent with, which the runtime applies to the listener before every send. The full server fill policy and wildcard-bind packet-info handling remain unimplemented; see `crates/irtt-server/AGENTS.md`. The rest of this section describes the implemented boundary and the parts not yet written.
 
 - It shares wire semantics with the client through `irtt-proto` and follows the same clean-room boundary.
 - It must **not** depend on `irtt-client` merely to reuse protocol behavior.

Cargo.lock

@@ -735,6 +735,7 @@ dependencies = [
  "getrandom 0.4.3",
  "irtt-client",
  "irtt-proto",
+ "socket2",
  "thiserror 2.0.20",
  "tokio",
 ]

README.md

@@ -304,6 +304,7 @@ The client, event stream, machine-readable output, multi-target execution, local
 
 The server library and the server applet are implemented: open negotiation,
 session state, echo processing, per-session rate limiting, idle expiry, the
-maximum-duration close, and HMAC authentication. DSCP application, the full
+maximum-duration close, HMAC authentication, and the negotiated per-session
+reply traffic class, which is applied on sockets that support it. The full
 server fill policy, per-packet source-address selection for wildcard binds on
 multi-homed hosts, and multiple listeners in one process are not implemented.

crates/irtt-server/AGENTS.md

@@ -157,8 +157,63 @@ 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 marking 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.
+- **One exception, and it is what keeps the server usable on hosts without the
+  option**: an *unmarked* reply may still be sent when this server has never
+  successfully applied a nonzero class to the socket, because then there is no
+  marking of ours for it to inherit. Some Windows builds do not support
+  `IP_TOS`, and a few targets have no safe setter at all; without this, such a
+  host would drop every open reply and no session could ever start. The flag
+  behind it is consulted only on failure — it never elides a call.
+- The platform helper must report failure honestly, including for zero. "The
+  socket was cleared" and "nothing was written" are different statements, and a
+  socket passed to `Server::from_socket` may have been marked by its creator.
+  Whether an unappliable zero is nevertheless safe is the runtime's decision,
+  made from what this server itself applied.
+- 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. Keep the
+  target lists identical to the pinned `socket2` version's own gates — check
+  its source rather than copying another crate's older matrix.
 
 ## 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

@@ -32,6 +32,11 @@ use crate::{
 ///                                                 -> None        -> send nothing
 /// ```
 ///
+/// A reply is an [`OutboundDatagram`]: the bytes to send plus the transport
+/// policy that sending them requires. The core is where session policy lives, so
+/// it hands that policy out with the packet rather than leaving the runtime to
+/// rediscover it from the session table or from the packet it is about to send.
+///
 /// It is public as a low-level engine for a pre-1.0 crate, not as a promise of
 /// runtime independence. The server is intentionally Tokio-native; this is not
 /// a transport abstraction, and there is no blocking or alternate-runtime
@@ -41,11 +46,12 @@ use crate::{
 ///
 /// The core implements open handling, session creation and negotiation,
 /// normal echo processing, per-session rate limiting, idle expiry,
-/// maximum-duration close and client-initiated close. DSCP application and the
-/// full server fill policy are separate slices, and echo payloads are
-/// zero-filled. Incoming authenticated requests enforce logical expiry exactly;
-/// the runtime also calls a private maintenance hook to reclaim idle sessions
-/// when no traffic arrives.
+/// maximum-duration close and client-initiated close. It decides the traffic
+/// class each reply is to be sent with; actually applying it to a socket is the
+/// runtime's job. The full server fill policy is a separate slice, and echo
+/// payloads are zero-filled. Incoming authenticated requests enforce logical
+/// expiry exactly; the runtime also calls a private maintenance hook to reclaim
+/// idle sessions when no traffic arrives.
 #[derive(Debug)]
 pub struct ServerCore {
     config: ServerConfig,
@@ -111,6 +117,14 @@ impl ServerCore {
 
     /// Handles one received datagram, returning the reply to send, if any.
     ///
+    /// A reply is an [`OutboundDatagram`]: the encoded packet and the raw
+    /// traffic class it must be sent with. Open and no-test replies are always
+    /// unmarked, and an echo reply — including one carrying the
+    /// maximum-duration close — takes its session's negotiated class. The
+    /// runtime applies that class to the socket before sending; nothing here
+    /// touches a socket, and a class this core hands out is a request, not a
+    /// promise that the host can honor it.
+    ///
     /// `Ok(None)` means "answer nothing". The protocol defines no error reply,
     /// no reset and no NACK, so every rejection is a silent discard and a
     /// client distinguishes "rejected" from "lost" only by timing out. A
@@ -149,7 +163,7 @@ impl ServerCore {
         &mut self,
         peer: SocketAddr,
         packet: &[u8],
-    ) -> Result<Option<Vec<u8>>, ServerError> {
+    ) -> Result<Option<OutboundDatagram>, ServerError> {
         let Ok(request) = decode_request(packet) else {
             return Ok(None);
         };
@@ -295,7 +309,7 @@ impl ServerCore {
         &mut self,
         peer: SocketAddr,
         request: EchoRequest,
-    ) -> Result<Option<Vec<u8>>, ServerError> {
+    ) -> Result<Option<OutboundDatagram>, ServerError> {
         // Resource policy, applied to the length actually received. This is
         // `irtt-rs` policy rather than upstream-mechanism emulation: there is
         // no interface-MTU effective length here, just the datagram the caller
@@ -341,6 +355,10 @@ impl ServerCore {
         // it.
         let sent_at = self.clock.sample();
         let params = session.params();
+        // Read before the session can be released below: a maximum-duration
+        // close is an ordinary echo reply and carries the session's marking like
+        // any other.
+        let traffic_class = transport_traffic_class(params);
         let reply = EchoReply {
             // Open must be clear or upstream clients abort. Close is set only
             // for the maximum-duration limit, which is the sole trigger this
@@ -376,7 +394,7 @@ impl ServerCore {
             session.commit_receive_state(next);
             session.commit_lifetime_state(next_lifetime);
         }
-        Ok(Some(packet))
+        Ok(Some(OutboundDatagram::new(packet, traffic_class)))
     }
 
     /// Negotiates an open request and, unless it is a no-test one, creates the
@@ -393,7 +411,7 @@ impl ServerCore {
         no_test: bool,
         encoded_params: &[u8],
         now_ns: i64,
-    ) -> Result<Option<Vec<u8>>, ServerError> {
+    ) -> Result<Option<OutboundDatagram>, ServerError> {
         // The parameter decoder already rejects a truncated tag or value, a
         // varint overflow, an out-of-range stats/stamp-at/clock enum, and a
         // server-fill value that is oversized, short-buffered or not UTF-8.
@@ -419,7 +437,9 @@ impl ServerCore {
                 token: 0,
                 params,
             };
-            return self.encode_reply(&reply).map(Some);
+            return self
+                .encode_reply(&reply)
+                .map(|packet| Some(OutboundDatagram::unmarked(packet)));
         }
 
         if self.sessions.len() >= self.config.max_sessions() {
@@ -442,7 +462,7 @@ impl ServerCore {
             token,
             Session::new(peer, reply.params, &self.config, now_ns),
         );
-        Ok(Some(packet))
+        Ok(Some(OutboundDatagram::unmarked(packet)))
     }
 
     /// Releases the session a close request names, if that request owns one.
@@ -493,6 +513,82 @@ impl ServerCore {
     }
 }
 
+/// One reply the core prepared, together with the transport policy it must be
+/// sent under.
+///
+/// The bytes alone are not a complete instruction. A session negotiates a
+/// traffic class, and only the core knows which session a reply belongs to, so
+/// the class travels out with the packet instead of the runtime decoding the
+/// outgoing datagram again or keeping a second copy of session state.
+///
+/// [`traffic_class`](Self::traffic_class) is the **raw 8-bit IPv4 TOS / IPv6
+/// Traffic Class byte**, not a six-bit codepoint: the codepoint is its upper six
+/// bits and the low two are ECN. Zero is a value like any other and means
+/// unmarked — it is deliberately not `Option<u8>`, because a runtime sharing one
+/// socket between sessions has to apply zero as explicitly as it applies
+/// anything else, or an unmarked reply inherits the previous one's marking.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct OutboundDatagram {
+    bytes: Vec<u8>,
+    traffic_class: u8,
+}
+
+impl OutboundDatagram {
+    /// A reply to send with an explicit raw traffic-class byte.
+    fn new(bytes: Vec<u8>, traffic_class: u8) -> Self {
+        Self {
+            bytes,
+            traffic_class,
+        }
+    }
+
+    /// A reply to send unmarked.
+    fn unmarked(bytes: Vec<u8>) -> Self {
+        Self::new(bytes, 0)
+    }
+
+    /// The encoded reply.
+    #[must_use]
+    pub fn bytes(&self) -> &[u8] {
+        &self.bytes
+    }
+
+    /// The raw IPv4 TOS / IPv6 Traffic Class byte this reply is to be sent
+    /// with. Zero means unmarked.
+    #[must_use]
+    pub fn traffic_class(&self) -> u8 {
+        self.traffic_class
+    }
+
+    /// Consumes the datagram, yielding its encoded reply.
+    #[must_use]
+    pub fn into_bytes(self) -> Vec<u8> {
+        self.bytes
+    }
+}
+
+impl AsRef<[u8]> for OutboundDatagram {
+    fn as_ref(&self) -> &[u8] {
+        self.bytes()
+    }
+}
+
+/// The raw traffic-class byte a session's negotiated DSCP parameter asks for.
+///
+/// The wire parameter is an `i64` and this server deliberately accepts values
+/// outside `0..=255` during negotiation, returning and storing them unchanged.
+/// A value that is not a byte cannot be a TOS / Traffic Class byte, so the
+/// transport treats it as unmarked. That is **`irtt-rs` policy, not observed
+/// behavior**: the clean evidence records the reference host's handling of
+/// negative and oversized values as platform-specific and explicitly not a
+/// compatibility requirement, so there is nothing to reproduce. Zero is chosen
+/// over `as`-style truncation or two's-complement wrapping because neither
+/// −1 → 255 nor 256 → 0 is a marking the client asked for, and over refusing the
+/// session because a malformed-but-accepted open should stay usable.
+fn transport_traffic_class(params: &Params) -> u8 {
+    u8::try_from(params.dscp).unwrap_or(0)
+}
+
 /// A structurally decoded request kind, with an echo's arrival instant already
 /// sampled.
 ///
@@ -622,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

@@ -10,11 +10,12 @@
 //! receive state, rate and lifetime policy, resource decisions and reply
 //! construction. It performs no I/O; the clock its timestamps and deadlines
 //! come from is a private injected seam, not a runtime abstraction, so the
-//! engine stays testable. [`Server`] owns one Tokio UDP listener and one core,
-//! and provides sequential receive, reply, scheduled expiry maintenance and
-//! caller-controlled shutdown. The crate is intentionally Tokio-native; there
-//! is no blocking or alternate-runtime counterpart and no transport
-//! abstraction.
+//! engine stays testable. Each reply it produces is an [`OutboundDatagram`]:
+//! the packet and the raw traffic class it must be sent with. [`Server`] owns
+//! one Tokio UDP listener and one core, and provides sequential receive, reply,
+//! scheduled expiry maintenance and caller-controlled shutdown. The crate is
+//! intentionally Tokio-native; there is no blocking or alternate-runtime
+//! counterpart and no transport abstraction.
 //!
 //! # Rejection is silence
 //!
@@ -42,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;
@@ -54,6 +56,7 @@ mod error;
 mod negotiate;
 mod runtime;
 mod session;
+mod socket_options;
 mod token;
 
 #[cfg(test)]
@@ -63,6 +66,6 @@ pub use config::{
     ServerConfig, DEFAULT_BURST_ALLOWANCE, DEFAULT_IDLE_TIMEOUT, DEFAULT_MAX_PACKET_LENGTH,
     DEFAULT_MAX_SESSIONS, DEFAULT_MIN_SEND_INTERVAL,
 };
-pub use core::ServerCore;
+pub use core::{OutboundDatagram, ServerCore};
 pub use error::ServerError;
 pub use runtime::{Server, ServerRuntimeError};

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,19 @@ 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>,
+    /// Whether this server has ever successfully applied a nonzero traffic
+    /// class to the socket.
+    ///
+    /// This is not a cache and never elides a call — the class is applied
+    /// before every send regardless. It is consulted only when applying one
+    /// *fails*, to tell "the socket may be carrying a marking of ours" from
+    /// "it cannot be carrying one".
+    socket_is_marked: bool,
 }
 
 impl Server {
@@ -45,13 +67,50 @@ 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,
+            socket_is_marked: false,
         }
     }
 
+    /// Applies one reply's raw traffic class to the listener socket, and
+    /// reports whether that reply may be sent.
+    ///
+    /// The class is applied before *every* send, and zero 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.
+    ///
+    /// When the option cannot be applied, [`may_send_unappliable`] decides. A
+    /// marking this server asked for and did not get must never be replaced by
+    /// silently sending under the previous one, so those replies are dropped;
+    /// but a host that refuses the option outright — some Windows builds do not
+    /// support `IP_TOS`, and a few targets have no safe setter at all — must
+    /// still be able to run a server, and it can, because a socket this server
+    /// has never marked has no marking of ours to clear.
+    fn prepare_reply_traffic_class(&mut self, traffic_class: u8) -> bool {
+        match self.apply_reply_traffic_class(traffic_class) {
+            Ok(()) => {
+                self.socket_is_marked = traffic_class != 0;
+                true
+            }
+            Err(_) => may_send_unappliable(traffic_class, self.socket_is_marked),
+        }
+    }
+
+    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 +123,16 @@ 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 a marking that reply needed —
+    /// 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 marking could not be applied is not sent instead: sending
+    /// it would put it on the wire under the previous reply's marking. An
+    /// *unmarked* reply on a socket this server has never marked is the one
+    /// exception, so a host that does not support the option at all — some
+    /// Windows builds do not support `IP_TOS` — still runs a working server.
     pub async fn run<F>(&mut self, shutdown: F) -> Result<(), ServerRuntimeError>
     where
         F: Future<Output = ()>,
@@ -90,7 +157,10 @@ impl Server {
                     }
 
                     if let Some(reply) = self.core.handle_datagram(peer, &self.recv_buffer[..len])? {
-                        let send = self.socket.send_to(&reply, peer);
+                        if !self.prepare_reply_traffic_class(reply.traffic_class()) {
+                            continue;
+                        }
+                        let send = self.socket.send_to(reply.bytes(), peer);
                         tokio::pin!(send);
                         let sent = loop {
                             tokio::select! {
@@ -99,7 +169,7 @@ impl Server {
                                 sent = &mut send => break sent,
                             }
                         };
-                        if !matches!(sent, Ok(len) if len == reply.len()) {
+                        if !matches!(sent, Ok(len) if len == reply.bytes().len()) {
                             continue;
                         }
                     }
@@ -109,6 +179,21 @@ impl Server {
     }
 }
 
+/// Whether a reply may still be sent after its traffic class could not be
+/// applied.
+///
+/// Only an unmarked reply on a socket this server has never successfully marked
+/// qualifies. Both halves matter: a reply that wanted a marking must not go out
+/// without one, and once this server has put a nonzero class on the socket, a
+/// failure to clear it means the socket may still be carrying it.
+///
+/// A socket handed to [`Server::from_socket`] pre-marked by its creator is
+/// outside what this can know; explicit marking is the server's own to manage
+/// from a fresh listener.
+fn may_send_unappliable(traffic_class: u8, socket_is_marked: bool) -> bool {
+    traffic_class == 0 && !socket_is_marked
+}
+
 /// Failure to create or run a Tokio UDP server.
 #[derive(Debug, Error)]
 pub enum ServerRuntimeError {
@@ -284,4 +369,33 @@ mod tests {
     fn unthrottled() -> ServerConfig {
         ServerConfig::default().with_min_send_interval(Duration::ZERO)
     }
+
+    /// The rule that decides what an unappliable traffic class means, which no
+    /// normal interface can reach: it needs a host that refuses the socket
+    /// option, and the hosts these tests run on do not.
+    #[test]
+    fn only_an_unmarked_reply_on_a_never_marked_socket_survives_an_apply_failure() {
+        for (traffic_class, socket_is_marked, sendable, why) in [
+            (
+                0,
+                false,
+                true,
+                "no marking wanted, and none of ours to clear",
+            ),
+            (0, true, false, "our own marking may still be on the socket"),
+            (
+                0xb8,
+                false,
+                false,
+                "a marking that was wanted but not applied",
+            ),
+            (0xb8, true, false, "and likewise over a marking already set"),
+        ] {
+            assert_eq!(
+                may_send_unappliable(traffic_class, socket_is_marked),
+                sendable,
+                "{why}"
+            );
+        }
+    }
 }

crates/irtt-server/src/socket_options.rs

@@ -0,0 +1,121 @@
+//! 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 = "haiku",
+    target_os = "wasi",
+)))]
+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 = "haiku",
+    target_os = "wasi",
+))]
+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",
+    target_os = "illumos",
+))]
+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",
+    target_os = "illumos",
+)))]
+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.
+///
+/// This reports [`io::ErrorKind::Unsupported`] for *every* value, zero
+/// included. Returning success for zero would claim the socket had been cleared
+/// when nothing was written to it, which is not the same statement — a caller
+/// handing [`Server::from_socket`](crate::Server::from_socket) an already
+/// prepared socket may have marked it by some other means. Whether an
+/// unappliable zero is nevertheless safe to send is the runtime's decision, and
+/// it is made from what this server itself has applied; this helper's job is
+/// only to report honestly.
+#[cfg(any(
+    target_os = "fuchsia",
+    target_os = "redox",
+    target_os = "solaris",
+    target_os = "haiku",
+    target_os = "wasi",
+    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",
+        target_os = "illumos",
+    )),
+))]
+fn unsupported_traffic_class(_traffic_class: u32, option: &'static str) -> io::Result<()> {
+    Err(io::Error::new(
+        io::ErrorKind::Unsupported,
+        format!("{option} are unsupported on this target"),
+    ))
+}

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

@@ -93,8 +93,15 @@ fn a_valid_echo_is_answered_from_the_negotiated_params() {
         "no timestamps were negotiated, so none are emitted"
     );
     assert!(reply.payload.is_empty());
-    assert_eq!(packet.len(), echo_packet_len(false, &negotiated).unwrap());
-    assert_eq!(packet.len(), 16, "magic, flags, token and sequence only");
+    assert_eq!(
+        packet.bytes().len(),
+        echo_packet_len(false, &negotiated).unwrap()
+    );
+    assert_eq!(
+        packet.bytes().len(),
+        16,
+        "magic, flags, token and sequence only"
+    );
 
     assert_eq!(
         core.session_count(),
@@ -149,7 +156,7 @@ fn a_reply_is_sized_by_the_session_not_by_the_request() {
         let reply = expect_echo_reply(&packet, &negotiated, None);
 
         assert_eq!(
-            packet.len(),
+            packet.bytes().len(),
             64,
             "{name}: the reply is the negotiated length"
         );
@@ -418,7 +425,7 @@ fn an_authenticated_session_answers_only_authenticated_echoes() {
         .unwrap()
         .expect("a correctly authenticated echo must be answered");
     assert!(
-        verify_packet_hmac(KEY, &packet).is_ok(),
+        verify_packet_hmac(KEY, packet.bytes()).is_ok(),
         "the reply must carry a MAC that verifies with the server's key"
     );
     // `expect_echo_reply` pins the HMAC flag and verifies the MAC again while
@@ -596,7 +603,7 @@ fn only_the_negotiated_timestamps_are_emitted() {
         // failure as much as a selected one carrying the wrong instant.
         assert_eq!(reply.timestamps, expected, "{stamp_at:?} on {clock:?}");
         assert_eq!(
-            packet.len(),
+            packet.bytes().len(),
             echo_packet_len(false, &negotiated).unwrap(),
             "{stamp_at:?} on {clock:?} must be the negotiated layout's length"
         );
@@ -704,12 +711,12 @@ fn a_single_clock_midpoint_emits_exactly_one_field() {
         let reply = expect_echo_reply(&packet, &negotiated, None);
 
         assert_eq!(
-            packet.len(),
+            packet.bytes().len(),
             echo_packet_len(false, &negotiated).unwrap(),
             "{clock:?}: the negotiated length, not the compatibility one"
         );
         assert_eq!(
-            packet.len(),
+            packet.bytes().len(),
             24,
             "{clock:?}: one 8-byte midpoint field, not the upstream two"
         );
@@ -734,7 +741,7 @@ fn a_reply_payload_is_zero_filled_and_never_reflects_the_request() {
         .expect("an admissible echo must be answered");
     let reply = expect_echo_reply(&packet, &negotiated, None);
 
-    assert_eq!(packet.len(), 64);
+    assert_eq!(packet.bytes().len(), 64);
     assert_eq!(reply.payload.len(), 48);
     assert!(
         reply.payload.iter().all(|byte| *byte == 0),

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

@@ -94,8 +94,12 @@ fn an_authenticating_server_answers_a_valid_open_with_an_authenticated_reply() {
 
     let reply = expect_normal_open_reply(&packet, Some(KEY));
     assert_eq!(reply.token, TOKEN_A);
-    assert_ne!(packet[3] & FLAG_HMAC, 0, "the reply must set FLAG_HMAC");
-    verify_packet_hmac(KEY, &packet).expect("the reply must carry a valid MAC");
+    assert_ne!(
+        packet.bytes()[3] & FLAG_HMAC,
+        0,
+        "the reply must set FLAG_HMAC"
+    );
+    verify_packet_hmac(KEY, packet.bytes()).expect("the reply must carry a valid MAC");
     assert_eq!(core.session_count(), 1);
 }
 
@@ -109,7 +113,7 @@ fn a_non_authenticating_server_answers_an_ordinary_open() {
         .expect("an unauthenticated open must be answered");
 
     expect_normal_open_reply(&packet, None);
-    assert_eq!(packet[3] & FLAG_HMAC, 0);
+    assert_eq!(packet.bytes()[3] & FLAG_HMAC, 0);
     assert_eq!(core.session_count(), 1);
 }
 

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

@@ -22,7 +22,7 @@ use super::support::{
     expect_closing_echo_reply, expect_echo_reply, expect_no_test_reply, manual_core,
     no_test_request, open_request, other_peer, peer, ManualClock, ScriptedTokens, KEY,
 };
-use crate::ServerConfig;
+use crate::{OutboundDatagram, ServerConfig};
 
 const TOKEN_A: u64 = 0x0102_0304_0506_0708;
 const TOKEN_B: u64 = 0x1112_1314_1516_1718;
@@ -50,9 +50,9 @@ fn idle_after(idle: Duration) -> ServerConfig {
         .with_min_send_interval(Duration::ZERO)
 }
 
-fn count_of(packet: Option<Vec<u8>>, params: &Params) -> Option<u32> {
-    let packet = packet?;
-    expect_echo_reply(&packet, params, None).recv_count
+fn count_of(reply: Option<OutboundDatagram>, params: &Params) -> Option<u32> {
+    let reply = reply?;
+    expect_echo_reply(reply, params, None).recv_count
 }
 
 #[test]
@@ -451,10 +451,15 @@ fn the_first_echo_past_the_maximum_duration_carries_close_and_ends_the_session()
     );
     assert!(reply.payload.is_empty());
     assert_eq!(
-        packet.len(),
+        packet.bytes().len(),
         irtt_proto::echo_packet_len(false, &negotiated).unwrap(),
         "the negotiated layout, unchanged by the flag"
     );
+    assert_eq!(
+        i64::from(packet.traffic_class()),
+        negotiated.dscp,
+        "a server close is an echo reply and keeps the session's marking"
+    );
 
     // The token is unusable from that reply onward.
     assert_eq!(core.session_count(), 0);
@@ -595,6 +600,7 @@ fn an_authenticated_session_closes_through_the_ordinary_reply_path() {
     // together.
     let reply = expect_closing_echo_reply(&packet, &negotiated, Some(KEY));
     assert_eq!(reply.recv_count, Some(2));
-    irtt_proto::verify_packet_hmac(KEY, &packet).expect("the close-flagged reply is authenticated");
+    irtt_proto::verify_packet_hmac(KEY, packet.bytes())
+        .expect("the close-flagged reply is authenticated");
     assert_eq!(core.session_count(), 0);
 }

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

@@ -24,3 +24,4 @@ mod rate;
 mod sessions;
 mod support;
 mod tokens;
+mod traffic_class;

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

@@ -194,7 +194,7 @@ fn an_authenticated_no_test_reply_is_authenticated_and_still_creates_nothing() {
         .expect("an authenticated no-test open must be answered");
 
     expect_no_test_reply(&packet, Some(KEY));
-    verify_packet_hmac(KEY, &packet).expect("the no-test reply must carry a valid MAC");
+    verify_packet_hmac(KEY, packet.bytes()).expect("the no-test reply must carry a valid MAC");
     assert_eq!(core.session_count(), 0);
 }
 

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

@@ -98,7 +98,7 @@ fn an_empty_parameter_payload_is_answered_with_only_a_version() {
     // header (4) + token (8) + ProtocolVersion=1 (2). Every other parameter
     // negotiates to its wire-default zero and is omitted entirely, which is
     // only reachable if absent parameters really did decode as zero.
-    assert_eq!(packet.len(), 14);
+    assert_eq!(packet.bytes().len(), 14);
 
     let reply = expect_normal_open_reply(&packet, None);
     assert_eq!(
@@ -139,7 +139,7 @@ fn unknown_parameter_tags_are_ignored_and_not_reflected_in_the_reply() {
     );
     // The reply is the same 14 bytes an empty payload produces: no unknown tag
     // was copied through.
-    assert_eq!(packet.len(), 14);
+    assert_eq!(packet.bytes().len(), 14);
     assert_eq!(core.session_count(), 1);
 }
 
@@ -366,9 +366,9 @@ fn authentication_counts_toward_the_executable_echo_size() {
 #[test]
 fn negative_and_out_of_range_length_and_dscp_survive_negotiation() {
     // These lengths are all at or below the default maximum, and DSCP is
-    // accepted as decoded in this slice, including values a socket could never
-    // carry. Restricting a DSCP the server cannot apply is a later policy
-    // decision, not an admission rule.
+    // accepted as decoded, including values a socket could never carry:
+    // negotiation restricts neither, and a value that is not a byte is simply
+    // transported unmarked (see `traffic_class`).
     //
     // A negative length in particular stays negative in both the reply and the
     // stored session; the packet-length policy reduces oversized *positive*

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

@@ -18,7 +18,7 @@ use super::support::{
     core_with_sources, echo_at, echo_params, echo_request, expect_echo_reply, manual_core,
     open_negotiated, other_peer, peer, ManualClock, ScriptedTokens,
 };
-use crate::ServerConfig;
+use crate::{OutboundDatagram, ServerConfig};
 
 const TOKEN_A: u64 = 0x0102_0304_0506_0708;
 const TOKEN_B: u64 = 0x1112_1314_1516_1718;
@@ -41,9 +41,9 @@ fn params_at(interval_ns: i64) -> Params {
 }
 
 /// The received count a reply reported, or `None` if there was no reply.
-fn count_of(packet: Option<Vec<u8>>, params: &Params) -> Option<u32> {
-    let packet = packet?;
-    expect_echo_reply(&packet, params, None).recv_count
+fn count_of(reply: Option<OutboundDatagram>, params: &Params) -> Option<u32> {
+    let reply = reply?;
+    expect_echo_reply(reply, params, None).recv_count
 }
 
 #[test]

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

@@ -12,7 +12,7 @@ use irtt_proto::{
 
 use crate::{
     clock::{ClockSample, ClockSource},
-    core::ServerCore,
+    core::{OutboundDatagram, ServerCore},
     error::ServerError,
     token::TokenSource,
     ServerConfig,
@@ -231,7 +231,7 @@ pub(crate) fn echo_at(
     sequence: u32,
     params: &Params,
     hmac_key: Option<&[u8]>,
-) -> Option<Vec<u8>> {
+) -> Option<OutboundDatagram> {
     clock.set(at_ns);
     core.handle_datagram(
         peer(),
@@ -314,11 +314,12 @@ pub(crate) fn echo_request(
 /// Decodes an echo reply the server produced and checks the flags every reply
 /// must carry. Decoding verifies the MAC against `hmac_key` on the way.
 pub(crate) fn expect_echo_reply(
-    packet: &[u8],
+    packet: impl AsRef<[u8]>,
     params: &Params,
     hmac_key: Option<&[u8]>,
 ) -> EchoReply {
-    let reply = decode_echo_reply(packet, params, hmac_key).expect("server echo reply must decode");
+    let reply = decode_echo_reply(packet.as_ref(), params, hmac_key)
+        .expect("server echo reply must decode");
     // Exact equality, so Open and Close are pinned clear as well: Open makes an
     // upstream client abort, and Close belongs to the server-initiated close
     // slice.
@@ -334,12 +335,12 @@ pub(crate) fn expect_echo_reply(
 /// by the same path as any other, which is the point: the close is not a
 /// separate packet kind.
 pub(crate) fn expect_closing_echo_reply(
-    packet: &[u8],
+    packet: impl AsRef<[u8]>,
     params: &Params,
     hmac_key: Option<&[u8]>,
 ) -> EchoReply {
-    let reply =
-        decode_echo_reply(packet, params, hmac_key).expect("a closing echo reply must decode");
+    let reply = decode_echo_reply(packet.as_ref(), params, hmac_key)
+        .expect("a closing echo reply must decode");
     let expected = FLAG_REPLY | FLAG_CLOSE | if hmac_key.is_some() { FLAG_HMAC } else { 0 };
     assert_eq!(reply.flags, expected, "closing echo reply flags");
     reply
@@ -452,11 +453,14 @@ pub(crate) fn client_params() -> Params {
 }
 
 /// Decodes a reply the server produced, checking it against the server's key.
-pub(crate) fn decode_reply(packet: &[u8], hmac_key: Option<&[u8]>) -> OpenReply {
-    decode_open_reply(packet, hmac_key).expect("server reply must decode")
+pub(crate) fn decode_reply(packet: impl AsRef<[u8]>, hmac_key: Option<&[u8]>) -> OpenReply {
+    decode_open_reply(packet.as_ref(), hmac_key).expect("server reply must decode")
 }
 
-pub(crate) fn expect_normal_open_reply(packet: &[u8], hmac_key: Option<&[u8]>) -> OpenReply {
+pub(crate) fn expect_normal_open_reply(
+    packet: impl AsRef<[u8]>,
+    hmac_key: Option<&[u8]>,
+) -> OpenReply {
     let reply = decode_reply(packet, hmac_key);
     let expected = FLAG_OPEN | FLAG_REPLY | if hmac_key.is_some() { FLAG_HMAC } else { 0 };
     assert_eq!(reply.flags, expected, "normal open reply flags");
@@ -465,7 +469,7 @@ pub(crate) fn expect_normal_open_reply(packet: &[u8], hmac_key: Option<&[u8]>) -
     reply
 }
 
-pub(crate) fn expect_no_test_reply(packet: &[u8], hmac_key: Option<&[u8]>) -> OpenReply {
+pub(crate) fn expect_no_test_reply(packet: impl AsRef<[u8]>, hmac_key: Option<&[u8]>) -> OpenReply {
     let reply = decode_reply(packet, hmac_key);
     let expected =
         FLAG_OPEN | FLAG_REPLY | FLAG_CLOSE | if hmac_key.is_some() { FLAG_HMAC } else { 0 };

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

@@ -0,0 +1,134 @@
+//! Which traffic class the core asks for each reply to be sent with.
+//!
+//! The negotiated DSCP parameter is the **raw IPv4 TOS / IPv6 Traffic Class
+//! byte**, not a six-bit codepoint: the codepoint is its upper six bits and the
+//! low two are ECN. Nothing here shifts, masks or rounds it, and these tests
+//! exist to keep it that way.
+//!
+//! The clean specification's Section 20 settles the observable rules for values
+//! in `0..=255`: an echo reply carries its own session's value verbatim, the
+//! marking follows the session rather than the listener, and an open reply is
+//! unmarked even on a listener that has already sent a marked echo reply. The
+//! server-close case is covered where the maximum-duration close is tested, and
+//! the socket-level proof that these bytes reach the wire lives in the runtime's
+//! Linux tests.
+
+use irtt_proto::Params;
+
+use super::support::{
+    client_params, core_with_tokens, echo_request, expect_echo_reply, expect_no_test_reply,
+    expect_normal_open_reply, no_test_request, open_request, peer, unthrottled, ScriptedTokens,
+};
+
+const TOKEN_A: u64 = 0x7896_b6ab_8771_5213;
+
+/// A raw byte whose low two bits are set on purpose: ECN is part of the value,
+/// and a server rebuilding the byte from a codepoint would lose them.
+const RAW_TRAFFIC_CLASS: i64 = 0xbb;
+
+#[test]
+fn an_open_reply_is_unmarked_whatever_the_session_negotiates() {
+    let requested = Params {
+        dscp: 0xb8,
+        ..client_params()
+    };
+    let mut core = core_with_tokens(unthrottled(), ScriptedTokens::new([TOKEN_A]));
+
+    let reply = core
+        .handle_datagram(peer(), &open_request(&requested, None))
+        .unwrap()
+        .expect("an ordinary open must be answered");
+
+    assert_eq!(
+        expect_normal_open_reply(&reply, None).params.dscp,
+        0xb8,
+        "the negotiated value is returned to the client"
+    );
+    assert_eq!(
+        reply.traffic_class(),
+        0,
+        "an open reply is unmarked however the session it creates is marked"
+    );
+
+    let no_test = core
+        .handle_datagram(peer(), &no_test_request(&requested, None))
+        .unwrap()
+        .expect("a no-test open must be answered");
+    assert_eq!(expect_no_test_reply(&no_test, None).params.dscp, 0xb8);
+    assert_eq!(
+        no_test.traffic_class(),
+        0,
+        "a no-test reply is unmarked too"
+    );
+}
+
+#[test]
+fn an_echo_reply_carries_the_sessions_raw_negotiated_byte() {
+    let requested = Params {
+        dscp: RAW_TRAFFIC_CLASS,
+        ..client_params()
+    };
+    let mut core = core_with_tokens(unthrottled(), ScriptedTokens::new([TOKEN_A]));
+    let open = core
+        .handle_datagram(peer(), &open_request(&requested, None))
+        .unwrap()
+        .expect("the session-creating open must be answered");
+    let negotiated = expect_normal_open_reply(&open, None).params;
+
+    let reply = core
+        .handle_datagram(peer(), &echo_request(TOKEN_A, 0, &negotiated, &[], None))
+        .unwrap()
+        .expect("an admissible echo must be answered");
+
+    expect_echo_reply(&reply, &negotiated, None);
+    assert_eq!(
+        i64::from(reply.traffic_class()),
+        RAW_TRAFFIC_CLASS,
+        "the raw byte, unshifted and unmasked"
+    );
+}
+
+#[test]
+fn a_negotiated_value_that_is_not_a_byte_is_transported_unmarked() {
+    // `irtt-rs` policy, not observed behavior. The server deliberately accepts
+    // a DSCP parameter outside `0..=255` and returns and stores it unchanged;
+    // the clean specification records the reference host's handling of such
+    // values as platform-specific and explicitly not a compatibility
+    // requirement. A value that is not a byte cannot be a traffic-class byte,
+    // so the transport asks for none — rather than wrapping −1 into 255,
+    // truncating 256 to 0, or refusing a session the client can otherwise use.
+    for dscp in [-1, 256, i64::MAX] {
+        let requested = Params {
+            dscp,
+            ..client_params()
+        };
+        let mut core = core_with_tokens(unthrottled(), ScriptedTokens::new([TOKEN_A]));
+
+        let open = core
+            .handle_datagram(peer(), &open_request(&requested, None))
+            .unwrap()
+            .unwrap_or_else(|| panic!("an open requesting DSCP {dscp} must still be answered"));
+        let negotiated = expect_normal_open_reply(&open, None).params;
+        assert_eq!(negotiated.dscp, dscp, "the value is negotiated unchanged");
+        assert_eq!(open.traffic_class(), 0);
+        assert_eq!(
+            core.session(TOKEN_A)
+                .expect("the session must be live")
+                .params()
+                .dscp,
+            dscp,
+            "and stored unchanged"
+        );
+
+        let reply = core
+            .handle_datagram(peer(), &echo_request(TOKEN_A, 0, &negotiated, &[], None))
+            .unwrap()
+            .unwrap_or_else(|| panic!("the session must still serve echoes with DSCP {dscp}"));
+        assert_eq!(
+            expect_echo_reply(&reply, &negotiated, None).recv_count,
+            Some(1),
+            "an unmarkable session is an ordinary session in every other way"
+        );
+        assert_eq!(reply.traffic_class(), 0, "transported unmarked");
+    }
+}

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,
+    }
+}