fix(tests): break the irtt-client/irtt-server dev-dependency cycle

irtt-client and irtt-server mutually dev-depended on each other for
cross-crate integration tests. That cycle blocks `cargo publish` for
either crate whenever the other doesn't already exist on the registry
at the required version, which is exactly the situation for a crate's
first-ever release (irtt-server has never been published before).

Introduce a new `irtt-integration-tests` crate (publish = false, no
library target) to own tests whose actual subject is the interaction
between two or more irtt-* production crates:

  crate-local behavior          -> crate's own tests/
  cross-crate/system behavior   -> irtt-integration-tests

Move every test that imports both irtt_client and irtt_server there:
the client-side support harness (hmac, large_packets, protocol_options,
fill_negotiation) and the server-side real-socket tests that drive a
real AsyncClient against a real Server (async_client, negotiation_policy,
fill_wire_bytes, traffic_class, wildcard_source). server_set.rs stays in
irtt-server/tests since it only uses irtt_proto as test machinery to
drive irtt-server directly, which is a normal one-directional test
dependency, not a peer relationship.

irtt-server's dev-dependency on irtt-client is dropped entirely now
that nothing under crates/irtt-server/tests needs it, which fully
breaks the cycle from that side.

irtt-client keeps a one-directional dev-dependency on irtt-server: a
handful of white-box unit tests in src/async_client/tests.rs assert on
private AsyncClient internals (socket, remote, test_hooks) while also
needing a real in-tree server for blocking/async equivalence checks, so
they can't move to an external crate without losing that access. Their
InTreeServer helper is duplicated locally rather than shared, matching
the existing precedent of irtt-cli's own tests/support/in_tree_server.rs.

crates/irtt-integration-tests/tests/async_client.rs

@@ -0,0 +1,95 @@
+use std::time::Duration;
+
+use irtt_client::{AsyncClient, ClientConfig, ClientEvent, OpenOutcome, SocketConfig};
+use irtt_proto::{Clock, ReceivedStats, StampAt};
+use irtt_server::{Server, ServerConfig};
+use tokio::{sync::oneshot, time::timeout};
+
+#[tokio::test(flavor = "current_thread")]
+async fn async_client_opens_echoes_receives_and_closes() {
+    timeout(Duration::from_secs(2), exercise_client_server())
+        .await
+        .expect("client/server integration exceeded its bounded runtime");
+}
+
+async fn exercise_client_server() {
+    let mut server = Server::bind("127.0.0.1:0".parse().unwrap(), ServerConfig::default())
+        .await
+        .unwrap();
+    let server_addr = server.local_addr().unwrap();
+    let (shutdown_tx, shutdown_rx) = oneshot::channel();
+    let server_task = tokio::spawn(async move {
+        server
+            .run(async {
+                let _ = shutdown_rx.await;
+            })
+            .await
+    });
+
+    let config = ClientConfig {
+        server_addr: server_addr.to_string(),
+        duration: Some(Duration::from_secs(1)),
+        interval: Duration::from_millis(100),
+        length: 64,
+        received_stats: ReceivedStats::Both,
+        stamp_at: StampAt::Both,
+        clock: Clock::Both,
+        open_timeouts: vec![Duration::from_millis(200)],
+        socket_config: SocketConfig {
+            recv_timeout: Some(Duration::from_millis(200)),
+            ..SocketConfig::default()
+        },
+        probe_timeout: Duration::from_millis(200),
+        ..ClientConfig::default()
+    };
+    let mut client = AsyncClient::connect(config).await.unwrap();
+
+    let opened = client.open().await.unwrap();
+    match &opened {
+        OpenOutcome::Started {
+            remote,
+            token,
+            negotiated,
+            event,
+        } => {
+            assert_eq!(*remote, server_addr);
+            assert_ne!(*token, 0);
+            assert_eq!(negotiated.params.received_stats, ReceivedStats::Both);
+            assert_eq!(negotiated.params.stamp_at, StampAt::Both);
+            assert_eq!(negotiated.params.clock, Clock::Both);
+            // A default client requests no fill, and this open succeeded under
+            // the default strict negotiation policy — which it could not have
+            // if the server had answered an absent request with its own default
+            // descriptor. The server still fills the payload with it.
+            assert_eq!(negotiated.params.server_fill, None);
+            assert!(matches!(event, ClientEvent::SessionStarted { .. }));
+        }
+        OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"),
+    }
+
+    assert!(matches!(
+        client.send_probe().await.unwrap().as_slice(),
+        [ClientEvent::EchoSent {
+            seq: 0,
+            remote,
+            ..
+        }] if *remote == server_addr
+    ));
+    assert!(matches!(
+        client.recv().await.unwrap().as_slice(),
+        [ClientEvent::EchoReply {
+            seq: 0,
+            remote,
+            server_timing: Some(_),
+            received_stats: Some(_),
+            ..
+        }] if *remote == server_addr
+    ));
+    assert!(matches!(
+        client.close().await.unwrap().as_slice(),
+        [ClientEvent::SessionClosed { remote, .. }] if *remote == server_addr
+    ));
+
+    shutdown_tx.send(()).unwrap();
+    server_task.await.unwrap().unwrap();
+}

crates/irtt-integration-tests/tests/fill_negotiation.rs

@@ -0,0 +1,163 @@
+mod support;
+
+use irtt_client::{Client, ClientConfig, ClientError, NegotiationPolicy};
+use irtt_proto::{Params, TimestampFields};
+
+use support::{
+    config_for_params, default_params, run_one_probe, run_one_probe_with_config, server_fill,
+    start_open_server, InTreeServer, OneProbeRun, ServerObservation,
+};
+
+use crate::support::test_echo_packet_len;
+
+fn open_observation_params(run: &OneProbeRun) -> &Params {
+    match &run.observations[0] {
+        ServerObservation::Open { params, .. } => params,
+        other => panic!("expected Open observation, got {other:?}"),
+    }
+}
+
+fn echo_observation(run: &OneProbeRun) -> (usize, bool, u64, u32) {
+    match run.observations[1] {
+        ServerObservation::Echo {
+            len,
+            hmac,
+            token,
+            sequence,
+        } => (len, hmac, token, sequence),
+        ref other => panic!("expected Echo observation, got {other:?}"),
+    }
+}
+
+// ─── Config mapping ───
+
+#[test]
+fn server_fill_none_short_and_max_map_into_open_params() {
+    for fill in [None, Some("abc"), Some("0123456789abcdef0123456789abcdef")] {
+        let mut params = default_params();
+        params.server_fill = fill.and_then(server_fill);
+
+        let run = run_one_probe(params, TimestampFields::default());
+        assert_eq!(
+            open_observation_params(&run)
+                .server_fill
+                .as_ref()
+                .map(|f| f.value.as_str()),
+            fill
+        );
+    }
+}
+
+#[test]
+fn server_fill_empty_and_oversized_values_are_rejected_at_config_boundary() {
+    for fill in ["", "0123456789abcdef0123456789abcdefx"] {
+        let config = ClientConfig {
+            server_fill: Some(fill.to_owned()),
+            ..ClientConfig::default()
+        };
+        assert!(matches!(
+            Client::connect(config),
+            Err(ClientError::InvalidConfig { .. })
+        ));
+    }
+}
+
+// ─── Open negotiation ───
+
+#[test]
+fn server_fill_negotiated_params_reflect_accepted_value() {
+    let mut params = default_params();
+    params.server_fill = server_fill("rand");
+    let server = InTreeServer::start(irtt_server::ServerConfig::default());
+    let mut client = Client::connect(config_for_params(server.addr, &params)).unwrap();
+    let outcome = client.open().unwrap();
+    let negotiated = match outcome {
+        irtt_client::OpenOutcome::Started { negotiated, .. } => negotiated,
+        other => panic!("expected started outcome, got {other:?}"),
+    };
+    assert_eq!(negotiated.params.server_fill, params.server_fill);
+
+    client.send_probe().unwrap();
+    assert!(matches!(
+        client.recv_once().unwrap().as_slice(),
+        [irtt_client::ClientEvent::EchoReply { .. }]
+    ));
+    client.close().unwrap();
+}
+
+#[test]
+fn server_fill_strict_rejects_removed_changed_or_unexpected_fill() {
+    for (requested_fill, returned_fill) in [
+        (Some("rand"), None),
+        (Some("abc"), Some("xyz")),
+        (None, Some("unexpected")),
+    ] {
+        let mut requested = default_params();
+        requested.server_fill = requested_fill.and_then(server_fill);
+
+        let mut returned = requested.clone();
+        returned.server_fill = returned_fill.and_then(server_fill);
+
+        let server = start_open_server(returned, None);
+        let mut client = Client::connect(config_for_params(server.addr, &requested)).unwrap();
+        assert!(matches!(
+            client.open(),
+            Err(ClientError::NegotiationRejected { .. })
+        ));
+        server.join();
+    }
+}
+
+#[test]
+fn server_fill_loose_allows_server_to_remove_fill() {
+    let mut requested = default_params();
+    requested.server_fill = server_fill("rand");
+
+    let mut returned = requested.clone();
+    returned.server_fill = None;
+
+    let requested_for_config = requested.clone();
+    let run = run_one_probe_with_config(returned, TimestampFields::default(), None, |addr| {
+        ClientConfig {
+            negotiation_policy: NegotiationPolicy::Loose,
+            ..config_for_params(addr, &requested_for_config)
+        }
+    });
+
+    assert_eq!(
+        open_observation_params(&run).server_fill,
+        requested.server_fill
+    );
+    assert_eq!(run.negotiated.params.server_fill, None);
+}
+
+// ─── Echo behavior ───
+
+// server_fill is a negotiation/config parameter only; it does not affect
+// client-observable reply payload layout. The client sends server_fill in
+// the open request and expects the server to use it for reply payload
+// content, but the client's decode/echo layout logic is independent of
+// server_fill. Only the length parameter expands the packet beyond the
+// protocol header.
+#[test]
+fn server_fill_does_not_affect_client_side_packet_layout() {
+    let mut params_with_fill = default_params();
+    params_with_fill.server_fill = server_fill("rand");
+
+    let params_without_fill = default_params();
+
+    let run_with = run_one_probe(params_with_fill, TimestampFields::default());
+    let run_without = run_one_probe(params_without_fill, TimestampFields::default());
+
+    let (echo_len_with, _, _, _) = echo_observation(&run_with);
+    let (echo_len_without, _, _, _) = echo_observation(&run_without);
+    assert_eq!(echo_len_with, echo_len_without);
+    assert_eq!(
+        test_echo_packet_len(false, &run_with.negotiated.params),
+        echo_len_with
+    );
+    assert_eq!(
+        test_echo_packet_len(false, &run_without.negotiated.params),
+        echo_len_without
+    );
+}

crates/irtt-integration-tests/tests/fill_wire_bytes.rs

@@ -0,0 +1,216 @@
+//! A negotiated fill over a real socket.
+//!
+//! The core tests are authoritative for fill bytes; this exists to prove the
+//! whole path — negotiation, session, echo reply, UDP — puts the requested
+//! pattern on the wire. It drives the server with the production protocol codecs
+//! over a plain Tokio socket rather than through `AsyncClient`, because a
+//! client's reply payload is not part of its event API and exposing it merely to
+//! be asserted here would be the wrong trade.
+
+use std::time::Duration;
+
+use irtt_client::{
+    AsyncClient, ClientConfig, NegotiationPolicy, NegotiationRestriction, OpenOutcome, SocketConfig,
+};
+use irtt_proto::{
+    decode_echo_reply, decode_open_reply, echo_header_len, encode_request, Clock, Params,
+    ReceivedStats, RequestToEncode, ServerFill, StampAt,
+};
+use irtt_server::{Server, ServerConfig};
+use tokio::{net::UdpSocket, sync::oneshot, time::timeout};
+
+const DEADBEEF: &[u8] = &[0xde, 0xad, 0xbe, 0xef];
+
+#[tokio::test(flavor = "current_thread")]
+async fn a_negotiated_pattern_reaches_the_wire() {
+    timeout(Duration::from_secs(2), exercise_patterned_session())
+        .await
+        .expect("the fill integration exceeded its bounded runtime");
+}
+
+async fn exercise_patterned_session() {
+    let mut server = Server::bind("127.0.0.1:0".parse().unwrap(), ServerConfig::default())
+        .await
+        .unwrap();
+    let server_addr = server.local_addr().unwrap();
+    let (shutdown_tx, shutdown_rx) = oneshot::channel();
+    let server_task = tokio::spawn(async move {
+        server
+            .run(async {
+                let _ = shutdown_rx.await;
+            })
+            .await
+    });
+
+    let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+    socket.connect(server_addr).await.unwrap();
+
+    let requested = requested_params(24);
+    socket
+        .send(
+            &encode_request(
+                RequestToEncode::Open {
+                    params: &requested,
+                    no_test: false,
+                },
+                None,
+            )
+            .unwrap(),
+        )
+        .await
+        .unwrap();
+    let reply = decode_open_reply(&receive(&socket).await, None).unwrap();
+    let negotiated = reply.params;
+    assert_eq!(
+        negotiated.server_fill, requested.server_fill,
+        "a valid descriptor is returned unchanged, so strict negotiation holds"
+    );
+
+    // The request carries its own payload bytes, none of which may come back.
+    socket
+        .send(
+            &encode_request(
+                RequestToEncode::Echo {
+                    token: reply.token,
+                    sequence: 0,
+                    params: &negotiated,
+                    payload: &[0xa5; 24],
+                },
+                None,
+            )
+            .unwrap(),
+        )
+        .await
+        .unwrap();
+    let packet = receive(&socket).await;
+    let echo = decode_echo_reply(&packet, &negotiated, None).unwrap();
+
+    assert_eq!(packet.len() as i64, negotiated.length);
+    assert_eq!(
+        echo.payload,
+        DEADBEEF
+            .iter()
+            .copied()
+            .cycle()
+            .take(24)
+            .collect::<Vec<_>>()
+    );
+
+    socket
+        .send(&encode_request(RequestToEncode::Close { token: reply.token }, None).unwrap())
+        .await
+        .unwrap();
+
+    shutdown_tx.send(()).unwrap();
+    server_task.await.unwrap().unwrap();
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn a_client_sees_the_fallback_as_a_negotiation_restriction() {
+    timeout(Duration::from_secs(2), exercise_fallback_negotiation())
+        .await
+        .expect("the negotiation integration exceeded its bounded runtime");
+}
+
+/// The fallback is deliberately visible to a client rather than hidden: a
+/// descriptor this server could not honor comes back as the default one, which a
+/// strict client refuses and a loose client accepts with a restriction recorded.
+async fn exercise_fallback_negotiation() {
+    let mut server = Server::bind("127.0.0.1:0".parse().unwrap(), ServerConfig::default())
+        .await
+        .unwrap();
+    let server_addr = server.local_addr().unwrap();
+    let (shutdown_tx, shutdown_rx) = oneshot::channel();
+    let server_task = tokio::spawn(async move {
+        server
+            .run(async {
+                let _ = shutdown_rx.await;
+            })
+            .await
+    });
+
+    // `ClientConfig` accepts any non-empty descriptor within the wire bound, so
+    // an unknown one genuinely reaches the server.
+    let config = |policy| ClientConfig {
+        server_addr: server_addr.to_string(),
+        duration: Some(Duration::from_secs(1)),
+        interval: Duration::from_millis(100),
+        length: 64,
+        server_fill: Some("bogus".to_owned()),
+        negotiation_policy: policy,
+        open_timeouts: vec![Duration::from_millis(200)],
+        socket_config: SocketConfig {
+            recv_timeout: Some(Duration::from_millis(200)),
+            ..SocketConfig::default()
+        },
+        probe_timeout: Duration::from_millis(200),
+        ..ClientConfig::default()
+    };
+
+    let mut strict = AsyncClient::connect(config(NegotiationPolicy::Strict))
+        .await
+        .unwrap();
+    let rejected = strict
+        .open()
+        .await
+        .expect_err("a strict client must refuse the changed fill");
+    assert!(
+        rejected.to_string().contains("fill"),
+        "the restriction names the fill: {rejected}"
+    );
+
+    let mut loose = AsyncClient::connect(config(NegotiationPolicy::Loose))
+        .await
+        .unwrap();
+    match loose.open().await.unwrap() {
+        OpenOutcome::Started { negotiated, .. } => {
+            assert_eq!(
+                negotiated
+                    .params
+                    .server_fill
+                    .map(|fill| fill.value)
+                    .as_deref(),
+                Some("pattern:69727474"),
+                "the server reports the default descriptor it fell back to"
+            );
+            assert!(negotiated
+                .restrictions
+                .contains(&NegotiationRestriction::ServerFillChanged));
+        }
+        OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"),
+    }
+    loose.close().await.unwrap();
+
+    shutdown_tx.send(()).unwrap();
+    server_task.await.unwrap().unwrap();
+}
+
+/// Params requesting a `deadbeef` fill and exactly `payload_len` payload bytes.
+fn requested_params(payload_len: usize) -> Params {
+    let mut params = Params {
+        protocol_version: 1,
+        duration_ns: 1_000_000_000,
+        interval_ns: 100_000_000,
+        length: 0,
+        received_stats: ReceivedStats::Both,
+        stamp_at: StampAt::Both,
+        clock: Clock::Both,
+        dscp: 0,
+        server_fill: Some(ServerFill {
+            value: "pattern:deadbeef".to_owned(),
+        }),
+    };
+    params.length = (echo_header_len(false, &params) + payload_len) as i64;
+    params
+}
+
+/// One datagram, with a bounded wait so a lost reply fails rather than hangs.
+async fn receive(socket: &UdpSocket) -> Vec<u8> {
+    let mut buffer = vec![0; 2048];
+    let read = timeout(Duration::from_secs(1), socket.recv(&mut buffer))
+        .await
+        .expect("the server must answer within the test's bound")
+        .unwrap();
+    buffer.truncate(read);
+    buffer
+}

crates/irtt-integration-tests/tests/hmac.rs

@@ -0,0 +1,309 @@
+mod support;
+
+use std::time::Duration;
+
+use irtt_client::{
+    Client, ClientConfig, ClientError, ClientEvent, OpenOutcome, SocketConfig, WarningKind,
+};
+use irtt_proto::TimestampFields;
+
+use support::{
+    config_for_params, default_params, params_for_modes, run_one_probe_with_config,
+    standard_timestamps, start_bad_hmac_echo_reply_server, start_hmac_close_server,
+    start_hmac_required_open_drop_server, BackendPeer, ServerObservation, TOKEN,
+};
+
+use crate::support::test_echo_packet_len;
+
+#[test]
+fn hmac_open_success_negotiates_without_warnings() {
+    let key = b"compat-secret".to_vec();
+    let params = default_params();
+    let config_params = params.clone();
+    let config_key = key.clone();
+
+    let run = run_one_probe_with_config(
+        params.clone(),
+        TimestampFields::default(),
+        Some(key),
+        |addr| ClientConfig {
+            hmac_key: Some(config_key),
+            ..config_for_params(addr, &config_params)
+        },
+    );
+
+    assert_eq!(run.negotiated.params, params);
+    match &run.observations[0] {
+        ServerObservation::Open { params: got, hmac } => {
+            assert_eq!(got, &params);
+            assert!(*hmac);
+        }
+        other => panic!("expected open observation, got {other:?}"),
+    }
+    assert_no_warning_or_error(&[run.sent, run.reply]);
+}
+
+#[test]
+fn hmac_echo_success_verifies_request_and_accepts_reply() {
+    let key = b"compat-secret".to_vec();
+    let params = default_params();
+    let config_params = params.clone();
+    let config_key = key.clone();
+
+    let run = run_one_probe_with_config(params.clone(), standard_timestamps(), Some(key), |addr| {
+        ClientConfig {
+            hmac_key: Some(config_key),
+            ..config_for_params(addr, &config_params)
+        }
+    });
+
+    match &run.observations[1] {
+        ServerObservation::Echo {
+            len,
+            hmac,
+            token,
+            sequence,
+        } => {
+            assert_eq!(*len, test_echo_packet_len(true, &params));
+            assert!(*hmac);
+            assert_eq!(*token, TOKEN);
+            assert_eq!(*sequence, 0);
+        }
+        other => panic!("expected echo observation, got {other:?}"),
+    }
+
+    match &run.reply {
+        ClientEvent::EchoReply {
+            seq,
+            bytes,
+            server_timing,
+            ..
+        } => {
+            assert_eq!(*seq, 0);
+            assert_eq!(*bytes, test_echo_packet_len(true, &params));
+            assert!(server_timing.is_some());
+        }
+        other => panic!("expected EchoReply, got {other:?}"),
+    }
+    assert_no_duplicate_late_or_warning(&[run.reply]);
+}
+
+#[test]
+fn hmac_close_success_sends_authenticated_close_and_closes_session() {
+    let key = b"compat-secret".to_vec();
+    let params = default_params();
+    let server = start_hmac_close_server(params.clone(), key.clone());
+    let mut config = config_for_params(server.addr, &params);
+    config.hmac_key = Some(key);
+
+    let mut client = Client::connect(config).unwrap();
+    let outcome = client.open().unwrap();
+    assert_started(outcome, &params);
+
+    let events = client.close().unwrap();
+    assert!(matches!(
+        events.as_slice(),
+        [ClientEvent::SessionClosed { token: TOKEN, .. }]
+    ));
+    assert!(matches!(
+        client.send_probe(),
+        Err(ClientError::AlreadyClosed)
+    ));
+
+    let observations = server.observations(2);
+    match &observations[1] {
+        ServerObservation::Close { hmac, token } => {
+            assert!(*hmac);
+            assert_eq!(*token, TOKEN);
+        }
+        other => panic!("expected close observation, got {other:?}"),
+    }
+    server.join();
+}
+
+#[test]
+fn hmac_required_server_rejects_missing_or_wrong_client_key() {
+    // Both rejections leave the client timing out, but they are different
+    // failures on the wire: a client with no key presents no HMAC field at
+    // all, while a client with the wrong key presents one that does not
+    // authenticate.
+    for (hmac_key, expected_hmac_flag, expected_bad_hmac) in [
+        (None, false, false),
+        (Some(b"wrong-secret".to_vec()), true, true),
+    ] {
+        let server_key = b"compat-secret".to_vec();
+        let server = start_hmac_required_open_drop_server(server_key, Duration::from_millis(250));
+        let mut config = config_for_params(server.addr, &default_params());
+        config.open_timeouts = vec![Duration::from_millis(200)];
+        config.hmac_key = hmac_key;
+
+        let mut client = Client::connect(config).unwrap();
+        assert!(matches!(client.open(), Err(ClientError::OpenTimeout)));
+
+        let observations = server.observations(1);
+        assert_eq!(
+            observations.as_slice(),
+            [ServerObservation::RejectedHmac {
+                hmac: expected_hmac_flag,
+                bad_hmac: expected_bad_hmac,
+            }]
+        );
+        server.join();
+    }
+}
+
+#[test]
+fn bad_hmac_echo_reply_is_rejected_without_echo_reply_event() {
+    let key = b"compat-secret".to_vec();
+    let params = default_params();
+    let server = start_bad_hmac_echo_reply_server(params.clone(), key.clone());
+    let mut config = config_for_params(server.addr, &params);
+    config.hmac_key = Some(key);
+    config.socket_config = SocketConfig {
+        recv_timeout: Some(Duration::from_millis(500)),
+        ..Default::default()
+    };
+
+    let mut client = Client::connect(config).unwrap();
+    let outcome = client.open().unwrap();
+    assert_started(outcome, &params);
+
+    let sent = client.send_probe().unwrap();
+    assert_eq!(sent.len(), 1);
+    let events = client.recv_once().unwrap();
+    assert!(matches!(
+        events.as_slice(),
+        [ClientEvent::Warning {
+            kind: WarningKind::MalformedOrUnrelatedPacket,
+            ..
+        }]
+    ));
+    assert!(!events
+        .iter()
+        .any(|event| matches!(event, ClientEvent::EchoReply { .. })));
+
+    let observations = server.observations(2);
+    assert!(matches!(
+        observations.as_slice(),
+        [
+            ServerObservation::Open { hmac: true, .. },
+            ServerObservation::Echo {
+                hmac: true,
+                token: TOKEN,
+                sequence: 0,
+                ..
+            }
+        ]
+    ));
+    server.join();
+}
+
+#[test]
+fn hmac_open_reply_with_bad_hmac_is_ignored_until_timeout() {
+    let key = b"compat-secret".to_vec();
+    let wrong_key = b"wrong-secret".to_vec();
+    let params = default_params();
+    let server = support::start_bad_hmac_open_reply_server(params.clone(), key.clone(), wrong_key);
+    let mut config = config_for_params(server.addr, &params);
+    config.hmac_key = Some(key);
+
+    let mut client = Client::connect(config).unwrap();
+    assert!(matches!(client.open(), Err(ClientError::OpenTimeout)));
+    server.join();
+}
+
+#[test]
+fn non_hmac_client_open_does_not_set_hmac_flag() {
+    let params = params_for_modes(
+        irtt_proto::ReceivedStats::None,
+        irtt_proto::StampAt::None,
+        irtt_proto::Clock::Both,
+    );
+    let server = support::start_open_server(params.clone(), None);
+    let mut config = config_for_params(server.addr, &params);
+    config.hmac_key = None;
+
+    let mut client = Client::connect(config).unwrap();
+    let outcome = client.open().unwrap();
+    assert_started(outcome, &params);
+
+    let observations = server.observations(1);
+    assert!(matches!(
+        observations.as_slice(),
+        [ServerObservation::Open { hmac: false, .. }]
+    ));
+    server.join();
+}
+
+fn assert_started(outcome: OpenOutcome, expected: &irtt_proto::Params) {
+    match outcome {
+        OpenOutcome::Started {
+            token,
+            negotiated,
+            event:
+                ClientEvent::SessionStarted {
+                    token: event_token,
+                    negotiated: event_negotiated,
+                    ..
+                },
+            ..
+        } => {
+            assert_eq!(token, TOKEN);
+            assert_eq!(event_token, TOKEN);
+            assert_eq!(negotiated.params, *expected);
+            assert_eq!(event_negotiated.params, *expected);
+        }
+        other => panic!("expected started open outcome, got {other:?}"),
+    }
+}
+
+fn assert_no_warning_or_error(events: &[ClientEvent]) {
+    assert!(!events.iter().any(|event| {
+        matches!(
+            event,
+            ClientEvent::Warning { .. }
+                | ClientEvent::DuplicateReply { .. }
+                | ClientEvent::LateReply { .. }
+        )
+    }));
+}
+
+fn assert_no_duplicate_late_or_warning(events: &[ClientEvent]) {
+    assert_no_warning_or_error(events);
+}
+
+#[test]
+fn backend_hmac_correct_key_succeeds() {
+    let key = b"compat-secret".to_vec();
+    let params = default_params();
+    let peer = BackendPeer::start_open_echo(Some(key.clone()));
+    let mut config = config_for_params(peer.addr(), &params);
+    config.hmac_key = Some(key);
+
+    let mut client = Client::connect(config).unwrap();
+    let outcome = client.open().unwrap();
+    assert!(matches!(outcome, OpenOutcome::Started { .. }));
+
+    let sent = client.send_probe().unwrap();
+    assert_eq!(sent.len(), 1);
+
+    let events = client.recv_once().unwrap();
+    assert_eq!(events.len(), 1);
+    assert!(matches!(events[0], ClientEvent::EchoReply { .. }));
+
+    client.close().unwrap();
+}
+
+#[test]
+fn backend_hmac_required_rejects_missing_or_wrong_client_key() {
+    for hmac_key in [None, Some(b"wrong-secret".to_vec())] {
+        let server_key = b"compat-secret".to_vec();
+        let peer = BackendPeer::start_hmac_required(server_key);
+        let mut config = config_for_params(peer.addr(), &default_params());
+        config.open_timeouts = vec![Duration::from_millis(200)];
+        config.hmac_key = hmac_key;
+
+        let mut client = Client::connect(config).unwrap();
+        assert!(matches!(client.open(), Err(ClientError::OpenTimeout)));
+    }
+}

crates/irtt-integration-tests/tests/large_packets.rs

@@ -0,0 +1,103 @@
+mod support;
+
+use irtt_client::{Client, ClientConfig, ClientEvent, NegotiationPolicy};
+use irtt_proto::{echo_header_len, Params, TimestampFields};
+
+use support::{config_for_params, default_params, run_one_probe, BackendPeer, ServerObservation};
+
+use crate::support::test_echo_packet_len;
+
+fn params_with_length(length: i64) -> Params {
+    Params {
+        length,
+        ..default_params()
+    }
+}
+
+fn assert_echo_size(run: &support::OneProbeRun, expected: usize) {
+    match run.observations[1] {
+        ServerObservation::Echo { len, .. } => assert_eq!(len, expected),
+        ref other => panic!("expected Echo observation, got {other:?}"),
+    }
+}
+
+fn assert_reply(event: &ClientEvent) {
+    assert!(matches!(event, ClientEvent::EchoReply { .. }));
+}
+
+fn assert_sent_bytes(event: &ClientEvent, expected: usize) {
+    match event {
+        ClientEvent::EchoSent { bytes, .. } => assert_eq!(*bytes, expected),
+        other => panic!("expected EchoSent, got {other:?}"),
+    }
+}
+
+fn assert_reply_bytes(event: &ClientEvent, expected: usize) {
+    match event {
+        ClientEvent::EchoReply { bytes, .. } => assert_eq!(*bytes, expected),
+        other => panic!("expected EchoReply, got {other:?}"),
+    }
+}
+
+// ─── Echo request size: FakeServer observes exact packet size ───
+
+#[test]
+fn echo_request_size_matches_echo_packet_len_for_zero_length() {
+    let params = params_with_length(0);
+    let run = run_one_probe(params.clone(), TimestampFields::default());
+    let expected = test_echo_packet_len(false, &params);
+    assert_echo_size(&run, expected);
+    assert_sent_bytes(&run.sent, expected);
+}
+
+#[test]
+fn echo_request_size_matches_echo_packet_len_above_header() {
+    for length in [16, 60, 128, 1472, 4096] {
+        let params = params_with_length(length);
+        let run = run_one_probe(params.clone(), TimestampFields::default());
+        let expected = test_echo_packet_len(false, &params);
+        assert_eq!(
+            expected,
+            echo_header_len(false, &params).max(length as usize)
+        );
+        assert_echo_size(&run, expected);
+        assert_sent_bytes(&run.sent, expected);
+    }
+}
+
+// ─── Echo reply handling ───
+
+#[test]
+fn echo_reply_bytes_consistent_with_echo_packet_len_for_various_lengths() {
+    for length in [0, 1472, 4096] {
+        let params = params_with_length(length);
+        let run = run_one_probe(params.clone(), TimestampFields::default());
+        let expected = test_echo_packet_len(false, &params);
+        assert_reply(&run.reply);
+        assert_reply_bytes(&run.reply, expected);
+    }
+}
+
+// ─── Backend-neutral smoke ───
+
+#[test]
+fn backend_large_packet_smoke() {
+    let params = params_with_length(1472);
+    let peer = BackendPeer::start_open_echo(None);
+    let mut client = Client::connect(ClientConfig {
+        negotiation_policy: NegotiationPolicy::Loose,
+        length: 1472,
+        ..config_for_params(peer.addr(), &params)
+    })
+    .unwrap();
+
+    let outcome = client.open().unwrap();
+    assert!(matches!(outcome, irtt_client::OpenOutcome::Started { .. }));
+
+    client.send_probe().unwrap();
+    let events = client.recv_once().unwrap();
+    assert_eq!(events.len(), 1);
+    assert!(matches!(events[0], ClientEvent::EchoReply { .. }));
+
+    client.close().unwrap();
+}

crates/irtt-integration-tests/tests/negotiation_policy.rs

@@ -0,0 +1,225 @@
+//! The server's capability restrictions as a real client sees them.
+//!
+//! The core tests are authoritative for the negotiated values; these exist to
+//! prove the two ends agree over a socket. A restriction this server applies has
+//! to arrive at `irtt-client` as one of the restrictions it already models: a
+//! strict client refuses the changed value, and a loose one accepts it, records
+//! it and runs the session. Nothing in the client changed for this — that is
+//! what is being checked.
+
+use std::time::Duration;
+
+use irtt_client::{
+    AsyncClient, ClientConfig, NegotiationPolicy, NegotiationRestriction, OpenOutcome, SocketConfig,
+};
+use irtt_proto::{Clock, ReceivedStats, StampAt};
+use irtt_server::{Server, ServerConfig, TimestampAllowance};
+use tokio::{sync::oneshot, time::timeout};
+
+#[tokio::test(flavor = "current_thread")]
+async fn a_single_timestamp_allowance_reaches_a_strict_and_a_loose_client() {
+    timeout(Duration::from_secs(2), exercise_timestamp_restriction())
+        .await
+        .expect("the timestamp negotiation integration exceeded its bounded runtime");
+}
+
+async fn exercise_timestamp_restriction() {
+    let server = ServerConfig::default().with_timestamp_allowance(TimestampAllowance::Single);
+    let (server_addr, shutdown_tx, server_task) = serve(server).await;
+
+    let mut strict = client(server_addr, NegotiationPolicy::Strict).await;
+    let rejected = strict
+        .open()
+        .await
+        .expect_err("a strict client must refuse the reduced timestamp placement");
+    assert!(
+        rejected.to_string().contains("stamp-at"),
+        "the restriction names the timestamp placement: {rejected}"
+    );
+
+    let mut loose = client(server_addr, NegotiationPolicy::Loose).await;
+    let negotiated = match loose.open().await.unwrap() {
+        OpenOutcome::Started { negotiated, .. } => negotiated,
+        OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"),
+    };
+    assert_eq!(
+        negotiated.params.stamp_at,
+        StampAt::Midpoint,
+        "a request for both instants is answered with the midpoint"
+    );
+    assert_eq!(
+        negotiated.params.clock,
+        Clock::Both,
+        "the clock is not restricted with it"
+    );
+    assert!(negotiated
+        .restrictions
+        .contains(&NegotiationRestriction::StampAtChanged {
+            requested: StampAt::Both,
+            negotiated: StampAt::Midpoint,
+        }));
+
+    probe(&mut loose).await;
+    loose.close().await.unwrap();
+
+    shutdown_tx.send(()).unwrap();
+    server_task.await.unwrap().unwrap();
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn a_disallowed_dscp_reaches_a_strict_and_a_loose_client() {
+    timeout(Duration::from_secs(2), exercise_dscp_restriction())
+        .await
+        .expect("the DSCP negotiation integration exceeded its bounded runtime");
+}
+
+async fn exercise_dscp_restriction() {
+    let (server_addr, shutdown_tx, server_task) = serve(
+        ServerConfig::default()
+            .with_min_send_interval(Duration::ZERO)
+            .with_dscp_allowed(false),
+    )
+    .await;
+
+    // The client's configuration is a DSCP codepoint; 46 (EF) is raw byte 184 on
+    // the wire, and this test does not change either convention.
+    let with_dscp = |policy| ClientConfig {
+        dscp: 46,
+        ..test_config(server_addr, policy)
+    };
+
+    let mut strict = AsyncClient::connect(with_dscp(NegotiationPolicy::Strict))
+        .await
+        .unwrap();
+    let rejected = strict
+        .open()
+        .await
+        .expect_err("a strict client must refuse the removed marking");
+    assert!(
+        rejected.to_string().contains("DSCP"),
+        "the restriction names DSCP: {rejected}"
+    );
+
+    let mut loose = AsyncClient::connect(with_dscp(NegotiationPolicy::Loose))
+        .await
+        .unwrap();
+    let negotiated = match loose.open().await.unwrap() {
+        OpenOutcome::Started { negotiated, .. } => negotiated,
+        OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"),
+    };
+    assert_eq!(
+        negotiated.params.dscp, 0,
+        "the raw wire parameter is negotiated to zero"
+    );
+    assert!(negotiated
+        .restrictions
+        .contains(&NegotiationRestriction::DscpChanged {
+            requested: 46,
+            negotiated: 0,
+        }));
+
+    probe(&mut loose).await;
+    loose.close().await.unwrap();
+
+    shutdown_tx.send(()).unwrap();
+    server_task.await.unwrap().unwrap();
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn a_default_server_restricts_nothing_an_ordinary_client_asks_for() {
+    timeout(Duration::from_secs(2), exercise_default_policy())
+        .await
+        .expect("the default policy integration exceeded its bounded runtime");
+}
+
+/// The controls are opt-in, and this is what that has to mean: an ordinary
+/// client against an unconfigured server negotiates with no restriction at all,
+/// under the default *strict* policy that would refuse one.
+async fn exercise_default_policy() {
+    let (server_addr, shutdown_tx, server_task) = serve(ServerConfig::default()).await;
+
+    let mut client = AsyncClient::connect(ClientConfig {
+        dscp: 46,
+        ..test_config(server_addr, NegotiationPolicy::Strict)
+    })
+    .await
+    .unwrap();
+    let negotiated = match client.open().await.unwrap() {
+        OpenOutcome::Started { negotiated, .. } => negotiated,
+        OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"),
+    };
+
+    assert_eq!(negotiated.restrictions, Vec::new());
+    assert_eq!(negotiated.params.stamp_at, StampAt::Both);
+    assert_eq!(negotiated.params.clock, Clock::Both);
+    assert_eq!(negotiated.params.dscp, 184, "the raw byte for codepoint 46");
+
+    probe(&mut client).await;
+    client.close().await.unwrap();
+
+    shutdown_tx.send(()).unwrap();
+    server_task.await.unwrap().unwrap();
+}
+
+/// Starts one listener and returns its address, its shutdown trigger and its
+/// task.
+async fn serve(
+    config: ServerConfig,
+) -> (
+    std::net::SocketAddr,
+    oneshot::Sender<()>,
+    tokio::task::JoinHandle<Result<(), irtt_server::ServerRuntimeError>>,
+) {
+    let mut server = Server::bind("127.0.0.1:0".parse().unwrap(), config)
+        .await
+        .unwrap();
+    let server_addr = server.local_addr().unwrap();
+    let (shutdown_tx, shutdown_rx) = oneshot::channel();
+    let task = tokio::spawn(async move {
+        server
+            .run(async {
+                let _ = shutdown_rx.await;
+            })
+            .await
+    });
+    (server_addr, shutdown_tx, task)
+}
+
+/// A short deterministic client requesting both timestamps from both clocks.
+fn test_config(server_addr: std::net::SocketAddr, policy: NegotiationPolicy) -> ClientConfig {
+    ClientConfig {
+        server_addr: server_addr.to_string(),
+        duration: Some(Duration::from_secs(1)),
+        interval: Duration::from_millis(100),
+        length: 64,
+        received_stats: ReceivedStats::Both,
+        stamp_at: StampAt::Both,
+        clock: Clock::Both,
+        negotiation_policy: policy,
+        open_timeouts: vec![Duration::from_millis(200)],
+        socket_config: SocketConfig {
+            recv_timeout: Some(Duration::from_millis(200)),
+            ..SocketConfig::default()
+        },
+        probe_timeout: Duration::from_millis(200),
+        ..ClientConfig::default()
+    }
+}
+
+async fn client(server_addr: std::net::SocketAddr, policy: NegotiationPolicy) -> AsyncClient {
+    AsyncClient::connect(test_config(server_addr, policy))
+        .await
+        .unwrap()
+}
+
+/// One probe, answered: a restricted session is an ordinary session.
+async fn probe(client: &mut AsyncClient) {
+    client.send_probe().await.unwrap();
+    let events = client.recv().await.unwrap();
+    assert!(
+        events
+            .iter()
+            .any(|event| matches!(event, irtt_client::ClientEvent::EchoReply { seq: 0, .. })),
+        "the restricted session must answer an ordinary probe: {events:?}"
+    );
+}

crates/irtt-integration-tests/tests/protocol_options.rs

@@ -0,0 +1,374 @@
+mod support;
+mod protocol_options {
+    pub mod backend_smoke;
+}
+
+use std::time::Duration;
+
+use irtt_client::{
+    Client, ClientConfig, ClientError, ClientEvent, OneWayDelaySample, ReceivedStatsSample,
+    RttSample, ServerTiming, SignedDuration,
+};
+use irtt_proto::{Clock, Params, ReceivedStats, StampAt, TimestampFields};
+
+use support::{
+    config_for_params, default_params, params_for_modes, run_one_probe, run_one_probe_with_config,
+    server_fill, standard_timestamps, start_open_server, OneProbeRun, ServerObservation,
+    RECV_COUNT, RECV_WINDOW, TOKEN,
+};
+
+use crate::support::test_echo_packet_len;
+
+struct ReplyView<'a> {
+    seq: u32,
+    rtt: &'a RttSample,
+    server_timing: Option<&'a ServerTiming>,
+    one_way: Option<&'a OneWayDelaySample>,
+    received_stats: Option<&'a ReceivedStatsSample>,
+    bytes: usize,
+}
+
+#[test]
+fn received_stats_modes_drive_negotiated_echo_and_reply_events() {
+    for (mode, expected_count, expected_window) in [
+        (ReceivedStats::None, None, None),
+        (ReceivedStats::Count, Some(RECV_COUNT), None),
+        (ReceivedStats::Window, None, Some(RECV_WINDOW)),
+        (ReceivedStats::Both, Some(RECV_COUNT), Some(RECV_WINDOW)),
+    ] {
+        let params = params_for_modes(mode, StampAt::None, Clock::Both);
+        let run = run_one_probe(params.clone(), TimestampFields::default());
+        assert_negotiated_echo_use(&run, &params, false);
+
+        let reply = expect_echo_reply(&run.reply);
+        assert_eq!(reply.seq, 0);
+        assert_eq!(reply.bytes, test_echo_packet_len(false, &params));
+        assert!(reply.server_timing.is_none());
+        assert!(reply.one_way.is_none());
+        assert_received_stats(reply.received_stats, expected_count, expected_window);
+    }
+}
+
+#[test]
+fn timestamp_modes_drive_reply_event_timing_fields() {
+    for mode in [
+        StampAt::None,
+        StampAt::Send,
+        StampAt::Receive,
+        StampAt::Both,
+        StampAt::Midpoint,
+    ] {
+        let params = params_for_modes(ReceivedStats::None, mode, Clock::Both);
+        let run = run_one_probe(params.clone(), standard_timestamps());
+        assert_negotiated_echo_use(&run, &params, false);
+
+        let reply = expect_echo_reply(&run.reply);
+        assert!(reply.received_stats.is_none());
+        match mode {
+            StampAt::None => {
+                assert!(reply.server_timing.is_none());
+                assert!(reply.one_way.is_none());
+                assert!(reply.rtt.adjusted.is_none());
+                assert_eq!(
+                    reply.rtt.effective,
+                    SignedDuration::from_duration(reply.rtt.raw)
+                );
+                assert!(reply.rtt.raw > Duration::ZERO);
+            }
+            StampAt::Send => {
+                assert_timing_fields(reply.server_timing, false, false, false, false, true, true);
+                assert_one_way_presence(reply.one_way, false, true);
+                assert!(reply.rtt.adjusted.is_none());
+            }
+            StampAt::Receive => {
+                assert_timing_fields(reply.server_timing, true, true, false, false, false, false);
+                assert_one_way_presence(reply.one_way, true, false);
+                assert!(reply.rtt.adjusted.is_none());
+            }
+            StampAt::Both => {
+                assert_timing_fields(reply.server_timing, true, true, false, false, true, true);
+                assert_one_way_presence(reply.one_way, true, true);
+                assert_processing_subtracted(reply);
+            }
+            StampAt::Midpoint => {
+                assert_timing_fields(reply.server_timing, false, false, true, true, false, false);
+                assert_one_way_presence(reply.one_way, true, true);
+                assert!(reply.rtt.adjusted.is_none());
+            }
+        }
+    }
+}
+
+#[test]
+fn clock_modes_drive_reply_event_clock_fields() {
+    for clock in [Clock::Wall, Clock::Monotonic, Clock::Both] {
+        let params = params_for_modes(ReceivedStats::None, StampAt::Both, clock);
+        let run = run_one_probe(params.clone(), standard_timestamps());
+        assert_negotiated_echo_use(&run, &params, false);
+
+        let reply = expect_echo_reply(&run.reply);
+        assert!(reply.received_stats.is_none());
+        match clock {
+            Clock::Wall => {
+                assert_timing_fields(reply.server_timing, true, false, false, false, true, false);
+                assert_one_way_presence(reply.one_way, true, true);
+            }
+            Clock::Monotonic => {
+                assert_timing_fields(reply.server_timing, false, true, false, false, false, true);
+                assert!(reply.one_way.is_none());
+            }
+            Clock::Both => {
+                assert_timing_fields(reply.server_timing, true, true, false, false, true, true);
+                assert_one_way_presence(reply.one_way, true, true);
+            }
+            Clock::Unspecified => unreachable!("clients never request an unspecified clock"),
+        }
+        assert_processing_subtracted(reply);
+    }
+}
+
+#[test]
+fn rich_mode_exposes_stats_server_timing_and_adjusted_rtt() {
+    let params = params_for_modes(ReceivedStats::Both, StampAt::Both, Clock::Both);
+    let run = run_one_probe(params.clone(), standard_timestamps());
+    assert_negotiated_echo_use(&run, &params, false);
+
+    let reply = expect_echo_reply(&run.reply);
+    assert_received_stats(reply.received_stats, Some(RECV_COUNT), Some(RECV_WINDOW));
+    assert_timing_fields(reply.server_timing, true, true, false, false, true, true);
+    assert_one_way_presence(reply.one_way, true, true);
+    assert_processing_subtracted(reply);
+}
+
+#[test]
+fn hmac_rich_mode_uses_negotiated_echo_layout_and_decodes_reply() {
+    let key = b"compat-secret".to_vec();
+    let params = params_for_modes(ReceivedStats::Both, StampAt::Both, Clock::Both);
+    let config_params = params.clone();
+    let config_key = key.clone();
+    let run = run_one_probe_with_config(params.clone(), standard_timestamps(), Some(key), |addr| {
+        ClientConfig {
+            hmac_key: Some(config_key),
+            ..config_for_params(addr, &config_params)
+        }
+    });
+    assert_negotiated_echo_use(&run, &params, true);
+
+    let reply = expect_echo_reply(&run.reply);
+    assert_eq!(reply.bytes, test_echo_packet_len(true, &params));
+    assert_received_stats(reply.received_stats, Some(RECV_COUNT), Some(RECV_WINDOW));
+    assert_timing_fields(reply.server_timing, true, true, false, false, true, true);
+}
+
+#[test]
+fn strict_open_rejects_changed_compatibility_params_from_server() {
+    let mut requested = default_params();
+    requested.length = 256;
+    // Raw wire TOS/Traffic Class byte for DSCP codepoint 46 (EF).
+    requested.dscp = 184;
+    requested.server_fill = server_fill("rand");
+
+    for returned in changed_compatibility_params(&requested) {
+        let server = start_open_server(returned, None);
+        let mut config = config_for_params(server.addr, &requested);
+        config.server_addr = server.addr.to_string();
+        let mut client = Client::connect(config).unwrap();
+        assert!(matches!(
+            client.open(),
+            Err(ClientError::NegotiationRejected { .. })
+        ));
+        server.join();
+    }
+}
+
+#[test]
+fn loose_open_uses_returned_params_for_echo_layout_and_reply_parsing() {
+    let mut requested = default_params();
+    requested.length = 128;
+    // Raw wire TOS/Traffic Class byte for DSCP codepoint 46 (EF).
+    requested.dscp = 184;
+    requested.server_fill = server_fill("rand");
+
+    let mut returned = requested.clone();
+    returned.length = 28;
+    returned.received_stats = ReceivedStats::Count;
+    returned.stamp_at = StampAt::Receive;
+    returned.clock = Clock::Wall;
+    returned.dscp = 0;
+    returned.server_fill = None;
+
+    let requested_for_config = requested.clone();
+    let run = run_one_probe_with_config(returned.clone(), standard_timestamps(), None, |addr| {
+        ClientConfig {
+            negotiation_policy: irtt_client::NegotiationPolicy::Loose,
+            ..config_for_params(addr, &requested_for_config)
+        }
+    });
+
+    assert_eq!(open_params(&run), &requested);
+    assert_eq!(run.negotiated.params, returned);
+    assert_echo_uses_params(&run, &returned, false);
+
+    let reply = expect_echo_reply(&run.reply);
+    assert_received_stats(reply.received_stats, Some(RECV_COUNT), None);
+    assert_timing_fields(reply.server_timing, true, false, false, false, false, false);
+    assert_one_way_presence(reply.one_way, true, false);
+    assert!(reply.rtt.adjusted.is_none());
+}
+
+fn changed_compatibility_params(requested: &Params) -> Vec<Params> {
+    let mut changed = Vec::new();
+
+    let mut returned = requested.clone();
+    returned.received_stats = ReceivedStats::Count;
+    changed.push(returned);
+
+    let mut returned = requested.clone();
+    returned.stamp_at = StampAt::Midpoint;
+    changed.push(returned);
+
+    let mut returned = requested.clone();
+    returned.clock = Clock::Wall;
+    changed.push(returned);
+
+    let mut returned = requested.clone();
+    returned.server_fill = None;
+    changed.push(returned);
+
+    let mut returned = requested.clone();
+    returned.dscp = 8;
+    changed.push(returned);
+
+    let mut returned = requested.clone();
+    returned.length = 128;
+    changed.push(returned);
+
+    changed
+}
+
+fn assert_negotiated_echo_use(run: &OneProbeRun, params: &Params, hmac: bool) {
+    assert_eq!(run.negotiated.params, *params);
+    assert_eq!(open_params(run), params);
+    assert_echo_uses_params(run, params, hmac);
+}
+
+fn assert_echo_uses_params(run: &OneProbeRun, params: &Params, hmac: bool) {
+    let (echo_len, echo_hmac, token, sequence) = echo_observation(run);
+    assert_eq!(echo_len, test_echo_packet_len(hmac, params));
+    assert_eq!(echo_hmac, hmac);
+    assert_eq!(token, TOKEN);
+    assert_eq!(sequence, 0);
+
+    match &run.sent {
+        ClientEvent::EchoSent { seq, bytes, .. } => {
+            assert_eq!(*seq, 0);
+            assert_eq!(*bytes, test_echo_packet_len(hmac, params));
+        }
+        other => panic!("expected EchoSent, got {other:?}"),
+    }
+}
+
+fn open_params(run: &OneProbeRun) -> &Params {
+    match &run.observations[0] {
+        ServerObservation::Open { params, .. } => params,
+        other => panic!("expected open observation, got {other:?}"),
+    }
+}
+
+fn echo_observation(run: &OneProbeRun) -> (usize, bool, u64, u32) {
+    match run.observations[1] {
+        ServerObservation::Echo {
+            len,
+            hmac,
+            token,
+            sequence,
+        } => (len, hmac, token, sequence),
+        ref other => panic!("expected echo observation, got {other:?}"),
+    }
+}
+
+fn expect_echo_reply(event: &ClientEvent) -> ReplyView<'_> {
+    match event {
+        ClientEvent::EchoReply {
+            seq,
+            rtt,
+            server_timing,
+            one_way,
+            received_stats,
+            bytes,
+            ..
+        } => ReplyView {
+            seq: *seq,
+            rtt,
+            server_timing: server_timing.as_ref(),
+            one_way: one_way.as_ref(),
+            received_stats: received_stats.as_ref(),
+            bytes: *bytes,
+        },
+        other => panic!("expected EchoReply, got {other:?}"),
+    }
+}
+
+fn assert_received_stats(
+    sample: Option<&ReceivedStatsSample>,
+    expected_count: Option<u32>,
+    expected_window: Option<u64>,
+) {
+    if expected_count.is_none() && expected_window.is_none() {
+        assert!(sample.is_none());
+    } else {
+        let sample = sample.unwrap();
+        assert_eq!(sample.count, expected_count);
+        assert_eq!(sample.window, expected_window);
+    }
+}
+
+fn assert_timing_fields(
+    sample: Option<&ServerTiming>,
+    recv_wall: bool,
+    recv_mono: bool,
+    midpoint_wall: bool,
+    midpoint_mono: bool,
+    send_wall: bool,
+    send_mono: bool,
+) {
+    if !(recv_wall || recv_mono || midpoint_wall || midpoint_mono || send_wall || send_mono) {
+        assert!(sample.is_none());
+        return;
+    }
+
+    let sample = sample.unwrap();
+    assert_eq!(sample.receive_wall_ns.is_some(), recv_wall);
+    assert_eq!(sample.receive_mono_ns.is_some(), recv_mono);
+    assert_eq!(sample.midpoint_wall_ns.is_some(), midpoint_wall);
+    assert_eq!(sample.midpoint_mono_ns.is_some(), midpoint_mono);
+    assert_eq!(sample.send_wall_ns.is_some(), send_wall);
+    assert_eq!(sample.send_mono_ns.is_some(), send_mono);
+    assert_eq!(
+        sample.processing.is_some(),
+        (recv_mono && send_mono) || (recv_wall && send_wall)
+    );
+}
+
+fn assert_processing_subtracted(reply: ReplyView<'_>) {
+    let processing = reply.server_timing.unwrap().processing.unwrap();
+    assert_eq!(processing, Duration::from_nanos(1));
+    let adjusted = reply.rtt.adjusted.unwrap();
+    assert_eq!(reply.rtt.effective, adjusted);
+    assert!(adjusted < SignedDuration::from_duration(reply.rtt.raw));
+}
+
+fn assert_one_way_presence(
+    sample: Option<&OneWayDelaySample>,
+    client_to_server: bool,
+    server_to_client: bool,
+) {
+    if !(client_to_server || server_to_client) {
+        assert!(sample.is_none());
+        return;
+    }
+
+    let sample = sample.unwrap();
+    assert_eq!(sample.client_to_server.is_some(), client_to_server);
+    assert_eq!(sample.server_to_client.is_some(), server_to_client);
+}

crates/irtt-integration-tests/tests/protocol_options/backend_smoke.rs

@@ -0,0 +1,140 @@
+use irtt_client::{Client, ClientEvent};
+use irtt_proto::{Clock, ReceivedStats, StampAt};
+
+use crate::support::{config_for_params, default_params, params_for_modes, BackendPeer};
+
+#[test]
+fn backend_basic_open_echo_close() {
+    let params = default_params();
+    let peer = BackendPeer::start_open_echo(None);
+    let mut client = Client::connect(config_for_params(peer.addr(), &params)).unwrap();
+
+    let outcome = client.open().unwrap();
+    assert!(matches!(outcome, irtt_client::OpenOutcome::Started { .. }));
+
+    let sent = client.send_probe().unwrap();
+    assert_eq!(sent.len(), 1);
+
+    let events = client.recv_once().unwrap();
+    assert_eq!(events.len(), 1);
+    assert!(matches!(events[0], ClientEvent::EchoReply { .. }));
+
+    client.close().unwrap();
+}
+
+#[test]
+fn backend_received_stats_smoke() {
+    for mode in [
+        ReceivedStats::None,
+        ReceivedStats::Count,
+        ReceivedStats::Window,
+        ReceivedStats::Both,
+    ] {
+        let params = params_for_modes(mode, StampAt::None, Clock::Both);
+        let peer = BackendPeer::start_open_echo(None);
+        let mut client = Client::connect(config_for_params(peer.addr(), &params)).unwrap();
+
+        client.open().unwrap();
+        client.send_probe().unwrap();
+        let events = client.recv_once().unwrap();
+        assert_eq!(events.len(), 1);
+        assert!(matches!(events[0], ClientEvent::EchoReply { .. }));
+
+        client.close().unwrap();
+    }
+}
+
+#[test]
+fn backend_timestamp_smoke() {
+    for mode in [
+        StampAt::None,
+        StampAt::Send,
+        StampAt::Receive,
+        StampAt::Both,
+        StampAt::Midpoint,
+    ] {
+        let params = params_for_modes(ReceivedStats::None, mode, Clock::Both);
+        let peer = BackendPeer::start_open_echo(None);
+        let mut client = Client::connect(config_for_params(peer.addr(), &params)).unwrap();
+
+        client.open().unwrap();
+        client.send_probe().unwrap();
+        let events = client.recv_once().unwrap();
+        assert_eq!(events.len(), 1);
+
+        match &events[0] {
+            ClientEvent::EchoReply {
+                server_timing,
+                one_way,
+                ..
+            } => match mode {
+                StampAt::None => {
+                    assert!(server_timing.is_none());
+                    assert!(one_way.is_none());
+                }
+                StampAt::Send => {
+                    let st = server_timing.as_ref().unwrap();
+                    assert!(st.send_wall_ns.is_some() || st.send_mono_ns.is_some());
+                }
+                StampAt::Receive => {
+                    let st = server_timing.as_ref().unwrap();
+                    assert!(st.receive_wall_ns.is_some() || st.receive_mono_ns.is_some());
+                }
+                StampAt::Both => {
+                    let st = server_timing.as_ref().unwrap();
+                    assert!(st.receive_wall_ns.is_some() || st.receive_mono_ns.is_some());
+                    assert!(st.send_wall_ns.is_some() || st.send_mono_ns.is_some());
+                }
+                StampAt::Midpoint => {
+                    let st = server_timing.as_ref().unwrap();
+                    assert!(st.midpoint_wall_ns.is_some() || st.midpoint_mono_ns.is_some());
+                }
+            },
+            other => panic!("expected EchoReply, got {other:?}"),
+        }
+
+        client.close().unwrap();
+    }
+}
+
+#[test]
+fn backend_clock_smoke() {
+    for clock in [Clock::Wall, Clock::Monotonic, Clock::Both] {
+        let params = params_for_modes(ReceivedStats::None, StampAt::Both, clock);
+        let peer = BackendPeer::start_open_echo(None);
+        let mut client = Client::connect(config_for_params(peer.addr(), &params)).unwrap();
+
+        client.open().unwrap();
+        client.send_probe().unwrap();
+        let events = client.recv_once().unwrap();
+        assert_eq!(events.len(), 1);
+
+        match &events[0] {
+            ClientEvent::EchoReply { server_timing, .. } => {
+                let st = server_timing.as_ref().unwrap();
+                match clock {
+                    Clock::Wall => {
+                        assert!(st.receive_wall_ns.is_some());
+                        assert!(st.send_wall_ns.is_some());
+                    }
+                    Clock::Monotonic => {
+                        assert!(st.receive_mono_ns.is_some());
+                        assert!(st.send_mono_ns.is_some());
+                    }
+                    Clock::Both => {
+                        assert!(st.receive_wall_ns.is_some());
+                        assert!(st.receive_mono_ns.is_some());
+                        assert!(st.send_wall_ns.is_some());
+                        assert!(st.send_mono_ns.is_some());
+                    }
+                    Clock::Unspecified => {
+                        unreachable!("clients never request an unspecified clock")
+                    }
+                }
+            }
+            other => panic!("expected EchoReply, got {other:?}"),
+        }
+
+        client.close().unwrap();
+    }
+}

crates/irtt-integration-tests/tests/support/in_tree_server.rs

@@ -0,0 +1,74 @@
+use std::{
+    net::{IpAddr, Ipv4Addr, SocketAddr},
+    sync::mpsc,
+    thread::{self, JoinHandle},
+    time::Duration,
+};
+
+use irtt_server::{Server, ServerConfig};
+use tokio::sync::oneshot;
+
+const STARTUP_TIMEOUT: Duration = Duration::from_secs(2);
+
+pub struct InTreeServer {
+    pub addr: SocketAddr,
+    shutdown: Option<oneshot::Sender<()>>,
+    done: Option<JoinHandle<Result<(), irtt_server::ServerRuntimeError>>>,
+}
+
+impl InTreeServer {
+    pub fn start(config: ServerConfig) -> Self {
+        let (startup_tx, startup_rx) = mpsc::sync_channel(1);
+        let (shutdown_tx, shutdown_rx) = oneshot::channel();
+        let done = thread::spawn(move || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("failed to build in-tree test server runtime");
+            runtime.block_on(async move {
+                let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0);
+                let mut server = Server::bind(bind_addr, config).await?;
+                startup_tx
+                    .send(server.local_addr()?)
+                    .expect("test stopped waiting for in-tree server startup");
+                server
+                    .run(async {
+                        let _ = shutdown_rx.await;
+                    })
+                    .await
+            })
+        });
+
+        let addr = match startup_rx.recv_timeout(STARTUP_TIMEOUT) {
+            Ok(addr) => addr,
+            Err(error) => {
+                let _ = shutdown_tx.send(());
+                let result = done.join().expect("in-tree test server thread panicked");
+                panic!("in-tree test server failed before startup ({error}): {result:?}");
+            }
+        };
+
+        Self {
+            addr,
+            shutdown: Some(shutdown_tx),
+            done: Some(done),
+        }
+    }
+
+    fn stop_inner(&mut self) {
+        if let Some(shutdown) = self.shutdown.take() {
+            let _ = shutdown.send(());
+        }
+        if let Some(done) = self.done.take() {
+            done.join()
+                .expect("in-tree test server thread panicked")
+                .expect("in-tree test server stopped with an error");
+        }
+    }
+}
+
+impl Drop for InTreeServer {
+    fn drop(&mut self) {
+        self.stop_inner();
+    }
+}

crates/irtt-integration-tests/tests/support/mod.rs

@@ -0,0 +1,638 @@
+#![allow(dead_code)]
+
+mod in_tree_server;
+mod real_irtt;
+
+use std::{
+    io::{self, ErrorKind},
+    net::{SocketAddr, UdpSocket},
+    process::Command,
+    sync::mpsc,
+    thread::{self, JoinHandle},
+    time::{Duration, Instant, SystemTime},
+};
+
+use irtt_client::{Client, ClientConfig, ClientEvent, NegotiatedParams, OpenOutcome, SocketConfig};
+use irtt_proto::{
+    compute_hmac_in_place, decode_request, echo_packet_len, encode_echo_reply, encode_open_reply,
+    flags, verify_packet_hmac, Clock, DecodedRequest, DecodedRequestKind, EchoReply, OpenReply,
+    PacketLayout, Params, ProtoError, ReceivedStats, ServerFill, StampAt, TimestampFields,
+    HMAC_SIZE, MAGIC, PROTOCOL_VERSION,
+};
+use irtt_server::ServerConfig;
+
+pub use in_tree_server::InTreeServer;
+pub use real_irtt::RealIrtServer;
+
+const HMAC_OFFSET: usize = 4;
+pub const TOKEN: u64 = 0x1234_5678_90ab_cdef;
+pub const RECV_COUNT: u32 = 42;
+pub const RECV_WINDOW: u64 = 0x07;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TestBackendKind {
+    InTree,
+    Real,
+}
+
+pub fn selected_backend() -> TestBackendKind {
+    match std::env::var("IRTT_TEST_BACKEND").as_deref() {
+        Ok("real") => {
+            let irtt_bin = std::env::var("IRTT_BIN").unwrap_or_else(|_| "irtt".to_string());
+            match Command::new(&irtt_bin).arg("version").output() {
+                Ok(output) if output.status.success() => {
+                    debug_backend("[backend] selected backend=real");
+                    TestBackendKind::Real
+                }
+                _ => panic!("IRTT_TEST_BACKEND=real but irtt binary not found at '{irtt_bin}'"),
+            }
+        }
+        Ok("in-tree") | Err(_) => {
+            debug_backend("[backend] selected backend=in-tree");
+            TestBackendKind::InTree
+        }
+        Ok(other) => panic!("unknown IRTT_TEST_BACKEND value: {other}"),
+    }
+}
+
+fn debug_backend(message: &str) {
+    if std::env::var("IRTT_TEST_BACKEND_DEBUG").as_deref() == Ok("1") {
+        eprintln!("{message}");
+    }
+}
+
+pub enum BackendPeer {
+    InTree(InTreeServer),
+    Real(RealIrtServer),
+}
+
+impl BackendPeer {
+    pub fn addr(&self) -> SocketAddr {
+        match self {
+            BackendPeer::InTree(s) => s.addr,
+            BackendPeer::Real(s) => s.addr(),
+        }
+    }
+
+    pub fn start_open_echo(hmac_key: Option<Vec<u8>>) -> Self {
+        match selected_backend() {
+            TestBackendKind::InTree => BackendPeer::InTree(InTreeServer::start(
+                hmac_key.map_or_else(ServerConfig::default, |key| {
+                    ServerConfig::default().with_hmac_key(key)
+                }),
+            )),
+            TestBackendKind::Real => {
+                BackendPeer::Real(RealIrtServer::start(hmac_key.as_deref()).unwrap())
+            }
+        }
+    }
+
+    pub fn start_hmac_required(key: Vec<u8>) -> Self {
+        match selected_backend() {
+            TestBackendKind::InTree => BackendPeer::InTree(InTreeServer::start(
+                ServerConfig::default().with_hmac_key(key),
+            )),
+            TestBackendKind::Real => BackendPeer::Real(RealIrtServer::start(Some(&key)).unwrap()),
+        }
+    }
+}
+
+pub struct FakeServer {
+    pub addr: SocketAddr,
+    rx: mpsc::Receiver<ServerObservation>,
+    done: JoinHandle<()>,
+}
+
+impl FakeServer {
+    pub fn observations(&self, count: usize) -> Vec<ServerObservation> {
+        self.rx.iter().take(count).collect()
+    }
+
+    pub fn join(self) {
+        self.done.join().unwrap();
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ServerObservation {
+    Open {
+        params: Params,
+        hmac: bool,
+    },
+    RejectedHmac {
+        hmac: bool,
+        bad_hmac: bool,
+    },
+    Echo {
+        len: usize,
+        hmac: bool,
+        token: u64,
+        sequence: u32,
+    },
+    Close {
+        hmac: bool,
+        token: u64,
+    },
+}
+
+pub struct OneProbeRun {
+    pub negotiated: NegotiatedParams,
+    pub sent: ClientEvent,
+    pub reply: ClientEvent,
+    pub observations: Vec<ServerObservation>,
+}
+
+pub fn default_params() -> Params {
+    Params {
+        protocol_version: PROTOCOL_VERSION,
+        duration_ns: 3_000_000_000,
+        interval_ns: 1_000_000_000,
+        received_stats: ReceivedStats::Both,
+        stamp_at: StampAt::Both,
+        clock: Clock::Both,
+        ..Params::default()
+    }
+}
+
+pub fn params_for_modes(received_stats: ReceivedStats, stamp_at: StampAt, clock: Clock) -> Params {
+    Params {
+        received_stats,
+        stamp_at,
+        clock,
+        ..default_params()
+    }
+}
+
+/// Builds a [`ClientConfig`] whose negotiated `Params` will match `params`.
+///
+/// `params.dscp` is the raw IP TOS / Traffic Class byte, while
+/// [`ClientConfig::dscp`] is the six-bit codepoint that produces it; shift
+/// right by two to recover the codepoint the resulting config must request.
+/// Panics if `params.dscp` has any ECN bits set, since those cannot be
+/// recovered from a codepoint and this helper's contract would silently stop
+/// holding otherwise.
+pub fn config_for_params(addr: SocketAddr, params: &Params) -> ClientConfig {
+    let traffic_class =
+        u8::try_from(params.dscp).expect("test DSCP must fit the raw traffic-class byte");
+    assert_eq!(
+        traffic_class & 0b11,
+        0,
+        "test Params::dscp must be representable by ClientConfig's DSCP codepoint"
+    );
+    ClientConfig {
+        server_addr: addr.to_string(),
+        duration: if params.duration_ns == 0 {
+            None
+        } else {
+            Some(Duration::from_nanos(
+                u64::try_from(params.duration_ns).expect("test duration must be non-negative"),
+            ))
+        },
+        interval: Duration::from_nanos(
+            u64::try_from(params.interval_ns).expect("test interval must be non-negative"),
+        ),
+        length: u32::try_from(params.length).unwrap(),
+        received_stats: params.received_stats,
+        stamp_at: params.stamp_at,
+        clock: params.clock,
+        dscp: traffic_class >> 2,
+        server_fill: params.server_fill.as_ref().map(|fill| fill.value.clone()),
+        open_timeouts: vec![Duration::from_millis(200)],
+        socket_config: SocketConfig {
+            recv_timeout: Some(Duration::from_millis(500)),
+            ..Default::default()
+        },
+        ..ClientConfig::default()
+    }
+}
+
+pub fn standard_timestamps() -> TimestampFields {
+    TimestampFields {
+        recv_wall: Some(0),
+        recv_mono: Some(5_000_000),
+        midpoint_wall: Some(0),
+        midpoint_mono: Some(5_000_000),
+        send_wall: Some(1),
+        send_mono: Some(5_000_001),
+    }
+}
+
+pub fn run_one_probe(params: Params, timestamps: TimestampFields) -> OneProbeRun {
+    let config_params = params.clone();
+    run_one_probe_with_config(params, timestamps, None, |addr| {
+        config_for_params(addr, &config_params)
+    })
+}
+
+pub fn run_one_probe_with_config<F>(
+    server_params: Params,
+    timestamps: TimestampFields,
+    hmac_key: Option<Vec<u8>>,
+    build_config: F,
+) -> OneProbeRun
+where
+    F: FnOnce(SocketAddr) -> ClientConfig,
+{
+    let server = start_one_probe_server(server_params, timestamps, hmac_key);
+    let mut client = Client::connect(build_config(server.addr)).unwrap();
+    let negotiated = assert_started(client.open().unwrap());
+
+    let sent_events = client.send_probe().unwrap();
+    assert_eq!(sent_events.len(), 1);
+
+    let reply_events = client.recv_once().unwrap();
+    assert_eq!(reply_events.len(), 1);
+
+    let observations = server.observations(2);
+    server.join();
+
+    OneProbeRun {
+        negotiated,
+        sent: sent_events.into_iter().next().unwrap(),
+        reply: reply_events.into_iter().next().unwrap(),
+        observations,
+    }
+}
+
+pub fn start_open_server(params: Params, hmac_key: Option<Vec<u8>>) -> FakeServer {
+    start_fake_server(move |socket, tx| {
+        let (request, peer) = recv_request(&socket);
+        let (open_params, hmac) = open_request_params(&request, hmac_key.as_deref());
+        tx.send(ServerObservation::Open {
+            params: open_params,
+            hmac,
+        })
+        .unwrap();
+
+        let reply = open_reply(
+            flags::FLAG_OPEN | flags::FLAG_REPLY,
+            TOKEN,
+            &params,
+            hmac_key.as_deref(),
+        );
+        socket.send_to(&reply, peer).unwrap();
+    })
+}
+
+pub fn start_hmac_required_open_drop_server(key: Vec<u8>, wait: Duration) -> FakeServer {
+    start_fake_server(move |socket, tx| {
+        socket.set_read_timeout(Some(wait)).unwrap();
+        while let Some(request) = recv_request_timeout(&socket) {
+            let hmac = decode_request(&request).is_ok_and(|request| request.hmac_present);
+            // A server with a key drops any request that fails authentication,
+            // whether it carried no HMAC field at all or a field signed with
+            // the wrong key.
+            match verify_packet_hmac(&key, &request) {
+                Ok(()) => {}
+                Err(error) => {
+                    tx.send(ServerObservation::RejectedHmac {
+                        hmac,
+                        bad_hmac: error == ProtoError::BadHmac,
+                    })
+                    .unwrap();
+                }
+            }
+        }
+    })
+}
+
+pub fn start_hmac_close_server(params: Params, key: Vec<u8>) -> FakeServer {
+    start_fake_server(move |socket, tx| {
+        let (request, peer) = recv_request(&socket);
+        let (open_params, hmac) = open_request_params(&request, Some(&key));
+        tx.send(ServerObservation::Open {
+            params: open_params,
+            hmac,
+        })
+        .unwrap();
+
+        let reply = open_reply(
+            flags::FLAG_OPEN | flags::FLAG_REPLY,
+            TOKEN,
+            &params,
+            Some(&key),
+        );
+        socket.send_to(&reply, peer).unwrap();
+
+        socket
+            .set_read_timeout(Some(Duration::from_secs(2)))
+            .unwrap();
+        let (request, peer) = recv_request(&socket);
+        let close = observe_close_request(&request, &key);
+        tx.send(close).unwrap();
+
+        let reply = close_reply(TOKEN, &key);
+        socket.send_to(&reply, peer).unwrap();
+    })
+}
+
+pub fn start_bad_hmac_echo_reply_server(params: Params, key: Vec<u8>) -> FakeServer {
+    start_fake_server(move |socket, tx| {
+        let (request, peer) = recv_request(&socket);
+        let (open_params, hmac) = open_request_params(&request, Some(&key));
+        tx.send(ServerObservation::Open {
+            params: open_params,
+            hmac,
+        })
+        .unwrap();
+
+        let reply = open_reply(
+            flags::FLAG_OPEN | flags::FLAG_REPLY,
+            TOKEN,
+            &params,
+            Some(&key),
+        );
+        socket.send_to(&reply, peer).unwrap();
+
+        socket
+            .set_read_timeout(Some(Duration::from_secs(2)))
+            .unwrap();
+        let (request, peer) = recv_request(&socket);
+        let echo = observe_echo_request(&request, Some(&key));
+        let sequence = match echo {
+            ServerObservation::Echo { sequence, .. } => sequence,
+            _ => unreachable!(),
+        };
+        tx.send(echo).unwrap();
+
+        let mut reply_packet = echo_reply_packet(
+            TOKEN,
+            sequence,
+            &params,
+            &TimestampFields::default(),
+            Some(&key),
+        );
+        reply_packet[HMAC_OFFSET] ^= 0xff;
+        socket.send_to(&reply_packet, peer).unwrap();
+    })
+}
+
+pub fn start_bad_hmac_open_reply_server(
+    params: Params,
+    request_key: Vec<u8>,
+    reply_key: Vec<u8>,
+) -> FakeServer {
+    start_fake_server(move |socket, tx| {
+        let (request, peer) = recv_request(&socket);
+        let (open_params, hmac) = open_request_params(&request, Some(&request_key));
+        tx.send(ServerObservation::Open {
+            params: open_params,
+            hmac,
+        })
+        .unwrap();
+
+        let reply = open_reply(
+            flags::FLAG_OPEN | flags::FLAG_REPLY,
+            TOKEN,
+            &params,
+            Some(&reply_key),
+        );
+        socket.send_to(&reply, peer).unwrap();
+    })
+}
+
+pub fn server_fill(value: &str) -> Option<ServerFill> {
+    Some(ServerFill {
+        value: value.to_owned(),
+    })
+}
+
+fn start_one_probe_server(
+    params: Params,
+    timestamps: TimestampFields,
+    hmac_key: Option<Vec<u8>>,
+) -> FakeServer {
+    start_fake_server(move |socket, tx| {
+        let (request, peer) = recv_request(&socket);
+        let (open_params, hmac) = open_request_params(&request, hmac_key.as_deref());
+        tx.send(ServerObservation::Open {
+            params: open_params,
+            hmac,
+        })
+        .unwrap();
+
+        let reply = open_reply(
+            flags::FLAG_OPEN | flags::FLAG_REPLY,
+            TOKEN,
+            &params,
+            hmac_key.as_deref(),
+        );
+        socket.send_to(&reply, peer).unwrap();
+
+        socket
+            .set_read_timeout(Some(Duration::from_secs(2)))
+            .unwrap();
+        let (request, peer) = recv_request(&socket);
+        let echo = observe_echo_request(&request, hmac_key.as_deref());
+        let sequence = match echo {
+            ServerObservation::Echo { sequence, .. } => sequence,
+            _ => unreachable!(),
+        };
+        tx.send(echo).unwrap();
+
+        let timestamps = materialize_wall_timestamps(timestamps);
+        let reply_packet =
+            echo_reply_packet(TOKEN, sequence, &params, &timestamps, hmac_key.as_deref());
+        socket.send_to(&reply_packet, peer).unwrap();
+    })
+}
+
+fn start_fake_server<F>(handler: F) -> FakeServer
+where
+    F: FnOnce(UdpSocket, mpsc::Sender<ServerObservation>) + Send + 'static,
+{
+    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+    let addr = socket.local_addr().unwrap();
+    let (tx, rx) = mpsc::channel();
+    let done = thread::spawn(move || handler(socket, tx));
+    FakeServer { addr, rx, done }
+}
+
+/// A `recv_from` that retries an interrupted syscall
+/// (`EINTR`/`ErrorKind::Interrupted`) instead of failing outright, keeping
+/// one absolute deadline across any number of interruptions exactly like the
+/// production blocking receive fix this test file exists to guard: each
+/// retry re-applies only the *remaining* time against the socket's
+/// currently configured read timeout, so a signal storm cannot turn this
+/// into an unbounded wait. The configured timeout is restored before
+/// returning so a later call is unaffected by an earlier interruption here.
+fn recv_from_retrying_interrupted(
+    socket: &UdpSocket,
+    buf: &mut [u8],
+) -> io::Result<(usize, SocketAddr)> {
+    let configured_timeout = socket.read_timeout()?;
+    let deadline = configured_timeout.and_then(|timeout| Instant::now().checked_add(timeout));
+    let mut timeout_narrowed = false;
+
+    let result = loop {
+        match socket.recv_from(buf) {
+            Ok(received) => break Ok(received),
+            Err(err) if err.kind() == ErrorKind::Interrupted => {
+                if let Some(deadline) = deadline {
+                    let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
+                        break Err(io::Error::from(ErrorKind::TimedOut));
+                    };
+                    if remaining.is_zero() {
+                        break Err(io::Error::from(ErrorKind::TimedOut));
+                    }
+                    if let Err(err) = socket.set_read_timeout(Some(remaining)) {
+                        break Err(err);
+                    }
+                    timeout_narrowed = true;
+                }
+            }
+            Err(err) => break Err(err),
+        }
+    };
+
+    if timeout_narrowed {
+        socket.set_read_timeout(configured_timeout)?;
+    }
+    result
+}
+
+fn recv_request(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
+    let mut buf = [0_u8; 8192];
+    match recv_from_retrying_interrupted(socket, &mut buf) {
+        Ok((size, peer)) => (buf[..size].to_vec(), peer),
+        Err(err) => panic!("fake server recv_from failed: {err}"),
+    }
+}
+
+/// Like [`recv_request`], but any error, including a configured read
+/// timeout, means "nothing more to receive" rather than a failure.
+fn recv_request_timeout(socket: &UdpSocket) -> Option<Vec<u8>> {
+    let mut buf = [0_u8; 8192];
+    recv_from_retrying_interrupted(socket, &mut buf)
+        .ok()
+        .map(|(size, _)| buf[..size].to_vec())
+}
+
+fn assert_started(outcome: OpenOutcome) -> NegotiatedParams {
+    match outcome {
+        OpenOutcome::Started { negotiated, .. } => negotiated,
+        OpenOutcome::NoTestCompleted { .. } => panic!("expected started outcome"),
+    }
+}
+
+/// Admits a request the way a compliant server does: structural decode first,
+/// then HMAC presence policy, then authentication.
+fn admit_request<'a>(packet: &'a [u8], hmac_key: Option<&[u8]>) -> DecodedRequest<'a> {
+    let request = decode_request(packet).expect("fake peers only receive well-formed requests");
+    assert_eq!(request.hmac_present, hmac_key.is_some());
+    if let Some(key) = hmac_key {
+        verify_packet_hmac(key, packet).unwrap();
+    }
+    request
+}
+
+fn open_request_params(packet: &[u8], hmac_key: Option<&[u8]>) -> (Params, bool) {
+    let request = admit_request(packet, hmac_key);
+    match request.kind {
+        DecodedRequestKind::Open { params, .. } => {
+            (Params::decode(params).unwrap(), request.hmac_present)
+        }
+        other => panic!("expected an open request, got {other:?}"),
+    }
+}
+
+fn observe_echo_request(packet: &[u8], hmac_key: Option<&[u8]>) -> ServerObservation {
+    let request = admit_request(packet, hmac_key);
+    match request.kind {
+        DecodedRequestKind::Echo {
+            token, sequence, ..
+        } => ServerObservation::Echo {
+            len: packet.len(),
+            hmac: request.hmac_present,
+            token,
+            sequence,
+        },
+        other => panic!("expected an echo request, got {other:?}"),
+    }
+}
+
+fn observe_close_request(packet: &[u8], hmac_key: &[u8]) -> ServerObservation {
+    let request = admit_request(packet, Some(hmac_key));
+    match request.kind {
+        DecodedRequestKind::Close { token } => ServerObservation::Close { hmac: true, token },
+        other => panic!("expected a close request, got {other:?}"),
+    }
+}
+
+// irtt-proto has no encode_close_reply/CloseReply API yet, so this stays a
+// manual encoder. Migrate it once server-side close encoding is added.
+fn close_reply(token: u64, hmac_key: &[u8]) -> Vec<u8> {
+    let mut packet = Vec::new();
+    packet.extend_from_slice(&MAGIC);
+    packet.push(flags::FLAG_REPLY | flags::FLAG_CLOSE | flags::FLAG_HMAC);
+    packet.extend_from_slice(&[0_u8; HMAC_SIZE]);
+    packet.extend_from_slice(&token.to_le_bytes());
+    compute_hmac_in_place(hmac_key, &mut packet, HMAC_OFFSET).unwrap();
+    packet
+}
+
+fn open_reply(flags: u8, token: u64, params: &Params, hmac_key: Option<&[u8]>) -> Vec<u8> {
+    encode_open_reply(
+        &OpenReply {
+            flags,
+            token,
+            params: params.clone(),
+        },
+        hmac_key,
+    )
+    .unwrap()
+}
+
+pub(super) fn test_echo_packet_len(hmac: bool, params: &Params) -> usize {
+    echo_packet_len(hmac, params).expect("test params must have a representable length")
+}
+
+fn echo_reply_packet(
+    token: u64,
+    seq: u32,
+    params: &Params,
+    timestamps: &TimestampFields,
+    hmac_key: Option<&[u8]>,
+) -> Vec<u8> {
+    let layout = PacketLayout::echo(hmac_key.is_some(), params);
+    let reply = EchoReply {
+        flags: flags::FLAG_REPLY,
+        token,
+        sequence: seq,
+        recv_count: layout.recv_count.then_some(RECV_COUNT),
+        recv_window: layout.recv_window.then_some(RECV_WINDOW),
+        timestamps: TimestampFields {
+            recv_wall: layout.recv_wall.then(|| timestamps.recv_wall.unwrap_or(0)),
+            recv_mono: layout.recv_mono.then(|| timestamps.recv_mono.unwrap_or(0)),
+            midpoint_wall: layout
+                .midpoint_wall
+                .then(|| timestamps.midpoint_wall.unwrap_or(0)),
+            midpoint_mono: layout
+                .midpoint_mono
+                .then(|| timestamps.midpoint_mono.unwrap_or(0)),
+            send_wall: layout.send_wall.then(|| timestamps.send_wall.unwrap_or(0)),
+            send_mono: layout.send_mono.then(|| timestamps.send_mono.unwrap_or(0)),
+        },
+        payload: Vec::new(),
+    };
+    encode_echo_reply(&reply, params, hmac_key).unwrap()
+}
+
+fn materialize_wall_timestamps(mut timestamps: TimestampFields) -> TimestampFields {
+    let now_ns: i64 = SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .unwrap()
+        .as_nanos()
+        .try_into()
+        .expect("current wall-clock time fits i64 nanoseconds");
+    if let Some(offset) = timestamps.recv_wall {
+        timestamps.recv_wall = Some(now_ns + offset);
+    }
+    if let Some(offset) = timestamps.midpoint_wall {
+        timestamps.midpoint_wall = Some(now_ns + offset);
+    }
+    if let Some(offset) = timestamps.send_wall {
+        timestamps.send_wall = Some(now_ns + offset);
+    }
+    timestamps
+}

crates/irtt-integration-tests/tests/support/real_irtt.rs

@@ -0,0 +1,277 @@
+#![allow(dead_code)]
+
+use std::io::{BufRead, BufReader, Read};
+use std::net::{SocketAddr, UdpSocket};
+use std::process::{Child, Command, Stdio};
+use std::sync::{mpsc, Arc, Mutex};
+use std::thread::{self, JoinHandle};
+use std::time::{Duration, Instant};
+
+const STARTUP_ATTEMPTS: usize = 5;
+const STARTUP_TIMEOUT: Duration = Duration::from_secs(5);
+
+pub struct RealIrtServer {
+    addr: SocketAddr,
+    child: Child,
+    keepalive_ms: Option<u64>,
+    stdout_thread: Option<JoinHandle<()>>,
+    stderr_thread: Option<JoinHandle<()>>,
+}
+
+impl RealIrtServer {
+    pub fn start(hmac_key: Option<&[u8]>) -> Result<Self, String> {
+        let irtt_bin = std::env::var("IRTT_BIN").unwrap_or_else(|_| "irtt".to_string());
+        let keepalive_ms = std::env::var("IRTT_TEST_KEEP_SERVER_MS")
+            .ok()
+            .and_then(|v| v.parse::<u64>().ok());
+
+        let mut failures = Vec::new();
+
+        for attempt in 1..=STARTUP_ATTEMPTS {
+            let port = Self::find_free_port()?;
+            let addr = SocketAddr::from(([127, 0, 0, 1], port));
+            let bind = format!("127.0.0.1:{port}");
+
+            debug_eprintln!(
+                "[real_irtt] selected backend=real irtt_bin={irtt_bin} bind={bind} attempt={attempt}/{STARTUP_ATTEMPTS}"
+            );
+
+            match Self::start_attempt(&irtt_bin, keepalive_ms, hmac_key, addr, &bind) {
+                Ok(server) => return Ok(server),
+                Err(failure) => {
+                    debug_eprintln!(
+                        "[real_irtt] startup attempt {attempt}/{STARTUP_ATTEMPTS} failed for {bind}:\n{}",
+                        failure.details()
+                    );
+                    failures.push(failure);
+                }
+            }
+        }
+
+        Err(format_startup_failures(&irtt_bin, &failures))
+    }
+
+    fn start_attempt(
+        irtt_bin: &str,
+        keepalive_ms: Option<u64>,
+        hmac_key: Option<&[u8]>,
+        addr: SocketAddr,
+        bind: &str,
+    ) -> Result<Self, StartupFailure> {
+        let mut cmd = Command::new(irtt_bin);
+        cmd.arg("server")
+            .arg("-b")
+            .arg(bind)
+            .arg("--tstamp=dual")
+            .stderr(Stdio::piped())
+            .stdout(Stdio::piped());
+
+        if let Some(key) = hmac_key {
+            let hex_key: String = key.iter().map(|b| format!("{:02x}", b)).collect();
+            cmd.arg(format!("--hmac=0x{hex_key}"));
+        }
+
+        let full_cmd = format!("{cmd:?}");
+        debug_eprintln!("[real_irtt] spawning: {full_cmd}");
+
+        let mut child = cmd.spawn().map_err(|e| StartupFailure {
+            bind: bind.to_owned(),
+            command: full_cmd.clone(),
+            output: String::new(),
+            exit_status: None,
+            error: Some(format!("failed to spawn irtt server: {e}")),
+        })?;
+
+        let pid = child.id();
+        debug_eprintln!("[real_irtt] child pid={pid}");
+
+        let stdout = child.stdout.take().ok_or_else(|| StartupFailure {
+            bind: bind.to_owned(),
+            command: full_cmd.clone(),
+            output: String::new(),
+            exit_status: None,
+            error: Some("failed to capture stdout".to_owned()),
+        })?;
+        let stderr = child.stderr.take().ok_or_else(|| StartupFailure {
+            bind: bind.to_owned(),
+            command: full_cmd.clone(),
+            output: String::new(),
+            exit_status: None,
+            error: Some("failed to capture stderr".to_owned()),
+        })?;
+
+        let captured = Arc::new(Mutex::new(Vec::new()));
+        let (line_tx, line_rx) = mpsc::channel();
+
+        let stdout_thread = {
+            let line_tx = line_tx.clone();
+            let captured = Arc::clone(&captured);
+            thread::spawn(move || drain_stream("stdout", stdout, line_tx, captured))
+        };
+
+        let stderr_thread = {
+            let captured = Arc::clone(&captured);
+            thread::spawn(move || drain_stream("stderr", stderr, line_tx, captured))
+        };
+
+        let ready = wait_for_ready(&line_rx, STARTUP_TIMEOUT);
+
+        if !ready {
+            let exit_status = child.try_wait().ok().flatten();
+            debug_eprintln!("[real_irtt] readiness result=false exit_status={exit_status:?}");
+            child.kill().ok();
+            child.wait().ok();
+            stdout_thread.join().ok();
+            stderr_thread.join().ok();
+
+            let output = captured.lock().unwrap().join("\n");
+            return Err(StartupFailure {
+                bind: bind.to_owned(),
+                command: full_cmd,
+                output,
+                exit_status,
+                error: Some("readiness timed out or server exited before ready".to_owned()),
+            });
+        }
+
+        debug_eprintln!("[real_irtt] readiness result=true addr={addr}");
+
+        Ok(Self {
+            addr,
+            child,
+            keepalive_ms,
+            stdout_thread: Some(stdout_thread),
+            stderr_thread: Some(stderr_thread),
+        })
+    }
+
+    pub fn addr(&self) -> SocketAddr {
+        self.addr
+    }
+
+    fn find_free_port() -> Result<u16, String> {
+        let socket =
+            UdpSocket::bind("127.0.0.1:0").map_err(|e| format!("failed to find free port: {e}"))?;
+        let port = socket
+            .local_addr()
+            .map_err(|e| format!("failed to get local addr: {e}"))?
+            .port();
+        drop(socket);
+        Ok(port)
+    }
+}
+
+struct StartupFailure {
+    bind: String,
+    command: String,
+    output: String,
+    exit_status: Option<std::process::ExitStatus>,
+    error: Option<String>,
+}
+
+impl StartupFailure {
+    fn details(&self) -> String {
+        let mut details = format!("bind: {}\ncommand: {}", self.bind, self.command);
+        if let Some(error) = &self.error {
+            details.push_str(&format!("\nerror: {error}"));
+        }
+        if let Some(status) = self.exit_status {
+            details.push_str(&format!("\nexit status: {status}"));
+        }
+        if !self.output.is_empty() {
+            details.push_str(&format!("\noutput:\n{}", self.output));
+        }
+        details
+    }
+}
+
+fn format_startup_failures(irtt_bin: &str, failures: &[StartupFailure]) -> String {
+    let mut message = format!(
+        "irtt server failed to start after {} attempt(s) with binary '{irtt_bin}'",
+        failures.len()
+    );
+
+    for (index, failure) in failures.iter().enumerate() {
+        message.push_str(&format!(
+            "\n\nattempt {}/{}:\n{}",
+            index + 1,
+            failures.len(),
+            failure.details()
+        ));
+    }
+
+    message
+}
+
+impl Drop for RealIrtServer {
+    fn drop(&mut self) {
+        if let Some(ms) = self.keepalive_ms {
+            debug_eprintln!(
+                "[real_irtt] keepalive: sleeping {ms}ms before kill (pid={})",
+                self.child.id()
+            );
+            thread::sleep(Duration::from_millis(ms));
+        }
+        debug_eprintln!("[real_irtt] killing child pid={}", self.child.id());
+        self.child.kill().ok();
+        self.child.wait().ok();
+        if let Some(thread) = self.stdout_thread.take() {
+            thread.join().ok();
+        }
+        if let Some(thread) = self.stderr_thread.take() {
+            thread.join().ok();
+        }
+    }
+}
+
+fn wait_for_ready(line_rx: &mpsc::Receiver<(&'static str, String)>, timeout: Duration) -> bool {
+    let start = Instant::now();
+    loop {
+        let remaining = timeout
+            .checked_sub(start.elapsed())
+            .unwrap_or(Duration::ZERO);
+
+        match line_rx.recv_timeout(remaining) {
+            Ok((stream, text)) => {
+                debug_eprintln!("[real_irtt] {stream}: {text}");
+                if text.contains("[ListenerStart]") {
+                    return true;
+                }
+                if (text.contains("[ListenerStop]") || text.contains("[ServerStop]"))
+                    && (text.contains("error") || text.contains("Error"))
+                {
+                    return false;
+                }
+            }
+            Err(_) => return false,
+        }
+    }
+}
+
+fn drain_stream(
+    stream: &'static str,
+    pipe: impl Read,
+    tx: mpsc::Sender<(&'static str, String)>,
+    captured: Arc<Mutex<Vec<String>>>,
+) {
+    let reader = BufReader::new(pipe);
+    for line in reader.lines() {
+        match line {
+            Ok(text) => {
+                captured.lock().unwrap().push(format!("{stream}: {text}"));
+                let _ = tx.send((stream, text));
+            }
+            Err(_) => break,
+        }
+    }
+}
+
+macro_rules! debug_eprintln {
+    ($($arg:tt)*) => {
+        if std::env::var("IRTT_TEST_BACKEND_DEBUG").as_deref() == Ok("1") {
+            eprintln!($($arg)*);
+        }
+    };
+}
+
+use debug_eprintln;

crates/irtt-integration-tests/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,
+    }
+}

crates/irtt-integration-tests/tests/wildcard_source.rs

@@ -0,0 +1,263 @@
+//! A wildcard listener answers from the address the request was sent to.
+//!
+//! Every test here drives a real [`AsyncClient`], which reads from a *connected*
+//! UDP socket. That connection is the assertion: the kernel discards any
+//! datagram whose source is not the endpoint the client sent to, so a reply
+//! that left from another of the host's addresses is indistinguishable from no
+//! reply at all. An open, an echo and a close completing therefore prove the
+//! source address, without the test having to inspect it.
+//!
+//! Compiled on the targets that implement wildcard source selection. Elsewhere
+//! a wildcard bind is refused outright, which the runtime's own tests cover.
+
+#![cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
+
+use std::{
+    io,
+    net::{Ipv4Addr, SocketAddr, UdpSocket},
+    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};
+
+/// EF, as the client's configuration takes it, and the raw TOS byte it occupies
+/// the upper six bits of.
+const EF_DSCP: u8 = 46;
+const EF_TRAFFIC_CLASS: u8 = 0xb8;
+
+#[tokio::test(flavor = "current_thread")]
+async fn a_wildcard_ipv4_listener_replies_from_the_requested_address() {
+    timeout(Duration::from_secs(5), async {
+        let server = Server::bind("0.0.0.0:0".parse().unwrap(), unthrottled())
+            .await
+            .expect("a wildcard IPv4 listener must be supported on this target");
+        let port = server.local_addr().unwrap().port();
+        // A second loopback address makes this a genuine multi-homed case: the
+        // listener is bound to none of the host's addresses in particular, and
+        // the client contacts one the routing table would not have picked on
+        // its own. Hosts carrying only 127.0.0.1 — macOS, unless an alias is
+        // configured — still exercise the whole ancillary path, one address
+        // wide.
+        let target = SocketAddr::from((secondary_ipv4_loopback(), port));
+        let running = spawn(server);
+
+        let mut client = connected_client(target).await;
+        assert_started(&mut client, target).await;
+
+        let meta = echo_once(&mut client).await;
+        // Source selection is per-packet ancillary data and the traffic class
+        // is a socket-wide setting applied before the send. Both act on the
+        // same reply, so this is where they have to agree.
+        if cfg!(target_os = "linux") {
+            assert_eq!(
+                meta.traffic_class,
+                Some(EF_TRAFFIC_CLASS),
+                "a source-selected reply still carries its session's marking"
+            );
+            assert_eq!(meta.dscp, Some(EF_DSCP));
+            assert_eq!(meta.ecn, Some(0));
+        }
+
+        assert!(matches!(
+            client.close().await.unwrap().as_slice(),
+            [ClientEvent::SessionClosed { .. }]
+        ));
+        running.stop().await;
+    })
+    .await
+    .expect("the wildcard IPv4 source-selection test exceeded its bounded runtime");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn an_ipv4_mapped_wildcard_listener_replies_from_the_requested_address() {
+    // `[::ffff:0.0.0.0]` is a wildcard that does not look like one. Linux
+    // accepts it as an IPv4 wildcard bind and `getsockname` reports it back
+    // unchanged, so a listener that only recognized `0.0.0.0` and `[::]` would
+    // take this for an explicit address and answer from a routing-selected
+    // source. macOS normalizes the bind to `[::]`, where this is simply the
+    // IPv6 wildcard case again.
+    timeout(Duration::from_secs(5), async {
+        let server = match Server::bind("[::ffff:0.0.0.0]:0".parse().unwrap(), unthrottled()).await
+        {
+            Ok(server) => server,
+            // FreeBSD refuses a mapped bind under its default `IPV6_V6ONLY`.
+            Err(ServerRuntimeError::Bind { source, .. }) if is_family_unavailable(&source) => {
+                eprintln!("skipping IPv4-mapped wildcard test: the bind is unavailable here");
+                return;
+            }
+            Err(error) => panic!("unexpected IPv4-mapped wildcard bind failure: {error}"),
+        };
+        let port = server.local_addr().unwrap().port();
+        let target = SocketAddr::from((secondary_ipv4_loopback(), port));
+        let running = spawn(server);
+
+        let mut client = connected_client(target).await;
+        assert_started(&mut client, target).await;
+        let meta = echo_once(&mut client).await;
+        // A mapped listener is `AF_INET6` but emits IPv4, and the IPv6 Traffic
+        // Class option does not reach those packets — it leaves their TOS at
+        // zero. Only the IPv4 option marks them, so this asserts the listener
+        // was read as the IPv4 one it is.
+        if cfg!(target_os = "linux") {
+            assert_eq!(
+                meta.traffic_class,
+                Some(EF_TRAFFIC_CLASS),
+                "a mapped listener's reply still carries its session's marking"
+            );
+            assert_eq!(meta.dscp, Some(EF_DSCP));
+        }
+        assert!(matches!(
+            client.close().await.unwrap().as_slice(),
+            [ClientEvent::SessionClosed { .. }]
+        ));
+
+        running.stop().await;
+    })
+    .await
+    .expect("the IPv4-mapped wildcard test exceeded its bounded runtime");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn a_wildcard_ipv6_listener_replies_from_the_requested_address() {
+    timeout(Duration::from_secs(5), async {
+        if !ipv6_loopback_available() {
+            eprintln!("skipping wildcard IPv6 test: IPv6 loopback unavailable");
+            return;
+        }
+        let server = match Server::bind("[::]:0".parse().unwrap(), unthrottled()).await {
+            Ok(server) => server,
+            Err(ServerRuntimeError::Bind { source, .. }) if is_family_unavailable(&source) => {
+                eprintln!("skipping wildcard IPv6 test: IPv6 unavailable");
+                return;
+            }
+            Err(error) => panic!("unexpected wildcard IPv6 bind failure: {error}"),
+        };
+        let port = server.local_addr().unwrap().port();
+        let target: SocketAddr = format!("[::1]:{port}").parse().unwrap();
+        let running = spawn(server);
+
+        // Loopback is one address, so this does not prove multi-homed IPv6. It
+        // does exercise the parts that differ from IPv4: the received packet
+        // info, the interface index carried with it, and the source-selected
+        // IPv6 send.
+        let mut client = connected_client(target).await;
+        assert_started(&mut client, target).await;
+        echo_once(&mut client).await;
+        assert!(matches!(
+            client.close().await.unwrap().as_slice(),
+            [ClientEvent::SessionClosed { .. }]
+        ));
+
+        running.stop().await;
+    })
+    .await
+    .expect("the wildcard IPv6 source-selection test exceeded its bounded runtime");
+}
+
+/// A local IPv4 address that is not the primary loopback one, where the host
+/// has one.
+///
+/// Bindability is the test: an address this process can bind is one the host
+/// accepts datagrams for. `127.0.0.0/8` supplies a second local destination on
+/// hosts that carry the whole range, without touching interface configuration.
+fn secondary_ipv4_loopback() -> Ipv4Addr {
+    let candidate = Ipv4Addr::new(127, 0, 0, 2);
+    if UdpSocket::bind((candidate, 0)).is_ok() {
+        return candidate;
+    }
+    eprintln!("{candidate} is not a local address on this host; using 127.0.0.1 instead");
+    Ipv4Addr::LOCALHOST
+}
+
+fn ipv6_loopback_available() -> bool {
+    match UdpSocket::bind("[::1]:0") {
+        Ok(_) => true,
+        Err(error) if is_family_unavailable(&error) => false,
+        Err(error) => panic!("unexpected IPv6 loopback bind failure: {error}"),
+    }
+}
+
+fn is_family_unavailable(error: &io::Error) -> bool {
+    matches!(
+        error.kind(),
+        io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported
+    )
+}
+
+fn unthrottled() -> ServerConfig {
+    ServerConfig::default().with_min_send_interval(Duration::ZERO)
+}
+
+async fn connected_client(server_addr: SocketAddr) -> AsyncClient {
+    let config = ClientConfig {
+        server_addr: server_addr.to_string(),
+        duration: Some(Duration::from_secs(30)),
+        interval: Duration::from_millis(10),
+        length: 64,
+        dscp: EF_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()
+    };
+    AsyncClient::connect(config).await.unwrap()
+}
+
+async fn assert_started(client: &mut AsyncClient, server_addr: SocketAddr) {
+    match client.open().await.unwrap() {
+        OpenOutcome::Started { remote, token, .. } => {
+            assert_eq!(remote, server_addr);
+            assert_ne!(token, 0);
+        }
+        OpenOutcome::NoTestCompleted { .. } => panic!("a normal client unexpectedly ran no-test"),
+    }
+}
+
+/// Sends one echo and returns its reply's receive metadata.
+async fn echo_once(client: &mut AsyncClient) -> PacketMeta {
+    client.send_probe().await.unwrap();
+    loop {
+        for event in client.recv().await.unwrap() {
+            if let ClientEvent::EchoReply {
+                seq, packet_meta, ..
+            } = event
+            {
+                assert_eq!(seq, 0);
+                return packet_meta;
+            }
+        }
+    }
+}
+
+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 }
+}