diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index a007269..bc45e7e 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -612,6 +612,8 @@ enum OpenSessionFailureCleanup { /// The sole authoritative zero-or-many target driver. #[must_use = "ManagedClientTask must be awaited or deliberately dropped"] pub struct ManagedClientTask { + #[cfg(test)] + timeout_inspections: std::cell::RefCell>, state: DriverState, lifecycle: ManagedLifecycle, config: ManagedClientConfig, @@ -1833,6 +1835,8 @@ impl ManagedClientTask { } fn target_has_due_timeout(&self, index: usize, now: Instant) -> bool { + #[cfg(test)] + self.timeout_inspections.borrow_mut().push(index); match &self.targets[index].state { TargetState::Active { client } => { self.state != DriverState::Stopping @@ -2318,6 +2322,8 @@ fn build_task( stop: Arc::clone(&stop), }; let task = ManagedClientTask { + #[cfg(test)] + timeout_inspections: std::cell::RefCell::new(Vec::new()), state: DriverState::NotStarted, lifecycle: ManagedLifecycle::NotStarted, config, @@ -2387,3 +2393,41 @@ fn close_timeout_failure() -> ManagedTargetFailure { message: Arc::from("best-effort close did not become writable before its deadline"), } } + +#[cfg(test)] +mod tests { + use super::*; + + // Poll latency through public events cannot distinguish an O(n) discovery + // scan from a bounded scan reliably. This tiny inspection trace measures + // work directly, including empty targets, and checks eventual coverage + // without asserting a cursor value or a particular visitation order. + #[test] + fn timeout_discovery_has_bounded_work_and_eventually_visits_every_target() { + let count = TIMEOUT_WORK_BUDGET * 2 + 1; + let targets = (0..count) + .map(|i| ManagedTargetConfig::new(i.to_string(), "127.0.0.1:2112")) + .collect(); + let (mut task, _handle) = ManagedClient::task( + ManagedClientConfig { + max_live_target_generations: count, + ..ManagedClientConfig::default() + }, + targets, + ) + .unwrap(); + let mut visited = HashSet::new(); + for _ in 0..count.div_ceil(TIMEOUT_WORK_BUDGET) { + task.timeout_inspections.borrow_mut().clear(); + task.poll_timeout_pass(Instant::now()); + let inspected = task.timeout_inspections.borrow(); + assert!(inspected.len() <= TIMEOUT_WORK_BUDGET); + visited.extend(inspected.iter().copied()); + } + assert_eq!( + visited.len(), + count, + "timeout discovery must reach all targets" + ); + } +} diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index ee68096..7ec2e9a 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -1133,3 +1133,173 @@ fn config_duration_to_ns(field: &str, duration: Duration) -> Result SessionMachine { + let mut machine = SessionMachine::new( + ClientConfig { + max_pending_probes: 2, + clock: Clock::Wall, + stamp_at: irtt_proto::StampAt::Receive, + ..ClientConfig::default() + }, + "127.0.0.1:2112".parse().unwrap(), + ) + .unwrap(); + let acceptance = machine + .prepare_open_acceptance( + OpenReply { + flags: FLAG_OPEN | FLAG_REPLY, + token: 1, + params: machine.requested.clone(), + }, + ClientTimestamp::now(), + ) + .unwrap(); + machine.commit_open(acceptance); + machine + } + + fn send(machine: &mut SessionMachine) -> ProbeSent { + let probe = machine.prepare_probe().unwrap(); + let preflight = machine.preflight_probe_commit(&probe).unwrap(); + let commit = machine + .finalize_probe_commit(preflight, ClientTimestamp::now()) + .unwrap(); + machine.commit_probe_sent(commit, ClientTimestamp::now(), probe.bytes.len()) + } + + fn reply(machine: &mut SessionMachine, seq: u32) -> ClientEvent { + let events = machine + .process_echo_reply( + EchoReply { + flags: FLAG_REPLY, + token: 1, + sequence: seq, + recv_count: None, + recv_window: None, + timestamps: TimestampFields { + recv_wall: Some(1), + ..TimestampFields::default() + }, + payload: vec![], + }, + 32, + ClientTimestamp::now(), + ReceiveMeta::default(), + ) + .unwrap(); + events.into_iter().next().unwrap() + } + + // Advancing a real client through 2^32 submissions is unreasonable. Only + // move the send/receive sequence epoch; construct completion/timeout history through + // ordinary machine transitions and assert emitted behavior on reuse. + #[test] + fn sequence_wrap_reuses_completed_and_timed_out_ids_as_new_probes() { + for timed_out in [false, true] { + let mut machine = opened(); + assert_eq!(send(&mut machine).seq, 0); + if timed_out { + let deadline = machine.next_probe_timeout_deadline().unwrap(); + assert!(matches!( + machine.poll_timeouts_at(deadline).unwrap()[0], + ClientEvent::EchoLoss { seq: 0, .. } + )); + } else { + assert!(matches!( + reply(&mut machine, 0), + ClientEvent::EchoReply { seq: 0, .. } + )); + } + let session = machine.open_session_mut().unwrap(); + session.next_wire_seq = u32::MAX; + session.highest_received_seq = Some(u32::MAX - 1); + assert_eq!(send(&mut machine).seq, u32::MAX); + assert!(matches!( + reply(&mut machine, u32::MAX), + ClientEvent::EchoReply { seq: u32::MAX, .. } + )); + assert_eq!(send(&mut machine).seq, 0); + assert!(matches!( + reply(&mut machine, 0), + ClientEvent::EchoReply { seq: 0, .. } + )); + assert!(matches!( + reply(&mut machine, 0), + ClientEvent::DuplicateReply { seq: 0, .. } + )); + assert_eq!(send(&mut machine).seq, 1); + assert!(matches!( + reply(&mut machine, 1), + ClientEvent::EchoReply { seq: 1, .. } + )); + assert!( + matches!( + reply(&mut machine, 0), + ClientEvent::DuplicateReply { seq: 0, .. } + ), + "reused sequence must receive a fresh completion retention lifetime" + ); + assert_eq!(send(&mut machine).seq, 2); + assert!(matches!( + reply(&mut machine, 2), + ClientEvent::EchoReply { seq: 2, .. } + )); + assert!( + matches!( + reply(&mut machine, 0), + ClientEvent::LateReply { + sent_at: None, + rtt: None, + .. + } + ), + "evicted reused sequence must not resurrect a previous epoch's timeout" + ); + } + } + + // A recoverable send failure that consumes an OPT_ID, followed by a + // plausible wrong-ID completion, cannot be induced reliably with public + // sockets. Exercise that failure boundary directly, without a socket hook + // or inspecting the stored correlation flag/metadata. + #[test] + fn failed_submission_disables_later_tx_endpoints_for_the_session() { + let mut machine = opened(); + for attempt in 0..3 { + if attempt == 1 { + machine.invalidate_kernel_tx_correlation(); + } + let prepared = machine.prepare_probe().unwrap(); + let preflight = machine.preflight_probe_commit(&prepared).unwrap(); + let pre_send = ClientTimestamp::now(); + let commit = machine.finalize_probe_commit(preflight, pre_send).unwrap(); + std::thread::sleep(Duration::from_millis(1)); + let sent = + machine.commit_probe_sent(commit, ClientTimestamp::now(), prepared.bytes.len()); + machine.record_kernel_tx_timestamp(sent.seq, pre_send.wall); + let event = reply(&mut machine, sent.seq); + let ClientEvent::EchoReply { + one_way: Some(one_way), + .. + } = event + else { + panic!("expected measured reply") + }; + let expected_wall = if attempt == 0 { + pre_send.wall + } else { + sent.sent_at.wall + }; + let wall_ns = + i128::try_from(expected_wall.duration_since(UNIX_EPOCH).unwrap().as_nanos()) + .unwrap(); + assert_eq!(one_way.client_to_server.unwrap().as_nanos(), 1 - wall_ns); + } + } +} diff --git a/crates/irtt-integration-tests/tests/managed_behavior.rs b/crates/irtt-integration-tests/tests/managed_behavior.rs index b202b04..c709cbe 100644 --- a/crates/irtt-integration-tests/tests/managed_behavior.rs +++ b/crates/irtt-integration-tests/tests/managed_behavior.rs @@ -16,8 +16,8 @@ async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() { for pacing in [ManagedPacing::Burst, ManagedPacing::Staggered] { let config = ManagedClientConfig { client: ClientConfig { - duration: Some(Duration::from_millis(200)), - interval: Duration::from_millis(20), + duration: Some(Duration::from_millis(600)), + interval: Duration::from_millis(100), probe_timeout: Duration::from_millis(100), open_timeouts: vec![Duration::from_millis(200)], ..ClientConfig::default() @@ -31,6 +31,7 @@ async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() { let mut events = handle.subscribe().unwrap(); let run = tokio::spawn(task); let mut sends: HashMap> = HashMap::new(); + let mut actual_sends = Vec::new(); while let Ok(event) = events.recv().await { if let ManagedEvent::Client { target, @@ -44,11 +45,12 @@ async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() { { let scheduled = scheduled_at.expect("managed probes expose their cadence"); assert!(sent_at.mono >= scheduled); + actual_sends.push((target.clone(), sent_at.mono)); let previous = sends.entry(target).or_default(); if let Some(last) = previous.last() { let elapsed: Duration = scheduled.duration_since(*last); assert!(!elapsed.is_zero()); - assert_eq!(elapsed.as_nanos() % Duration::from_millis(20).as_nanos(), 0); + assert_eq!(elapsed.as_nanos() % Duration::from_millis(100).as_nanos(), 0); } previous.push(scheduled); } @@ -57,6 +59,19 @@ async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() { assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete); assert_eq!(outcome.successful_target_outcomes, 2); assert_eq!(sends.len(), 2); + actual_sends.sort_by_key(|(_, at)| *at); + let cross_target_gaps = actual_sends.windows(2) + .filter(|pair| pair[0].0 != pair[1].0) + .map(|pair| pair[1].1.duration_since(pair[0].1)).collect::>(); + assert!(cross_target_gaps.len() >= 2); + // Allow timer wake jitter while separating the nominal 50 ms + // stagger spacing from back-to-back burst sends. + match pacing { + ManagedPacing::Burst => assert!(cross_target_gaps.iter().filter(|gap| **gap < Duration::from_millis(25)).count() >= 2, + "burst must repeatedly send different targets back-to-back: {cross_target_gaps:?}"), + ManagedPacing::Staggered => assert!(cross_target_gaps.iter().all(|gap| *gap >= Duration::from_millis(35)), + "staggered must space different targets across the interval: {cross_target_gaps:?}"), + } for target in outcome.recent_target_outcomes.iter() { assert!(sends[&target.target].len() >= 2); assert!(target.replies_received >= 2); diff --git a/crates/irtt-server/tests/protocol_contracts.rs b/crates/irtt-server/tests/protocol_contracts.rs index 99c35da..7c894a4 100644 --- a/crates/irtt-server/tests/protocol_contracts.rs +++ b/crates/irtt-server/tests/protocol_contracts.rs @@ -316,3 +316,42 @@ fn negotiated_timestamp_modes_emit_only_the_selected_fields_in_order() { } } } + +#[test] +fn duration_close_waits_for_rate_allowance_after_the_deadline() { + let refill = Duration::from_secs(4); + let maximum = Duration::from_millis(1); + let mut server = ServerCore::new( + ServerConfig::default() + .with_min_send_interval(refill) + .with_burst_allowance(1) + .with_idle_timeout(Duration::MAX) + .with_max_test_duration(maximum), + ); + let opened = open( + &mut server, + &Params { + received_stats: ReceivedStats::Both, + ..Params::with_protocol_defaults() + }, + ); + let first = echo(&mut server, &opened, 0).unwrap(); + assert_eq!(first.flags & FLAG_CLOSE, 0); + thread::sleep(maximum + Duration::from_secs(2)); + assert!( + echo(&mut server, &opened, 63).is_none(), + "exhausted allowance must suppress the deadline-crossing reply" + ); + assert_eq!( + server.session_count(), + 1, + "rate refusal must not release the session" + ); + thread::sleep(refill); + let closed = echo(&mut server, &opened, 1).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_eq!(server.session_count(), 0); +} diff --git a/crates/irtt-server/tests/runtime.rs b/crates/irtt-server/tests/runtime.rs index a91216c..047f3e9 100644 --- a/crates/irtt-server/tests/runtime.rs +++ b/crates/irtt-server/tests/runtime.rs @@ -155,3 +155,109 @@ async fn wildcard_construction_is_refused_without_reply_source_selection() { Err(ServerRuntimeError::WildcardSourceSelectionUnsupported { .. }) )); } + +// A current-thread runtime cannot receive while this test holds its thread. +// Queue real packets before delaying the server: no synthetic ancillary data or +// transport hooks. A short queue delay selects kernel wall time; a delay beyond +// the documented one-second sanity bound falls back to userspace. Monotonic and +// midpoint timestamps must always remain at the later userspace observation. +#[cfg(target_os = "linux")] +#[tokio::test(flavor = "current_thread")] +async fn queued_echoes_select_kernel_receive_wall_only_when_plausible() { + use std::time::{SystemTime, UNIX_EPOCH}; + let ns = || { + i64::try_from( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(), + ) + .unwrap() + }; + for (placement, delay, kernel_expected) in [ + (StampAt::Both, Duration::from_millis(100), true), + (StampAt::Both, Duration::from_millis(1100), false), + (StampAt::Midpoint, Duration::from_millis(100), false), + ] { + let mut server = Server::bind("127.0.0.1:0".parse().unwrap(), unthrottled()) + .await + .unwrap(); + let addr = server.local_addr().unwrap(); + let client = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + client.connect(addr).await.unwrap(); + let (stop_tx, stop_rx) = oneshot::channel(); + let run = tokio::spawn(async move { + server + .run(async { + let _ = stop_rx.await; + }) + .await + }); + let params = Params { + stamp_at: placement, + clock: Clock::Both, + received_stats: ReceivedStats::Both, + ..Params::with_protocol_defaults() + }; + let open = encode_request( + RequestToEncode::Open { + params: ¶ms, + no_test: false, + }, + None, + ) + .unwrap(); + client.send(&open).await.unwrap(); + let mut buffer = [0; 2048]; + let len = timeout(Duration::from_secs(2), client.recv(&mut buffer)) + .await + .unwrap() + .unwrap(); + let opened = decode_open_reply(&buffer[..len], None).unwrap(); + let echo = encode_request( + RequestToEncode::Echo { + token: opened.token, + sequence: 0, + params: &opened.params, + payload: &[], + }, + None, + ) + .unwrap(); + client.writable().await.unwrap(); + let before_send = ns(); + client.try_send(&echo).unwrap(); // No await from submission through resume marker. + let queued_by = ns(); + std::thread::sleep(delay); + let resumed_at = ns(); + let len = timeout(Duration::from_secs(2), client.recv(&mut buffer)) + .await + .unwrap() + .unwrap(); + let reply = decode_echo_reply(&buffer[..len], &opened.params, None).unwrap(); + assert_eq!(reply.recv_count, Some(1)); + assert_eq!(reply.recv_window, Some(1)); + let timestamps = reply.timestamps; + if kernel_expected { + let receive = timestamps.recv_wall.unwrap(); + assert!( + receive >= before_send && receive <= queued_by, + "receive wall must identify the kernel arrival before the queue delay" + ); + let wall_elapsed = timestamps.send_wall.unwrap() - receive; + let mono_elapsed = timestamps.send_mono.unwrap() - timestamps.recv_mono.unwrap(); + assert!( + wall_elapsed - mono_elapsed >= 50_000_000, + "queue delay must not move the monotonic endpoint" + ); + } else { + let receive = timestamps.recv_wall.or(timestamps.midpoint_wall).unwrap(); + assert!( + receive >= resumed_at, + "fallback and midpoint must use userspace time" + ); + } + stop_tx.send(()).unwrap(); + run.await.unwrap().unwrap(); + } +} diff --git a/crates/irtt-server/tests/session_behavior.rs b/crates/irtt-server/tests/session_behavior.rs index fcf8983..db63b22 100644 --- a/crates/irtt-server/tests/session_behavior.rs +++ b/crates/irtt-server/tests/session_behavior.rs @@ -1,4 +1,7 @@ -use std::{net::SocketAddr, time::Duration}; +use std::{ + net::{SocketAddr, SocketAddrV6}, + time::Duration, +}; use irtt_proto::{decode_echo_reply, decode_open_reply, encode_request, Params, RequestToEncode}; use irtt_server::{ServerConfig, ServerCore}; @@ -116,3 +119,143 @@ fn zero_burst_and_zero_idle_timeout_silently_refuse_echoes() { assert!(server.handle_datagram(peer, &echo).unwrap().is_none()); } } + +#[test] +fn ipv6_flow_labels_share_a_session_but_zones_do_not() { + let endpoint = |flow, scope| { + SocketAddr::V6(SocketAddrV6::new( + "fe80::1234".parse().unwrap(), + 40000, + flow, + scope, + )) + }; + let mut server = + ServerCore::new(ServerConfig::default().with_min_send_interval(Duration::ZERO)); + let params = Params { + received_stats: irtt_proto::ReceivedStats::Both, + ..Params::with_protocol_defaults() + }; + let packet = encode_request( + RequestToEncode::Open { + params: ¶ms, + no_test: false, + }, + None, + ) + .unwrap(); + let reply = server + .handle_datagram(endpoint(1, 2), &packet) + .unwrap() + .unwrap(); + let opened = decode_open_reply(reply.bytes(), None).unwrap(); + let echo = encode_request( + RequestToEncode::Echo { + token: opened.token, + sequence: 0, + params: &opened.params, + payload: &[], + }, + None, + ) + .unwrap(); + let close = encode_request( + RequestToEncode::Close { + token: opened.token, + }, + None, + ) + .unwrap(); + assert!(server + .handle_datagram(endpoint(1, 3), &echo) + .unwrap() + .is_none()); + assert!(server + .handle_datagram(endpoint(1, 3), &close) + .unwrap() + .is_none()); + assert_eq!(server.session_count(), 1); + let reply = server + .handle_datagram(endpoint(99, 2), &echo) + .unwrap() + .unwrap(); + assert_eq!( + decode_echo_reply(reply.bytes(), &opened.params, None) + .unwrap() + .recv_count, + Some(1) + ); + assert!(server + .handle_datagram(endpoint(99, 2), &close) + .unwrap() + .is_none()); + assert_eq!(server.session_count(), 0); +} + +#[test] +fn oversized_authenticated_echo_does_not_advance_session_state() { + let key = b"oversized-echo-test"; + let peer = "127.0.0.1:40000".parse().unwrap(); + let mut server = ServerCore::new( + ServerConfig::default() + .with_hmac_key(key) + .with_max_packet_length(96) + .with_burst_allowance(1) + .with_min_send_interval(Duration::from_secs(60)) + .with_max_test_duration(Duration::from_millis(1)) + .with_idle_timeout(Duration::MAX), + ); + let params = Params { + length: 96, + received_stats: irtt_proto::ReceivedStats::Both, + ..Params::with_protocol_defaults() + }; + let open = encode_request( + RequestToEncode::Open { + params: ¶ms, + no_test: false, + }, + Some(key), + ) + .unwrap(); + let reply = server.handle_datagram(peer, &open).unwrap().unwrap(); + let opened = decode_open_reply(reply.bytes(), Some(key)).unwrap(); + let oversized_params = Params { + length: 128, + ..opened.params.clone() + }; + let oversized = encode_request( + RequestToEncode::Echo { + token: opened.token, + sequence: 63, + params: &oversized_params, + payload: &[], + }, + Some(key), + ) + .unwrap(); + assert!(oversized.len() > 96); + assert!(server.handle_datagram(peer, &oversized).unwrap().is_none()); + // If the oversized request started maximum duration, the first valid + // echo would now close. Oversleep preserves this distinction. + std::thread::sleep(Duration::from_millis(2001)); + let valid = encode_request( + RequestToEncode::Echo { + token: opened.token, + sequence: 0, + params: &opened.params, + payload: &[], + }, + Some(key), + ) + .unwrap(); + let reply = server + .handle_datagram(peer, &valid) + .unwrap() + .expect("oversized request must not consume allowance"); + let reply = decode_echo_reply(reply.bytes(), &opened.params, Some(key)).unwrap(); + assert_eq!(reply.flags & irtt_proto::FLAG_CLOSE, 0); + assert_eq!(reply.recv_count, Some(1)); + assert_eq!(reply.recv_window, Some(1)); + assert_eq!(server.session_count(), 1); +} diff --git a/crates/irtt-stats/tests/rolling_properties.rs b/crates/irtt-stats/tests/rolling_properties.rs index 561cdb4..91222e5 100644 --- a/crates/irtt-stats/tests/rolling_properties.rs +++ b/crates/irtt-stats/tests/rolling_properties.rs @@ -1,6 +1,6 @@ //! Independent count and time windows must match a filtered replay through //! StatsCollector, while cumulative accounting retains the complete history. -//! Generated events use non-decreasing timestamps, as required by the collector. +//! Generated events have non-decreasing normalized window timestamps. use std::time::{Duration, Instant, UNIX_EPOCH}; @@ -170,7 +170,7 @@ fn build_event(op: &Op, seq: u32, clock_ms: u64, base: Instant) -> ClientEvent { Op::Loss => ClientEvent::EchoLoss { seq, sent_at: ts_at(base, clock_ms), - timeout_at: ts_at(base, clock_ms).mono, + timeout_at: ts_at(base, clock_ms).mono - Duration::from_millis(10), }, Op::Warning => ClientEvent::Warning { kind: irtt_client::WarningKind::UntrackedReply, @@ -233,6 +233,11 @@ proptest! { // scratch after every step. let mut history: Vec<(ClientEvent, u64)> = Vec::new(); + // Force a loss whose clamped time evicts the initial send, while its + // raw pre-send deadline would retain it. Every generated case must + // therefore distinguish the two normalization choices. + let ops = [Op::Send, Op::Advance(u16::try_from(time_limit_ms + 5).unwrap()), Op::Loss, Op::Advance(10), Op::Warning] + .into_iter().chain(ops).collect::>(); for (idx, op) in ops.iter().enumerate() { if let Op::Advance(delta) = op { clock_ms += u64::from(*delta);