From e642f962d4c323f47ff49634e1cf82f197364070 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 11:36:25 +0700 Subject: [PATCH] fix(client): separate probe admission and send timing --- crates/irtt-client/src/client.rs | 89 +++++++---- crates/irtt-client/src/client/schedule.rs | 46 +++--- .../tests/probes_replies/probe_schedule.rs | 148 ++++++++++++++++-- crates/irtt-client/src/event.rs | 6 +- crates/irtt-client/src/managed/group.rs | 134 +++++++++++++--- crates/irtt-client/src/session/machine.rs | 132 ++++++++++++---- 6 files changed, 433 insertions(+), 122 deletions(-) diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 14fc88a..a674ad8 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -20,7 +20,7 @@ use crate::{ pub(crate) mod schedule; -use schedule::ProbeSchedule; +use schedule::{instant_abs_diff, ProbeSchedule}; #[derive(Debug, Clone, Copy)] enum ProbeScheduleMode { @@ -28,6 +28,15 @@ enum ProbeScheduleMode { Managed { scheduled_at: Instant }, } +#[cfg(test)] +#[derive(Debug, Clone, Copy)] +pub(crate) struct ProbeSendTimestamps { + pub(crate) permission_at: Instant, + pub(crate) sent_at: ClientTimestamp, + pub(crate) send_call_start: Instant, + pub(crate) send_finished_at: Instant, +} + #[cfg(test)] use crate::{ session::machine::{ @@ -55,6 +64,8 @@ pub struct Client { probe_reported_len: Option, #[cfg(test)] probe_send_error: bool, + #[cfg(test)] + probe_send_timestamps: Option, } impl Client { @@ -79,6 +90,8 @@ impl Client { probe_reported_len: None, #[cfg(test)] probe_send_error: false, + #[cfg(test)] + probe_send_timestamps: None, }) } @@ -184,7 +197,7 @@ impl Client { /// list when the run duration has elapsed and no further probe should be /// sent. pub fn send_probe(&mut self) -> Result, ClientError> { - self.send_probe_inner(None) + self.send_probe_transaction(ProbeScheduleMode::CallerPaced) } /// Receive and classify at most one datagram from the socket. @@ -309,31 +322,16 @@ impl Client { &mut self, scheduled_at: Instant, ) -> Result, ClientError> { - self.send_managed_probe_inner(scheduled_at, None) - } - - fn send_probe_inner( - &mut self, - override_ts: Option, - ) -> Result, ClientError> { - self.send_probe_transaction(override_ts, ProbeScheduleMode::CallerPaced) - } - - fn send_managed_probe_inner( - &mut self, - scheduled_at: Instant, - override_ts: Option, - ) -> Result, ClientError> { - self.send_probe_transaction(override_ts, ProbeScheduleMode::Managed { scheduled_at }) + self.send_probe_transaction(ProbeScheduleMode::Managed { scheduled_at }) } fn send_probe_transaction( &mut self, - override_ts: Option, mode: ProbeScheduleMode, ) -> Result, ClientError> { self.runtime.ensure_open()?; - let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now); + #[cfg(test)] + let test_timestamps = self.probe_send_timestamps.take(); #[cfg(test)] let fail_send = std::mem::take(&mut self.probe_send_error); let remote = self.remote; @@ -341,20 +339,26 @@ impl Client { let schedule = schedule .as_mut() .expect("open sessions always have a probe schedule"); - if !schedule.permit_probe_at(sent_at.mono) { + #[cfg(not(test))] + let permission_at = Instant::now(); + #[cfg(test)] + let permission_at = test_timestamps + .map(|timestamps| timestamps.permission_at) + .unwrap_or_else(Instant::now); + if !schedule.permit_probe_at(permission_at) { return Ok(Vec::new()); } let Some(prepared) = runtime.prepare_probe()? else { return Ok(Vec::new()); }; - let machine_commit = runtime.preflight_probe_commit(&prepared, sent_at)?; + let machine_preflight = runtime.preflight_probe_commit(&prepared)?; let schedule_commit = match mode { ProbeScheduleMode::CallerPaced => { - schedule.preflight_caller_commit(sent_at.mono, machine_commit.next_packets_sent)? + schedule.preflight_caller_commit(machine_preflight.next_packets_sent)? } ProbeScheduleMode::Managed { scheduled_at } => { - schedule.preflight_managed_commit(scheduled_at, sent_at.mono)? + schedule.preflight_managed_commit(scheduled_at, permission_at)? } }; let mut events = Vec::new(); @@ -367,21 +371,39 @@ impl Client { let expected_bytes = prepared.bytes.len(); let scheduled_at = schedule_commit.scheduled_at; - let timer_error = schedule_commit.timer_error; #[cfg(test)] let reported_bytes = self.probe_reported_len.take(); + #[cfg(not(test))] + let sent_at = ClientTimestamp::now(); + #[cfg(test)] + let sent_at = test_timestamps + .map(|timestamps| timestamps.sent_at) + .unwrap_or_else(ClientTimestamp::now); + let machine_commit = runtime.finalize_probe_commit(machine_preflight, sent_at)?; + #[cfg(not(test))] + let send_call_start = Instant::now(); + #[cfg(test)] + let send_call_start = test_timestamps + .map(|timestamps| timestamps.send_call_start) + .unwrap_or_else(Instant::now); #[cfg(test)] if fail_send { return Err(ClientError::Socket(io::Error::other( "injected probe send failure", ))); } - let send_call_start = Instant::now(); let bytes = socket.send(&prepared.bytes)?; + #[cfg(not(test))] + let send_finished_at = Instant::now(); + #[cfg(test)] + let send_finished_at = test_timestamps + .map(|timestamps| timestamps.send_finished_at) + .unwrap_or_else(Instant::now); let sent = runtime.commit_probe_sent(machine_commit, bytes); schedule.commit(schedule_commit); - let send_call = send_call_start.elapsed(); + let send_call = send_finished_at.saturating_duration_since(send_call_start); + let timer_error = instant_abs_diff(sent_at.mono, scheduled_at); #[cfg(test)] let bytes = reported_bytes.unwrap_or(bytes); validate_datagram_length(expected_bytes, bytes)?; @@ -425,16 +447,21 @@ pub(crate) fn validate_datagram_length(expected: usize, actual: usize) -> Result #[cfg(test)] impl Client { - fn send_probe_at(&mut self, ts: ClientTimestamp) -> Result, ClientError> { - self.send_probe_inner(Some(ts)) + fn send_probe_at( + &mut self, + timestamps: ProbeSendTimestamps, + ) -> Result, ClientError> { + self.probe_send_timestamps = Some(timestamps); + self.send_probe() } fn send_managed_probe_at( &mut self, scheduled_at: Instant, - ts: ClientTimestamp, + timestamps: ProbeSendTimestamps, ) -> Result, ClientError> { - self.send_managed_probe_inner(scheduled_at, Some(ts)) + self.probe_send_timestamps = Some(timestamps); + self.send_managed_probe(scheduled_at) } } diff --git a/crates/irtt-client/src/client/schedule.rs b/crates/irtt-client/src/client/schedule.rs index f300437..19fb7b3 100644 --- a/crates/irtt-client/src/client/schedule.rs +++ b/crates/irtt-client/src/client/schedule.rs @@ -14,7 +14,6 @@ pub(crate) struct ProbeSchedule { pub(crate) struct ScheduleCommit { pub(crate) scheduled_at: Instant, pub(crate) next_send_at: Option, - pub(crate) timer_error: Duration, } impl ProbeSchedule { @@ -64,7 +63,6 @@ impl ProbeSchedule { pub(crate) fn preflight_caller_commit( &self, - sent_at: Instant, next_packets_sent: u64, ) -> Result { let scheduled_at = self @@ -73,17 +71,17 @@ impl ProbeSchedule { let interval_ns = u64::try_from(self.interval.as_nanos()).map_err(|_| ClientError::DurationOverflow)?; let next_send_at = next_probe_deadline(self.start_at, interval_ns, next_packets_sent)?; - Ok(self.finish_commit(scheduled_at, next_send_at, sent_at)) + Ok(self.finish_commit(scheduled_at, next_send_at)) } pub(crate) fn preflight_managed_commit( &self, scheduled_at: Instant, - sent_at: Instant, + permission_at: Instant, ) -> Result { let (scheduled_at, next_send_at, _) = - advance_cadence(scheduled_at, self.interval, sent_at)?; - Ok(self.finish_commit(scheduled_at, next_send_at, sent_at)) + advance_cadence(scheduled_at, self.interval, permission_at)?; + Ok(self.finish_commit(scheduled_at, next_send_at)) } pub(crate) fn commit(&mut self, commit: ScheduleCommit) { @@ -94,12 +92,7 @@ impl ProbeSchedule { self.next_send_at.is_none() } - fn finish_commit( - &self, - scheduled_at: Instant, - next_send_at: Instant, - sent_at: Instant, - ) -> ScheduleCommit { + fn finish_commit(&self, scheduled_at: Instant, next_send_at: Instant) -> ScheduleCommit { ScheduleCommit { scheduled_at, next_send_at: if self.end_at.is_some_and(|end| next_send_at >= end) { @@ -107,7 +100,6 @@ impl ProbeSchedule { } else { Some(next_send_at) }, - timer_error: instant_abs_diff(sent_at, scheduled_at), } } } @@ -173,7 +165,7 @@ fn duration_from_nanos(nanos: u128) -> Result { Ok(Duration::new(seconds, subsec_nanos)) } -fn instant_abs_diff(left: Instant, right: Instant) -> Duration { +pub(crate) fn instant_abs_diff(left: Instant, right: Instant) -> Duration { left.checked_duration_since(right) .or_else(|| right.checked_duration_since(left)) .unwrap_or(Duration::ZERO) @@ -209,13 +201,11 @@ mod tests { fn caller_preflight_keeps_absolute_cadence_without_mutation() { let start = Instant::now(); let schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 0)).unwrap(); - let sent_at = start + Duration::from_millis(4); - let commit = schedule.preflight_caller_commit(sent_at, 3).unwrap(); + let commit = schedule.preflight_caller_commit(3).unwrap(); assert_eq!(commit.scheduled_at, start); assert_eq!(commit.next_send_at, Some(start + Duration::from_millis(30))); - assert_eq!(commit.timer_error, Duration::from_millis(4)); assert_eq!(schedule.next_send_deadline(), Some(start)); } @@ -224,7 +214,7 @@ mod tests { let start = Instant::now(); let mut schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 0)).unwrap(); - let commit = schedule.preflight_caller_commit(start, 1).unwrap(); + let commit = schedule.preflight_caller_commit(1).unwrap(); assert_eq!(schedule.next_send_deadline(), Some(start)); schedule.commit(commit); @@ -239,16 +229,14 @@ mod tests { let start = Instant::now(); let mut schedule = ProbeSchedule::new(start, &negotiated(10_000_000, 20_000_000)).unwrap(); - let first = schedule.preflight_caller_commit(start, 1).unwrap(); + let first = schedule.preflight_caller_commit(1).unwrap(); schedule.commit(first); assert_eq!( schedule.next_send_deadline(), Some(start + Duration::from_millis(10)) ); - let second = schedule - .preflight_caller_commit(start + Duration::from_millis(10), 2) - .unwrap(); + let second = schedule.preflight_caller_commit(2).unwrap(); schedule.commit(second); assert!(schedule.is_finished()); @@ -276,7 +264,6 @@ mod tests { assert_eq!(commit.scheduled_at, start + Duration::from_millis(40)); assert_eq!(commit.next_send_at, Some(start + Duration::from_millis(50))); - assert_eq!(commit.timer_error, Duration::from_millis(5)); assert_eq!(schedule.next_send_deadline(), Some(start)); } @@ -304,9 +291,20 @@ mod tests { let schedule = ProbeSchedule::new(start, &negotiated(2, 0)).unwrap(); assert!(matches!( - schedule.preflight_caller_commit(start, u64::MAX), + schedule.preflight_caller_commit(u64::MAX), Err(ClientError::DurationOverflow) )); assert_eq!(schedule.next_send_deadline(), Some(start)); } + + #[test] + fn timer_error_is_adapter_owned() { + let scheduled_at = Instant::now(); + let sent_at = scheduled_at + Duration::from_millis(7); + + assert_eq!( + instant_abs_diff(sent_at, scheduled_at), + Duration::from_millis(7) + ); + } } 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 b1d6815..78ec3b1 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 @@ -1,5 +1,14 @@ use super::*; +fn probe_timestamps(permission_at: Instant, sent_at: ClientTimestamp) -> ProbeSendTimestamps { + ProbeSendTimestamps { + permission_at, + sent_at, + send_call_start: sent_at.mono, + send_finished_at: sent_at.mono, + } +} + #[test] fn send_probe_fails_before_open() { let server = start_fake_server(|_socket, _tx| {}); @@ -103,7 +112,40 @@ fn blocking_failed_probe_send_preserves_machine_and_schedule() { } #[test] -fn blocking_probe_uses_one_timestamp_for_permission_and_schedule_commit() { +fn blocking_timeout_finalization_overflow_does_not_send_or_commit() { + let params = default_params(); + let server = silent_open_server(params); + let config = ClientConfig { + probe_timeout: Duration::MAX, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open().unwrap()); + + let initial_deadline = client.next_send_deadline().unwrap(); + let sent_at = ClientTimestamp { + mono: Instant::now(), + wall: SystemTime::now(), + }; + + assert!(matches!( + client.send_probe_at(probe_timestamps(initial_deadline, sent_at)), + Err(ClientError::DurationOverflow) + )); + assert_eq!(client.runtime.packets_sent(), 0); + assert_eq!(client.next_send_deadline(), Some(initial_deadline)); + thread::sleep(Duration::from_millis(30)); + assert!(server + .rx + .try_iter() + .all(|packet| packet.len() >= 4 && packet[3] & FLAG_OPEN != 0)); + + client.close().unwrap(); + server.join(); +} + +#[test] +fn blocking_probe_separates_permission_committed_send_and_send_call_timing() { let duration = Duration::from_secs(1); let interval = Duration::from_millis(100); let params = Params { @@ -133,26 +175,83 @@ fn blocking_probe_uses_one_timestamp_for_permission_and_schedule_commit() { ) .unwrap(), ); + let permission_at = opened_mono + Duration::from_millis(500); let sent_at = ClientTimestamp { wall: SystemTime::now(), - mono: opened_mono + Duration::from_millis(500), + mono: opened_mono + Duration::from_millis(525), + }; + let timestamps = ProbeSendTimestamps { + permission_at, + sent_at, + send_call_start: opened_mono + Duration::from_millis(526), + send_finished_at: opened_mono + Duration::from_millis(533), }; - let events = client.send_probe_at(sent_at).unwrap(); + let events = client.send_probe_at(timestamps).unwrap(); assert!(matches!( events.as_slice(), [ClientEvent::EchoSent { scheduled_at, sent_at: event_sent_at, + send_call, timer_error, .. }] if *scheduled_at == opened_mono && *event_sent_at == sent_at - && *timer_error == Duration::from_millis(500) + && *send_call == Duration::from_millis(7) + && *timer_error == Duration::from_millis(525) )); assert_eq!(client.next_send_deadline(), Some(opened_mono + interval)); assert_eq!(client.runtime.packets_sent(), 1); + let timeout_at = sent_at.mono + client.probe_timeout(); + assert!(client + .poll_timeouts_at(timeout_at - Duration::from_nanos(1)) + .unwrap() + .is_empty()); + assert!(matches!( + client.poll_timeouts_at(timeout_at).unwrap().as_slice(), + [ClientEvent::EchoLoss { + sent_at: loss_sent_at, + .. + }] if *loss_sent_at == sent_at + )); + + client.close().unwrap(); + server.join(); +} + +#[test] +fn blocking_send_call_excludes_wrapped_history_cleanup() { + let params = default_params(); + let server = silent_open_server(params); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + assert_open_started(client.open().unwrap()); + + let scheduled_at = client.next_send_deadline().unwrap(); + let sent_at = ClientTimestamp { + wall: SystemTime::now(), + mono: scheduled_at + Duration::from_millis(5), + }; + client.runtime.seed_wrapped_probe_history_for_test(sent_at); + let timestamps = ProbeSendTimestamps { + permission_at: scheduled_at, + sent_at, + send_call_start: scheduled_at + Duration::from_millis(6), + send_finished_at: scheduled_at + Duration::from_millis(13), + }; + + let events = client.send_probe_at(timestamps).unwrap(); + + assert!(matches!( + events.as_slice(), + [ClientEvent::EchoSent { + seq: 0, + send_call, + .. + }] if *send_call == Duration::from_millis(7) + )); + assert!(!client.runtime.has_timed_out_metadata()); client.close().unwrap(); server.join(); @@ -189,19 +288,25 @@ fn send_probe_respects_finite_duration_exclusive_end() { mono: start, wall: SystemTime::now(), }; - assert!(client.send_probe_at(now0).is_ok()); + assert!(client + .send_probe_at(probe_timestamps(now0.mono, now0)) + .is_ok()); let now1 = ClientTimestamp { mono: start + interval, wall: SystemTime::now(), }; - assert!(client.send_probe_at(now1).is_ok()); + assert!(client + .send_probe_at(probe_timestamps(now1.mono, now1)) + .is_ok()); let now2 = ClientTimestamp { mono: start + Duration::from_secs(1), wall: SystemTime::now(), }; - let events = client.send_probe_at(now2).unwrap(); + let events = client + .send_probe_at(probe_timestamps(now2.mono, now2)) + .unwrap(); assert!(events.is_empty()); assert!(client.next_send_deadline().is_none()); @@ -231,12 +336,19 @@ fn managed_probe_skips_missed_schedule_slots() { assert_open_started(client.open().unwrap()); let first_deadline = client.next_send_deadline().unwrap(); + let permission_at = first_deadline + Duration::from_millis(45); let delayed_send = ClientTimestamp { - mono: first_deadline + Duration::from_millis(45), + mono: first_deadline + Duration::from_millis(48), wall: SystemTime::now(), }; + let timestamps = ProbeSendTimestamps { + permission_at, + sent_at: delayed_send, + send_call_start: first_deadline + Duration::from_millis(49), + send_finished_at: first_deadline + Duration::from_millis(55), + }; let events = client - .send_managed_probe_at(first_deadline, delayed_send) + .send_managed_probe_at(first_deadline, timestamps) .unwrap(); assert_eq!(events.len(), 1); @@ -246,9 +358,13 @@ fn managed_probe_skips_missed_schedule_slots() { seq: 0, scheduled_at, sent_at, + send_call, + timer_error, .. } if scheduled_at == first_deadline + Duration::from_millis(40) && sent_at == delayed_send + && send_call == Duration::from_millis(6) + && timer_error == Duration::from_millis(8) )); assert_eq!( client.next_send_deadline(), @@ -287,15 +403,19 @@ fn managed_probe_skip_preserves_finite_run_end() { wall: SystemTime::now(), }; let first = client - .send_managed_probe_at(first_deadline, delayed_send) + .send_managed_probe_at( + first_deadline, + probe_timestamps(delayed_send.mono, delayed_send), + ) .unwrap(); + let second_sent_at = ClientTimestamp { + mono: first_deadline + Duration::from_millis(46), + wall: SystemTime::now(), + }; let second = client .send_managed_probe_at( first_deadline + duration, - ClientTimestamp { - mono: first_deadline + Duration::from_millis(46), - wall: SystemTime::now(), - }, + probe_timestamps(second_sent_at.mono, second_sent_at), ) .unwrap(); diff --git a/crates/irtt-client/src/event.rs b/crates/irtt-client/src/event.rs index 2bb6803..7d2dd13 100644 --- a/crates/irtt-client/src/event.rs +++ b/crates/irtt-client/src/event.rs @@ -94,11 +94,13 @@ pub enum ClientEvent { remote: SocketAddr, /// Monotonic deadline at which the probe was scheduled to be sent. scheduled_at: Instant, - /// Client wall/monotonic timestamp captured for the send. + /// Client wall/monotonic timestamp captured after probe preparation, + /// immediately before timeout finalization and the socket send. sent_at: ClientTimestamp, /// Number of bytes passed to the UDP socket. bytes: usize, - /// Elapsed time spent in the socket send call. + /// Elapsed time spent in the socket send call, excluding protocol and + /// scheduler commit work. send_call: Duration, /// Absolute difference between the scheduled send time and actual send /// timestamp. diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index 7a24a7c..012a249 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -14,7 +14,7 @@ use std::{ use crate::{ client::{ echo_sent_event, - schedule::{advance_cadence, ProbeSchedule}, + schedule::{advance_cadence, instant_abs_diff, ProbeSchedule}, validate_datagram_length, }, config::{ClientAuthConfig, ClientConfig}, @@ -28,6 +28,9 @@ use crate::{ timing::ClientTimestamp, }; +#[cfg(test)] +use crate::client::ProbeSendTimestamps; + use super::{ cancellation::CancellationToken, hub::{EventHub, EventSubscription, SubscriberConfig}, @@ -641,6 +644,8 @@ struct TargetState { probe_reported_len: Option, #[cfg(test)] probe_send_error: bool, + #[cfg(test)] + probe_send_timestamps: Option, } type ScheduledTarget = (Arc>, Instant); @@ -739,6 +744,8 @@ impl TargetState { probe_reported_len: None, #[cfg(test)] probe_send_error: false, + #[cfg(test)] + probe_send_timestamps: None, }) } @@ -1344,18 +1351,30 @@ fn send_echo_to_locked_target( target: &mut TargetState, scheduled_at: Instant, ) -> Result, ClientError> { - send_echo_to_locked_target_at(socket, target, scheduled_at, ClientTimestamp::now()) + send_echo_to_locked_target_inner(socket, target, scheduled_at) } +#[cfg(test)] fn send_echo_to_locked_target_at( socket: &UdpSocket, target: &mut TargetState, scheduled_at: Instant, - sent_at: ClientTimestamp, + timestamps: ProbeSendTimestamps, +) -> Result, ClientError> { + target.probe_send_timestamps = Some(timestamps); + send_echo_to_locked_target_inner(socket, target, scheduled_at) +} + +fn send_echo_to_locked_target_inner( + socket: &UdpSocket, + target: &mut TargetState, + scheduled_at: Instant, ) -> Result, ClientError> { target.runtime.ensure_open()?; let remote = target.remote; #[cfg(test)] + let test_timestamps = target.probe_send_timestamps.take(); + #[cfg(test)] let reported_bytes = target.probe_reported_len.take(); #[cfg(test)] let fail_send = std::mem::take(&mut target.probe_send_error); @@ -1365,15 +1384,21 @@ fn send_echo_to_locked_target_at( let schedule = schedule .as_mut() .expect("active targets always have a probe schedule"); - if !schedule.permit_probe_at(sent_at.mono) { + #[cfg(not(test))] + let permission_at = Instant::now(); + #[cfg(test)] + let permission_at = test_timestamps + .map(|timestamps| timestamps.permission_at) + .unwrap_or_else(Instant::now); + if !schedule.permit_probe_at(permission_at) { return Ok(Vec::new()); } let Some(prepared) = runtime.prepare_probe()? else { return Ok(Vec::new()); }; - let machine_commit = runtime.preflight_probe_commit(&prepared, sent_at)?; - let schedule_commit = schedule.preflight_managed_commit(scheduled_at, sent_at.mono)?; + let machine_preflight = runtime.preflight_probe_commit(&prepared)?; + let schedule_commit = schedule.preflight_managed_commit(scheduled_at, permission_at)?; let mut events = Vec::new(); events .try_reserve(1) @@ -1384,19 +1409,37 @@ fn send_echo_to_locked_target_at( let expected_bytes = prepared.bytes.len(); let event_scheduled_at = schedule_commit.scheduled_at; - let timer_error = schedule_commit.timer_error; + #[cfg(not(test))] + let sent_at = ClientTimestamp::now(); + #[cfg(test)] + let sent_at = test_timestamps + .map(|timestamps| timestamps.sent_at) + .unwrap_or_else(ClientTimestamp::now); + let machine_commit = runtime.finalize_probe_commit(machine_preflight, sent_at)?; + #[cfg(not(test))] + let send_call_start = Instant::now(); + #[cfg(test)] + let send_call_start = test_timestamps + .map(|timestamps| timestamps.send_call_start) + .unwrap_or_else(Instant::now); #[cfg(test)] if fail_send { return Err(ClientError::Socket(std::io::Error::other( "injected managed probe send failure", ))); } - let send_call_start = Instant::now(); let bytes = socket.send_to(&prepared.bytes, remote)?; + #[cfg(not(test))] + let send_finished_at = Instant::now(); + #[cfg(test)] + let send_finished_at = test_timestamps + .map(|timestamps| timestamps.send_finished_at) + .unwrap_or_else(Instant::now); let sent = runtime.commit_probe_sent(machine_commit, bytes); schedule.commit(schedule_commit); - let send_call = send_call_start.elapsed(); + let send_call = send_finished_at.saturating_duration_since(send_call_start); + let timer_error = instant_abs_diff(sent_at.mono, event_scheduled_at); #[cfg(test)] let bytes = reported_bytes.unwrap_or(bytes); validate_datagram_length(expected_bytes, bytes)?; @@ -1812,6 +1855,15 @@ mod tests { const TOKEN: u64 = 0x1234_5678_90ab_cdef; + fn probe_timestamps(permission_at: Instant, sent_at: ClientTimestamp) -> ProbeSendTimestamps { + ProbeSendTimestamps { + permission_at, + sent_at, + send_call_start: sent_at.mono, + send_finished_at: sent_at.mono, + } + } + #[derive(Debug)] struct FakeServer { addr: SocketAddr, @@ -1922,14 +1974,17 @@ mod tests { let schedule = schedule.as_mut().unwrap(); assert!(schedule.permit_probe_at(sent_at.mono)); let prepared = runtime.prepare_probe().unwrap().unwrap(); - let machine_commit = runtime.preflight_probe_commit(&prepared, sent_at).unwrap(); + let machine_preflight = runtime.preflight_probe_commit(&prepared).unwrap(); let schedule_commit = schedule .preflight_managed_commit(scheduled_at, sent_at.mono) .unwrap(); let event_scheduled_at = schedule_commit.scheduled_at; - let timer_error = schedule_commit.timer_error; + let machine_commit = runtime + .finalize_probe_commit(machine_preflight, sent_at) + .unwrap(); let sent = runtime.commit_probe_sent(machine_commit, prepared.bytes.len()); schedule.commit(schedule_commit); + let timer_error = instant_abs_diff(sent_at.mono, event_scheduled_at); vec![echo_sent_event( remote, sent, @@ -2975,7 +3030,7 @@ mod tests { } #[test] - fn group_probe_uses_one_timestamp_for_permission_and_schedule_commit() { + fn group_probe_separates_permission_committed_send_and_send_call_timing() { let duration = Duration::from_secs(1); let interval = Duration::from_millis(100); let params = test_params(Some(duration), interval); @@ -2990,25 +3045,34 @@ mod tests { mono: opened_mono, }; let mut target = active_probe_target(&config, ¶ms, remote, opened_at); + let permission_at = opened_mono + Duration::from_millis(500); let sent_at = ClientTimestamp { wall: std::time::SystemTime::now(), - mono: opened_mono + Duration::from_millis(500), + mono: opened_mono + Duration::from_millis(525), + }; + let timestamps = ProbeSendTimestamps { + permission_at, + sent_at, + send_call_start: opened_mono + Duration::from_millis(526), + send_finished_at: opened_mono + Duration::from_millis(533), }; let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); let events = - send_echo_to_locked_target_at(&socket, &mut target, opened_mono, sent_at).unwrap(); + send_echo_to_locked_target_at(&socket, &mut target, opened_mono, timestamps).unwrap(); assert!(matches!( events.as_slice(), [ClientEvent::EchoSent { scheduled_at, sent_at: event_sent_at, + send_call, timer_error, .. - }] if *scheduled_at == sent_at.mono + }] if *scheduled_at == permission_at && *event_sent_at == sent_at - && *timer_error == Duration::ZERO + && *send_call == Duration::from_millis(7) + && *timer_error == Duration::from_millis(25) )); assert_eq!( target @@ -3018,6 +3082,19 @@ mod tests { Some(opened_mono + Duration::from_millis(600)) ); assert_eq!(target.runtime.packets_sent(), 1); + let timeout_at = sent_at.mono + target.runtime.probe_timeout(); + assert!(target + .runtime + .poll_timeouts_at(timeout_at - Duration::from_nanos(1)) + .unwrap() + .is_empty()); + assert!(matches!( + target.runtime.poll_timeouts_at(timeout_at).unwrap().as_slice(), + [ClientEvent::EchoLoss { + sent_at: loss_sent_at, + .. + }] if *loss_sent_at == sent_at + )); } #[test] @@ -3044,7 +3121,12 @@ mod tests { target.probe_send_error = true; assert!(matches!( - send_echo_to_locked_target_at(&socket, &mut target, opened_mono, sent_at), + send_echo_to_locked_target_at( + &socket, + &mut target, + opened_mono, + probe_timestamps(sent_at.mono, sent_at), + ), Err(ClientError::Socket(_)) )); assert_eq!(target.runtime.packets_sent(), 0); @@ -3057,9 +3139,14 @@ mod tests { ); assert!(matches!( - send_echo_to_locked_target_at(&socket, &mut target, opened_mono, sent_at) - .unwrap() - .as_slice(), + send_echo_to_locked_target_at( + &socket, + &mut target, + opened_mono, + probe_timestamps(sent_at.mono, sent_at), + ) + .unwrap() + .as_slice(), [ClientEvent::EchoSent { seq: 0, .. }] )); } @@ -3089,7 +3176,12 @@ mod tests { let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); assert!(matches!( - send_echo_to_locked_target_at(&socket, &mut target, opened_mono, sent_at), + send_echo_to_locked_target_at( + &socket, + &mut target, + opened_mono, + probe_timestamps(sent_at.mono, sent_at), + ), Err(ClientError::DatagramLengthMismatch { expected: error_expected, actual, diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index e60fa46..2c53753 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -67,11 +67,18 @@ pub(crate) struct PreparedProbe { pub(crate) seq: u32, } +#[derive(Debug)] +pub(crate) struct ProbeCommitPreflight { + seq: u32, + next_wire_seq: u32, + pub(crate) next_packets_sent: u64, +} + #[derive(Debug)] pub(crate) struct ProbeCommit { pending: PendingProbe, next_wire_seq: u32, - pub(crate) next_packets_sent: u64, + next_packets_sent: u64, } #[derive(Debug, Clone, Copy)] @@ -193,9 +200,7 @@ impl SessionMachine { pub(crate) fn preflight_probe_commit( &mut self, prepared: &PreparedProbe, - sent_at: ClientTimestamp, - ) -> Result { - let probe_timeout = self.config.probe_timeout; + ) -> Result { let session = self.open_session_mut()?; if prepared.seq != session.next_wire_seq { return Err(ClientError::StalePreparedProbe { @@ -211,19 +216,31 @@ impl SessionMachine { .ok_or(ClientError::CounterOverflow { counter: "packets_sent", })?; + Ok(ProbeCommitPreflight { + seq: prepared.seq, + next_wire_seq: prepared.seq.wrapping_add(1), + next_packets_sent, + }) + } + + pub(crate) fn finalize_probe_commit( + &self, + preflight: ProbeCommitPreflight, + sent_at: ClientTimestamp, + ) -> Result { let timeout_at = sent_at .mono - .checked_add(probe_timeout) + .checked_add(self.config.probe_timeout) .ok_or(ClientError::DurationOverflow)?; Ok(ProbeCommit { pending: PendingProbe { - wire_seq: prepared.seq, + wire_seq: preflight.seq, sent_at, timeout_at, }, - next_wire_seq: prepared.seq.wrapping_add(1), - next_packets_sent, + next_wire_seq: preflight.next_wire_seq, + next_packets_sent: preflight.next_packets_sent, }) } @@ -812,6 +829,22 @@ fn config_duration_to_ns(field: &str, duration: Duration) -> Result