feat(client): implement DSCP negotiation and socket traffic class application

Apply the negotiated DSCP value to the UDP socket after a successful session open and clear it before closing to support network-level quality of service. This change adds platform-specific socket option handling for IPv4 TOS and IPv6 Traffic Class.

crates/irtt-client/Cargo.toml

@@ -7,7 +7,7 @@ rust-version.workspace = true
 
 [dependencies]
 irtt-proto = { path = "../irtt-proto" }
-socket2 = "0.5"
+socket2 = { version = "0.5", features = ["all"] }
 thiserror = "2"
 
 [features]

crates/irtt-client/src/client.rs

@@ -20,6 +20,7 @@ use crate::{
     probe::{CompletedSet, PendingMap, PendingProbe, TimedOutMap},
     session::{validate_negotiated_params, ActiveSession, ClientPhase, NegotiatedParams},
     socket::{connect_udp_socket, resolve_remote, validate_open_timeouts},
+    socket_options::{apply_dscp_to_socket, clear_dscp_on_socket},
     timing::ClientTimestamp,
 };
 
@@ -126,6 +127,7 @@ impl Client {
             }
         };
 
+        clear_dscp_on_socket(&self.socket, self.remote)?;
         let packet =
             encode_close_request(&CloseRequest { token }, self.config.hmac_key.as_deref())?;
         self.socket.send(&packet)?;
@@ -543,6 +545,11 @@ impl Client {
         let negotiated = NegotiatedParams {
             params: reply.params.clone(),
         };
+        let negotiated_dscp =
+            u8::try_from(negotiated.params.dscp).map_err(|_| ClientError::InvalidConfig {
+                reason: "negotiated dscp must be in range 0..=63".to_owned(),
+            })?;
+        apply_dscp_to_socket(&self.socket, self.remote, negotiated_dscp)?;
         self.negotiated = Some(negotiated.clone());
         self.phase = ClientPhase::Open { token: reply.token };
 

crates/irtt-client/src/client/tests/dscp.rs

@@ -0,0 +1,68 @@
+use super::*;
+use crate::socket_options::socket_traffic_class;
+
+#[test]
+#[cfg(not(any(
+    target_os = "fuchsia",
+    target_os = "redox",
+    target_os = "solaris",
+    target_os = "illumos",
+    target_os = "haiku",
+)))]
+fn normal_open_applies_negotiated_dscp_after_open_and_close_clears_it() {
+    let mut params = default_params();
+    params.dscp = 46;
+    let server = start_fake_server(move |socket, tx| {
+        let (_, peer) = recv_request(&socket, &tx);
+        let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
+        socket.send_to(&reply, peer).unwrap();
+        let _ = recv_request(&socket, &tx);
+    });
+    let mut config = default_test_config(server.addr);
+    config.dscp = 46;
+    let mut client = Client::connect(config).unwrap();
+
+    assert_eq!(
+        socket_traffic_class(&client.socket, client.remote).unwrap() & 0xfc,
+        0
+    );
+    assert_open_started(client.open(ClientTimestamp::now()).unwrap());
+    assert_eq!(
+        socket_traffic_class(&client.socket, client.remote).unwrap() & 0xfc,
+        184
+    );
+
+    client.close(ClientTimestamp::now()).unwrap();
+    assert_eq!(
+        socket_traffic_class(&client.socket, client.remote).unwrap(),
+        0
+    );
+    server.join();
+}
+
+#[test]
+#[cfg(not(any(
+    target_os = "fuchsia",
+    target_os = "redox",
+    target_os = "solaris",
+    target_os = "illumos",
+    target_os = "haiku",
+)))]
+fn normal_open_uses_negotiated_dscp_not_requested_dscp() {
+    let mut returned = default_params();
+    returned.dscp = 0;
+    let server = open_success_server(returned);
+
+    let mut config = default_test_config(server.addr);
+    config.dscp = 46;
+    config.negotiation_policy = NegotiationPolicy::Loose;
+    let mut client = Client::connect(config).unwrap();
+
+    let negotiated = assert_open_started(client.open(ClientTimestamp::now()).unwrap());
+    assert_eq!(negotiated.params.dscp, 0);
+    assert_eq!(
+        socket_traffic_class(&client.socket, client.remote).unwrap(),
+        0
+    );
+    server.join();
+}

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

@@ -11,6 +11,7 @@ use support::*;
 
 mod close;
 mod config;
+mod dscp;
 mod hmac;
 mod negotiation;
 mod no_test;

crates/irtt-client/src/error.rs

@@ -8,6 +8,12 @@ pub enum ClientError {
     Resolve { addr: String },
     #[error("socket error: {0}")]
     Socket(#[from] io::Error),
+    #[error("socket option error while trying to {operation} for {remote}: {source}")]
+    SocketOption {
+        operation: &'static str,
+        remote: std::net::SocketAddr,
+        source: io::Error,
+    },
     #[error("protocol error: {0}")]
     Protocol(#[from] irtt_proto::ProtoError),
     #[error("all open requests timed out")]

crates/irtt-client/src/lib.rs

@@ -8,6 +8,7 @@ mod managed;
 mod probe;
 mod session;
 mod socket;
+mod socket_options;
 mod timing;
 
 pub use client::Client;

crates/irtt-client/src/socket_options.rs

@@ -0,0 +1,322 @@
+use std::{
+    io,
+    net::{SocketAddr, UdpSocket},
+};
+
+use crate::{config::MAX_DSCP_CODEPOINT, error::ClientError};
+
+pub(crate) fn dscp_codepoint_to_traffic_class(dscp: u8) -> Result<u32, ClientError> {
+    if dscp > MAX_DSCP_CODEPOINT {
+        return Err(ClientError::InvalidConfig {
+            reason: format!("dscp must be <= {MAX_DSCP_CODEPOINT}"),
+        });
+    }
+    Ok(u32::from(dscp) << 2)
+}
+
+pub(crate) fn apply_dscp_to_socket(
+    socket: &UdpSocket,
+    remote: SocketAddr,
+    dscp_codepoint: u8,
+) -> Result<(), ClientError> {
+    let traffic_class = dscp_codepoint_to_traffic_class(dscp_codepoint)?;
+    set_socket_traffic_class(socket, remote, traffic_class, "set negotiated DSCP")
+}
+
+pub(crate) fn clear_dscp_on_socket(
+    socket: &UdpSocket,
+    remote: SocketAddr,
+) -> Result<(), ClientError> {
+    set_socket_traffic_class(socket, remote, 0, "clear DSCP before close")
+}
+
+#[cfg(test)]
+pub(crate) fn socket_traffic_class(
+    socket: &UdpSocket,
+    remote: SocketAddr,
+) -> Result<u32, ClientError> {
+    get_socket_traffic_class(socket, remote).map_err(|source| ClientError::SocketOption {
+        operation: "read DSCP socket option",
+        remote,
+        source,
+    })
+}
+
+fn set_socket_traffic_class(
+    socket: &UdpSocket,
+    remote: SocketAddr,
+    traffic_class: u32,
+    operation: &'static str,
+) -> Result<(), ClientError> {
+    set_socket_traffic_class_io(socket, remote, traffic_class).map_err(|source| {
+        ClientError::SocketOption {
+            operation,
+            remote,
+            source,
+        }
+    })
+}
+
+fn set_socket_traffic_class_io(
+    socket: &UdpSocket,
+    remote: SocketAddr,
+    traffic_class: u32,
+) -> io::Result<()> {
+    if remote.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: &UdpSocket, traffic_class: u32) -> io::Result<()> {
+    socket2::SockRef::from(socket).set_tos(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: &UdpSocket, traffic_class: u32) -> io::Result<()> {
+    unsupported_traffic_class(traffic_class, "IPv4 DSCP 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: &UdpSocket, traffic_class: u32) -> io::Result<()> {
+    socket2::SockRef::from(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: &UdpSocket, traffic_class: u32) -> io::Result<()> {
+    unsupported_traffic_class(traffic_class, "IPv6 DSCP socket options")
+}
+
+#[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, feature: &'static str) -> io::Result<()> {
+    if traffic_class == 0 {
+        Ok(())
+    } else {
+        Err(io::Error::new(
+            io::ErrorKind::Unsupported,
+            format!("{feature} are unsupported on this target"),
+        ))
+    }
+}
+
+#[cfg(all(
+    test,
+    not(any(
+        target_os = "fuchsia",
+        target_os = "redox",
+        target_os = "solaris",
+        target_os = "illumos",
+        target_os = "haiku",
+    ))
+))]
+fn get_ipv4_traffic_class(socket: &UdpSocket) -> io::Result<u32> {
+    socket2::SockRef::from(socket).tos()
+}
+
+#[cfg(all(
+    test,
+    any(
+        target_os = "fuchsia",
+        target_os = "redox",
+        target_os = "solaris",
+        target_os = "illumos",
+        target_os = "haiku",
+    )
+))]
+fn get_ipv4_traffic_class(_socket: &UdpSocket) -> io::Result<u32> {
+    unsupported_readback("IPv4 DSCP socket option readback")
+}
+
+#[cfg(all(
+    test,
+    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 get_ipv6_traffic_class(socket: &UdpSocket) -> io::Result<u32> {
+    socket2::SockRef::from(socket).tclass_v6()
+}
+
+#[cfg(all(
+    test,
+    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 get_ipv6_traffic_class(_socket: &UdpSocket) -> io::Result<u32> {
+    unsupported_readback("IPv6 DSCP socket option readback")
+}
+
+#[cfg(test)]
+fn get_socket_traffic_class(socket: &UdpSocket, remote: SocketAddr) -> io::Result<u32> {
+    if remote.is_ipv4() {
+        get_ipv4_traffic_class(socket)
+    } else {
+        get_ipv6_traffic_class(socket)
+    }
+}
+
+#[cfg(all(
+    test,
+    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_readback(feature: &'static str) -> io::Result<u32> {
+    Err(io::Error::new(
+        io::ErrorKind::Unsupported,
+        format!("{feature} is unsupported on this target"),
+    ))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn converts_dscp_codepoint_to_traffic_class_byte() {
+        for (dscp, traffic_class) in [(0, 0), (1, 4), (46, 184), (63, 252)] {
+            let value = dscp_codepoint_to_traffic_class(dscp).unwrap();
+            assert_eq!(value, traffic_class);
+            assert_eq!(value & 0b11, 0);
+        }
+    }
+
+    #[test]
+    fn rejects_out_of_range_dscp_codepoint() {
+        assert!(matches!(
+            dscp_codepoint_to_traffic_class(64),
+            Err(ClientError::InvalidConfig { .. })
+        ));
+    }
+
+    #[test]
+    #[cfg(not(any(
+        target_os = "fuchsia",
+        target_os = "redox",
+        target_os = "solaris",
+        target_os = "illumos",
+        target_os = "haiku",
+    )))]
+    fn ipv4_socket_option_sets_and_clears_traffic_class() {
+        let remote = SocketAddr::from(([127, 0, 0, 1], 9));
+        let socket = UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap();
+        socket.connect(remote).unwrap();
+
+        apply_dscp_to_socket(&socket, remote, 46).unwrap();
+        let traffic_class = socket_traffic_class(&socket, remote).unwrap();
+        assert_eq!(traffic_class & 0xfc, 184);
+        assert_eq!(traffic_class & 0b11, 0);
+
+        clear_dscp_on_socket(&socket, remote).unwrap();
+        assert_eq!(socket_traffic_class(&socket, remote).unwrap(), 0);
+    }
+
+    #[test]
+    #[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 ipv6_socket_option_sets_and_clears_traffic_class() {
+        let remote = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], 9));
+        let socket = UdpSocket::bind(SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], 0))).unwrap();
+        socket.connect(remote).unwrap();
+
+        apply_dscp_to_socket(&socket, remote, 46).unwrap();
+        let traffic_class = socket_traffic_class(&socket, remote).unwrap();
+        assert_eq!(traffic_class & 0xfc, 184);
+        assert_eq!(traffic_class & 0b11, 0);
+
+        clear_dscp_on_socket(&socket, remote).unwrap();
+        assert_eq!(socket_traffic_class(&socket, remote).unwrap(), 0);
+    }
+}

crates/irtt-client/tests/protocol_options.rs

@@ -392,6 +392,19 @@ fn backend_basic_open_echo_close() {
     client.close(ClientTimestamp::now()).unwrap();
 }
 
+#[test]
+fn backend_dscp_smoke() {
+    let mut params = default_params();
+    params.dscp = 46;
+    let server = start_open_server(params.clone(), None);
+    let mut client = Client::connect(config_for_params(server.addr, &params)).unwrap();
+
+    let outcome = client.open(ClientTimestamp::now()).unwrap();
+    assert!(matches!(outcome, irtt_client::OpenOutcome::Started { .. }));
+    client.close(ClientTimestamp::now()).unwrap();
+    server.join();
+}
+
 #[test]
 fn backend_received_stats_smoke() {
     for mode in [