From 5420387846c26c1f87129f35b0ccba2afb5dd72b Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 13 Aug 2026 20:40:07 +0700 Subject: [PATCH] feat(server): apply reply traffic class --- AGENTS.md | 2 +- Cargo.lock | 1 + README.md | 3 +- crates/irtt-server/AGENTS.md | 45 ++++- crates/irtt-server/Cargo.toml | 3 +- crates/irtt-server/src/core.rs | 7 +- crates/irtt-server/src/lib.rs | 8 +- crates/irtt-server/src/runtime.rs | 45 ++++- crates/irtt-server/src/socket_options.rs | 120 ++++++++++++ crates/irtt-server/tests/traffic_class.rs | 225 ++++++++++++++++++++++ 10 files changed, 445 insertions(+), 14 deletions(-) create mode 100644 crates/irtt-server/src/socket_options.rs create mode 100644 crates/irtt-server/tests/traffic_class.rs diff --git a/AGENTS.md b/AGENTS.md index 47bd554..69fda30 100644 --- a/AGENTS.md +++ b/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. diff --git a/Cargo.lock b/Cargo.lock index 717dea6..1862351 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -735,6 +735,7 @@ dependencies = [ "getrandom 0.4.3", "irtt-client", "irtt-proto", + "socket2", "thiserror 2.0.20", "tokio", ] diff --git a/README.md b/README.md index fde01f1..36ab907 100644 --- a/README.md +++ b/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. diff --git a/crates/irtt-server/AGENTS.md b/crates/irtt-server/AGENTS.md index 59334f7..4ab041e 100644 --- a/crates/irtt-server/AGENTS.md +++ b/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 diff --git a/crates/irtt-server/Cargo.toml b/crates/irtt-server/Cargo.toml index 1689773..78b364e 100644 --- a/crates/irtt-server/Cargo.toml +++ b/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"] } diff --git a/crates/irtt-server/src/core.rs b/crates/irtt-server/src/core.rs index b13bd9f..cbb8c6f 100644 --- a/crates/irtt-server/src/core.rs +++ b/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) diff --git a/crates/irtt-server/src/lib.rs b/crates/irtt-server/src/lib.rs index b83d16c..5d5a1b3 100644 --- a/crates/irtt-server/src/lib.rs +++ b/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)] diff --git a/crates/irtt-server/src/runtime.rs b/crates/irtt-server/src/runtime.rs index 81ba79b..9573286 100644 --- a/crates/irtt-server/src/runtime.rs +++ b/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, + /// 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, } 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 { 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(&mut self, shutdown: F) -> Result<(), ServerRuntimeError> where F: Future, @@ -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 { diff --git a/crates/irtt-server/src/socket_options.rs b/crates/irtt-server/src/socket_options.rs new file mode 100644 index 0000000..77af726 --- /dev/null +++ b/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"), + )) + } +} diff --git a/crates/irtt-server/tests/traffic_class.rs b/crates/irtt-server/tests/traffic_class.rs new file mode 100644 index 0000000..129454d --- /dev/null +++ b/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>, +} + +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, + } +}