diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 2a8ac09..201a408 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -664,26 +664,6 @@ impl Client { fn send_probe_at(&mut self, ts: ClientTimestamp) -> Result, ClientError> { self.send_probe_inner(Some(ts)) } - - fn process_received_packet( - &mut self, - packet: &[u8], - received_at: ClientTimestamp, - meta: ReceiveMeta, - ) -> Result, ClientError> { - let Some(reply) = self.decode_received_packet(packet) else { - return Ok(vec![ClientEvent::Warning { - kind: WarningKind::MalformedOrUnrelatedPacket, - message: "dropped malformed or unrelated packet".to_owned(), - at: received_at, - }]); - }; - self.process_echo_reply(reply, packet.len(), received_at, meta) - } - - fn recv_buffer_size(&self) -> usize { - self.recv_buffer.len() - } } fn update_highest_received(highest_received_seq: &mut Option, wire_seq: u32) { diff --git a/crates/irtt-client/src/client/tests/probes_replies/packet_length.rs b/crates/irtt-client/src/client/tests/probes_replies/packet_length.rs index 82f4952..b577ce9 100644 --- a/crates/irtt-client/src/client/tests/probes_replies/packet_length.rs +++ b/crates/irtt-client/src/client/tests/probes_replies/packet_length.rs @@ -65,7 +65,20 @@ fn short_echo_reply_does_not_emit_echo_reply() { length: 64, ..default_params() }; - let server = silent_open_server(params.clone()); + let server = start_fake_server({ + let params = params.clone(); + 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 (request, _) = recv_request(&socket, &tx); + let seq = u32::from_le_bytes(request[12..16].try_into().unwrap()); + let mut reply = + echo_reply_packet(TOKEN, seq, ¶ms, &TimestampFields::default(), None); + reply.truncate(echo_packet_len(false, ¶ms) - 1); + socket.send_to(&reply, peer).unwrap(); + } + }); let config = ClientConfig { length: 64, socket_config: crate::SocketConfig { @@ -78,43 +91,7 @@ fn short_echo_reply_does_not_emit_echo_reply() { assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); - let mut packet = echo_reply_packet(TOKEN, 0, ¶ms, &TimestampFields::default(), None); - packet.truncate(echo_packet_len(false, ¶ms) - 1); - let events = client - .process_received_packet(&packet, ClientTimestamp::now(), ReceiveMeta::default()) - .unwrap(); - - assert!(matches!( - events.as_slice(), - [ClientEvent::Warning { - kind: WarningKind::MalformedOrUnrelatedPacket, - .. - }] - )); - client.close().unwrap(); - server.join(); -} - -#[test] -fn overlong_echo_reply_does_not_emit_echo_reply() { - 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().unwrap()); - client.send_probe().unwrap(); - - let mut packet = echo_reply_packet(TOKEN, 0, ¶ms, &TimestampFields::default(), None); - packet.push(0); - let events = client - .process_received_packet(&packet, ClientTimestamp::now(), ReceiveMeta::default()) - .unwrap(); + let events = client.recv_once().unwrap(); assert!(matches!( events.as_slice(), @@ -158,14 +135,6 @@ fn overlong_datagram_detection_uses_extra_receive_byte() { }; let mut client = Client::connect(config).unwrap(); assert_open_started(client.open().unwrap()); - assert_eq!( - client.recv_buffer_size(), - echo_packet_len(false, ¶ms) + 1 - ); - assert_eq!( - client.recv_buffer.len(), - echo_packet_len(false, ¶ms) + 1 - ); client.send_probe().unwrap(); let events = client.recv_once().unwrap(); diff --git a/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs b/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs index dfc9be9..5ee1e98 100644 --- a/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs +++ b/crates/irtt-client/src/client/tests/probes_replies/probe_schedule.rs @@ -50,83 +50,6 @@ fn send_probe_sends_valid_echo_request() { 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().unwrap()); - let session_start = client.next_send_deadline().unwrap(); - 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().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().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().unwrap(); - server.join(); -} - #[test] fn send_probe_respects_finite_duration_exclusive_end() { let params = Params { @@ -179,84 +102,20 @@ fn send_probe_respects_finite_duration_exclusive_end() { } #[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() +fn connect_rejects_invalid_probe_limits() { + for config in [ + ClientConfig { + max_pending_probes: 0, + ..ClientConfig::default() }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open().unwrap()); - - let start = client.next_send_deadline().unwrap(); - 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()); + ClientConfig { + probe_timeout: Duration::ZERO, + ..ClientConfig::default() + }, + ] { + assert!(matches!( + Client::connect(config), + Err(ClientError::InvalidConfig { .. }) + )); } - - client.close().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 deadline0 = client.next_send_deadline().unwrap(); - - client.send_probe().unwrap(); - let deadline1 = client.next_send_deadline().unwrap(); - assert_eq!(deadline1, deadline0 + Duration::from_secs(1)); - - client.send_probe().unwrap(); - let deadline2 = client.next_send_deadline().unwrap(); - assert_eq!(deadline2, deadline0 + Duration::from_secs(2)); - - client.close().unwrap(); - 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 { .. }) - )); } diff --git a/crates/irtt-client/src/client/tests/probes_replies/reply_classification.rs b/crates/irtt-client/src/client/tests/probes_replies/reply_classification.rs index b8221d1..198ed77 100644 --- a/crates/irtt-client/src/client/tests/probes_replies/reply_classification.rs +++ b/crates/irtt-client/src/client/tests/probes_replies/reply_classification.rs @@ -1,43 +1,5 @@ use super::*; -#[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().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(); @@ -149,114 +111,3 @@ fn out_of_order_reply_emits_late_event() { } 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().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().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 { .. })); - - 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(); -} diff --git a/crates/irtt-client/src/client/tests/probes_replies/reply_metadata.rs b/crates/irtt-client/src/client/tests/probes_replies/reply_metadata.rs index 21293f2..8edcf53 100644 --- a/crates/irtt-client/src/client/tests/probes_replies/reply_metadata.rs +++ b/crates/irtt-client/src/client/tests/probes_replies/reply_metadata.rs @@ -83,49 +83,3 @@ fn echo_reply_metadata_propagates_kernel_rx_timestamp_with_ancillary() { client.close().unwrap(); server.join(); } - -#[test] -fn process_received_packet_uses_supplied_receive_metadata() { - 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().unwrap()); - - client.send_probe_at(ClientTimestamp::now()).unwrap(); - let recv_ts = ClientTimestamp::now(); - let reply = echo_reply_packet(TOKEN, 0, ¶ms, &TimestampFields::default(), None); - let events = client - .process_received_packet( - &reply, - recv_ts, - ReceiveMeta { - traffic_class: Some(186), - kernel_rx_timestamp: Some(SystemTime::UNIX_EPOCH + Duration::new(12, 34)), - }, - ) - .unwrap(); - - assert_eq!(events.len(), 1); - match &events[0] { - ClientEvent::EchoReply { packet_meta, .. } => { - assert_eq!(packet_meta.traffic_class, Some(186)); - assert_eq!(packet_meta.dscp, Some(46)); - assert_eq!(packet_meta.ecn, Some(2)); - assert_eq!( - packet_meta.kernel_rx_timestamp, - Some(SystemTime::UNIX_EPOCH + Duration::new(12, 34)) - ); - } - other => panic!("expected EchoReply, got {other:?}"), - } - - client.close().unwrap(); - server.join(); -} diff --git a/crates/irtt-client/src/client/tests/probes_replies/timing_math.rs b/crates/irtt-client/src/client/tests/probes_replies/timing_math.rs index 1ef5d96..cc7ec6c 100644 --- a/crates/irtt-client/src/client/tests/probes_replies/timing_math.rs +++ b/crates/irtt-client/src/client/tests/probes_replies/timing_math.rs @@ -1,48 +1,5 @@ use super::*; -#[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().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().unwrap(); - server.join(); -} - #[test] fn server_processing_greater_than_raw_does_not_underflow() { let base = Instant::now(); @@ -70,14 +27,6 @@ fn server_processing_greater_than_raw_does_not_underflow() { assert_eq!(rtt.effective_signed, SignedDuration { ns: -999_999_999 }); } -#[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 unix_epoch_ns_i64_rejects_out_of_range_wall_time() { let i64_max_ns = u64::try_from(i64::MAX).unwrap(); @@ -110,49 +59,6 @@ fn compute_one_way_omits_samples_when_client_wall_time_overflows_i64() { assert_eq!(compute_one_way(&sent_at, &received_at, &ts), 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().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, ReceiveMeta::default()) - .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().unwrap(); - server.join(); -} - #[test] fn compute_one_way_returns_available_direction_samples() { let sent_at = ClientTimestamp {