From e115b96f21b3d1f6d5b642ae48eb8c38abbf4984 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Fri, 2 Oct 2026 15:14:05 +0700 Subject: [PATCH] test: strengthen durable behavior coverage after review --- Cargo.lock | 1 + crates/irtt-app/tests/cli.rs | 54 ++- crates/irtt-client/Cargo.toml | 3 + crates/irtt-client/tests/linux_transport.rs | 235 +++++++++++++ crates/irtt-client/tests/managed_control.rs | 160 +++++++++ .../tests/async_client_lifecycle.rs | 2 +- .../tests/managed_lifecycle.rs | 47 --- crates/irtt-server/AGENTS.md | 32 +- .../irtt-server/tests/protocol_contracts.rs | 318 ++++++++++++++++++ crates/irtt-server/tests/runtime.rs | 10 + crates/irtt-server/tests/traffic_class.rs | 133 ++++++++ crates/irtt-stats/src/ipdv.rs | 15 +- crates/irtt-stats/tests/ipdv.rs | 112 ++++++ crates/irtt-stats/tests/stats.rs | 26 +- 14 files changed, 1063 insertions(+), 85 deletions(-) create mode 100644 crates/irtt-client/tests/linux_transport.rs create mode 100644 crates/irtt-client/tests/managed_control.rs create mode 100644 crates/irtt-server/tests/protocol_contracts.rs create mode 100644 crates/irtt-server/tests/traffic_class.rs create mode 100644 crates/irtt-stats/tests/ipdv.rs diff --git a/Cargo.lock b/Cargo.lock index e45e9f9..c046d8e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -718,6 +718,7 @@ name = "irtt-client" version = "0.5.3" dependencies = [ "irtt-proto", + "irtt-server", "nix 0.31.3", "socket2", "thiserror 2.0.21", diff --git a/crates/irtt-app/tests/cli.rs b/crates/irtt-app/tests/cli.rs index 3eb2285..c648fa6 100644 --- a/crates/irtt-app/tests/cli.rs +++ b/crates/irtt-app/tests/cli.rs @@ -1,22 +1,49 @@ -#![cfg(all(feature = "client", feature = "server", feature = "tui"))] - +#[cfg(feature = "client")] #[path = "../../irtt-integration-tests/tests/support/in_tree_server.rs"] mod in_tree_server; +#[cfg(feature = "client")] use irtt_server::ServerConfig; use std::process::Command; -// The actual binaries own dispatch, argument handling, exit policy and output. -// A real authenticated server also detects broken argument-to-config mapping. +// Dispatch/help/availability belong to the application and run in every build. #[test] -fn applets_dispatch_validate_arguments_and_export_a_real_client_run() { +fn dispatcher_reports_and_runs_the_enabled_applets() { let dispatcher = env!("CARGO_BIN_EXE_irtt-rs"); - for binary in [ - dispatcher, - env!("CARGO_BIN_EXE_irtt-client"), - env!("CARGO_BIN_EXE_irtt-server"), - env!("CARGO_BIN_EXE_irtt-tui"), + let help = Command::new(dispatcher).arg("--help").output().unwrap(); + assert!(help.status.success()); + let help = String::from_utf8(help.stdout).unwrap(); + for (applet, enabled) in [ + ("client", cfg!(feature = "client")), + ("server", cfg!(feature = "server")), + ("tui", cfg!(feature = "tui")), ] { + let row = help + .lines() + .find(|line| line.trim_start().starts_with(applet)) + .unwrap(); + assert_eq!(row.contains("not available"), !enabled); + let output = Command::new(dispatcher) + .args([applet, "--help"]) + .output() + .unwrap(); + assert_eq!(output.status.success(), enabled, "{applet}"); + if !enabled { + assert!(String::from_utf8(output.stderr) + .unwrap() + .contains("not available")); + } + } + let binaries = [ + dispatcher, + #[cfg(feature = "client")] + env!("CARGO_BIN_EXE_irtt-client"), + #[cfg(feature = "server")] + env!("CARGO_BIN_EXE_irtt-server"), + #[cfg(feature = "tui")] + env!("CARGO_BIN_EXE_irtt-tui"), + ]; + for binary in binaries { assert!(Command::new(binary) .arg("--help") .output() @@ -30,6 +57,13 @@ fn applets_dispatch_validate_arguments_and_export_a_real_client_run() { .status .success()); } +} + +// A real authenticated server detects broken client argument-to-config mapping. +#[cfg(feature = "client")] +#[test] +fn client_arguments_export_a_real_client_run() { + let dispatcher = env!("CARGO_BIN_EXE_irtt-rs"); for option in ["--dscp", "--ttl"] { assert!(!Command::new(dispatcher) .args(["client", option, "256", "127.0.0.1:9"]) diff --git a/crates/irtt-client/Cargo.toml b/crates/irtt-client/Cargo.toml index 77d6bc2..c3cc153 100644 --- a/crates/irtt-client/Cargo.toml +++ b/crates/irtt-client/Cargo.toml @@ -25,6 +25,9 @@ default = [] ancillary = ["dep:nix"] tokio = ["dep:tokio"] +[dev-dependencies] +irtt-server.workspace = true + [package.metadata.docs.rs] all-features = true cargo-args = ["-Zunstable-options", "-Zrustdoc-scrape-examples"] diff --git a/crates/irtt-client/tests/linux_transport.rs b/crates/irtt-client/tests/linux_transport.rs new file mode 100644 index 0000000..bd89537 --- /dev/null +++ b/crates/irtt-client/tests/linux_transport.rs @@ -0,0 +1,235 @@ +//! Real Linux socket observations through the public client adapters. +#![cfg(all(target_os = "linux", feature = "ancillary"))] + +use std::{ + io::IoSliceMut, + net::{SocketAddr, UdpSocket}, + sync::mpsc, + thread, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use irtt_client::{Client, ClientConfig, ClientEvent, NegotiationPolicy, SocketConfig}; +use irtt_proto::{decode_request, Clock, DecodedRequestKind, ReceivedStats, StampAt}; +use irtt_server::{ServerConfig, ServerCore}; +use nix::sys::socket::{recvmsg, ControlMessageOwned, MsgFlags}; + +// A compliant packet peer uses the real server core. Its receive-side IP_TOS +// observation is independent of the client/server parameter agreement. +fn peer(dscp_allowed: bool, probes: usize) -> (SocketAddr, thread::JoinHandle>) { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + socket2::SockRef::from(&socket) + .set_recv_tos_v4(true) + .unwrap(); + let addr = socket.local_addr().unwrap(); + let task = thread::spawn(move || { + let mut core = ServerCore::new( + ServerConfig::default() + .with_min_send_interval(Duration::ZERO) + .with_dscp_allowed(dscp_allowed), + ); + let mut markings = Vec::new(); + // OPEN, every ECHO, then CLOSE. + for _ in 0..probes + 2 { + let mut bytes = [0; 2048]; + let mut control = nix::cmsg_space!(u8); + let mut iov = [IoSliceMut::new(&mut bytes)]; + let msg = recvmsg::( + std::os::fd::AsRawFd::as_raw_fd(&socket), + &mut iov, + Some(&mut control), + MsgFlags::empty(), + ) + .unwrap(); + let size = msg.bytes; + let source = msg.address.unwrap(); + let marking = msg + .cmsgs() + .unwrap() + .find_map(|message| match message { + ControlMessageOwned::Ipv4Tos(tos) => Some(tos), + _ => None, + }) + .expect("Linux must report IP_TOS"); + markings.push(marking); + let source = SocketAddr::from((source.ip(), source.port())); + if let Some(reply) = core.handle_datagram(source, &bytes[..size]).unwrap() { + socket.send_to(reply.bytes(), source).unwrap(); + } + } + assert_eq!(core.session_count(), 0); + markings + }); + (addr, task) +} + +fn config(addr: SocketAddr) -> ClientConfig { + ClientConfig { + server_addr: addr.to_string(), + dscp: 46, + negotiation_policy: NegotiationPolicy::Loose, + duration: None, + interval: Duration::from_millis(10), + clock: Clock::Both, + stamp_at: StampAt::Both, + received_stats: ReceivedStats::Both, + length: 128, + open_timeouts: vec![Duration::from_secs(1)], + socket_config: SocketConfig { + recv_timeout: Some(Duration::from_secs(1)), + ..SocketConfig::default() + }, + ..ClientConfig::default() + } +} + +// TX timestamps have no separate public field: the upstream OWD exposes the +// selected send endpoint. Reconstruct it and require a real kernel endpoint +// rather than accepting the userspace fallback. RTT must still use sent_at. +fn assert_kernel_send_endpoint(event: &ClientEvent, before_send: SystemTime) { + let ClientEvent::EchoReply { + sent_at, + received_at, + rtt, + server_timing: Some(timing), + one_way: Some(one_way), + packet_meta, + .. + } = event + else { + panic!("expected measured reply: {event:?}") + }; + let ns = |wall: SystemTime| { + i128::try_from(wall.duration_since(UNIX_EPOCH).unwrap().as_nanos()).unwrap() + }; + let selected = + i128::from(timing.receive_wall_ns.unwrap()) - one_way.client_to_server.unwrap().as_nanos(); + assert!(selected >= ns(before_send)); + assert!( + selected < ns(sent_at.wall), + "TX timestamp was not observed; userspace fallback is insufficient" + ); + assert!(selected <= ns(received_at.wall)); + assert!( + packet_meta.kernel_rx_timestamp.is_some(), + "Linux RX timestamp required" + ); + assert_eq!(rtt.raw, received_at.mono.duration_since(sent_at.mono)); +} + +#[test] +fn blocking_probes_use_kernel_tx_timing_and_negotiated_packet_marking() { + for allowed in [true, false] { + const PROBES: usize = 3; + let (addr, peer) = peer(allowed, PROBES); + let mut client = Client::connect(config(addr)).unwrap(); + client.open().unwrap(); + assert_eq!( + client.negotiated_params().unwrap().params.dscp, + if allowed { 184 } else { 0 } + ); + for _ in 0..PROBES { + let before = SystemTime::now(); + client.send_probe().unwrap(); + let events = client.recv_once().unwrap(); + assert_eq!(events.len(), 1); + assert_kernel_send_endpoint(&events[0], before); + } + client.close().unwrap(); + let markings = peer.join().unwrap(); + assert_eq!(markings[0], 0, "OPEN is unmarked"); + assert_eq!( + &markings[1..=PROBES], + vec![if allowed { 184 } else { 0 }; PROBES] + ); + assert_eq!( + markings[PROBES + 1], + 0, + "local CLOSE clears the negotiated marking" + ); + } +} + +#[cfg(feature = "tokio")] +#[test] +fn async_probes_use_kernel_tx_timing_and_negotiated_packet_marking() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + for allowed in [true, false] { + const PROBES: usize = 3; + let (addr, peer) = peer(allowed, PROBES); + runtime.block_on(async { + tokio::time::timeout(Duration::from_secs(5), async { + let mut client = irtt_client::AsyncClient::connect(config(addr)) + .await + .unwrap(); + client.open().await.unwrap(); + assert_eq!( + client.negotiated_params().unwrap().params.dscp, + if allowed { 184 } else { 0 } + ); + for _ in 0..PROBES { + let before = SystemTime::now(); + client.send_probe().await.unwrap(); + let events = client.recv().await.unwrap(); + assert_eq!(events.len(), 1); + assert_kernel_send_endpoint(&events[0], before); + } + client.close().await.unwrap(); + }) + .await + .expect("async transport observations timed out"); + }); + let markings = peer.join().unwrap(); + assert_eq!(markings[0], 0); + assert_eq!( + &markings[1..=PROBES], + vec![if allowed { 184 } else { 0 }; PROBES] + ); + assert_eq!(markings[PROBES + 1], 0); + } +} + +// Real ICMP refusal remains a socket error, never optional timestamp metadata. +// The bound peer closes only after the client has accepted a valid OPEN. +#[test] +fn a_closed_udp_peer_surfaces_a_socket_error() { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + let addr = socket.local_addr().unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + let (close_tx, close_rx) = mpsc::channel(); + let task = thread::spawn(move || { + let mut core = ServerCore::new(ServerConfig::default()); + let mut bytes = [0; 2048]; + let (size, source) = socket.recv_from(&mut bytes).unwrap(); + assert!(matches!( + decode_request(&bytes[..size]).unwrap().kind, + DecodedRequestKind::Open { .. } + )); + let reply = core + .handle_datagram(source, &bytes[..size]) + .unwrap() + .unwrap(); + socket.send_to(reply.bytes(), source).unwrap(); + close_rx.recv_timeout(Duration::from_secs(3)).unwrap(); + drop(socket); + }); + let mut client = Client::connect(config(addr)).unwrap(); + client.open().unwrap(); + close_tx.send(()).unwrap(); + task.join().unwrap(); + // Refusal can arrive during send's TX drain or the following receive. + let result = client.send_probe().and_then(|_| client.recv_once()); + assert!( + matches!(result, Err(irtt_client::ClientError::Socket(ref error)) + if error.kind() == std::io::ErrorKind::ConnectionRefused), + "{result:?}" + ); +} diff --git a/crates/irtt-client/tests/managed_control.rs b/crates/irtt-client/tests/managed_control.rs new file mode 100644 index 0000000..1e7b80d --- /dev/null +++ b/crates/irtt-client/tests/managed_control.rs @@ -0,0 +1,160 @@ +#![cfg(feature = "tokio")] + +use std::{ + future::Future, + pin::pin, + task::{Context, Poll, Waker}, + time::Duration, +}; + +use irtt_client::managed::*; + +fn config() -> ManagedClientConfig { + ManagedClientConfig { + completion: ManagedCompletionPolicy::ExplicitStop, + command_capacity: 3, + ..ManagedClientConfig::default() + } +} + +#[test] +fn stop_resolves_every_accepted_update_and_closes_admission() { + let (task, handle) = ManagedClient::task(config(), vec![]).unwrap(); + let receipts: Vec<_> = (0..3) + .map(|index| { + handle + .update_targets(vec![ManagedTargetConfig::new( + format!("queued-{index}"), + "127.0.0.1:9", + )]) + .unwrap() + }) + .collect(); + assert!(matches!( + handle.update_targets(vec![]), + Err(ManagedCommandError::QueueFull) + )); + + let stop = handle.stop(); + assert!(matches!( + handle.update_targets(vec![]), + Err(ManagedCommandError::Stopping) + )); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + tokio::time::timeout(Duration::from_secs(2), async { + let outcome = task.await; + stop.await; + for receipt in receipts { + assert!(matches!( + receipt.await, + Err(ManagedCommandApplyError::Stopping) + )); + } + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); + assert_eq!(outcome.applied_command_sequence, 0); + assert_eq!(outcome.total_target_outcomes, 0); + let status = handle.status(); + assert_eq!(status.lifecycle, ManagedLifecycle::Completed); + assert!(status.stop_requested); + assert_eq!(status.final_outcome.as_deref(), Some(&outcome)); + assert!(matches!( + handle.update_targets(vec![]), + Err(ManagedCommandError::DriverClosed) + )); + handle.stop().await; + }) + .await + .expect("stop must resolve every buffered update and repeated stop receipt"); + }); +} + +#[test] +fn failed_driver_resolves_every_accepted_update_with_its_terminal_failure() { + let (task, handle) = ManagedClient::task(config(), vec![]).unwrap(); + let receipts: Vec<_> = (0..3) + .map(|index| { + handle + .update_targets(vec![ManagedTargetConfig::new( + format!("queued-{index}"), + "127.0.0.1:9", + )]) + .unwrap() + }) + .collect(); + let mut context = Context::from_waker(Waker::noop()); + let Poll::Ready(outcome) = pin!(task).as_mut().poll(&mut context) else { + panic!("polling without Tokio must fail immediately"); + }; + assert_eq!( + outcome.end_reason, + ManagedEndReason::DriverFailed(ManagedDriverFailure::NoTokioRuntime) + ); + for receipt in receipts { + assert!(matches!( + pin!(receipt).as_mut().poll(&mut context), + Poll::Ready(Err(ManagedCommandApplyError::DriverFailed( + ManagedDriverFailure::NoTokioRuntime + ))) + )); + } + let status = handle.status(); + assert_eq!(status.lifecycle, ManagedLifecycle::Failed); + assert_eq!(status.final_outcome.as_deref(), Some(&outcome)); + assert!(matches!( + handle.update_targets(vec![]), + Err(ManagedCommandError::DriverClosed) + )); + assert!(pin!(handle.stop()).as_mut().poll(&mut context).is_ready()); +} + +#[test] +fn concurrent_submission_and_stop_or_seal_never_lose_an_accepted_receipt() { + let (task, handle) = ManagedClient::task(config(), vec![]).unwrap(); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(2)); + let submit_barrier = std::sync::Arc::clone(&barrier); + let submit_handle = handle.clone(); + let (sender, receiver) = std::sync::mpsc::channel(); + let submit = std::thread::spawn(move || { + submit_barrier.wait(); + sender.send(submit_handle.update_targets(vec![])).unwrap(); + }); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + // Stop is latched before the driver starts, while submission may cross + // admission before stop, after stop, or after terminal sealing. + barrier.wait(); + let stop = handle.stop(); + runtime.block_on(async { + tokio::time::timeout(Duration::from_secs(2), async { + let outcome = task.await; + stop.await; + assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); + assert_eq!(outcome.applied_command_sequence, 0); + assert_eq!(outcome.total_target_outcomes, 0); + let result = receiver + .recv_timeout(Duration::from_secs(2)) + .expect("concurrent submission must return"); + submit.join().unwrap(); + match result { + Ok(receipt) => assert!(matches!( + receipt.await, + Err(ManagedCommandApplyError::Stopping) + )), + Err(ManagedCommandError::Stopping | ManagedCommandError::DriverClosed) => {} + Err(error) => panic!("unexpected concurrent submission result: {error:?}"), + } + assert!(matches!( + handle.update_targets(vec![]), + Err(ManagedCommandError::DriverClosed) + )); + }) + .await + .expect("every accepted concurrent submission must receive a terminal disposition"); + }); +} diff --git a/crates/irtt-integration-tests/tests/async_client_lifecycle.rs b/crates/irtt-integration-tests/tests/async_client_lifecycle.rs index e1baf91..8349de9 100644 --- a/crates/irtt-integration-tests/tests/async_client_lifecycle.rs +++ b/crates/irtt-integration-tests/tests/async_client_lifecycle.rs @@ -381,7 +381,7 @@ fn recv_after_local_close_and_no_test_fails_on_first_poll() { } #[test] -fn authenticated_async_lifecycle_applies_negotiated_dscp() { +fn authenticated_async_lifecycle_negotiates_dscp() { let key = b"async-lifecycle-key".to_vec(); let server = InTreeServer::start(ServerConfig::default().with_hmac_key(key.clone())); runtime().block_on(async { diff --git a/crates/irtt-integration-tests/tests/managed_lifecycle.rs b/crates/irtt-integration-tests/tests/managed_lifecycle.rs index a9a15ad..6b8d054 100644 --- a/crates/irtt-integration-tests/tests/managed_lifecycle.rs +++ b/crates/irtt-integration-tests/tests/managed_lifecycle.rs @@ -250,53 +250,6 @@ fn explicit_empty_waits_for_stop() { assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); } -#[test] -fn accepted_update_before_stop_resolves_stopping() { - let mut managed = config(ManagedPacing::Staggered); - managed.completion = ManagedCompletionPolicy::ExplicitStop; - let (task, handle) = ManagedClient::task(managed, vec![]).unwrap(); - let receipt = handle - .update_targets(vec![ManagedTargetConfig::new("queued", "127.0.0.1:9")]) - .unwrap(); - let stop = handle.stop(); - let outcome = runtime().block_on(task); - runtime().block_on(stop); - assert_eq!(outcome.applied_command_sequence, 0); - assert!(matches!( - runtime().block_on(receipt), - Err(ManagedCommandApplyError::Stopping) - )); -} - -#[test] -fn accepted_update_before_driver_failure_resolves_driver_failed() { - let mut managed = config(ManagedPacing::Staggered); - managed.completion = ManagedCompletionPolicy::ExplicitStop; - let (task, handle) = ManagedClient::task(managed, vec![]).unwrap(); - let receipt = handle - .update_targets(vec![ManagedTargetConfig::new("queued", "127.0.0.1:9")]) - .unwrap(); - let mut task = Box::pin(task); - let Poll::Ready(outcome) = poll_once(task.as_mut()) else { - panic!("first poll outside Tokio did not fail") - }; - assert_eq!( - outcome.end_reason, - ManagedEndReason::DriverFailed(ManagedDriverFailure::NoTokioRuntime) - ); - let mut receipt = Box::pin(receipt); - assert!(matches!( - poll_once(Pin::as_mut(&mut receipt)), - Poll::Ready(Err(ManagedCommandApplyError::DriverFailed( - ManagedDriverFailure::NoTokioRuntime - ))) - )); - assert!(matches!( - handle.update_targets(vec![]), - Err(ManagedCommandError::DriverClosed) - )); -} - #[test] fn extreme_event_capacity_is_rejected_without_panic() { let mut managed = config(ManagedPacing::Staggered); diff --git a/crates/irtt-server/AGENTS.md b/crates/irtt-server/AGENTS.md index d3804b3..ce0f8e4 100644 --- a/crates/irtt-server/AGENTS.md +++ b/crates/irtt-server/AGENTS.md @@ -333,9 +333,9 @@ effective parameters must be safe for the server to run. - The transition is pure and is committed only after the reply has encoded, so an internal encoding failure cannot leave a session claiming to have answered a request it never did. -- Clock sampling is a private injected seam, like token generation, not a - runtime abstraction: the receive instant is taken as soon as a datagram is - classified as an echo and the send instant just before the reply is built, so +- Clock sampling uses the system clock directly, not a runtime abstraction: the + receive instant is taken as soon as a datagram is classified as an echo and + the send instant just before the reply is built, so they bracket the server's own handling. The monotonic origin belongs to the clock source and is stable for its life. - A reply's receive instant is held back to its send instant where the wall @@ -601,16 +601,22 @@ half-created session. maintain shadow encoders for well-formed wire data. - Hand-build packets only where the point is a payload a compliant encoder cannot produce — truncated varints, out-of-range enums, corrupted MACs. -- Token generation and clock sampling are the only nondeterministic parts; tests - inject a scripted source for each, so identity, collisions, allocation failure - and timestamp values are all assertable. Keep both seams private, and keep - timestamp, rate and lifetime tests free of sleeps, tolerances and real - wall-clock assertions. -- Two clock fakes, for two different questions. A scripted clock returns a fixed - list of samples and is for tests that assert individual timestamp *readings*; a - hand-moved clock stands still until the test moves it, and is for rate and - lifetime tests, which care when a datagram arrived and not how many times the - core read the clock on the way. Prefer the latter for anything about deadlines. +- Prefer the public `ServerCore` packet interface and real runtime sockets for + durable behavior. Use production encoders and decoders, actual OS-generated + tokens, and observable replies/session capacity rather than private state. +- Observe nonzero rate refill and lifetime deadlines with real time when an + oversleep cannot invalidate the assertion. Bound observation loops and report + useful failures. Do not restore fake clocks, random-source injection, race + gates, synthetic ancillary classifiers or private observability merely to + reproduce the deleted test structure. Focused fault injection remains allowed + under the root policy only when no reasonable production-interface alternative + can protect an important invariant. +- Random fill's OS-source failure cannot be forced through the packet interface. + Its zero-fill fallback remains an explicitly unexercised error path; ordinary + random-fill wire tests must not claim to cover it or add an injection seam + merely for coverage. Existing token-allocation fault tests protect bounded + allocation and transactional failure, which cannot be driven reliably with + actual random draws. - Assert rate and lifetime behavior from replies: which datagrams came back, the count and window they reported, and `session_count`. The token bucket is a black-box inference about the reference server, not a protocol requirement, so diff --git a/crates/irtt-server/tests/protocol_contracts.rs b/crates/irtt-server/tests/protocol_contracts.rs new file mode 100644 index 0000000..99c35da --- /dev/null +++ b/crates/irtt-server/tests/protocol_contracts.rs @@ -0,0 +1,318 @@ +//! Durable server contracts observed through encoded requests and public replies. +use std::{ + net::SocketAddr, + thread, + time::{Duration, Instant}, +}; + +use irtt_proto::{ + decode_echo_reply, decode_open_reply, encode_request, Clock, EchoReply, OpenReply, Params, + ReceivedStats, RequestToEncode, StampAt, FLAG_CLOSE, +}; +use irtt_server::{ServerConfig, ServerCore, TimestampAllowance}; + +fn peer() -> SocketAddr { + "127.0.0.1:40000".parse().unwrap() +} +fn open_packet(params: &Params, no_test: bool, key: Option<&[u8]>) -> Vec { + encode_request(RequestToEncode::Open { params, no_test }, key).unwrap() +} +fn open(server: &mut ServerCore, params: &Params) -> OpenReply { + let reply = server + .handle_datagram(peer(), &open_packet(params, false, None)) + .unwrap() + .unwrap(); + decode_open_reply(reply.bytes(), None).unwrap() +} +fn echo(server: &mut ServerCore, opened: &OpenReply, sequence: u32) -> Option { + let request = encode_request( + RequestToEncode::Echo { + token: opened.token, + sequence, + params: &opened.params, + payload: &[], + }, + None, + ) + .unwrap(); + server + .handle_datagram(peer(), &request) + .unwrap() + .map(|reply| decode_echo_reply(reply.bytes(), &opened.params, None).unwrap()) +} +fn unthrottled() -> ServerConfig { + ServerConfig::default().with_min_send_interval(Duration::ZERO) +} + +#[test] +fn no_test_authenticates_negotiates_and_uses_no_session_capacity() { + let key = b"no-test-contract"; + let mut server = ServerCore::new( + ServerConfig::default() + .with_hmac_key(key) + .with_max_sessions(0) + .with_max_packet_length(96) + .with_min_send_interval(Duration::from_millis(50)) + .with_timestamp_allowance(TimestampAllowance::Single) + .with_dscp_allowed(false), + ); + let params = Params { + protocol_version: 99, + length: 128, + dscp: 185, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Params::default() + }; + let request = open_packet(¶ms, true, Some(key)); + assert!(server + .handle_datagram(peer(), &open_packet(¶ms, true, None)) + .unwrap() + .is_none()); + let mut bad_mac = request.clone(); + bad_mac[4] ^= 1; + assert!(server.handle_datagram(peer(), &bad_mac).unwrap().is_none()); + let reply = server.handle_datagram(peer(), &request).unwrap().unwrap(); + assert_eq!(reply.traffic_class(), 0); + let negotiated = decode_open_reply(reply.bytes(), Some(key)).unwrap(); + assert_eq!(negotiated.token, 0); + assert_eq!(negotiated.params.protocol_version, 1); + assert_ne!(negotiated.flags & FLAG_CLOSE, 0); + assert_eq!(negotiated.params.length, 96); + assert_eq!(negotiated.params.interval_ns, 50_000_000); + assert_eq!(negotiated.params.stamp_at, StampAt::Midpoint); + assert_eq!(negotiated.params.clock, Clock::Both); + assert_eq!(negotiated.params.dscp, 0); + assert_eq!(server.session_count(), 0); + assert!(server + .handle_datagram(peer(), &open_packet(¶ms, false, Some(key))) + .unwrap() + .is_none()); +} + +#[test] +fn no_test_refuses_unexecutable_layout_but_validates_after_restriction() { + let params = Params { + stamp_at: StampAt::Both, + ..Params::with_protocol_defaults() + }; + let mut server = ServerCore::new(unthrottled()); + assert!(server + .handle_datagram(peer(), &open_packet(¶ms, true, None)) + .unwrap() + .is_none()); + let mut restricted = + ServerCore::new(unthrottled().with_timestamp_allowance(TimestampAllowance::None)); + let reply = restricted + .handle_datagram(peer(), &open_packet(¶ms, true, None)) + .unwrap() + .unwrap(); + assert_eq!( + decode_open_reply(reply.bytes(), None) + .unwrap() + .params + .stamp_at, + StampAt::None + ); + let rich = Params { + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Params::with_protocol_defaults() + }; + let mut too_small = ServerCore::new(unthrottled().with_max_packet_length(32)); + assert!(too_small + .handle_datagram(peer(), &open_packet(&rich, true, None)) + .unwrap() + .is_none()); + assert_eq!(too_small.session_count(), 0); + let invalid = Params { + interval_ns: -1, + ..Params::with_protocol_defaults() + }; + assert!(server + .handle_datagram(peer(), &open_packet(&invalid, true, None)) + .unwrap() + .is_none()); + assert_eq!(server.session_count(), 0); +} + +#[test] +fn receive_statistics_follow_verified_gap_duplicate_reordering_and_wrap_vectors() { + // SERVER_BEHAVIORAL_VECTORS.md Sections 1.2, 1.5, 1.6, 1.8 and 1.10. + let vectors: &[&[(u32, u64)]] = &[ + &[(0, 1), (1, 3), (3, 13), (4, 27)], + &[(0, 1), (1, 3), (1, 3), (2, 7)], + &[(0, 1), (1, 3), (2, 7), (1, 1), (3, 5)], + &[(0, 1), (63, 0x8000_0000_0000_0001)], + &[(0, 1), (64, 1)], + &[ + (u32::MAX - 2, 1), + (u32::MAX - 1, 3), + (u32::MAX, 7), + (0, 15), + (1, 31), + ], + ]; + for vector in vectors { + let mut server = ServerCore::new(unthrottled()); + let opened = open( + &mut server, + &Params { + received_stats: ReceivedStats::Both, + ..Params::with_protocol_defaults() + }, + ); + for (index, &(sequence, window)) in vector.iter().enumerate() { + let reply = echo(&mut server, &opened, sequence).unwrap(); + assert_eq!(reply.recv_count, Some(index as u32 + 1)); + assert_eq!( + reply.recv_window, + Some(window), + "sequence {sequence} in {vector:?}" + ); + } + } +} + +#[test] +fn a_spent_nonzero_burst_refills_without_counting_dropped_echoes() { + let refill = Duration::from_millis(100); + let mut server = ServerCore::new( + ServerConfig::default() + .with_min_send_interval(refill) + .with_burst_allowance(1) + .with_idle_timeout(Duration::MAX), + ); + let opened = open( + &mut server, + &Params { + received_stats: ReceivedStats::Both, + ..Params::with_protocol_defaults() + }, + ); + let mut count = echo(&mut server, &opened, 0).unwrap().recv_count.unwrap(); + // Observe actual refusal rather than assuming two calls fit a timing window. + let deadline = Instant::now() + Duration::from_secs(2); + while let Some(reply) = echo(&mut server, &opened, 0) { + count = reply.recv_count.unwrap(); + assert!( + Instant::now() < deadline, + "could not observe a spent burst within two seconds" + ); + } + thread::sleep(refill); // Oversleep only replenishes more; idle expiry is disabled in practice. + let reply = echo(&mut server, &opened, 1).unwrap(); + assert_eq!(reply.recv_count, Some(count + 1)); + assert_eq!(reply.recv_window, Some(3)); +} + +#[test] +fn a_nonzero_idle_deadline_reclaims_an_unused_session_before_another_open() { + let idle = Duration::from_millis(10); + let mut server = ServerCore::new(unthrottled().with_max_sessions(1).with_idle_timeout(idle)); + let params = Params::with_protocol_defaults(); + let old = open(&mut server, ¶ms); + thread::sleep(idle); // Any oversleep still requires silent expiry and capacity recovery. + assert!(echo(&mut server, &old, 0).is_none()); + assert_eq!(server.session_count(), 0); + open(&mut server, ¶ms); + assert_eq!(server.session_count(), 1); +} + +#[test] +fn maximum_duration_starts_at_the_first_served_echo_and_closes_with_its_reply() { + let maximum = Duration::from_millis(1); + let mut server = ServerCore::new( + unthrottled() + .with_idle_timeout(Duration::MAX) + .with_max_test_duration(maximum), + ); + let opened = open( + &mut server, + &Params { + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + dscp: 185, + ..Params::with_protocol_defaults() + }, + ); + assert_eq!(opened.params.duration_ns, 1_000_000); + let elapsed_deadline = maximum + Duration::from_secs(2); + thread::sleep(elapsed_deadline); + let first = echo(&mut server, &opened, 0).unwrap(); + assert_eq!( + first.flags & FLAG_CLOSE, + 0, + "OPEN must not start the maximum-duration clock" + ); + thread::sleep(elapsed_deadline); // Oversleep still requires the next served echo to close. + let request = encode_request( + RequestToEncode::Echo { + token: opened.token, + sequence: 1, + params: &opened.params, + payload: &[], + }, + None, + ) + .unwrap(); + let reply = server.handle_datagram(peer(), &request).unwrap().unwrap(); + assert_eq!(reply.traffic_class(), 185); + let closed = decode_echo_reply(reply.bytes(), &opened.params, None).unwrap(); + assert_ne!(closed.flags & FLAG_CLOSE, 0); + assert_eq!(closed.sequence, 1); + assert_eq!(closed.recv_count, Some(2)); + assert_eq!(closed.recv_window, Some(3)); + assert!(closed.timestamps.recv_wall.is_some()); + assert!(closed.timestamps.send_mono.is_some()); + assert_eq!(server.session_count(), 0); + assert!(echo(&mut server, &opened, 2).is_none()); +} + +#[test] +fn negotiated_timestamp_modes_emit_only_the_selected_fields_in_order() { + for clock in [Clock::Wall, Clock::Monotonic, Clock::Both] { + for placement in [ + StampAt::None, + StampAt::Receive, + StampAt::Send, + StampAt::Both, + StampAt::Midpoint, + ] { + let mut server = ServerCore::new(unthrottled()); + let opened = open( + &mut server, + &Params { + stamp_at: placement, + clock, + ..Params::with_protocol_defaults() + }, + ); + let t = echo(&mut server, &opened, 0).unwrap().timestamps; + let wall = clock != Clock::Monotonic; + let mono = clock != Clock::Wall; + let recv = matches!(placement, StampAt::Receive | StampAt::Both); + let send = matches!(placement, StampAt::Send | StampAt::Both); + let midpoint = placement == StampAt::Midpoint; + assert_eq!(t.recv_wall.is_some(), wall && recv); + assert_eq!(t.recv_mono.is_some(), mono && recv); + assert_eq!(t.send_wall.is_some(), wall && send); + assert_eq!(t.send_mono.is_some(), mono && send); + assert_eq!(t.midpoint_wall.is_some(), wall && midpoint); + assert_eq!(t.midpoint_mono.is_some(), mono && midpoint); + if let (Some(recv), Some(send)) = (t.recv_wall, t.send_wall) { + assert!(recv <= send); + } + if let (Some(recv), Some(send)) = (t.recv_mono, t.send_mono) { + assert!(0 <= recv && recv <= send); + let next = echo(&mut server, &opened, 1).unwrap().timestamps; + assert!( + next.recv_mono.unwrap() >= send, + "monotonic origin must remain stable" + ); + } + } + } +} diff --git a/crates/irtt-server/tests/runtime.rs b/crates/irtt-server/tests/runtime.rs index a19270d..a91216c 100644 --- a/crates/irtt-server/tests/runtime.rs +++ b/crates/irtt-server/tests/runtime.rs @@ -145,3 +145,13 @@ async fn exercise_bound_server(mut server: Server) { fn unthrottled() -> ServerConfig { ServerConfig::default().with_min_send_interval(Duration::ZERO) } + +#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "freebsd")))] +#[tokio::test(flavor = "current_thread")] +async fn wildcard_construction_is_refused_without_reply_source_selection() { + let addr = "0.0.0.0:0".parse().unwrap(); + assert!(matches!( + Server::bind(addr, unthrottled()).await, + Err(ServerRuntimeError::WildcardSourceSelectionUnsupported { .. }) + )); +} diff --git a/crates/irtt-server/tests/traffic_class.rs b/crates/irtt-server/tests/traffic_class.rs new file mode 100644 index 0000000..ce26d67 --- /dev/null +++ b/crates/irtt-server/tests/traffic_class.rs @@ -0,0 +1,133 @@ +//! Observe the emitted IP byte, independently of protocol negotiation. +#![cfg(target_os = "linux")] + +use std::{io::IoSliceMut, net::UdpSocket as StdSocket, os::fd::AsRawFd, time::Duration}; + +use irtt_proto::{ + decode_echo_reply, decode_open_reply, encode_request, OpenReply, Params, RequestToEncode, + FLAG_CLOSE, +}; +use irtt_server::{Server, ServerConfig}; +use nix::sys::socket::{recvmsg, ControlMessageOwned, MsgFlags}; +use tokio::{net::UdpSocket, sync::oneshot, time::timeout}; + +async fn exchange(client: &UdpSocket, request: &[u8], expected: u8) -> Vec { + client.send(request).await.unwrap(); + loop { + client.readable().await.unwrap(); + let packet = client.try_io(tokio::io::Interest::READABLE, || { + let mut bytes = [0; 2048]; + let mut control = nix::cmsg_space!(u8); + let mut iov = [IoSliceMut::new(&mut bytes)]; + let msg = recvmsg::<()>( + client.as_raw_fd(), + &mut iov, + Some(&mut control), + MsgFlags::MSG_DONTWAIT, + )?; + let size = msg.bytes; + let class = msg.cmsgs().unwrap().find_map(|message| match message { + ControlMessageOwned::Ipv4Tos(class) => Some(class), + _ => None, + }); + assert_eq!(class, Some(expected)); + Ok(bytes[..size].to_vec()) + }); + match packet { + Ok(packet) => return packet, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {} + Err(error) => panic!("receive failed: {error}"), + } + } +} + +fn echo(opened: &OpenReply, sequence: u32) -> Vec { + encode_request( + RequestToEncode::Echo { + token: opened.token, + sequence, + params: &opened.params, + payload: &[], + }, + None, + ) + .unwrap() +} + +#[tokio::test(flavor = "current_thread")] +async fn every_open_echo_and_server_close_applies_its_own_raw_traffic_class() { + timeout(Duration::from_secs(8), async { + let maximum = Duration::from_millis(1); + let mut server = Server::bind( + "127.0.0.1:0".parse().unwrap(), + ServerConfig::default() + .with_min_send_interval(Duration::ZERO) + .with_idle_timeout(Duration::MAX) + .with_max_test_duration(maximum), + ) + .await + .unwrap(); + let addr = server.local_addr().unwrap(); + let (shutdown, stopped) = oneshot::channel(); + let task = tokio::spawn(async move { + server + .run(async { + let _ = stopped.await; + }) + .await + }); + let socket = StdSocket::bind("127.0.0.1:0").unwrap(); + socket2::SockRef::from(&socket) + .set_recv_tos_v4(true) + .unwrap(); + socket.set_nonblocking(true).unwrap(); + let client = UdpSocket::from_std(socket).unwrap(); + client.connect(addr).await.unwrap(); + let mut sessions = Vec::new(); + // Each new OPEN follows a prior ECHO. The first raw byte also includes + // ECN bits: neither a codepoint shift nor masking would preserve it. + for (requested, class) in [(185, 185), (32, 32), (0, 0), (256, 0), (-1, 0)] { + let params = Params { + dscp: requested, + ..Params::with_protocol_defaults() + }; + let request = encode_request( + RequestToEncode::Open { + params: ¶ms, + no_test: false, + }, + None, + ) + .unwrap(); + let packet = exchange(&client, &request, 0).await; + let opened = decode_open_reply(&packet, None).unwrap(); + assert_eq!(opened.params.dscp, requested); + let packet = exchange(&client, &echo(&opened, 0), class).await; + assert_eq!( + decode_echo_reply(&packet, &opened.params, None) + .unwrap() + .flags + & FLAG_CLOSE, + 0 + ); + sessions.push((opened, class)); + } + // Every session has now served its first echo. Waiting past all duration + // deadlines is safe under oversleep; idle expiry cannot intervene. + tokio::time::sleep(maximum + Duration::from_secs(2)).await; + for (opened, class) in sessions { + let packet = exchange(&client, &echo(&opened, 1), class).await; + assert_ne!( + decode_echo_reply(&packet, &opened.params, None) + .unwrap() + .flags + & FLAG_CLOSE, + 0 + ); + } + shutdown.send(()).unwrap(); + task.await.unwrap().unwrap(); + }) + .await + .expect("marked runtime exchanges did not complete"); +} diff --git a/crates/irtt-stats/src/ipdv.rs b/crates/irtt-stats/src/ipdv.rs index 81c2bfb..606e465 100644 --- a/crates/irtt-stats/src/ipdv.rs +++ b/crates/irtt-stats/src/ipdv.rs @@ -199,10 +199,23 @@ mod tests { ) { let mut tracker = IpdvTracker::new(Some(limit)); for seq in seqs { - tracker.insert(ipdv_sample(seq, i128::from(seq))); + let completed_before = tracker.completed_pairs.clone(); + let pairs = tracker.insert(ipdv_sample(seq, i128::from(seq))); + let emitted: HashSet<_> = pairs.iter().map(|pair| pair.current_seq).collect(); + prop_assert_eq!(emitted.len(), pairs.len()); + prop_assert!(emitted.is_disjoint(&completed_before), "a retained pair must not be emitted twice"); prop_assert!(tracker.samples.len() <= limit); prop_assert!(tracker.sample_order.len() <= limit); prop_assert!(tracker.completed_pairs.len() <= limit); + // Eviction must remove stale pair markers as well as samples; + // otherwise retained state can prevent future adjacent pairs. + prop_assert_eq!(tracker.sample_order.len(), tracker.samples.len()); + prop_assert_eq!(tracker.sample_order.iter().collect::>().len(), tracker.samples.len()); + prop_assert!(tracker.sample_order.iter().all(|seq| tracker.samples.contains_key(seq))); + prop_assert!(tracker.completed_pairs.iter().all(|seq| { + tracker.samples.contains_key(seq) + && tracker.samples.contains_key(&seq.wrapping_sub(1)) + }), "completed pairs must retain both sequence endpoints"); } } diff --git a/crates/irtt-stats/tests/ipdv.rs b/crates/irtt-stats/tests/ipdv.rs new file mode 100644 index 0000000..2480e1a --- /dev/null +++ b/crates/irtt-stats/tests/ipdv.rs @@ -0,0 +1,112 @@ +use std::time::{Duration, Instant, UNIX_EPOCH}; + +use irtt_client::{ + ClientEvent, ClientTimestamp, PacketMeta, RttSample, ServerTiming, SignedDuration, +}; +use irtt_stats::{StatsCollector, StatsConfig}; + +fn reply( + base: Instant, + seq: u32, + send_ms: u64, + receive_ms: u64, + timing: ServerTiming, +) -> ClientEvent { + let raw = Duration::from_millis(receive_ms - send_ms); + ClientEvent::EchoReply { + seq, + remote: "127.0.0.1:2112".parse().unwrap(), + sent_at: ClientTimestamp { + mono: base + Duration::from_millis(send_ms), + wall: UNIX_EPOCH + Duration::from_millis(200 + u64::from(seq) * 10), + }, + received_at: ClientTimestamp { + mono: base + Duration::from_millis(receive_ms), + wall: UNIX_EPOCH + Duration::from_millis(240 + u64::from(seq) * 20), + }, + rtt: RttSample { + raw, + adjusted: None, + effective: SignedDuration::from_duration(raw), + }, + server_timing: Some(timing), + one_way: None, + received_stats: None, + bytes: 64, + packet_meta: PacketMeta::default(), + } +} + +#[test] +fn directional_ipdv_uses_matching_clock_domains_and_preserves_zero_and_absence() { + let first = ServerTiming { + receive_mono_ns: Some(100_000_000), + send_mono_ns: Some(101_000_000), + receive_wall_ns: Some(1_000_000_000), + send_wall_ns: Some(1_001_000_000), + midpoint_mono_ns: None, + midpoint_wall_ns: None, + processing: None, + }; + let second = ServerTiming { + receive_mono_ns: Some(105_000_000), + send_mono_ns: Some(108_000_000), + receive_wall_ns: Some(1_020_000_000), + send_wall_ns: Some(1_026_000_000), + ..first + }; + // Wall and monotonic deltas deliberately disagree: availability alone + // cannot prove the formulas or which clock domain wins. + let mut wall_only = second; + wall_only.receive_mono_ns = None; + wall_only.send_mono_ns = None; + let mut no_send = wall_only; + no_send.receive_wall_ns = None; + let mut no_receive = wall_only; + no_receive.send_wall_ns = None; + let zero = ServerTiming { + receive_mono_ns: Some(110_000_000), + send_mono_ns: Some(111_000_000), + ..second + }; + for (label, current, send_ms, expected_send_ms, expected_receive_ms) in [ + ("monotonic preferred", second, 30, Some(5), Some(3)), + ("wall fallback", wall_only, 30, Some(10), Some(5)), + ("missing send endpoint", no_send, 30, None, Some(5)), + ("missing receive endpoint", no_receive, 30, Some(10), None), + ("reordered client send times", second, 10, Some(15), Some(3)), + ("zero variation", zero, 30, Some(0), Some(0)), + ] { + let base = Instant::now(); + let mut collector = StatsCollector::new(StatsConfig::finite()); + assert!(collector + .process(&reply(base, 0, 20, 40, first)) + .ipdv_pairs + .is_empty()); + let update = collector.process(&reply(base, 1, send_ms, 50, current)); + assert_eq!(update.ipdv_pairs.len(), 1, "{label}"); + let pair = &update.ipdv_pairs[0]; + assert_eq!( + pair.send_ipdv, + expected_send_ms.map(Duration::from_millis), + "{label}" + ); + assert_eq!( + pair.receive_ipdv, + expected_receive_ms.map(Duration::from_millis), + "{label}" + ); + let snapshot = collector.snapshot(); + for (expected, actual) in [ + (expected_send_ms, snapshot.ipdv.send), + (expected_receive_ms, snapshot.ipdv.receive), + ] { + assert_eq!(actual.count, u64::from(expected.is_some()), "{label}"); + assert_eq!( + actual.total_ns, + i128::from(expected.unwrap_or(0)) * 1_000_000, + "{label}" + ); + } + } +} diff --git a/crates/irtt-stats/tests/stats.rs b/crates/irtt-stats/tests/stats.rs index 4818210..6a2d1e5 100644 --- a/crates/irtt-stats/tests/stats.rs +++ b/crates/irtt-stats/tests/stats.rs @@ -319,6 +319,19 @@ fn ipdv_wraparound_sequence_is_adjacent() { assert!(wrapped.contributed_sample); assert_one_ipdv_pair(&wrapped, u32::MAX, 0, Duration::from_millis(4)); + // The client classifies duplicate datagrams before publishing events; + // their accounting must not emit an adjacent pair a second time. + for seq in [u32::MAX, 0] { + let duplicate = collector.process(&ClientEvent::DuplicateReply { + seq, + remote: "127.0.0.1:2112".parse().unwrap(), + received_at: ts(30), + bytes: 64, + }); + assert!(!duplicate.contributed_sample); + assert_no_ipdv_pairs(&duplicate); + } + let snapshot = collector.snapshot(); assert_eq!(snapshot.ipdv.round_trip.count, 1); assert_eq!(snapshot.ipdv.round_trip.total_ns, 4_000_000); @@ -355,19 +368,6 @@ fn late_reply_can_complete_ipdv_pair() { assert_eq!(snapshot.ipdv.round_trip.total_ns, 10_000_000); } -#[test] -fn update_exposes_directional_ipdv_when_available() { - let mut collector = StatsCollector::new(StatsConfig::finite()); - collector.process(&unadjusted_reply(0, 10)); - let update = collector.process(&unadjusted_reply(1, 13)); - - assert!(update.contributed_sample); - - let pair = assert_one_ipdv_pair(&update, 0, 1, Duration::from_millis(3)); - assert!(pair.send_ipdv.is_some()); - assert!(pair.receive_ipdv.is_some()); -} - #[test] fn gap_fill_update_exposes_both_completed_ipdv_pairs() { let mut collector = StatsCollector::new(StatsConfig::finite());