diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index f4362ad..276dd8b 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -790,2250 +790,4 @@ fn duration_to_ns(duration: Duration) -> Result { } #[cfg(test)] -mod tests { - use super::*; - use crate::config::{NegotiationPolicy, DEFAULT_OPEN_TIMEOUTS}; - use irtt_proto::{ - compute_hmac_in_place, echo_packet_len, flags::FLAG_HMAC, flags::FLAG_OPEN, - flags::FLAG_REPLY, layout::PacketLayout, verify_hmac, Clock, ReceivedStats, StampAt, - HMAC_SIZE, MAGIC, - }; - use std::{ - net::UdpSocket, - sync::mpsc, - thread::{self, JoinHandle}, - time::SystemTime, - }; - - const TOKEN: u64 = 0x1234_5678_90ab_cdef; - const HMAC_OFFSET: usize = 4; - - struct FakeServer { - addr: SocketAddr, - rx: mpsc::Receiver>, - done: JoinHandle<()>, - } - - impl FakeServer { - fn join(self) { - self.done.join().unwrap(); - } - } - - fn default_test_config(addr: SocketAddr) -> ClientConfig { - ClientConfig { - server_addr: addr.to_string(), - open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], - ..ClientConfig::default() - } - } - - fn start_fake_server(handler: F) -> FakeServer - where - F: FnOnce(UdpSocket, mpsc::Sender>) + 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 } - } - - fn recv_request(socket: &UdpSocket, tx: &mpsc::Sender>) -> (Vec, SocketAddr) { - let mut buf = [0_u8; 512]; - let (size, peer) = socket.recv_from(&mut buf).unwrap(); - let packet = buf[..size].to_vec(); - tx.send(packet.clone()).unwrap(); - (packet, peer) - } - - fn open_reply(flags: u8, token: u64, params: &Params, hmac_key: Option<&[u8]>) -> Vec { - let mut packet = Vec::new(); - packet.extend_from_slice(&MAGIC); - packet.push(flags | hmac_key.map_or(0, |_| FLAG_HMAC)); - if hmac_key.is_some() { - packet.extend_from_slice(&[0_u8; HMAC_SIZE]); - } - packet.extend_from_slice(&token.to_le_bytes()); - packet.extend_from_slice(¶ms.encode()); - if let Some(key) = hmac_key { - compute_hmac_in_place(key, &mut packet, HMAC_OFFSET).unwrap(); - } - packet - } - - fn echo_reply_packet( - token: u64, - seq: u32, - params: &Params, - timestamps: &TimestampFields, - hmac_key: Option<&[u8]>, - ) -> Vec { - let has_hmac = hmac_key.is_some(); - let layout = PacketLayout::echo(has_hmac, params); - let packet_len = echo_packet_len(has_hmac, params); - let mut packet = Vec::with_capacity(packet_len); - - let mut flags = FLAG_REPLY; - if has_hmac { - flags |= FLAG_HMAC; - } - packet.extend_from_slice(&MAGIC); - packet.push(flags); - if has_hmac { - packet.extend_from_slice(&[0_u8; HMAC_SIZE]); - } - packet.extend_from_slice(&token.to_le_bytes()); - packet.extend_from_slice(&seq.to_le_bytes()); - - if layout.recv_count { - packet.extend_from_slice(&42_u32.to_le_bytes()); - } - if layout.recv_window { - packet.extend_from_slice(&0x07_u64.to_le_bytes()); - } - - if layout.recv_wall { - packet.extend_from_slice(×tamps.recv_wall.unwrap_or(0).to_le_bytes()); - } - if layout.recv_mono { - packet.extend_from_slice(×tamps.recv_mono.unwrap_or(0).to_le_bytes()); - } - if layout.midpoint_wall { - packet.extend_from_slice(×tamps.midpoint_wall.unwrap_or(0).to_le_bytes()); - } - if layout.midpoint_mono { - packet.extend_from_slice(×tamps.midpoint_mono.unwrap_or(0).to_le_bytes()); - } - if layout.send_wall { - packet.extend_from_slice(×tamps.send_wall.unwrap_or(0).to_le_bytes()); - } - if layout.send_mono { - packet.extend_from_slice(×tamps.send_mono.unwrap_or(0).to_le_bytes()); - } - - packet.resize(packet_len, 0); - - if let Some(key) = hmac_key { - compute_hmac_in_place(key, &mut packet, HMAC_OFFSET).unwrap(); - } - packet - } - - fn open_success_server(params: Params) -> FakeServer { - start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - }) - } - - fn no_test_server(params: Params, token: u64) -> FakeServer { - start_fake_server(move |socket, tx| { - let (request, peer) = recv_request(&socket, &tx); - assert_eq!(request[3] & flags::FLAG_CLOSE, flags::FLAG_CLOSE); - let reply = open_reply( - FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, - token, - ¶ms, - None, - ); - socket.send_to(&reply, peer).unwrap(); - }) - } - - fn timeout_server(wait: Duration) -> FakeServer { - start_fake_server(move |socket, tx| { - socket.set_read_timeout(Some(wait)).unwrap(); - while recv_request_timeout(&socket, &tx).is_some() {} - }) - } - - fn recv_request_timeout( - socket: &UdpSocket, - tx: &mpsc::Sender>, - ) -> Option<(Vec, SocketAddr)> { - let mut buf = [0_u8; 512]; - match socket.recv_from(&mut buf) { - Ok((size, peer)) => { - let packet = buf[..size].to_vec(); - tx.send(packet.clone()).unwrap(); - Some((packet, peer)) - } - Err(_) => None, - } - } - - fn echo_server(params: Params) -> FakeServer { - start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - - socket - .set_read_timeout(Some(Duration::from_secs(5))) - .unwrap(); - loop { - let mut buf = [0_u8; 2048]; - let size = match socket.recv_from(&mut buf) { - Ok((size, _)) => size, - Err(_) => break, - }; - let packet = buf[..size].to_vec(); - tx.send(packet.clone()).unwrap(); - - if buf[3] & flags::FLAG_CLOSE != 0 { - break; - } - - let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); - let ts = TimestampFields { - recv_wall: Some(1_000_000_000), - recv_mono: Some(100_000), - send_wall: Some(1_000_100_000), - send_mono: Some(200_000), - ..Default::default() - }; - let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, None); - socket.send_to(&reply_packet, peer).unwrap(); - } - }) - } - - fn assert_open_started(outcome: OpenOutcome) -> NegotiatedParams { - match outcome { - OpenOutcome::Started { - token, negotiated, .. - } => { - assert_eq!(token, TOKEN); - negotiated - } - OpenOutcome::NoTestCompleted { .. } => panic!("unexpected no-test outcome"), - } - } - - fn assert_no_test_completed(outcome: OpenOutcome) -> NegotiatedParams { - match outcome { - OpenOutcome::NoTestCompleted { - negotiated, event, .. - } => { - assert!(matches!( - event, - ClientEvent::NoTestCompleted { - negotiated: ref event_params, - .. - } if *event_params == negotiated - )); - negotiated - } - OpenOutcome::Started { .. } => panic!("unexpected started outcome"), - } - } - - fn open_client_with_echo_server(params: &Params) -> (Client, FakeServer) { - let server = echo_server(params.clone()); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - (client, server) - } - - fn default_params() -> Params { - Params { - protocol_version: 1, - 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() - } - } - - // ---------- Existing Milestone 2 tests ---------- - - #[test] - fn client_config_default() { - let config = ClientConfig::default(); - assert_eq!(config.duration, Some(Duration::from_secs(3))); - assert_eq!(config.interval, Duration::from_secs(1)); - assert_eq!(config.length, 0); - assert_eq!(config.received_stats, ReceivedStats::Both); - assert_eq!(config.stamp_at, StampAt::Both); - assert_eq!(config.clock, Clock::Both); - assert_eq!(config.dscp, 0); - assert_eq!(config.hmac_key, None); - assert_eq!(config.server_fill, None); - assert_eq!(config.open_timeouts, DEFAULT_OPEN_TIMEOUTS); - assert_eq!(config.run_mode, RunMode::Normal); - assert_eq!(config.negotiation_policy, NegotiationPolicy::Strict); - assert_eq!(config.probe_timeout, Duration::from_secs(4)); - assert_eq!(config.max_pending_probes, 4096); - } - - #[test] - fn params_from_config_maps_compatibility_fields() { - let config = ClientConfig { - duration: Some(Duration::from_secs(5)), - interval: Duration::from_millis(250), - length: 1472, - received_stats: ReceivedStats::Window, - stamp_at: StampAt::Midpoint, - clock: Clock::Wall, - dscp: 46, - hmac_key: Some(b"secret".to_vec()), - server_fill: Some("rand".to_owned()), - ..ClientConfig::default() - }; - - let params = params_from_config(&config).unwrap(); - assert_eq!(params.protocol_version, PROTOCOL_VERSION); - assert_eq!(params.duration_ns, 5_000_000_000); - assert_eq!(params.interval_ns, 250_000_000); - assert_eq!(params.length, 1472); - assert_eq!(params.received_stats, ReceivedStats::Window); - assert_eq!(params.stamp_at, StampAt::Midpoint); - assert_eq!(params.clock, Clock::Wall); - assert_eq!(params.dscp, 46, "config DSCP codepoint must not be shifted"); - assert_eq!( - params.server_fill.as_ref().map(|fill| fill.value.as_str()), - Some("rand") - ); - assert_eq!(config.hmac_key.as_deref(), Some(b"secret".as_slice())); - } - - #[test] - fn params_from_config_encodes_continuous_duration_as_zero() { - let config = ClientConfig { - duration: None, - ..ClientConfig::default() - }; - assert_eq!(params_from_config(&config).unwrap().duration_ns, 0); - } - - #[test] - fn params_from_config_accepts_max_dscp_codepoint() { - let config = ClientConfig { - dscp: 63, - ..ClientConfig::default() - }; - assert_eq!(params_from_config(&config).unwrap().dscp, 63); - } - - #[test] - fn params_from_config_rejects_invalid_dscp_codepoint() { - let config = ClientConfig { - dscp: 64, - ..ClientConfig::default() - }; - assert!(matches!( - params_from_config(&config), - Err(ClientError::InvalidConfig { .. }) - )); - } - - #[test] - fn params_from_config_rejects_oversized_server_fill() { - let config = ClientConfig { - server_fill: Some("0123456789abcdef0123456789abcdefx".to_owned()), - ..ClientConfig::default() - }; - assert!(matches!( - params_from_config(&config), - Err(ClientError::InvalidConfig { .. }) - )); - } - - #[test] - fn address_resolution_connects_to_local_fake_server() { - let server = start_fake_server(|socket, tx| { - let _ = recv_request(&socket, &tx); - }); - let client = Client::connect(default_test_config(server.addr)).unwrap(); - client.socket.send(b"ping").unwrap(); - assert_eq!(server.rx.recv().unwrap(), b"ping"); - server.join(); - } - - #[test] - fn successful_open_handshake() { - let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - let params = params_from_config(&config).unwrap(); - let server = open_success_server(params.clone()); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - - let negotiated = assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - assert_eq!(negotiated.params, params); - assert!(matches!(client.phase, ClientPhase::Open { token: TOKEN })); - server.join(); - } - - #[test] - fn open_fails_when_already_open() { - let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - let params = params_from_config(&config).unwrap(); - let server = open_success_server(params); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::AlreadyOpen) - )); - server.join(); - } - - #[test] - fn open_fails_after_close() { - let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - let _ = recv_request(&socket, &tx); - }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.close(ClientTimestamp::now()).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::AlreadyClosed) - )); - server.join(); - } - - #[test] - fn open_fails_after_no_test_completed() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - let params = params_from_config(&config).unwrap(); - let server = no_test_server(params, 0); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert_no_test_completed(client.open(ClientTimestamp::now()).unwrap()); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::AlreadyCompleted) - )); - server.join(); - } - - #[test] - fn open_retries_after_first_timeout() { - let server = start_fake_server(|socket, tx| { - let (first, _) = recv_request(&socket, &tx); - let (_, peer) = recv_request(&socket, &tx); - let params = Params::decode(&first[4..]).unwrap(); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - assert_eq!(first[3] & FLAG_OPEN, FLAG_OPEN); - socket.send_to(&reply, peer).unwrap(); - }); - let config = ClientConfig { - open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(500)], - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - let outcome = client.open(ClientTimestamp::now()).unwrap(); - assert_open_started(outcome); - assert_eq!(server.rx.iter().take(2).count(), 2); - server.join(); - } - - #[test] - fn open_timeout_after_all_timeouts() { - let server = timeout_server(Duration::from_millis(700)); - let config = ClientConfig { - open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::OpenTimeout) - )); - assert_eq!(server.rx.iter().take(2).count(), 2); - server.join(); - } - - #[test] - fn open_restores_configured_read_timeout_after_timeout() { - let server = timeout_server(Duration::from_millis(700)); - let config = ClientConfig { - open_timeouts: vec![Duration::from_millis(200)], - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(450)), - ..crate::SocketConfig::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::OpenTimeout) - )); - - let start = std::time::Instant::now(); - let mut buf = [0_u8; 1]; - assert!(client.socket.recv(&mut buf).is_err()); - assert!(start.elapsed() >= Duration::from_millis(350)); - server.join(); - } - - #[test] - fn strict_negotiation_accepts_identical_params() { - let config = ClientConfig::default(); - let params = params_from_config(&config).unwrap(); - assert!(validate_negotiated_params(¶ms, ¶ms, NegotiationPolicy::Strict).is_ok()); - } - - #[test] - fn strict_negotiation_rejects_changed_params() { - let config = ClientConfig::default(); - let requested = params_from_config(&config).unwrap(); - let mut returned = requested.clone(); - returned.dscp = 1; - assert!(matches!( - validate_negotiated_params(&requested, &returned, NegotiationPolicy::Strict), - Err(ClientError::NegotiationRejected { .. }) - )); - } - - #[test] - fn loose_negotiation_accepts_server_restricted_params() { - let config = ClientConfig::default(); - let requested = params_from_config(&config).unwrap(); - let mut returned = requested.clone(); - returned.duration_ns /= 2; - returned.length = 0; - assert!( - validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose).is_ok() - ); - } - - #[test] - fn protocol_version_mismatch_fails() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.negotiation_policy = NegotiationPolicy::Loose; - let mut params = params_from_config(&config).unwrap(); - params.protocol_version = 2; - let server = open_success_server(params); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::ProtocolVersionMismatch { received: 2, .. }) - )); - server.join(); - } - - #[test] - fn server_rejection_fails_in_normal_mode() { - let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::ServerRejected) - )); - server.join(); - } - - #[test] - fn no_test_open_close_succeeds_on_open_reply_close() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - let params = params_from_config(&config).unwrap(); - let server = no_test_server(params.clone(), 0); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - let negotiated = assert_no_test_completed(client.open(ClientTimestamp::now()).unwrap()); - assert_eq!(negotiated.params, params); - assert_eq!(client.negotiated.as_ref(), Some(&negotiated)); - assert!(matches!( - client.close(ClientTimestamp::now()), - Err(ClientError::NotOpen) - )); - server.join(); - } - - #[test] - fn no_test_success_validates_params() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - let params = params_from_config(&config).unwrap(); - let server = no_test_server(params.clone(), 0); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - let negotiated = assert_no_test_completed(client.open(ClientTimestamp::now()).unwrap()); - assert_eq!(negotiated.params, params); - server.join(); - } - - #[test] - fn no_test_rejects_non_close_open_reply() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - let params = params_from_config(&config).unwrap(); - let server = open_success_server(params); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::UnexpectedNoTestReply) - )); - server.join(); - } - - #[test] - fn no_test_rejects_non_zero_token_with_close_reply() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - let params = params_from_config(&config).unwrap(); - let server = no_test_server(params, TOKEN); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::NonZeroNoTestToken { token: TOKEN }) - )); - server.join(); - } - - #[test] - fn no_test_strict_negotiation_rejects_changed_params() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - let mut params = params_from_config(&config).unwrap(); - params.dscp = 1; - let server = no_test_server(params, 0); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::NegotiationRejected { .. }) - )); - server.join(); - } - - #[test] - fn no_test_loose_negotiation_accepts_restricted_params() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - config.negotiation_policy = NegotiationPolicy::Loose; - let mut params = params_from_config(&config).unwrap(); - params.duration_ns /= 2; - let server = no_test_server(params.clone(), 0); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - let negotiated = assert_no_test_completed(client.open(ClientTimestamp::now()).unwrap()); - assert_eq!(negotiated.params, params); - server.join(); - } - - #[test] - fn close_sends_one_close_packet_with_negotiated_token() { - let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - let _ = recv_request(&socket, &tx); - }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - let events = client.close(ClientTimestamp::now()).unwrap(); - assert_eq!(events.len(), 1); - let packets: Vec<_> = server.rx.iter().take(2).collect(); - let close = &packets[1]; - assert_eq!(close[3], flags::FLAG_CLOSE); - assert_eq!(u64::from_le_bytes(close[4..12].try_into().unwrap()), TOKEN); - server.join(); - } - - #[test] - fn hmac_open_success() { - let key = b"secret".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key.clone()); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (request, peer) = recv_request(&socket, &tx); - verify_hmac(&key, &request, HMAC_OFFSET).unwrap(); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&key)); - socket.send_to(&reply, peer).unwrap(); - }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - server.join(); - } - - #[test] - fn hmac_open_rejects_missing_hmac() { - let key = b"secret".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::Protocol( - irtt_proto::ProtoError::HmacPresenceMismatch - )) - )); - server.join(); - } - - #[test] - fn hmac_open_rejects_bad_hmac() { - let key = b"secret".to_vec(); - let wrong_key = b"wrong".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&wrong_key)); - socket.send_to(&reply, peer).unwrap(); - }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(ClientTimestamp::now()), - Err(ClientError::Protocol(irtt_proto::ProtoError::BadHmac)) - )); - server.join(); - } - - #[test] - fn hmac_close_packet_includes_valid_hmac() { - let key = b"secret".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key.clone()); - let params = params_from_config(&config).unwrap(); - let server_key = key.clone(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&server_key)); - socket.send_to(&reply, peer).unwrap(); - let _ = recv_request(&socket, &tx); - }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.close(ClientTimestamp::now()).unwrap(); - let packets: Vec<_> = server.rx.iter().take(2).collect(); - let close = &packets[1]; - assert_eq!(close[3], flags::FLAG_CLOSE | FLAG_HMAC); - verify_hmac(&key, close, HMAC_OFFSET).unwrap(); - assert_eq!( - u64::from_le_bytes(close[4 + HMAC_SIZE..12 + HMAC_SIZE].try_into().unwrap()), - TOKEN - ); - server.join(); - } - - #[test] - fn minimum_open_timeout_under_200ms_is_rejected() { - let config = ClientConfig { - open_timeouts: vec![Duration::from_millis(199)], - ..ClientConfig::default() - }; - assert!(matches!( - Client::connect(config), - Err(ClientError::OpenTimeoutTooSmall { .. }) - )); - } - - #[test] - fn empty_open_timeouts_is_rejected() { - let config = ClientConfig { - open_timeouts: vec![], - ..ClientConfig::default() - }; - assert!(matches!( - Client::connect(config), - Err(ClientError::NoOpenTimeouts) - )); - } - - // ---------- Milestone 3 probe tests ---------- - - #[test] - fn send_probe_fails_before_open() { - let server = start_fake_server(|_socket, _tx| {}); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - assert!(matches!(client.send_probe(), Err(ClientError::NotOpen))); - server.join(); - } - - #[test] - fn send_probe_fails_after_no_test_completed() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - let params = params_from_config(&config).unwrap(); - let server = no_test_server(params, 0); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert_no_test_completed(client.open(ClientTimestamp::now()).unwrap()); - assert!(matches!( - client.send_probe(), - Err(ClientError::AlreadyCompleted) - )); - server.join(); - } - - #[test] - fn send_probe_fails_after_close() { - let params = default_params(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - socket - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - loop { - let mut buf = [0_u8; 512]; - match socket.recv_from(&mut buf) { - Ok((size, _)) => { - tx.send(buf[..size].to_vec()).unwrap(); - } - Err(_) => break, - } - } - }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.close(ClientTimestamp::now()).unwrap(); - assert!(matches!( - client.send_probe(), - Err(ClientError::AlreadyClosed) - )); - server.join(); - } - - fn silent_open_server(params: Params) -> FakeServer { - start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - socket - .set_read_timeout(Some(Duration::from_secs(5))) - .unwrap(); - loop { - let mut buf = [0_u8; 2048]; - match socket.recv_from(&mut buf) { - Ok((size, _)) => { - tx.send(buf[..size].to_vec()).unwrap(); - } - Err(_) => break, - } - } - }) - } - - #[test] - fn send_probe_sends_valid_echo_request() { - let params = default_params(); - let server = silent_open_server(params.clone()); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - let events = client.send_probe().unwrap(); - assert_eq!(events.len(), 1); - match &events[0] { - ClientEvent::EchoSent { - seq, - logical_seq, - remote, - bytes, - .. - } => { - assert_eq!(*seq, 0); - assert_eq!(*logical_seq, 0); - assert_eq!(*remote, server.addr); - assert_eq!(*bytes, echo_packet_len(false, ¶ms)); - } - other => panic!("expected EchoSent, got {other:?}"), - } - thread::sleep(Duration::from_millis(30)); - let packets: Vec<_> = server.rx.try_iter().collect(); - let echo_reqs: Vec<_> = packets - .iter() - .filter(|p| p.len() >= 4 && p[3] & FLAG_OPEN == 0) - .collect(); - let echo_req = echo_reqs.first().unwrap(); - assert_eq!(&echo_req[..3], &MAGIC); - assert_eq!(echo_req[3], 0x00); - let req_token = u64::from_le_bytes(echo_req[4..12].try_into().unwrap()); - assert_eq!(req_token, TOKEN); - let seq = u32::from_le_bytes(echo_req[12..16].try_into().unwrap()); - assert_eq!(seq, 0); - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn echo_sent_reports_schedule_and_timer_error() { - let params = default_params(); - let server = silent_open_server(params); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - let start = ClientTimestamp { - mono: Instant::now(), - wall: SystemTime::now(), - }; - assert_open_started(client.open(start).unwrap()); - let session_start = client.session.as_ref().unwrap().start_mono; - assert!( - session_start >= start.mono, - "probe schedule must start after open begins" - ); - let first_probe_at = ClientTimestamp { - mono: session_start, - wall: SystemTime::now(), - }; - - let events = client.send_probe_at(first_probe_at).unwrap(); - match &events[0] { - ClientEvent::EchoSent { - scheduled_at, - sent_at, - timer_error, - .. - } => { - assert_eq!(*scheduled_at, session_start); - assert_eq!(*sent_at, first_probe_at); - assert_eq!(*timer_error, Duration::ZERO); - } - other => panic!("expected EchoSent, got {other:?}"), - } - - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn send_probe_starts_seq_at_zero_and_increments() { - let params = default_params(); - let server = silent_open_server(params); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); - client.send_probe().unwrap(); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(30)); - let packets: Vec<_> = server.rx.try_iter().collect(); - let echo_reqs: Vec<_> = packets - .iter() - .filter(|p| p.len() >= 4 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0) - .collect(); - assert_eq!(echo_reqs.len(), 3); - for (i, pkt) in echo_reqs.iter().enumerate() { - let seq = u32::from_le_bytes(pkt[12..16].try_into().unwrap()); - assert_eq!(seq, i as u32); - } - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn send_probe_respects_finite_duration_exclusive_end() { - let params = Params { - protocol_version: 1, - duration_ns: 1_000_000_000, - interval_ns: 500_000_000, - received_stats: ReceivedStats::Both, - stamp_at: StampAt::Both, - clock: Clock::Both, - ..Params::default() - }; - let server = silent_open_server(params.clone()); - let config = ClientConfig { - duration: Some(Duration::from_secs(1)), - interval: Duration::from_millis(500), - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - - let session = client.session.as_ref().unwrap(); - let start = session.start_mono; - let interval = Duration::from_millis(500); - - let now0 = ClientTimestamp { - mono: start, - wall: SystemTime::now(), - }; - assert!(client.send_probe_at(now0).is_ok()); - - let now1 = ClientTimestamp { - mono: start + interval, - wall: SystemTime::now(), - }; - assert!(client.send_probe_at(now1).is_ok()); - - let now2 = ClientTimestamp { - mono: start + Duration::from_secs(1), - wall: SystemTime::now(), - }; - let events = client.send_probe_at(now2).unwrap(); - assert!(events.is_empty()); - assert!(client.session.as_ref().unwrap().sending_done); - assert!(client.next_send_deadline().is_none()); - - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn continuous_duration_keeps_generating_send_deadlines() { - let params = Params { - protocol_version: 1, - duration_ns: 0, - interval_ns: 500_000_000, - received_stats: ReceivedStats::Both, - stamp_at: StampAt::Both, - clock: Clock::Both, - ..Params::default() - }; - let server = silent_open_server(params); - let config = ClientConfig { - duration: None, - interval: Duration::from_millis(500), - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - - let start = client.session.as_ref().unwrap().start_mono; - let interval = Duration::from_millis(500); - for seq in 0..4 { - let now = ClientTimestamp { - mono: start + interval * seq, - wall: SystemTime::now(), - }; - let events = client.send_probe_at(now).unwrap(); - assert_eq!(events.len(), 1); - assert!(client.next_send_deadline().is_some()); - assert!(!client.session.as_ref().unwrap().sending_done); - } - - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn next_send_deadline_advances_by_interval() { - let params = default_params(); - let (mut client, server) = open_client_with_echo_server(¶ms); - - let session = client.session.as_ref().unwrap(); - let start = session.start_mono; - let deadline0 = client.next_send_deadline().unwrap(); - assert_eq!(deadline0, start); - - client.send_probe().unwrap(); - let deadline1 = client.next_send_deadline().unwrap(); - assert_eq!(deadline1, start + Duration::from_secs(1)); - - client.send_probe().unwrap(); - let deadline2 = client.next_send_deadline().unwrap(); - assert_eq!(deadline2, start + Duration::from_secs(2)); - - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn recv_once_returns_empty_on_timeout() { - let params = default_params(); - let server = open_success_server(params); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - let events = client.recv_once().unwrap(); - assert!(events.is_empty()); - server.join(); - } - - #[test] - fn recv_once_decodes_echo_reply_and_emits_event() { - let params = default_params(); - let (mut client, server) = open_client_with_echo_server(¶ms); - - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(50)); - - let events = client.recv_once().unwrap(); - assert_eq!(events.len(), 1); - match &events[0] { - ClientEvent::EchoReply { - seq, - logical_seq, - rtt, - received_stats, - server_timing, - bytes, - .. - } => { - assert_eq!(*seq, 0); - assert_eq!(*logical_seq, 0); - assert_eq!(*bytes, echo_packet_len(false, ¶ms)); - assert!(rtt.raw > Duration::ZERO); - assert_eq!(rtt.effective, rtt.adjusted.unwrap_or(rtt.raw)); - assert!(received_stats.is_some()); - let stats = received_stats.as_ref().unwrap(); - assert_eq!(stats.count, Some(42)); - assert_eq!(stats.window, Some(0x07)); - assert!(server_timing.is_some()); - let st = server_timing.as_ref().unwrap(); - assert!(st.processing.is_some()); - } - other => panic!("expected EchoReply, got {other:?}"), - } - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn echo_reply_rtt_uses_client_monotonic() { - let params = default_params(); - let (mut client, server) = open_client_with_echo_server(¶ms); - - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(20)); - - let events = client.recv_once().unwrap(); - if let ClientEvent::EchoReply { rtt, .. } = &events[0] { - assert!(rtt.raw >= Duration::from_millis(15)); - } else { - panic!("expected EchoReply"); - } - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn server_processing_subtracted_when_valid() { - let params = default_params(); - let (mut client, server) = open_client_with_echo_server(¶ms); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(20)); - let events = client.recv_once().unwrap(); - if let ClientEvent::EchoReply { - rtt, server_timing, .. - } = &events[0] - { - let st = server_timing.as_ref().unwrap(); - let processing = st.processing.unwrap(); - assert!(processing > Duration::ZERO); - if let Some(adj) = rtt.adjusted { - assert!(adj < rtt.raw); - assert_eq!(rtt.effective, adj); - } - } else { - panic!("expected EchoReply"); - } - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn server_processing_greater_than_raw_does_not_underflow() { - let base = Instant::now(); - let rtt = compute_rtt( - &ClientTimestamp { - mono: base, - wall: SystemTime::now(), - }, - &ClientTimestamp { - mono: base + Duration::from_nanos(1), - wall: SystemTime::now(), - }, - &TimestampFields { - recv_mono: Some(0), - send_mono: Some(1_000_000_000), - ..Default::default() - }, - ); - assert!(rtt.adjusted.is_none()); - assert_eq!(rtt.effective, rtt.raw); - assert_eq!( - rtt.adjusted_signed, - Some(SignedDuration { ns: -999_999_999 }) - ); - assert_eq!(rtt.effective_signed, SignedDuration { ns: -999_999_999 }); - } - - #[test] - fn received_stats_parsed_into_sample() { - let params = default_params(); - let (mut client, server) = open_client_with_echo_server(¶ms); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(30)); - let events = client.recv_once().unwrap(); - if let ClientEvent::EchoReply { received_stats, .. } = &events[0] { - let rs = received_stats.as_ref().unwrap(); - assert_eq!(rs.count, Some(42)); - assert_eq!(rs.window, Some(0x07)); - } else { - panic!("expected EchoReply"); - } - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn wrong_token_reply_is_dropped() { - let params = default_params(); - let wrong_token: u64 = 0xDEAD_BEEF_CAFE_BABE; - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - - socket - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - let mut buf = [0_u8; 2048]; - if let Ok((size, _)) = socket.recv_from(&mut buf) { - tx.send(buf[..size].to_vec()).unwrap(); - let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); - let ts = TimestampFields::default(); - let reply_packet = echo_reply_packet(wrong_token, seq, ¶ms, &ts, None); - socket.send_to(&reply_packet, peer).unwrap(); - } - }); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(30)); - let events = client.recv_once().unwrap(); - assert_eq!(events.len(), 1); - assert!(matches!(&events[0], ClientEvent::Warning { .. })); - server.join(); - } - - #[test] - fn bad_hmac_reply_is_dropped() { - let key = b"secret".to_vec(); - let wrong_key = b"wrong".to_vec(); - let params = default_params(); - let server_key = key.clone(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&server_key)); - socket.send_to(&reply, peer).unwrap(); - - socket - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - let mut buf = [0_u8; 2048]; - if let Ok((size, _)) = socket.recv_from(&mut buf) { - tx.send(buf[..size].to_vec()).unwrap(); - let seq = u32::from_le_bytes( - buf[4 + HMAC_SIZE + 8..4 + HMAC_SIZE + 12] - .try_into() - .unwrap(), - ); - let ts = TimestampFields::default(); - let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, Some(&wrong_key)); - socket.send_to(&reply_packet, peer).unwrap(); - } - }); - let config = ClientConfig { - hmac_key: Some(key), - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(30)); - let events = client.recv_once().unwrap(); - assert_eq!(events.len(), 1); - assert!(matches!(&events[0], ClientEvent::Warning { .. })); - server.join(); - } - - #[test] - fn duplicate_reply_emits_duplicate_event() { - let params = default_params(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - - socket - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - let mut buf = [0_u8; 2048]; - if let Ok((size, _)) = socket.recv_from(&mut buf) { - tx.send(buf[..size].to_vec()).unwrap(); - let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); - let ts = TimestampFields::default(); - let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, None); - socket.send_to(&reply_packet, peer).unwrap(); - thread::sleep(Duration::from_millis(10)); - socket.send_to(&reply_packet, peer).unwrap(); - } - }); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(50)); - - let events1 = client.recv_once().unwrap(); - assert_eq!(events1.len(), 1); - assert!(matches!(&events1[0], ClientEvent::EchoReply { .. })); - - thread::sleep(Duration::from_millis(30)); - let events2 = client.recv_once().unwrap(); - assert_eq!(events2.len(), 1); - assert!(matches!( - &events2[0], - ClientEvent::DuplicateReply { seq: 0, .. } - )); - server.join(); - } - - #[test] - fn out_of_order_reply_emits_late_event() { - let params = default_params(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - - socket - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - - let mut seqs = Vec::new(); - for _ in 0..2 { - let mut buf = [0_u8; 2048]; - if let Ok((size, _)) = socket.recv_from(&mut buf) { - tx.send(buf[..size].to_vec()).unwrap(); - let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); - seqs.push(seq); - } - } - let ts = TimestampFields::default(); - let reply1 = echo_reply_packet(TOKEN, seqs[1], ¶ms, &ts, None); - socket.send_to(&reply1, peer).unwrap(); - thread::sleep(Duration::from_millis(10)); - let reply0 = echo_reply_packet(TOKEN, seqs[0], ¶ms, &ts, None); - socket.send_to(&reply0, peer).unwrap(); - }); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(50)); - - let ev1 = client.recv_once().unwrap(); - assert_eq!(ev1.len(), 1); - assert!(matches!(&ev1[0], ClientEvent::EchoReply { seq: 1, .. })); - - thread::sleep(Duration::from_millis(30)); - let ev2 = client.recv_once().unwrap(); - assert_eq!(ev2.len(), 1); - match &ev2[0] { - ClientEvent::LateReply { - seq, - highest_seen, - rtt, - .. - } => { - assert_eq!(*seq, 0); - assert_eq!(*highest_seen, 1); - assert!(rtt.is_some()); - } - other => panic!("expected LateReply, got {other:?}"), - } - server.join(); - } - - #[test] - fn poll_timeouts_emits_echo_loss() { - let params = default_params(); - let server = silent_open_server(params); - let config = ClientConfig { - probe_timeout: Duration::from_millis(100), - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - - client.send_probe().unwrap(); - client.send_probe().unwrap(); - - let no_loss = client.poll_timeouts(ClientTimestamp::now()).unwrap(); - assert!(no_loss.is_empty()); - - thread::sleep(Duration::from_millis(150)); - let events = client.poll_timeouts(ClientTimestamp::now()).unwrap(); - assert_eq!(events.len(), 2); - for event in &events { - assert!(matches!(event, ClientEvent::EchoLoss { .. })); - } - server.join(); - } - - #[test] - fn poll_timeouts_removes_expired_pending() { - let params = default_params(); - let server = silent_open_server(params); - let config = ClientConfig { - probe_timeout: Duration::from_millis(100), - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(150)); - client.poll_timeouts(ClientTimestamp::now()).unwrap(); - - let session = client.session.as_ref().unwrap(); - assert_eq!(session.pending.len(), 0); - assert_eq!(session.timed_out.len(), 1); - server.join(); - } - - #[test] - fn late_reply_after_timeout_preserves_measurement_metadata() { - let params = default_params(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - - let mut buf = [0_u8; 2048]; - let (size, _) = socket.recv_from(&mut buf).unwrap(); - tx.send(buf[..size].to_vec()).unwrap(); - let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); - - thread::sleep(Duration::from_millis(90)); - let ts = TimestampFields { - recv_wall: Some(1_000_000_000), - recv_mono: Some(100_000), - send_wall: Some(1_000_100_000), - send_mono: Some(200_000), - ..Default::default() - }; - let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, None); - socket.send_to(&reply_packet, peer).unwrap(); - socket.send_to(&reply_packet, peer).unwrap(); - }); - let config = ClientConfig { - probe_timeout: Duration::from_millis(40), - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(60)); - let losses = client.poll_timeouts(ClientTimestamp::now()).unwrap(); - assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. })); - assert_eq!(client.session.as_ref().unwrap().pending.len(), 0); - assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 1); - - let late = client.recv_once().unwrap(); - match &late[0] { - ClientEvent::LateReply { - seq, - logical_seq, - sent_at, - rtt, - server_timing, - one_way, - bytes, - .. - } => { - assert_eq!(*seq, 0); - assert_eq!(*logical_seq, Some(0)); - assert!(sent_at.is_some()); - assert!(rtt.is_some()); - assert!(server_timing.is_some()); - assert!(one_way.is_some()); - assert_eq!(*bytes, echo_packet_len(false, &default_params())); - } - other => panic!("expected stats-eligible LateReply, got {other:?}"), - } - assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 0); - - let duplicate = client.recv_once().unwrap(); - assert!(matches!( - &duplicate[0], - ClientEvent::DuplicateReply { - seq: 0, - bytes, - .. - } if *bytes == echo_packet_len(false, &default_params()) - )); - - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn close_clears_timed_out_metadata() { - let params = default_params(); - let server = silent_open_server(params); - let config = ClientConfig { - probe_timeout: Duration::from_millis(40), - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(60)); - client.poll_timeouts(ClientTimestamp::now()).unwrap(); - assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 1); - - client.close(ClientTimestamp::now()).unwrap(); - assert!(client.session.is_none()); - server.join(); - } - - #[test] - fn pending_map_bounded() { - let params = Params { - duration_ns: 60_000_000_000, - ..default_params() - }; - let server = silent_open_server(params); - let config = ClientConfig { - duration: Some(Duration::from_secs(60)), - max_pending_probes: 3, - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); - client.send_probe().unwrap(); - client.send_probe().unwrap(); - assert!(matches!( - client.send_probe(), - Err(ClientError::PendingLimitExceeded { limit: 3 }) - )); - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn minimal_negotiated_layout_works() { - let params = Params { - protocol_version: 1, - duration_ns: 3_000_000_000, - interval_ns: 1_000_000_000, - received_stats: ReceivedStats::None, - stamp_at: StampAt::None, - clock: Clock::Both, - ..Params::default() - }; - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - - socket - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - let mut buf = [0_u8; 2048]; - if let Ok((size, _)) = socket.recv_from(&mut buf) { - tx.send(buf[..size].to_vec()).unwrap(); - let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); - let ts = TimestampFields::default(); - let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, None); - socket.send_to(&reply_packet, peer).unwrap(); - } - }); - let config = ClientConfig { - received_stats: ReceivedStats::None, - stamp_at: StampAt::None, - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(30)); - let events = client.recv_once().unwrap(); - assert_eq!(events.len(), 1); - if let ClientEvent::EchoReply { - received_stats, - server_timing, - .. - } = &events[0] - { - assert!(received_stats.is_none()); - assert!(server_timing.is_none()); - } else { - panic!("expected EchoReply"); - } - server.join(); - } - - #[test] - fn hmac_echo_request_reply_works() { - let key = b"testkey".to_vec(); - let params = default_params(); - let server_key = key.clone(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&server_key)); - socket.send_to(&reply, peer).unwrap(); - - socket - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - let mut buf = [0_u8; 2048]; - if let Ok((size, _)) = socket.recv_from(&mut buf) { - tx.send(buf[..size].to_vec()).unwrap(); - verify_hmac(&server_key, &buf[..size], HMAC_OFFSET).unwrap(); - let seq = u32::from_le_bytes( - buf[4 + HMAC_SIZE + 8..4 + HMAC_SIZE + 12] - .try_into() - .unwrap(), - ); - let ts = TimestampFields { - recv_mono: Some(100), - send_mono: Some(200), - ..Default::default() - }; - let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, Some(&server_key)); - socket.send_to(&reply_packet, peer).unwrap(); - } - }); - let config = ClientConfig { - hmac_key: Some(key), - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(50)); - let events = client.recv_once().unwrap(); - assert_eq!(events.len(), 1); - assert!(matches!(&events[0], ClientEvent::EchoReply { .. })); - server.join(); - } - - #[test] - fn late_reply_with_pending_preserves_rtt() { - let params = default_params(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - - socket - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - let mut seqs = Vec::new(); - for _ in 0..3 { - let mut buf = [0_u8; 2048]; - if let Ok((size, _)) = socket.recv_from(&mut buf) { - tx.send(buf[..size].to_vec()).unwrap(); - let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); - seqs.push(seq); - } - } - let ts = TimestampFields::default(); - let reply2 = echo_reply_packet(TOKEN, seqs[2], ¶ms, &ts, None); - socket.send_to(&reply2, peer).unwrap(); - thread::sleep(Duration::from_millis(10)); - let reply0 = echo_reply_packet(TOKEN, seqs[0], ¶ms, &ts, None); - socket.send_to(&reply0, peer).unwrap(); - }); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); - client.send_probe().unwrap(); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(50)); - - let ev1 = client.recv_once().unwrap(); - assert!(matches!(&ev1[0], ClientEvent::EchoReply { seq: 2, .. })); - - thread::sleep(Duration::from_millis(30)); - let ev2 = client.recv_once().unwrap(); - match &ev2[0] { - ClientEvent::LateReply { - seq, rtt, sent_at, .. - } => { - assert_eq!(*seq, 0); - assert!(rtt.is_some()); - assert!(sent_at.is_some()); - } - other => panic!("expected LateReply, got {other:?}"), - } - server.join(); - } - - // ---------- Correctness cleanup regression tests ---------- - - #[test] - fn pending_full_does_not_send_packet() { - let params = Params { - duration_ns: 60_000_000_000, - ..default_params() - }; - let server = silent_open_server(params); - let config = ClientConfig { - duration: Some(Duration::from_secs(60)), - max_pending_probes: 2, - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - - client.send_probe().unwrap(); - client.send_probe().unwrap(); - - thread::sleep(Duration::from_millis(30)); - let before_count: Vec<_> = server.rx.try_iter().collect(); - let echo_before: Vec<_> = before_count - .iter() - .filter(|p| p.len() >= 4 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0) - .collect(); - assert_eq!(echo_before.len(), 2); - - assert!(matches!( - client.send_probe(), - Err(ClientError::PendingLimitExceeded { limit: 2 }) - )); - - thread::sleep(Duration::from_millis(30)); - let after: Vec<_> = server.rx.try_iter().collect(); - let echo_after: Vec<_> = after - .iter() - .filter(|p| p.len() >= 4 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0) - .collect(); - assert_eq!( - echo_after.len(), - 0, - "no packet should be sent when pending is full" - ); - - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn unmatched_future_reply_emits_warning_not_late() { - let params = default_params(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - - socket - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - let mut buf = [0_u8; 2048]; - if let Ok((size, _)) = socket.recv_from(&mut buf) { - tx.send(buf[..size].to_vec()).unwrap(); - let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); - let ts = TimestampFields::default(); - let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, None); - socket.send_to(&reply_packet, peer).unwrap(); - thread::sleep(Duration::from_millis(10)); - let future_reply = echo_reply_packet(TOKEN, 999, ¶ms, &ts, None); - socket.send_to(&future_reply, peer).unwrap(); - } - }); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(50)); - - let ev1 = client.recv_once().unwrap(); - assert!(matches!(&ev1[0], ClientEvent::EchoReply { seq: 0, .. })); - - thread::sleep(Duration::from_millis(30)); - let ev2 = client.recv_once().unwrap(); - assert_eq!(ev2.len(), 1); - assert!( - matches!(&ev2[0], ClientEvent::Warning { .. }), - "unmatched future reply should emit Warning, got {:?}", - ev2[0] - ); - server.join(); - } - - #[test] - fn unmatched_future_reply_does_not_update_highest_received_seq() { - let params = default_params(); - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - - socket - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - - let mut seqs = Vec::new(); - for _ in 0..2 { - let mut buf = [0_u8; 2048]; - if let Ok((size, _)) = socket.recv_from(&mut buf) { - tx.send(buf[..size].to_vec()).unwrap(); - let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); - seqs.push(seq); - } - } - - let ts = TimestampFields::default(); - let reply0 = echo_reply_packet(TOKEN, seqs[0], ¶ms, &ts, None); - socket.send_to(&reply0, peer).unwrap(); - thread::sleep(Duration::from_millis(10)); - let future_reply = echo_reply_packet(TOKEN, 999, ¶ms, &ts, None); - socket.send_to(&future_reply, peer).unwrap(); - thread::sleep(Duration::from_millis(10)); - let reply1 = echo_reply_packet(TOKEN, seqs[1], ¶ms, &ts, None); - socket.send_to(&reply1, peer).unwrap(); - }); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(50)); - - let ev0 = client.recv_once().unwrap(); - assert!(matches!(&ev0[0], ClientEvent::EchoReply { seq: 0, .. })); - - thread::sleep(Duration::from_millis(30)); - let ev_future = client.recv_once().unwrap(); - assert!(matches!(&ev_future[0], ClientEvent::Warning { .. })); - - assert_eq!( - client.session.as_ref().unwrap().highest_received_seq, - Some(0), - "highest_received_seq should not be updated by unmatched future reply" - ); - - thread::sleep(Duration::from_millis(30)); - let ev1 = client.recv_once().unwrap(); - assert!( - matches!(&ev1[0], ClientEvent::EchoReply { seq: 1, .. }), - "valid pending reply seq=1 should not be poisoned, got {:?}", - ev1[0] - ); - server.join(); - } - - #[test] - fn connect_rejects_zero_max_pending_probes() { - let config = ClientConfig { - max_pending_probes: 0, - ..ClientConfig::default() - }; - assert!(matches!( - Client::connect(config), - Err(ClientError::InvalidConfig { .. }) - )); - } - - #[test] - fn connect_rejects_zero_probe_timeout() { - let config = ClientConfig { - probe_timeout: Duration::ZERO, - ..ClientConfig::default() - }; - assert!(matches!( - Client::connect(config), - Err(ClientError::InvalidConfig { .. }) - )); - } - - #[test] - fn compute_one_way_returns_none_when_both_directions_fail() { - let ts = TimestampFields::default(); - let now = ClientTimestamp::now(); - let result = compute_one_way(&now, &now, &ts); - assert!(result.is_none()); - } - - #[test] - fn matched_reply_with_reversed_monotonic_time_still_emits_event() { - let params = default_params(); - let server = silent_open_server(params.clone()); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - - let base = Instant::now() + Duration::from_secs(1); - let send_ts = ClientTimestamp { - mono: base, - wall: SystemTime::now(), - }; - client.send_probe_at(send_ts).unwrap(); - - let recv_ts = ClientTimestamp { - mono: send_ts.mono - Duration::from_millis(500), - wall: send_ts.wall + Duration::from_millis(10), - }; - let reply = echo_reply_packet(TOKEN, 0, ¶ms, &TimestampFields::default(), None); - let events = client.process_received_packet(&reply, recv_ts).unwrap(); - - assert_eq!(events.len(), 1); - match &events[0] { - ClientEvent::EchoReply { rtt, .. } => { - assert_eq!(rtt.raw, Duration::ZERO); - assert_eq!(rtt.effective, Duration::ZERO); - } - other => panic!("expected EchoReply, got {other:?}"), - } - - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn recv_available_drains_burst_replies() { - let params = default_params(); - let (mut client, server) = open_client_with_echo_server(¶ms); - - client.send_probe().unwrap(); - client.send_probe().unwrap(); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(80)); - - let events = client - .recv_available(RecvBudget { max_packets: 8 }) - .unwrap(); - assert_eq!(events.len(), 3); - for (seq, event) in events.iter().enumerate() { - assert!(matches!( - event, - ClientEvent::EchoReply { - seq: actual_seq, - .. - } if *actual_seq == seq as u32 - )); - } - - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn recv_available_respects_packet_budget() { - let params = default_params(); - let (mut client, server) = open_client_with_echo_server(¶ms); - - client.send_probe().unwrap(); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(80)); - - let first = client - .recv_available(RecvBudget { max_packets: 1 }) - .unwrap(); - assert_eq!(first.len(), 1); - assert!(matches!(&first[0], ClientEvent::EchoReply { seq: 0, .. })); - - let second = client - .recv_available(RecvBudget { max_packets: 8 }) - .unwrap(); - assert_eq!(second.len(), 1); - assert!(matches!(&second[0], ClientEvent::EchoReply { seq: 1, .. })); - - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn send_probe_wraps_wire_sequence_at_u32_max() { - let params = default_params(); - let server = silent_open_server(params); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - - let session = client.session.as_mut().unwrap(); - session.next_wire_seq = u32::MAX; - session.next_logical_seq = 41; - - client.send_probe().unwrap(); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(30)); - - let packets: Vec<_> = server.rx.try_iter().collect(); - let seqs: Vec = packets - .iter() - .filter(|p| p.len() >= 16 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0) - .map(|p| u32::from_le_bytes(p[12..16].try_into().unwrap())) - .collect(); - assert_eq!(seqs, vec![u32::MAX, 0]); - - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn send_probe_after_sending_done_is_noop() { - let params = default_params(); - let server = silent_open_server(params); - let config = ClientConfig { - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - - client.session.as_mut().unwrap().sending_done = true; - let events = client.send_probe().unwrap(); - assert!(events.is_empty()); - assert_eq!(client.session.as_ref().unwrap().packets_sent, 0); - - thread::sleep(Duration::from_millis(30)); - let packets: Vec<_> = server.rx.try_iter().collect(); - let echo_count = packets - .iter() - .filter(|p| p.len() >= 4 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0) - .count(); - assert_eq!(echo_count, 0); - - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn compute_one_way_returns_available_direction_samples() { - let sent_at = ClientTimestamp { - mono: Instant::now(), - wall: SystemTime::UNIX_EPOCH + Duration::from_secs(10), - }; - let received_at = ClientTimestamp { - mono: sent_at.mono + Duration::from_millis(40), - wall: SystemTime::UNIX_EPOCH + Duration::from_secs(10) + Duration::from_millis(40), - }; - let ts = TimestampFields { - recv_wall: Some(10_000_000_000 + 15_000_000), - send_wall: Some(10_000_000_000 + 25_000_000), - ..Default::default() - }; - - let sample = compute_one_way(&sent_at, &received_at, &ts).unwrap(); - assert_eq!(sample.client_to_server, Some(Duration::from_millis(15))); - assert_eq!(sample.server_to_client, Some(Duration::from_millis(15))); - } - - #[test] - fn recv_buffer_uses_negotiated_packet_length() { - let params = Params { - protocol_version: 1, - duration_ns: 3_000_000_000, - interval_ns: 1_000_000_000, - length: 4096, - received_stats: ReceivedStats::Both, - stamp_at: StampAt::Both, - clock: Clock::Both, - ..Params::default() - }; - let server = start_fake_server(move |socket, tx| { - let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); - socket - .set_read_timeout(Some(Duration::from_secs(2))) - .unwrap(); - loop { - let mut buf = [0_u8; 8192]; - match socket.recv_from(&mut buf) { - Ok((size, _)) => { - tx.send(buf[..size].to_vec()).unwrap(); - } - Err(_) => break, - } - } - }); - let config = ClientConfig { - length: 4096, - socket_config: crate::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - let buf_size = client.recv_buffer_size(); - assert!( - buf_size >= 4096, - "recv buffer should be at least negotiated length, got {buf_size}" - ); - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } - - #[test] - fn recv_once_at_test_helper_provides_deterministic_timestamp() { - let params = default_params(); - let (mut client, server) = open_client_with_echo_server(¶ms); - client.send_probe().unwrap(); - thread::sleep(Duration::from_millis(50)); - - let fixed_ts = ClientTimestamp { - mono: Instant::now(), - wall: SystemTime::now(), - }; - let events = client.recv_once_at(fixed_ts).unwrap(); - assert_eq!(events.len(), 1); - if let ClientEvent::EchoReply { received_at, .. } = &events[0] { - assert_eq!(*received_at, fixed_ts); - } else { - panic!("expected EchoReply"); - } - client.close(ClientTimestamp::now()).unwrap(); - server.join(); - } -} +mod tests; diff --git a/crates/irtt-client/src/client/tests/config_open.rs b/crates/irtt-client/src/client/tests/config_open.rs new file mode 100644 index 0000000..fc3423f --- /dev/null +++ b/crates/irtt-client/src/client/tests/config_open.rs @@ -0,0 +1,515 @@ +use super::*; + +#[test] +fn client_config_default() { + let config = ClientConfig::default(); + assert_eq!(config.duration, Some(Duration::from_secs(3))); + assert_eq!(config.interval, Duration::from_secs(1)); + assert_eq!(config.length, 0); + assert_eq!(config.received_stats, ReceivedStats::Both); + assert_eq!(config.stamp_at, StampAt::Both); + assert_eq!(config.clock, Clock::Both); + assert_eq!(config.dscp, 0); + assert_eq!(config.hmac_key, None); + assert_eq!(config.server_fill, None); + assert_eq!(config.open_timeouts, DEFAULT_OPEN_TIMEOUTS); + assert_eq!(config.run_mode, RunMode::Normal); + assert_eq!(config.negotiation_policy, NegotiationPolicy::Strict); + assert_eq!(config.probe_timeout, Duration::from_secs(4)); + assert_eq!(config.max_pending_probes, 4096); +} + +#[test] +fn params_from_config_maps_compatibility_fields() { + let config = ClientConfig { + duration: Some(Duration::from_secs(5)), + interval: Duration::from_millis(250), + length: 1472, + received_stats: ReceivedStats::Window, + stamp_at: StampAt::Midpoint, + clock: Clock::Wall, + dscp: 46, + hmac_key: Some(b"secret".to_vec()), + server_fill: Some("rand".to_owned()), + ..ClientConfig::default() + }; + + let params = params_from_config(&config).unwrap(); + assert_eq!(params.protocol_version, PROTOCOL_VERSION); + assert_eq!(params.duration_ns, 5_000_000_000); + assert_eq!(params.interval_ns, 250_000_000); + assert_eq!(params.length, 1472); + assert_eq!(params.received_stats, ReceivedStats::Window); + assert_eq!(params.stamp_at, StampAt::Midpoint); + assert_eq!(params.clock, Clock::Wall); + assert_eq!(params.dscp, 46, "config DSCP codepoint must not be shifted"); + assert_eq!( + params.server_fill.as_ref().map(|fill| fill.value.as_str()), + Some("rand") + ); + assert_eq!(config.hmac_key.as_deref(), Some(b"secret".as_slice())); +} + +#[test] +fn params_from_config_encodes_continuous_duration_as_zero() { + let config = ClientConfig { + duration: None, + ..ClientConfig::default() + }; + assert_eq!(params_from_config(&config).unwrap().duration_ns, 0); +} + +#[test] +fn params_from_config_accepts_max_dscp_codepoint() { + let config = ClientConfig { + dscp: 63, + ..ClientConfig::default() + }; + assert_eq!(params_from_config(&config).unwrap().dscp, 63); +} + +#[test] +fn params_from_config_rejects_invalid_dscp_codepoint() { + let config = ClientConfig { + dscp: 64, + ..ClientConfig::default() + }; + assert!(matches!( + params_from_config(&config), + Err(ClientError::InvalidConfig { .. }) + )); +} + +#[test] +fn params_from_config_rejects_oversized_server_fill() { + let config = ClientConfig { + server_fill: Some("0123456789abcdef0123456789abcdefx".to_owned()), + ..ClientConfig::default() + }; + assert!(matches!( + params_from_config(&config), + Err(ClientError::InvalidConfig { .. }) + )); +} + +#[test] +fn address_resolution_connects_to_local_fake_server() { + let server = start_fake_server(|socket, tx| { + let _ = recv_request(&socket, &tx); + }); + let client = Client::connect(default_test_config(server.addr)).unwrap(); + client.socket.send(b"ping").unwrap(); + assert_eq!(server.rx.recv().unwrap(), b"ping"); + server.join(); +} + +#[test] +fn successful_open_handshake() { + let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + let params = params_from_config(&config).unwrap(); + let server = open_success_server(params.clone()); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + + let negotiated = assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + assert_eq!(negotiated.params, params); + assert!(matches!(client.phase, ClientPhase::Open { token: TOKEN })); + server.join(); +} + +#[test] +fn open_fails_when_already_open() { + let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + let params = params_from_config(&config).unwrap(); + let server = open_success_server(params); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::AlreadyOpen) + )); + server.join(); +} + +#[test] +fn open_fails_after_close() { + let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + let params = params_from_config(&config).unwrap(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.close(ClientTimestamp::now()).unwrap(); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::AlreadyClosed) + )); + server.join(); +} + +#[test] +fn open_fails_after_no_test_completed() { + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.run_mode = RunMode::NoTest; + let params = params_from_config(&config).unwrap(); + let server = no_test_server(params, 0); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert_no_test_completed(client.open(ClientTimestamp::now()).unwrap()); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::AlreadyCompleted) + )); + server.join(); +} + +#[test] +fn open_retries_after_first_timeout() { + let server = start_fake_server(|socket, tx| { + let (first, _) = recv_request(&socket, &tx); + let (_, peer) = recv_request(&socket, &tx); + let params = Params::decode(&first[4..]).unwrap(); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + assert_eq!(first[3] & FLAG_OPEN, FLAG_OPEN); + socket.send_to(&reply, peer).unwrap(); + }); + let config = ClientConfig { + open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(500)], + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + let outcome = client.open(ClientTimestamp::now()).unwrap(); + assert_open_started(outcome); + assert_eq!(server.rx.iter().take(2).count(), 2); + server.join(); +} + +#[test] +fn open_timeout_after_all_timeouts() { + let server = timeout_server(Duration::from_millis(700)); + let config = ClientConfig { + open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::OpenTimeout) + )); + assert_eq!(server.rx.iter().take(2).count(), 2); + server.join(); +} + +#[test] +fn open_restores_configured_read_timeout_after_timeout() { + let server = timeout_server(Duration::from_millis(700)); + let config = ClientConfig { + open_timeouts: vec![Duration::from_millis(200)], + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(450)), + ..crate::SocketConfig::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::OpenTimeout) + )); + + let start = std::time::Instant::now(); + let mut buf = [0_u8; 1]; + assert!(client.socket.recv(&mut buf).is_err()); + assert!(start.elapsed() >= Duration::from_millis(350)); + server.join(); +} + +#[test] +fn strict_negotiation_accepts_identical_params() { + let config = ClientConfig::default(); + let params = params_from_config(&config).unwrap(); + assert!(validate_negotiated_params(¶ms, ¶ms, NegotiationPolicy::Strict).is_ok()); +} + +#[test] +fn strict_negotiation_rejects_changed_params() { + let config = ClientConfig::default(); + let requested = params_from_config(&config).unwrap(); + let mut returned = requested.clone(); + returned.dscp = 1; + assert!(matches!( + validate_negotiated_params(&requested, &returned, NegotiationPolicy::Strict), + Err(ClientError::NegotiationRejected { .. }) + )); +} + +#[test] +fn loose_negotiation_accepts_server_restricted_params() { + let config = ClientConfig::default(); + let requested = params_from_config(&config).unwrap(); + let mut returned = requested.clone(); + returned.duration_ns /= 2; + returned.length = 0; + assert!(validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose).is_ok()); +} + +#[test] +fn protocol_version_mismatch_fails() { + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.negotiation_policy = NegotiationPolicy::Loose; + let mut params = params_from_config(&config).unwrap(); + params.protocol_version = 2; + let server = open_success_server(params); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::ProtocolVersionMismatch { received: 2, .. }) + )); + server.join(); +} + +#[test] +fn server_rejection_fails_in_normal_mode() { + let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + let params = params_from_config(&config).unwrap(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::ServerRejected) + )); + server.join(); +} + +#[test] +fn no_test_open_close_succeeds_on_open_reply_close() { + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.run_mode = RunMode::NoTest; + let params = params_from_config(&config).unwrap(); + let server = no_test_server(params.clone(), 0); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + let negotiated = assert_no_test_completed(client.open(ClientTimestamp::now()).unwrap()); + assert_eq!(negotiated.params, params); + assert_eq!(client.negotiated.as_ref(), Some(&negotiated)); + assert!(matches!( + client.close(ClientTimestamp::now()), + Err(ClientError::NotOpen) + )); + server.join(); +} + +#[test] +fn no_test_success_validates_params() { + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.run_mode = RunMode::NoTest; + let params = params_from_config(&config).unwrap(); + let server = no_test_server(params.clone(), 0); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + let negotiated = assert_no_test_completed(client.open(ClientTimestamp::now()).unwrap()); + assert_eq!(negotiated.params, params); + server.join(); +} + +#[test] +fn no_test_rejects_non_close_open_reply() { + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.run_mode = RunMode::NoTest; + let params = params_from_config(&config).unwrap(); + let server = open_success_server(params); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::UnexpectedNoTestReply) + )); + server.join(); +} + +#[test] +fn no_test_rejects_non_zero_token_with_close_reply() { + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.run_mode = RunMode::NoTest; + let params = params_from_config(&config).unwrap(); + let server = no_test_server(params, TOKEN); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::NonZeroNoTestToken { token: TOKEN }) + )); + server.join(); +} + +#[test] +fn no_test_strict_negotiation_rejects_changed_params() { + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.run_mode = RunMode::NoTest; + let mut params = params_from_config(&config).unwrap(); + params.dscp = 1; + let server = no_test_server(params, 0); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::NegotiationRejected { .. }) + )); + server.join(); +} + +#[test] +fn no_test_loose_negotiation_accepts_restricted_params() { + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.run_mode = RunMode::NoTest; + config.negotiation_policy = NegotiationPolicy::Loose; + let mut params = params_from_config(&config).unwrap(); + params.duration_ns /= 2; + let server = no_test_server(params.clone(), 0); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + let negotiated = assert_no_test_completed(client.open(ClientTimestamp::now()).unwrap()); + assert_eq!(negotiated.params, params); + server.join(); +} + +#[test] +fn close_sends_one_close_packet_with_negotiated_token() { + let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + let params = params_from_config(&config).unwrap(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + let events = client.close(ClientTimestamp::now()).unwrap(); + assert_eq!(events.len(), 1); + let packets: Vec<_> = server.rx.iter().take(2).collect(); + let close = &packets[1]; + assert_eq!(close[3], flags::FLAG_CLOSE); + assert_eq!(u64::from_le_bytes(close[4..12].try_into().unwrap()), TOKEN); + server.join(); +} + +#[test] +fn hmac_open_success() { + let key = b"secret".to_vec(); + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.hmac_key = Some(key.clone()); + let params = params_from_config(&config).unwrap(); + let server = start_fake_server(move |socket, tx| { + let (request, peer) = recv_request(&socket, &tx); + verify_hmac(&key, &request, HMAC_OFFSET).unwrap(); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&key)); + socket.send_to(&reply, peer).unwrap(); + }); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + server.join(); +} + +#[test] +fn hmac_open_rejects_missing_hmac() { + let key = b"secret".to_vec(); + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.hmac_key = Some(key); + let params = params_from_config(&config).unwrap(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + }); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::Protocol( + irtt_proto::ProtoError::HmacPresenceMismatch + )) + )); + server.join(); +} + +#[test] +fn hmac_open_rejects_bad_hmac() { + let key = b"secret".to_vec(); + let wrong_key = b"wrong".to_vec(); + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.hmac_key = Some(key); + let params = params_from_config(&config).unwrap(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&wrong_key)); + socket.send_to(&reply, peer).unwrap(); + }); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert!(matches!( + client.open(ClientTimestamp::now()), + Err(ClientError::Protocol(irtt_proto::ProtoError::BadHmac)) + )); + server.join(); +} + +#[test] +fn hmac_close_packet_includes_valid_hmac() { + let key = b"secret".to_vec(); + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.hmac_key = Some(key.clone()); + let params = params_from_config(&config).unwrap(); + let server_key = key.clone(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&server_key)); + socket.send_to(&reply, peer).unwrap(); + let _ = recv_request(&socket, &tx); + }); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.close(ClientTimestamp::now()).unwrap(); + let packets: Vec<_> = server.rx.iter().take(2).collect(); + let close = &packets[1]; + assert_eq!(close[3], flags::FLAG_CLOSE | FLAG_HMAC); + verify_hmac(&key, close, HMAC_OFFSET).unwrap(); + assert_eq!( + u64::from_le_bytes(close[4 + HMAC_SIZE..12 + HMAC_SIZE].try_into().unwrap()), + TOKEN + ); + server.join(); +} + +#[test] +fn minimum_open_timeout_under_200ms_is_rejected() { + let config = ClientConfig { + open_timeouts: vec![Duration::from_millis(199)], + ..ClientConfig::default() + }; + assert!(matches!( + Client::connect(config), + Err(ClientError::OpenTimeoutTooSmall { .. }) + )); +} + +#[test] +fn empty_open_timeouts_is_rejected() { + let config = ClientConfig { + open_timeouts: vec![], + ..ClientConfig::default() + }; + assert!(matches!( + Client::connect(config), + Err(ClientError::NoOpenTimeouts) + )); +} diff --git a/crates/irtt-client/src/client/tests/mod.rs b/crates/irtt-client/src/client/tests/mod.rs new file mode 100644 index 0000000..53964eb --- /dev/null +++ b/crates/irtt-client/src/client/tests/mod.rs @@ -0,0 +1,13 @@ +use super::*; +use crate::config::{NegotiationPolicy, DEFAULT_OPEN_TIMEOUTS}; +use irtt_proto::{ + echo_packet_len, flags::FLAG_HMAC, flags::FLAG_OPEN, flags::FLAG_REPLY, verify_hmac, Clock, + ReceivedStats, StampAt, HMAC_SIZE, MAGIC, +}; +use std::{thread, time::SystemTime}; + +mod support; +use support::*; + +mod config_open; +mod probes_replies; diff --git a/crates/irtt-client/src/client/tests/probes_replies.rs b/crates/irtt-client/src/client/tests/probes_replies.rs new file mode 100644 index 0000000..be11a8d --- /dev/null +++ b/crates/irtt-client/src/client/tests/probes_replies.rs @@ -0,0 +1,1462 @@ +use super::*; + +#[test] +fn send_probe_fails_before_open() { + let server = start_fake_server(|_socket, _tx| {}); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + assert!(matches!(client.send_probe(), Err(ClientError::NotOpen))); + server.join(); +} + +#[test] +fn send_probe_fails_after_no_test_completed() { + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.run_mode = RunMode::NoTest; + let params = params_from_config(&config).unwrap(); + let server = no_test_server(params, 0); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert_no_test_completed(client.open(ClientTimestamp::now()).unwrap()); + assert!(matches!( + client.send_probe(), + Err(ClientError::AlreadyCompleted) + )); + server.join(); +} + +#[test] +fn send_probe_fails_after_close() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + loop { + let mut buf = [0_u8; 512]; + match socket.recv_from(&mut buf) { + Ok((size, _)) => { + tx.send(buf[..size].to_vec()).unwrap(); + } + Err(_) => break, + } + } + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.close(ClientTimestamp::now()).unwrap(); + assert!(matches!( + client.send_probe(), + Err(ClientError::AlreadyClosed) + )); + server.join(); +} + +fn silent_open_server(params: Params) -> FakeServer { + start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + loop { + let mut buf = [0_u8; 2048]; + match socket.recv_from(&mut buf) { + Ok((size, _)) => { + tx.send(buf[..size].to_vec()).unwrap(); + } + Err(_) => break, + } + } + }) +} + +#[test] +fn send_probe_sends_valid_echo_request() { + let params = default_params(); + let server = silent_open_server(params.clone()); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + let events = client.send_probe().unwrap(); + assert_eq!(events.len(), 1); + match &events[0] { + ClientEvent::EchoSent { + seq, + logical_seq, + remote, + bytes, + .. + } => { + assert_eq!(*seq, 0); + assert_eq!(*logical_seq, 0); + assert_eq!(*remote, server.addr); + assert_eq!(*bytes, echo_packet_len(false, ¶ms)); + } + other => panic!("expected EchoSent, got {other:?}"), + } + thread::sleep(Duration::from_millis(30)); + let packets: Vec<_> = server.rx.try_iter().collect(); + let echo_reqs: Vec<_> = packets + .iter() + .filter(|p| p.len() >= 4 && p[3] & FLAG_OPEN == 0) + .collect(); + let echo_req = echo_reqs.first().unwrap(); + assert_eq!(&echo_req[..3], &MAGIC); + assert_eq!(echo_req[3], 0x00); + let req_token = u64::from_le_bytes(echo_req[4..12].try_into().unwrap()); + assert_eq!(req_token, TOKEN); + let seq = u32::from_le_bytes(echo_req[12..16].try_into().unwrap()); + assert_eq!(seq, 0); + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn echo_sent_reports_schedule_and_timer_error() { + let params = default_params(); + let server = silent_open_server(params); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + let start = ClientTimestamp { + mono: Instant::now(), + wall: SystemTime::now(), + }; + assert_open_started(client.open(start).unwrap()); + let session_start = client.session.as_ref().unwrap().start_mono; + assert!( + session_start >= start.mono, + "probe schedule must start after open begins" + ); + let first_probe_at = ClientTimestamp { + mono: session_start, + wall: SystemTime::now(), + }; + + let events = client.send_probe_at(first_probe_at).unwrap(); + match &events[0] { + ClientEvent::EchoSent { + scheduled_at, + sent_at, + timer_error, + .. + } => { + assert_eq!(*scheduled_at, session_start); + assert_eq!(*sent_at, first_probe_at); + assert_eq!(*timer_error, Duration::ZERO); + } + other => panic!("expected EchoSent, got {other:?}"), + } + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn send_probe_starts_seq_at_zero_and_increments() { + let params = default_params(); + let server = silent_open_server(params); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.send_probe().unwrap(); + client.send_probe().unwrap(); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(30)); + let packets: Vec<_> = server.rx.try_iter().collect(); + let echo_reqs: Vec<_> = packets + .iter() + .filter(|p| p.len() >= 4 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0) + .collect(); + assert_eq!(echo_reqs.len(), 3); + for (i, pkt) in echo_reqs.iter().enumerate() { + let seq = u32::from_le_bytes(pkt[12..16].try_into().unwrap()); + assert_eq!(seq, i as u32); + } + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn send_probe_respects_finite_duration_exclusive_end() { + let params = Params { + protocol_version: 1, + duration_ns: 1_000_000_000, + interval_ns: 500_000_000, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Params::default() + }; + let server = silent_open_server(params.clone()); + let config = ClientConfig { + duration: Some(Duration::from_secs(1)), + interval: Duration::from_millis(500), + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + + let session = client.session.as_ref().unwrap(); + let start = session.start_mono; + let interval = Duration::from_millis(500); + + let now0 = ClientTimestamp { + mono: start, + wall: SystemTime::now(), + }; + assert!(client.send_probe_at(now0).is_ok()); + + let now1 = ClientTimestamp { + mono: start + interval, + wall: SystemTime::now(), + }; + assert!(client.send_probe_at(now1).is_ok()); + + let now2 = ClientTimestamp { + mono: start + Duration::from_secs(1), + wall: SystemTime::now(), + }; + let events = client.send_probe_at(now2).unwrap(); + assert!(events.is_empty()); + assert!(client.session.as_ref().unwrap().sending_done); + assert!(client.next_send_deadline().is_none()); + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn continuous_duration_keeps_generating_send_deadlines() { + let params = Params { + protocol_version: 1, + duration_ns: 0, + interval_ns: 500_000_000, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Params::default() + }; + let server = silent_open_server(params); + let config = ClientConfig { + duration: None, + interval: Duration::from_millis(500), + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + + let start = client.session.as_ref().unwrap().start_mono; + let interval = Duration::from_millis(500); + for seq in 0..4 { + let now = ClientTimestamp { + mono: start + interval * seq, + wall: SystemTime::now(), + }; + let events = client.send_probe_at(now).unwrap(); + assert_eq!(events.len(), 1); + assert!(client.next_send_deadline().is_some()); + assert!(!client.session.as_ref().unwrap().sending_done); + } + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn next_send_deadline_advances_by_interval() { + let params = default_params(); + let (mut client, server) = open_client_with_echo_server(¶ms); + + let session = client.session.as_ref().unwrap(); + let start = session.start_mono; + let deadline0 = client.next_send_deadline().unwrap(); + assert_eq!(deadline0, start); + + client.send_probe().unwrap(); + let deadline1 = client.next_send_deadline().unwrap(); + assert_eq!(deadline1, start + Duration::from_secs(1)); + + client.send_probe().unwrap(); + let deadline2 = client.next_send_deadline().unwrap(); + assert_eq!(deadline2, start + Duration::from_secs(2)); + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn recv_once_returns_empty_on_timeout() { + let params = default_params(); + let server = open_success_server(params); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + let events = client.recv_once().unwrap(); + assert!(events.is_empty()); + server.join(); +} + +#[test] +fn recv_once_decodes_echo_reply_and_emits_event() { + let params = default_params(); + let (mut client, server) = open_client_with_echo_server(¶ms); + + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(50)); + + let events = client.recv_once().unwrap(); + assert_eq!(events.len(), 1); + match &events[0] { + ClientEvent::EchoReply { + seq, + logical_seq, + rtt, + received_stats, + server_timing, + bytes, + .. + } => { + assert_eq!(*seq, 0); + assert_eq!(*logical_seq, 0); + assert_eq!(*bytes, echo_packet_len(false, ¶ms)); + assert!(rtt.raw > Duration::ZERO); + assert_eq!(rtt.effective, rtt.adjusted.unwrap_or(rtt.raw)); + assert!(received_stats.is_some()); + let stats = received_stats.as_ref().unwrap(); + assert_eq!(stats.count, Some(42)); + assert_eq!(stats.window, Some(0x07)); + assert!(server_timing.is_some()); + let st = server_timing.as_ref().unwrap(); + assert!(st.processing.is_some()); + } + other => panic!("expected EchoReply, got {other:?}"), + } + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn echo_reply_rtt_uses_client_monotonic() { + let params = default_params(); + let (mut client, server) = open_client_with_echo_server(¶ms); + + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(20)); + + let events = client.recv_once().unwrap(); + if let ClientEvent::EchoReply { rtt, .. } = &events[0] { + assert!(rtt.raw >= Duration::from_millis(15)); + } else { + panic!("expected EchoReply"); + } + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn server_processing_subtracted_when_valid() { + let params = default_params(); + let (mut client, server) = open_client_with_echo_server(¶ms); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(20)); + let events = client.recv_once().unwrap(); + if let ClientEvent::EchoReply { + rtt, server_timing, .. + } = &events[0] + { + let st = server_timing.as_ref().unwrap(); + let processing = st.processing.unwrap(); + assert!(processing > Duration::ZERO); + if let Some(adj) = rtt.adjusted { + assert!(adj < rtt.raw); + assert_eq!(rtt.effective, adj); + } + } else { + panic!("expected EchoReply"); + } + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn server_processing_greater_than_raw_does_not_underflow() { + let base = Instant::now(); + let rtt = compute_rtt( + &ClientTimestamp { + mono: base, + wall: SystemTime::now(), + }, + &ClientTimestamp { + mono: base + Duration::from_nanos(1), + wall: SystemTime::now(), + }, + &TimestampFields { + recv_mono: Some(0), + send_mono: Some(1_000_000_000), + ..Default::default() + }, + ); + assert!(rtt.adjusted.is_none()); + assert_eq!(rtt.effective, rtt.raw); + assert_eq!( + rtt.adjusted_signed, + Some(SignedDuration { ns: -999_999_999 }) + ); + assert_eq!(rtt.effective_signed, SignedDuration { ns: -999_999_999 }); +} + +#[test] +fn received_stats_parsed_into_sample() { + let params = default_params(); + let (mut client, server) = open_client_with_echo_server(¶ms); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(30)); + let events = client.recv_once().unwrap(); + if let ClientEvent::EchoReply { received_stats, .. } = &events[0] { + let rs = received_stats.as_ref().unwrap(); + assert_eq!(rs.count, Some(42)); + assert_eq!(rs.window, Some(0x07)); + } else { + panic!("expected EchoReply"); + } + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn wrong_token_reply_is_dropped() { + let params = default_params(); + let wrong_token: u64 = 0xDEAD_BEEF_CAFE_BABE; + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut buf = [0_u8; 2048]; + if let Ok((size, _)) = socket.recv_from(&mut buf) { + tx.send(buf[..size].to_vec()).unwrap(); + let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); + let ts = TimestampFields::default(); + let reply_packet = echo_reply_packet(wrong_token, seq, ¶ms, &ts, None); + socket.send_to(&reply_packet, peer).unwrap(); + } + }); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(30)); + let events = client.recv_once().unwrap(); + assert_eq!(events.len(), 1); + assert!(matches!(&events[0], ClientEvent::Warning { .. })); + server.join(); +} + +#[test] +fn bad_hmac_reply_is_dropped() { + let key = b"secret".to_vec(); + let wrong_key = b"wrong".to_vec(); + let params = default_params(); + let server_key = key.clone(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&server_key)); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut buf = [0_u8; 2048]; + if let Ok((size, _)) = socket.recv_from(&mut buf) { + tx.send(buf[..size].to_vec()).unwrap(); + let seq = u32::from_le_bytes( + buf[4 + HMAC_SIZE + 8..4 + HMAC_SIZE + 12] + .try_into() + .unwrap(), + ); + let ts = TimestampFields::default(); + let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, Some(&wrong_key)); + socket.send_to(&reply_packet, peer).unwrap(); + } + }); + let config = ClientConfig { + hmac_key: Some(key), + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(30)); + let events = client.recv_once().unwrap(); + assert_eq!(events.len(), 1); + assert!(matches!(&events[0], ClientEvent::Warning { .. })); + server.join(); +} + +#[test] +fn duplicate_reply_emits_duplicate_event() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut buf = [0_u8; 2048]; + if let Ok((size, _)) = socket.recv_from(&mut buf) { + tx.send(buf[..size].to_vec()).unwrap(); + let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); + let ts = TimestampFields::default(); + let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, None); + socket.send_to(&reply_packet, peer).unwrap(); + thread::sleep(Duration::from_millis(10)); + socket.send_to(&reply_packet, peer).unwrap(); + } + }); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(50)); + + let events1 = client.recv_once().unwrap(); + assert_eq!(events1.len(), 1); + assert!(matches!(&events1[0], ClientEvent::EchoReply { .. })); + + thread::sleep(Duration::from_millis(30)); + let events2 = client.recv_once().unwrap(); + assert_eq!(events2.len(), 1); + assert!(matches!( + &events2[0], + ClientEvent::DuplicateReply { seq: 0, .. } + )); + server.join(); +} + +#[test] +fn out_of_order_reply_emits_late_event() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + + let mut seqs = Vec::new(); + for _ in 0..2 { + let mut buf = [0_u8; 2048]; + if let Ok((size, _)) = socket.recv_from(&mut buf) { + tx.send(buf[..size].to_vec()).unwrap(); + let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); + seqs.push(seq); + } + } + let ts = TimestampFields::default(); + let reply1 = echo_reply_packet(TOKEN, seqs[1], ¶ms, &ts, None); + socket.send_to(&reply1, peer).unwrap(); + thread::sleep(Duration::from_millis(10)); + let reply0 = echo_reply_packet(TOKEN, seqs[0], ¶ms, &ts, None); + socket.send_to(&reply0, peer).unwrap(); + }); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.send_probe().unwrap(); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(50)); + + let ev1 = client.recv_once().unwrap(); + assert_eq!(ev1.len(), 1); + assert!(matches!(&ev1[0], ClientEvent::EchoReply { seq: 1, .. })); + + thread::sleep(Duration::from_millis(30)); + let ev2 = client.recv_once().unwrap(); + assert_eq!(ev2.len(), 1); + match &ev2[0] { + ClientEvent::LateReply { + seq, + highest_seen, + rtt, + .. + } => { + assert_eq!(*seq, 0); + assert_eq!(*highest_seen, 1); + assert!(rtt.is_some()); + } + other => panic!("expected LateReply, got {other:?}"), + } + server.join(); +} + +#[test] +fn poll_timeouts_emits_echo_loss() { + let params = default_params(); + let server = silent_open_server(params); + let config = ClientConfig { + probe_timeout: Duration::from_millis(100), + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + + client.send_probe().unwrap(); + client.send_probe().unwrap(); + + let no_loss = client.poll_timeouts(ClientTimestamp::now()).unwrap(); + assert!(no_loss.is_empty()); + + thread::sleep(Duration::from_millis(150)); + let events = client.poll_timeouts(ClientTimestamp::now()).unwrap(); + assert_eq!(events.len(), 2); + for event in &events { + assert!(matches!(event, ClientEvent::EchoLoss { .. })); + } + server.join(); +} + +#[test] +fn poll_timeouts_removes_expired_pending() { + let params = default_params(); + let server = silent_open_server(params); + let config = ClientConfig { + probe_timeout: Duration::from_millis(100), + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(150)); + client.poll_timeouts(ClientTimestamp::now()).unwrap(); + + let session = client.session.as_ref().unwrap(); + assert_eq!(session.pending.len(), 0); + assert_eq!(session.timed_out.len(), 1); + server.join(); +} + +#[test] +fn late_reply_after_timeout_preserves_measurement_metadata() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + + let mut buf = [0_u8; 2048]; + let (size, _) = socket.recv_from(&mut buf).unwrap(); + tx.send(buf[..size].to_vec()).unwrap(); + let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); + + thread::sleep(Duration::from_millis(90)); + let ts = TimestampFields { + recv_wall: Some(1_000_000_000), + recv_mono: Some(100_000), + send_wall: Some(1_000_100_000), + send_mono: Some(200_000), + ..Default::default() + }; + let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, None); + socket.send_to(&reply_packet, peer).unwrap(); + socket.send_to(&reply_packet, peer).unwrap(); + }); + let config = ClientConfig { + probe_timeout: Duration::from_millis(40), + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(60)); + let losses = client.poll_timeouts(ClientTimestamp::now()).unwrap(); + assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. })); + assert_eq!(client.session.as_ref().unwrap().pending.len(), 0); + assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 1); + + let late = client.recv_once().unwrap(); + match &late[0] { + ClientEvent::LateReply { + seq, + logical_seq, + sent_at, + rtt, + server_timing, + one_way, + bytes, + .. + } => { + assert_eq!(*seq, 0); + assert_eq!(*logical_seq, Some(0)); + assert!(sent_at.is_some()); + assert!(rtt.is_some()); + assert!(server_timing.is_some()); + assert!(one_way.is_some()); + assert_eq!(*bytes, echo_packet_len(false, &default_params())); + } + other => panic!("expected stats-eligible LateReply, got {other:?}"), + } + assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 0); + + let duplicate = client.recv_once().unwrap(); + assert!(matches!( + &duplicate[0], + ClientEvent::DuplicateReply { + seq: 0, + bytes, + .. + } if *bytes == echo_packet_len(false, &default_params()) + )); + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn close_clears_timed_out_metadata() { + let params = default_params(); + let server = silent_open_server(params); + let config = ClientConfig { + probe_timeout: Duration::from_millis(40), + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(60)); + client.poll_timeouts(ClientTimestamp::now()).unwrap(); + assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 1); + + client.close(ClientTimestamp::now()).unwrap(); + assert!(client.session.is_none()); + server.join(); +} + +#[test] +fn pending_map_bounded() { + let params = Params { + duration_ns: 60_000_000_000, + ..default_params() + }; + let server = silent_open_server(params); + let config = ClientConfig { + duration: Some(Duration::from_secs(60)), + max_pending_probes: 3, + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.send_probe().unwrap(); + client.send_probe().unwrap(); + client.send_probe().unwrap(); + assert!(matches!( + client.send_probe(), + Err(ClientError::PendingLimitExceeded { limit: 3 }) + )); + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn minimal_negotiated_layout_works() { + let params = Params { + protocol_version: 1, + duration_ns: 3_000_000_000, + interval_ns: 1_000_000_000, + received_stats: ReceivedStats::None, + stamp_at: StampAt::None, + clock: Clock::Both, + ..Params::default() + }; + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut buf = [0_u8; 2048]; + if let Ok((size, _)) = socket.recv_from(&mut buf) { + tx.send(buf[..size].to_vec()).unwrap(); + let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); + let ts = TimestampFields::default(); + let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, None); + socket.send_to(&reply_packet, peer).unwrap(); + } + }); + let config = ClientConfig { + received_stats: ReceivedStats::None, + stamp_at: StampAt::None, + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(30)); + let events = client.recv_once().unwrap(); + assert_eq!(events.len(), 1); + if let ClientEvent::EchoReply { + received_stats, + server_timing, + .. + } = &events[0] + { + assert!(received_stats.is_none()); + assert!(server_timing.is_none()); + } else { + panic!("expected EchoReply"); + } + server.join(); +} + +#[test] +fn hmac_echo_request_reply_works() { + let key = b"testkey".to_vec(); + let params = default_params(); + let server_key = key.clone(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&server_key)); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut buf = [0_u8; 2048]; + if let Ok((size, _)) = socket.recv_from(&mut buf) { + tx.send(buf[..size].to_vec()).unwrap(); + verify_hmac(&server_key, &buf[..size], HMAC_OFFSET).unwrap(); + let seq = u32::from_le_bytes( + buf[4 + HMAC_SIZE + 8..4 + HMAC_SIZE + 12] + .try_into() + .unwrap(), + ); + let ts = TimestampFields { + recv_mono: Some(100), + send_mono: Some(200), + ..Default::default() + }; + let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, Some(&server_key)); + socket.send_to(&reply_packet, peer).unwrap(); + } + }); + let config = ClientConfig { + hmac_key: Some(key), + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(50)); + let events = client.recv_once().unwrap(); + assert_eq!(events.len(), 1); + assert!(matches!(&events[0], ClientEvent::EchoReply { .. })); + server.join(); +} + +#[test] +fn late_reply_with_pending_preserves_rtt() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut seqs = Vec::new(); + for _ in 0..3 { + let mut buf = [0_u8; 2048]; + if let Ok((size, _)) = socket.recv_from(&mut buf) { + tx.send(buf[..size].to_vec()).unwrap(); + let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); + seqs.push(seq); + } + } + let ts = TimestampFields::default(); + let reply2 = echo_reply_packet(TOKEN, seqs[2], ¶ms, &ts, None); + socket.send_to(&reply2, peer).unwrap(); + thread::sleep(Duration::from_millis(10)); + let reply0 = echo_reply_packet(TOKEN, seqs[0], ¶ms, &ts, None); + socket.send_to(&reply0, peer).unwrap(); + }); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.send_probe().unwrap(); + client.send_probe().unwrap(); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(50)); + + let ev1 = client.recv_once().unwrap(); + assert!(matches!(&ev1[0], ClientEvent::EchoReply { seq: 2, .. })); + + thread::sleep(Duration::from_millis(30)); + let ev2 = client.recv_once().unwrap(); + match &ev2[0] { + ClientEvent::LateReply { + seq, rtt, sent_at, .. + } => { + assert_eq!(*seq, 0); + assert!(rtt.is_some()); + assert!(sent_at.is_some()); + } + other => panic!("expected LateReply, got {other:?}"), + } + server.join(); +} + +// ---------- Regression tests ---------- + +#[test] +fn pending_full_does_not_send_packet() { + let params = Params { + duration_ns: 60_000_000_000, + ..default_params() + }; + let server = silent_open_server(params); + let config = ClientConfig { + duration: Some(Duration::from_secs(60)), + max_pending_probes: 2, + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + + client.send_probe().unwrap(); + client.send_probe().unwrap(); + + thread::sleep(Duration::from_millis(30)); + let before_count: Vec<_> = server.rx.try_iter().collect(); + let echo_before: Vec<_> = before_count + .iter() + .filter(|p| p.len() >= 4 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0) + .collect(); + assert_eq!(echo_before.len(), 2); + + assert!(matches!( + client.send_probe(), + Err(ClientError::PendingLimitExceeded { limit: 2 }) + )); + + thread::sleep(Duration::from_millis(30)); + let after: Vec<_> = server.rx.try_iter().collect(); + let echo_after: Vec<_> = after + .iter() + .filter(|p| p.len() >= 4 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0) + .collect(); + assert_eq!( + echo_after.len(), + 0, + "no packet should be sent when pending is full" + ); + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn unmatched_future_reply_emits_warning_not_late() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut buf = [0_u8; 2048]; + if let Ok((size, _)) = socket.recv_from(&mut buf) { + tx.send(buf[..size].to_vec()).unwrap(); + let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); + let ts = TimestampFields::default(); + let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, None); + socket.send_to(&reply_packet, peer).unwrap(); + thread::sleep(Duration::from_millis(10)); + let future_reply = echo_reply_packet(TOKEN, 999, ¶ms, &ts, None); + socket.send_to(&future_reply, peer).unwrap(); + } + }); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(50)); + + let ev1 = client.recv_once().unwrap(); + assert!(matches!(&ev1[0], ClientEvent::EchoReply { seq: 0, .. })); + + thread::sleep(Duration::from_millis(30)); + let ev2 = client.recv_once().unwrap(); + assert_eq!(ev2.len(), 1); + assert!( + matches!(&ev2[0], ClientEvent::Warning { .. }), + "unmatched future reply should emit Warning, got {:?}", + ev2[0] + ); + server.join(); +} + +#[test] +fn unmatched_future_reply_does_not_update_highest_received_seq() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + + let mut seqs = Vec::new(); + for _ in 0..2 { + let mut buf = [0_u8; 2048]; + if let Ok((size, _)) = socket.recv_from(&mut buf) { + tx.send(buf[..size].to_vec()).unwrap(); + let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); + seqs.push(seq); + } + } + + let ts = TimestampFields::default(); + let reply0 = echo_reply_packet(TOKEN, seqs[0], ¶ms, &ts, None); + socket.send_to(&reply0, peer).unwrap(); + thread::sleep(Duration::from_millis(10)); + let future_reply = echo_reply_packet(TOKEN, 999, ¶ms, &ts, None); + socket.send_to(&future_reply, peer).unwrap(); + thread::sleep(Duration::from_millis(10)); + let reply1 = echo_reply_packet(TOKEN, seqs[1], ¶ms, &ts, None); + socket.send_to(&reply1, peer).unwrap(); + }); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + client.send_probe().unwrap(); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(50)); + + let ev0 = client.recv_once().unwrap(); + assert!(matches!(&ev0[0], ClientEvent::EchoReply { seq: 0, .. })); + + thread::sleep(Duration::from_millis(30)); + let ev_future = client.recv_once().unwrap(); + assert!(matches!(&ev_future[0], ClientEvent::Warning { .. })); + + assert_eq!( + client.session.as_ref().unwrap().highest_received_seq, + Some(0), + "highest_received_seq should not be updated by unmatched future reply" + ); + + thread::sleep(Duration::from_millis(30)); + let ev1 = client.recv_once().unwrap(); + assert!( + matches!(&ev1[0], ClientEvent::EchoReply { seq: 1, .. }), + "valid pending reply seq=1 should not be poisoned, got {:?}", + ev1[0] + ); + server.join(); +} + +#[test] +fn connect_rejects_zero_max_pending_probes() { + let config = ClientConfig { + max_pending_probes: 0, + ..ClientConfig::default() + }; + assert!(matches!( + Client::connect(config), + Err(ClientError::InvalidConfig { .. }) + )); +} + +#[test] +fn connect_rejects_zero_probe_timeout() { + let config = ClientConfig { + probe_timeout: Duration::ZERO, + ..ClientConfig::default() + }; + assert!(matches!( + Client::connect(config), + Err(ClientError::InvalidConfig { .. }) + )); +} + +#[test] +fn compute_one_way_returns_none_when_both_directions_fail() { + let ts = TimestampFields::default(); + let now = ClientTimestamp::now(); + let result = compute_one_way(&now, &now, &ts); + assert!(result.is_none()); +} + +#[test] +fn matched_reply_with_reversed_monotonic_time_still_emits_event() { + let params = default_params(); + let server = silent_open_server(params.clone()); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + + let base = Instant::now() + Duration::from_secs(1); + let send_ts = ClientTimestamp { + mono: base, + wall: SystemTime::now(), + }; + client.send_probe_at(send_ts).unwrap(); + + let recv_ts = ClientTimestamp { + mono: send_ts.mono - Duration::from_millis(500), + wall: send_ts.wall + Duration::from_millis(10), + }; + let reply = echo_reply_packet(TOKEN, 0, ¶ms, &TimestampFields::default(), None); + let events = client.process_received_packet(&reply, recv_ts).unwrap(); + + assert_eq!(events.len(), 1); + match &events[0] { + ClientEvent::EchoReply { rtt, .. } => { + assert_eq!(rtt.raw, Duration::ZERO); + assert_eq!(rtt.effective, Duration::ZERO); + } + other => panic!("expected EchoReply, got {other:?}"), + } + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn recv_available_drains_burst_replies() { + let params = default_params(); + let (mut client, server) = open_client_with_echo_server(¶ms); + + client.send_probe().unwrap(); + client.send_probe().unwrap(); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(80)); + + let events = client + .recv_available(RecvBudget { max_packets: 8 }) + .unwrap(); + assert_eq!(events.len(), 3); + for (seq, event) in events.iter().enumerate() { + assert!(matches!( + event, + ClientEvent::EchoReply { + seq: actual_seq, + .. + } if *actual_seq == seq as u32 + )); + } + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn recv_available_respects_packet_budget() { + let params = default_params(); + let (mut client, server) = open_client_with_echo_server(¶ms); + + client.send_probe().unwrap(); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(80)); + + let first = client + .recv_available(RecvBudget { max_packets: 1 }) + .unwrap(); + assert_eq!(first.len(), 1); + assert!(matches!(&first[0], ClientEvent::EchoReply { seq: 0, .. })); + + let second = client + .recv_available(RecvBudget { max_packets: 8 }) + .unwrap(); + assert_eq!(second.len(), 1); + assert!(matches!(&second[0], ClientEvent::EchoReply { seq: 1, .. })); + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn send_probe_wraps_wire_sequence_at_u32_max() { + let params = default_params(); + let server = silent_open_server(params); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + + let session = client.session.as_mut().unwrap(); + session.next_wire_seq = u32::MAX; + session.next_logical_seq = 41; + + client.send_probe().unwrap(); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(30)); + + let packets: Vec<_> = server.rx.try_iter().collect(); + let seqs: Vec = packets + .iter() + .filter(|p| p.len() >= 16 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0) + .map(|p| u32::from_le_bytes(p[12..16].try_into().unwrap())) + .collect(); + assert_eq!(seqs, vec![u32::MAX, 0]); + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn send_probe_after_sending_done_is_noop() { + let params = default_params(); + let server = silent_open_server(params); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + + client.session.as_mut().unwrap().sending_done = true; + let events = client.send_probe().unwrap(); + assert!(events.is_empty()); + assert_eq!(client.session.as_ref().unwrap().packets_sent, 0); + + thread::sleep(Duration::from_millis(30)); + let packets: Vec<_> = server.rx.try_iter().collect(); + let echo_count = packets + .iter() + .filter(|p| p.len() >= 4 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0) + .count(); + assert_eq!(echo_count, 0); + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn compute_one_way_returns_available_direction_samples() { + let sent_at = ClientTimestamp { + mono: Instant::now(), + wall: SystemTime::UNIX_EPOCH + Duration::from_secs(10), + }; + let received_at = ClientTimestamp { + mono: sent_at.mono + Duration::from_millis(40), + wall: SystemTime::UNIX_EPOCH + Duration::from_secs(10) + Duration::from_millis(40), + }; + let ts = TimestampFields { + recv_wall: Some(10_000_000_000 + 15_000_000), + send_wall: Some(10_000_000_000 + 25_000_000), + ..Default::default() + }; + + let sample = compute_one_way(&sent_at, &received_at, &ts).unwrap(); + assert_eq!(sample.client_to_server, Some(Duration::from_millis(15))); + assert_eq!(sample.server_to_client, Some(Duration::from_millis(15))); +} + +#[test] +fn recv_buffer_uses_negotiated_packet_length() { + let params = Params { + protocol_version: 1, + duration_ns: 3_000_000_000, + interval_ns: 1_000_000_000, + length: 4096, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Params::default() + }; + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + loop { + let mut buf = [0_u8; 8192]; + match socket.recv_from(&mut buf) { + Ok((size, _)) => { + tx.send(buf[..size].to_vec()).unwrap(); + } + Err(_) => break, + } + } + }); + let config = ClientConfig { + length: 4096, + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + let buf_size = client.recv_buffer_size(); + assert!( + buf_size >= 4096, + "recv buffer should be at least negotiated length, got {buf_size}" + ); + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} + +#[test] +fn recv_once_at_test_helper_provides_deterministic_timestamp() { + let params = default_params(); + let (mut client, server) = open_client_with_echo_server(¶ms); + client.send_probe().unwrap(); + thread::sleep(Duration::from_millis(50)); + + let fixed_ts = ClientTimestamp { + mono: Instant::now(), + wall: SystemTime::now(), + }; + let events = client.recv_once_at(fixed_ts).unwrap(); + assert_eq!(events.len(), 1); + if let ClientEvent::EchoReply { received_at, .. } = &events[0] { + assert_eq!(*received_at, fixed_ts); + } else { + panic!("expected EchoReply"); + } + client.close(ClientTimestamp::now()).unwrap(); + server.join(); +} diff --git a/crates/irtt-client/src/client/tests/support.rs b/crates/irtt-client/src/client/tests/support.rs new file mode 100644 index 0000000..b5ee4d5 --- /dev/null +++ b/crates/irtt-client/src/client/tests/support.rs @@ -0,0 +1,269 @@ +use super::*; +use irtt_proto::{ + compute_hmac_in_place, echo_packet_len, flags::FLAG_HMAC, flags::FLAG_OPEN, flags::FLAG_REPLY, + layout::PacketLayout, Clock, ReceivedStats, StampAt, HMAC_SIZE, MAGIC, +}; +use std::{ + net::UdpSocket, + sync::mpsc, + thread::{self, JoinHandle}, +}; + +pub(super) const TOKEN: u64 = 0x1234_5678_90ab_cdef; +pub(super) const HMAC_OFFSET: usize = 4; + +pub(super) struct FakeServer { + pub(super) addr: SocketAddr, + pub(super) rx: mpsc::Receiver>, + done: JoinHandle<()>, +} + +impl FakeServer { + pub(super) fn join(self) { + self.done.join().unwrap(); + } +} + +pub(super) fn default_test_config(addr: SocketAddr) -> ClientConfig { + ClientConfig { + server_addr: addr.to_string(), + open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], + ..ClientConfig::default() + } +} + +pub(super) fn start_fake_server(handler: F) -> FakeServer +where + F: FnOnce(UdpSocket, mpsc::Sender>) + 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 } +} + +pub(super) fn recv_request( + socket: &UdpSocket, + tx: &mpsc::Sender>, +) -> (Vec, SocketAddr) { + let mut buf = [0_u8; 512]; + let (size, peer) = socket.recv_from(&mut buf).unwrap(); + let packet = buf[..size].to_vec(); + tx.send(packet.clone()).unwrap(); + (packet, peer) +} + +pub(super) fn open_reply( + flags: u8, + token: u64, + params: &Params, + hmac_key: Option<&[u8]>, +) -> Vec { + let mut packet = Vec::new(); + packet.extend_from_slice(&MAGIC); + packet.push(flags | hmac_key.map_or(0, |_| FLAG_HMAC)); + if hmac_key.is_some() { + packet.extend_from_slice(&[0_u8; HMAC_SIZE]); + } + packet.extend_from_slice(&token.to_le_bytes()); + packet.extend_from_slice(¶ms.encode()); + if let Some(key) = hmac_key { + compute_hmac_in_place(key, &mut packet, HMAC_OFFSET).unwrap(); + } + packet +} + +pub(super) fn echo_reply_packet( + token: u64, + seq: u32, + params: &Params, + timestamps: &TimestampFields, + hmac_key: Option<&[u8]>, +) -> Vec { + let has_hmac = hmac_key.is_some(); + let layout = PacketLayout::echo(has_hmac, params); + let packet_len = echo_packet_len(has_hmac, params); + let mut packet = Vec::with_capacity(packet_len); + + let mut flags = FLAG_REPLY; + if has_hmac { + flags |= FLAG_HMAC; + } + packet.extend_from_slice(&MAGIC); + packet.push(flags); + if has_hmac { + packet.extend_from_slice(&[0_u8; HMAC_SIZE]); + } + packet.extend_from_slice(&token.to_le_bytes()); + packet.extend_from_slice(&seq.to_le_bytes()); + + if layout.recv_count { + packet.extend_from_slice(&42_u32.to_le_bytes()); + } + if layout.recv_window { + packet.extend_from_slice(&0x07_u64.to_le_bytes()); + } + + if layout.recv_wall { + packet.extend_from_slice(×tamps.recv_wall.unwrap_or(0).to_le_bytes()); + } + if layout.recv_mono { + packet.extend_from_slice(×tamps.recv_mono.unwrap_or(0).to_le_bytes()); + } + if layout.midpoint_wall { + packet.extend_from_slice(×tamps.midpoint_wall.unwrap_or(0).to_le_bytes()); + } + if layout.midpoint_mono { + packet.extend_from_slice(×tamps.midpoint_mono.unwrap_or(0).to_le_bytes()); + } + if layout.send_wall { + packet.extend_from_slice(×tamps.send_wall.unwrap_or(0).to_le_bytes()); + } + if layout.send_mono { + packet.extend_from_slice(×tamps.send_mono.unwrap_or(0).to_le_bytes()); + } + + packet.resize(packet_len, 0); + + if let Some(key) = hmac_key { + compute_hmac_in_place(key, &mut packet, HMAC_OFFSET).unwrap(); + } + packet +} + +pub(super) fn open_success_server(params: Params) -> FakeServer { + start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + }) +} + +pub(super) fn no_test_server(params: Params, token: u64) -> FakeServer { + start_fake_server(move |socket, tx| { + let (request, peer) = recv_request(&socket, &tx); + assert_eq!(request[3] & flags::FLAG_CLOSE, flags::FLAG_CLOSE); + let reply = open_reply( + FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, + token, + ¶ms, + None, + ); + socket.send_to(&reply, peer).unwrap(); + }) +} + +pub(super) fn timeout_server(wait: Duration) -> FakeServer { + start_fake_server(move |socket, tx| { + socket.set_read_timeout(Some(wait)).unwrap(); + while recv_request_timeout(&socket, &tx).is_some() {} + }) +} + +pub(super) fn recv_request_timeout( + socket: &UdpSocket, + tx: &mpsc::Sender>, +) -> Option<(Vec, SocketAddr)> { + let mut buf = [0_u8; 512]; + match socket.recv_from(&mut buf) { + Ok((size, peer)) => { + let packet = buf[..size].to_vec(); + tx.send(packet.clone()).unwrap(); + Some((packet, peer)) + } + Err(_) => None, + } +} + +pub(super) fn echo_server(params: Params) -> FakeServer { + start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + loop { + let mut buf = [0_u8; 2048]; + let size = match socket.recv_from(&mut buf) { + Ok((size, _)) => size, + Err(_) => break, + }; + let packet = buf[..size].to_vec(); + tx.send(packet.clone()).unwrap(); + + if buf[3] & flags::FLAG_CLOSE != 0 { + break; + } + + let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap()); + let ts = TimestampFields { + recv_wall: Some(1_000_000_000), + recv_mono: Some(100_000), + send_wall: Some(1_000_100_000), + send_mono: Some(200_000), + ..Default::default() + }; + let reply_packet = echo_reply_packet(TOKEN, seq, ¶ms, &ts, None); + socket.send_to(&reply_packet, peer).unwrap(); + } + }) +} + +pub(super) fn assert_open_started(outcome: OpenOutcome) -> NegotiatedParams { + match outcome { + OpenOutcome::Started { + token, negotiated, .. + } => { + assert_eq!(token, TOKEN); + negotiated + } + OpenOutcome::NoTestCompleted { .. } => panic!("unexpected no-test outcome"), + } +} + +pub(super) fn assert_no_test_completed(outcome: OpenOutcome) -> NegotiatedParams { + match outcome { + OpenOutcome::NoTestCompleted { + negotiated, event, .. + } => { + assert!(matches!( + event, + ClientEvent::NoTestCompleted { + negotiated: ref event_params, + .. + } if *event_params == negotiated + )); + negotiated + } + OpenOutcome::Started { .. } => panic!("unexpected started outcome"), + } +} + +pub(super) fn open_client_with_echo_server(params: &Params) -> (Client, FakeServer) { + let server = echo_server(params.clone()); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open(ClientTimestamp::now()).unwrap()); + (client, server) +} + +pub(super) fn default_params() -> Params { + Params { + protocol_version: 1, + 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() + } +}