diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index cdb8a6e..14fc88a 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -12,7 +12,7 @@ use crate::{ error::ClientError, event::{ClientEvent, OpenOutcome}, receive::recv_datagram, - session::machine::{recv_buffer_size, SendProbeResult, SessionMachine, MAX_OPEN_PACKET_SIZE}, + session::machine::{recv_buffer_size, ProbeSent, SessionMachine, MAX_OPEN_PACKET_SIZE}, socket::{connect_udp_socket, resolve_remote, validate_open_timeouts}, socket_options::{apply_dscp_to_socket, clear_dscp_on_socket}, timing::ClientTimestamp, @@ -22,6 +22,12 @@ pub(crate) mod schedule; use schedule::ProbeSchedule; +#[derive(Debug, Clone, Copy)] +enum ProbeScheduleMode { + CallerPaced, + Managed { scheduled_at: Instant }, +} + #[cfg(test)] use crate::{ session::machine::{ @@ -45,6 +51,10 @@ pub struct Client { socket: UdpSocket, remote: SocketAddr, recv_buffer: Vec, + #[cfg(test)] + probe_reported_len: Option, + #[cfg(test)] + probe_send_error: bool, } impl Client { @@ -65,6 +75,10 @@ impl Client { socket, remote, recv_buffer: vec![0_u8; recv_buffer_size(false, None)?], + #[cfg(test)] + probe_reported_len: None, + #[cfg(test)] + probe_send_error: false, }) } @@ -302,24 +316,7 @@ impl Client { &mut self, override_ts: Option, ) -> Result, ClientError> { - let socket = &self.socket; - self.runtime.ensure_open()?; - let schedule = self - .schedule - .as_mut() - .expect("open sessions always have a probe schedule"); - self.runtime - .send_probe_with(schedule, override_ts, |packet| { - let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now); - let send_call_start = Instant::now(); - let bytes = socket.send(packet)?; - let send_call = send_call_start.elapsed(); - Ok(SendProbeResult { - sent_at, - bytes, - send_call, - }) - }) + self.send_probe_transaction(override_ts, ProbeScheduleMode::CallerPaced) } fn send_managed_probe_inner( @@ -327,24 +324,102 @@ impl Client { scheduled_at: Instant, override_ts: Option, ) -> Result, ClientError> { - let socket = &self.socket; + self.send_probe_transaction(override_ts, ProbeScheduleMode::Managed { scheduled_at }) + } + + fn send_probe_transaction( + &mut self, + override_ts: Option, + mode: ProbeScheduleMode, + ) -> Result, ClientError> { self.runtime.ensure_open()?; - let schedule = self - .schedule + let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now); + #[cfg(test)] + let fail_send = std::mem::take(&mut self.probe_send_error); + let remote = self.remote; + let (runtime, schedule, socket) = (&mut self.runtime, &mut self.schedule, &self.socket); + let schedule = schedule .as_mut() .expect("open sessions always have a probe schedule"); - self.runtime - .send_probe_for_deadline(schedule, override_ts, scheduled_at, |packet| { - let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now); - let send_call_start = Instant::now(); - let bytes = socket.send(packet)?; - let send_call = send_call_start.elapsed(); - Ok(SendProbeResult { - sent_at, - bytes, - send_call, - }) - }) + if !schedule.permit_probe_at(sent_at.mono) { + 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 = match mode { + ProbeScheduleMode::CallerPaced => { + schedule.preflight_caller_commit(sent_at.mono, machine_commit.next_packets_sent)? + } + ProbeScheduleMode::Managed { scheduled_at } => { + schedule.preflight_managed_commit(scheduled_at, sent_at.mono)? + } + }; + let mut events = Vec::new(); + events + .try_reserve(1) + .map_err(|source| ClientError::AllocationFailed { + operation: "probe event result", + source, + })?; + + 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(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)?; + let sent = runtime.commit_probe_sent(machine_commit, bytes); + schedule.commit(schedule_commit); + + let send_call = send_call_start.elapsed(); + #[cfg(test)] + let bytes = reported_bytes.unwrap_or(bytes); + validate_datagram_length(expected_bytes, bytes)?; + + events.push(echo_sent_event( + remote, + sent, + send_call, + scheduled_at, + timer_error, + )); + Ok(events) + } +} + +pub(crate) fn echo_sent_event( + remote: SocketAddr, + sent: ProbeSent, + send_call: Duration, + scheduled_at: Instant, + timer_error: Duration, +) -> ClientEvent { + ClientEvent::EchoSent { + seq: sent.seq, + remote, + scheduled_at, + sent_at: sent.sent_at, + bytes: sent.bytes, + send_call, + timer_error, + } +} + +pub(crate) fn validate_datagram_length(expected: usize, actual: usize) -> Result<(), ClientError> { + if actual == expected { + Ok(()) + } else { + Err(ClientError::DatagramLengthMismatch { expected, actual }) } } 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 4b6d7cc..b1d6815 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 @@ -51,6 +51,113 @@ fn send_probe_sends_valid_echo_request() { server.join(); } +#[test] +fn blocking_short_probe_commits_before_presentation_length_error() { + let params = default_params(); + let server = silent_open_server(params.clone()); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + assert_open_started(client.open().unwrap()); + + let expected = echo_packet_len(false, ¶ms); + let initial_deadline = client.next_send_deadline().unwrap(); + client.probe_reported_len = Some(expected - 1); + + assert!(matches!( + client.send_probe(), + Err(ClientError::DatagramLengthMismatch { + expected: error_expected, + actual, + }) if error_expected == expected && actual + 1 == expected + )); + assert_eq!(client.runtime.packets_sent(), 1); + assert_eq!( + client.next_send_deadline(), + Some(initial_deadline + client.runtime.config().interval) + ); + + client.close().unwrap(); + server.join(); +} + +#[test] +fn blocking_failed_probe_send_preserves_machine_and_schedule() { + 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 initial_deadline = client.next_send_deadline().unwrap(); + client.probe_send_error = true; + + assert!(matches!(client.send_probe(), Err(ClientError::Socket(_)))); + assert_eq!(client.runtime.packets_sent(), 0); + assert_eq!(client.next_send_deadline(), Some(initial_deadline)); + + assert!(matches!( + client.send_probe().unwrap().as_slice(), + [ClientEvent::EchoSent { seq: 0, .. }] + )); + + client.close().unwrap(); + server.join(); +} + +#[test] +fn blocking_probe_uses_one_timestamp_for_permission_and_schedule_commit() { + let duration = Duration::from_secs(1); + let interval = Duration::from_millis(100); + let params = Params { + duration_ns: 1_000_000_000, + interval_ns: 100_000_000, + ..default_params() + }; + let server = silent_open_server(params.clone()); + let config = ClientConfig { + duration: Some(duration), + interval, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open().unwrap()); + + let opened_mono = Instant::now() + .checked_sub(Duration::from_secs(2)) + .expect("test host monotonic clock has at least two seconds of history"); + client.schedule = Some( + ProbeSchedule::new( + opened_mono, + &NegotiatedParams { + params, + restrictions: Vec::new(), + }, + ) + .unwrap(), + ); + let sent_at = ClientTimestamp { + wall: SystemTime::now(), + mono: opened_mono + Duration::from_millis(500), + }; + + let events = client.send_probe_at(sent_at).unwrap(); + + assert!(matches!( + events.as_slice(), + [ClientEvent::EchoSent { + scheduled_at, + sent_at: event_sent_at, + timer_error, + .. + }] if *scheduled_at == opened_mono + && *event_sent_at == sent_at + && *timer_error == Duration::from_millis(500) + )); + assert_eq!(client.next_send_deadline(), Some(opened_mono + interval)); + assert_eq!(client.runtime.packets_sent(), 1); + + client.close().unwrap(); + server.join(); +} + #[test] fn send_probe_respects_finite_duration_exclusive_end() { let params = Params { diff --git a/crates/irtt-client/src/error.rs b/crates/irtt-client/src/error.rs index 9aa8fe3..325946c 100644 --- a/crates/irtt-client/src/error.rs +++ b/crates/irtt-client/src/error.rs @@ -152,6 +152,16 @@ pub enum ClientError { /// can be reused. #[error("wire sequence {seq} is still pending")] PendingSequenceCollision { seq: u32 }, + /// A prepared probe no longer matches the session's next wire sequence. + /// + /// Prepare the probe again after any intervening successful send. + #[error( + "prepared probe sequence {prepared_seq} is stale; next wire sequence is {next_wire_seq}" + )] + StalePreparedProbe { + prepared_seq: u32, + next_wire_seq: u32, + }, /// Reserving memory required by an operation failed before any datagram was /// sent or logical protocol state was changed. #[error("failed to reserve memory for {operation}: {source}")] @@ -160,6 +170,13 @@ pub enum ClientError { #[source] source: TryReserveError, }, + /// UDP reported successful acceptance of a datagram with an unexpected + /// byte count. + /// + /// The send is committed locally before this transport invariant failure + /// is returned because the kernel reported success. + #[error("UDP accepted {actual} bytes for a {expected}-byte datagram")] + DatagramLengthMismatch { expected: usize, actual: usize }, /// Client configuration failed validation before the requested operation. /// /// The reported reason names a value that must be changed before retrying. diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index 4335272..7a24a7c 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -12,13 +12,17 @@ use std::{ }; use crate::{ - client::schedule::{advance_cadence, ProbeSchedule}, + client::{ + echo_sent_event, + schedule::{advance_cadence, ProbeSchedule}, + validate_datagram_length, + }, config::{ClientAuthConfig, ClientConfig}, error::ClientError, event::{ClientEvent, OpenOutcome}, metadata::ReceiveMeta, receive::recv_datagram_from, - session::machine::{params_from_config, SendProbeResult, SessionMachine}, + session::machine::{params_from_config, SessionMachine}, socket::{bind_unconnected_udp_socket, validate_open_timeouts}, socket_options::apply_dscp_to_socket, timing::ClientTimestamp, @@ -633,6 +637,10 @@ struct TargetState { counters: TargetCounters, order: u64, final_reason: Option, + #[cfg(test)] + probe_reported_len: Option, + #[cfg(test)] + probe_send_error: bool, } type ScheduledTarget = (Arc>, Instant); @@ -727,6 +735,10 @@ impl TargetState { counters: TargetCounters::default(), order, final_reason: None, + #[cfg(test)] + probe_reported_len: None, + #[cfg(test)] + probe_send_error: false, }) } @@ -1311,28 +1323,7 @@ fn send_echo_to_target( if !matches!(target.status, TargetStatus::Active) { return; } - let remote = target.remote; - target - .runtime - .ensure_open() - .expect("active targets are open"); - let TargetState { - runtime, schedule, .. - } = &mut *target; - let schedule = schedule - .as_mut() - .expect("active targets always have a probe schedule"); - let result = runtime.send_probe_for_deadline(schedule, None, scheduled_at, |packet| { - let sent_at = ClientTimestamp::now(); - let send_call_start = Instant::now(); - let bytes = socket.send_to(packet, remote)?; - let send_call = send_call_start.elapsed(); - Ok(SendProbeResult { - sent_at, - bytes, - send_call, - }) - }); + let result = send_echo_to_locked_target(socket, &mut target, scheduled_at); match result { Ok(events) => { @@ -1348,6 +1339,77 @@ fn send_echo_to_target( } } +fn send_echo_to_locked_target( + socket: &UdpSocket, + target: &mut TargetState, + scheduled_at: Instant, +) -> Result, ClientError> { + send_echo_to_locked_target_at(socket, target, scheduled_at, ClientTimestamp::now()) +} + +fn send_echo_to_locked_target_at( + socket: &UdpSocket, + target: &mut TargetState, + scheduled_at: Instant, + sent_at: ClientTimestamp, +) -> Result, ClientError> { + target.runtime.ensure_open()?; + let remote = target.remote; + #[cfg(test)] + let reported_bytes = target.probe_reported_len.take(); + #[cfg(test)] + let fail_send = std::mem::take(&mut target.probe_send_error); + let TargetState { + runtime, schedule, .. + } = target; + let schedule = schedule + .as_mut() + .expect("active targets always have a probe schedule"); + if !schedule.permit_probe_at(sent_at.mono) { + 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 mut events = Vec::new(); + events + .try_reserve(1) + .map_err(|source| ClientError::AllocationFailed { + operation: "managed probe event result", + source, + })?; + + let expected_bytes = prepared.bytes.len(); + let event_scheduled_at = schedule_commit.scheduled_at; + let timer_error = schedule_commit.timer_error; + #[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)?; + let sent = runtime.commit_probe_sent(machine_commit, bytes); + schedule.commit(schedule_commit); + + let send_call = send_call_start.elapsed(); + #[cfg(test)] + let bytes = reported_bytes.unwrap_or(bytes); + validate_datagram_length(expected_bytes, bytes)?; + events.push(echo_sent_event( + remote, + sent, + send_call, + event_scheduled_at, + timer_error, + )); + Ok(events) +} + fn collect_finished_targets(shared: &GroupShared, records: &mut TargetOutcomeHistory) { let finished = { let mut registry = shared @@ -1853,23 +1915,29 @@ mod tests { let scheduled_at = Instant::now(); let sent_at = ClientTimestamp::now(); let sent_events = { + let remote = target.remote; let TargetState { runtime, schedule, .. } = &mut target; - runtime.send_probe_for_deadline( - schedule.as_mut().unwrap(), - Some(sent_at), - scheduled_at, - |packet| { - Ok(SendProbeResult { - sent_at, - bytes: packet.len(), - send_call: Duration::ZERO, - }) - }, - ) - } - .unwrap(); + 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 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 sent = runtime.commit_probe_sent(machine_commit, prepared.bytes.len()); + schedule.commit(schedule_commit); + vec![echo_sent_event( + remote, + sent, + Duration::ZERO, + event_scheduled_at, + timer_error, + )] + }; publish_events(&hub, &mut target, sent_events); ActiveTargetFixture { @@ -1934,6 +2002,29 @@ mod tests { } } + fn active_probe_target( + config: &ClientConfig, + params: &Params, + remote: SocketAddr, + opened_at: ClientTimestamp, + ) -> TargetState { + let mut target = + TargetState::new(config, target("peer", remote), 0, opened_at.mono).unwrap(); + let packet = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, params); + let reply = target.runtime.decode_open_reply(&packet).unwrap(); + let mut schedule = None; + target + .runtime + .accept_open_reply(reply, opened_at, |negotiated| { + schedule = Some(ProbeSchedule::new(opened_at.mono, negotiated)?); + Ok(()) + }) + .unwrap(); + target.schedule = schedule; + target.status = TargetStatus::Active; + target + } + fn target_outcome( id: &str, remote: SocketAddr, @@ -2883,6 +2974,137 @@ mod tests { .all(|target| target.end_reason.failure().is_some())); } + #[test] + fn group_probe_uses_one_timestamp_for_permission_and_schedule_commit() { + let duration = Duration::from_secs(1); + let interval = Duration::from_millis(100); + let params = test_params(Some(duration), interval); + let config = group_config(Some(duration), interval, ManagedGroupPacing::Staggered).client; + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let opened_mono = Instant::now() + .checked_sub(Duration::from_secs(2)) + .expect("test host monotonic clock has at least two seconds of history"); + let opened_at = ClientTimestamp { + wall: std::time::SystemTime::now(), + mono: opened_mono, + }; + let mut target = active_probe_target(&config, ¶ms, remote, opened_at); + let sent_at = ClientTimestamp { + wall: std::time::SystemTime::now(), + mono: opened_mono + Duration::from_millis(500), + }; + 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(); + + assert!(matches!( + events.as_slice(), + [ClientEvent::EchoSent { + scheduled_at, + sent_at: event_sent_at, + timer_error, + .. + }] if *scheduled_at == sent_at.mono + && *event_sent_at == sent_at + && *timer_error == Duration::ZERO + )); + assert_eq!( + target + .schedule + .as_ref() + .and_then(ProbeSchedule::next_send_deadline), + Some(opened_mono + Duration::from_millis(600)) + ); + assert_eq!(target.runtime.packets_sent(), 1); + } + + #[test] + fn group_failed_probe_send_preserves_machine_and_schedule() { + let duration = Duration::from_secs(1); + let interval = Duration::from_millis(100); + let params = test_params(Some(duration), interval); + let config = group_config(Some(duration), interval, ManagedGroupPacing::Staggered).client; + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let opened_mono = Instant::now() + .checked_sub(Duration::from_secs(2)) + .expect("test host monotonic clock has at least two seconds of history"); + let opened_at = ClientTimestamp { + wall: std::time::SystemTime::now(), + mono: opened_mono, + }; + let mut target = active_probe_target(&config, ¶ms, remote, opened_at); + let sent_at = ClientTimestamp { + wall: std::time::SystemTime::now(), + mono: opened_mono + Duration::from_millis(500), + }; + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + target.probe_send_error = true; + + assert!(matches!( + send_echo_to_locked_target_at(&socket, &mut target, opened_mono, sent_at), + Err(ClientError::Socket(_)) + )); + assert_eq!(target.runtime.packets_sent(), 0); + assert_eq!( + target + .schedule + .as_ref() + .and_then(ProbeSchedule::next_send_deadline), + Some(opened_mono) + ); + + assert!(matches!( + send_echo_to_locked_target_at(&socket, &mut target, opened_mono, sent_at) + .unwrap() + .as_slice(), + [ClientEvent::EchoSent { seq: 0, .. }] + )); + } + + #[test] + fn group_short_probe_commits_before_presentation_length_error() { + let duration = Duration::from_secs(1); + let interval = Duration::from_millis(100); + let params = test_params(Some(duration), interval); + let config = group_config(Some(duration), interval, ManagedGroupPacing::Staggered).client; + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let opened_mono = Instant::now() + .checked_sub(Duration::from_secs(2)) + .expect("test host monotonic clock has at least two seconds of history"); + let opened_at = ClientTimestamp { + wall: std::time::SystemTime::now(), + mono: opened_mono, + }; + let mut target = active_probe_target(&config, ¶ms, remote, opened_at); + let sent_at = ClientTimestamp { + wall: std::time::SystemTime::now(), + mono: opened_mono + Duration::from_millis(500), + }; + let expected = echo_packet_len(false, ¶ms).unwrap(); + target.probe_reported_len = Some(expected - 1); + 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), + Err(ClientError::DatagramLengthMismatch { + expected: error_expected, + actual, + }) if error_expected == expected && actual + 1 == expected + )); + assert_eq!(target.runtime.packets_sent(), 1); + assert_eq!( + target + .schedule + .as_ref() + .and_then(ProbeSchedule::next_send_deadline), + Some(opened_mono + Duration::from_millis(600)) + ); + } + #[test] fn silent_opening_peer_times_out_all_attempts_and_publishes_once() { let silent_peer = UdpSocket::bind("127.0.0.1:0").unwrap(); diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index 26ae1b2..c14916b 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -72,13 +72,13 @@ impl PendingMap { } #[cfg(test)] - fn capacity(&self) -> usize { - self.map.capacity() + pub fn contains(&self, wire_seq: u32) -> bool { + self.map.contains_key(&wire_seq) } #[cfg(test)] - fn contains(&self, wire_seq: u32) -> bool { - self.map.contains_key(&wire_seq) + pub fn capacity(&self) -> usize { + self.map.capacity() } } @@ -139,6 +139,11 @@ impl TimedOutMap { self.map.len() } + #[cfg(test)] + pub fn contains(&self, wire_seq: u32) -> bool { + self.map.contains_key(&wire_seq) + } + #[cfg(test)] fn insertion_order_len(&self) -> usize { self.insertion_order.len() @@ -186,6 +191,14 @@ impl CompletedSet { self.set.contains(&seq) } + pub fn remove(&mut self, seq: u32) -> bool { + let removed = self.set.remove(&seq); + if removed { + self.insertion_order.retain(|entry| *entry != seq); + } + removed + } + fn evict_oldest(&mut self) { while let Some(oldest_seq) = self.insertion_order.pop_front() { if self.set.remove(&oldest_seq) { diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index ba56f41..e60fa46 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -7,7 +7,6 @@ use irtt_proto::{ }; use crate::{ - client::schedule::ProbeSchedule, config::{ ClientConfig, RunMode, MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_UDP_PAYLOAD_LENGTH, }, @@ -62,11 +61,24 @@ struct ActiveSession { completed: CompletedSet, } +#[derive(Debug)] +pub(crate) struct PreparedProbe { + pub(crate) bytes: Box<[u8]>, + pub(crate) seq: u32, +} + +#[derive(Debug)] +pub(crate) struct ProbeCommit { + pending: PendingProbe, + next_wire_seq: u32, + pub(crate) next_packets_sent: u64, +} + #[derive(Debug, Clone, Copy)] -pub(crate) struct SendProbeResult { +pub(crate) struct ProbeSent { + pub(crate) seq: u32, pub(crate) sent_at: ClientTimestamp, pub(crate) bytes: usize, - pub(crate) send_call: Duration, } impl SessionMachine { @@ -160,110 +172,79 @@ impl SessionMachine { self.config.probe_timeout } - pub(crate) fn send_probe_with( - &mut self, - schedule: &mut ProbeSchedule, - override_ts: Option, - send: F, - ) -> Result, ClientError> - where - F: FnOnce(&[u8]) -> Result, - { - self.send_probe_inner(schedule, override_ts, None, false, send) - } - - pub(crate) fn send_probe_for_deadline( - &mut self, - schedule: &mut ProbeSchedule, - override_ts: Option, - scheduled_at: Instant, - send: F, - ) -> Result, ClientError> - where - F: FnOnce(&[u8]) -> Result, - { - self.send_probe_inner(schedule, override_ts, Some(scheduled_at), true, send) - } - - fn send_probe_inner( - &mut self, - schedule: &mut ProbeSchedule, - override_ts: Option, - scheduled_at_override: Option, - skip_missed_slots: bool, - send: F, - ) -> Result, ClientError> - where - F: FnOnce(&[u8]) -> Result, - { - let now = override_ts.unwrap_or_else(ClientTimestamp::now); - if !schedule.permit_probe_at(now.mono) { - return Ok(vec![]); - } - - let (config, state) = (&self.config, &mut self.state); - let probe_timeout = config.probe_timeout; - let hmac_key = config.hmac_key.as_deref(); - let session = match state { - MachineState::Open(session) => session, - MachineState::Closed { .. } => return Err(ClientError::AlreadyClosed), - MachineState::Connected => return Err(ClientError::NotOpen), - MachineState::NoTestCompleted => return Err(ClientError::AlreadyCompleted), - }; - - session.pending.preflight_insert(session.next_wire_seq)?; - - let wire_seq = session.next_wire_seq; - + pub(crate) fn prepare_probe(&self) -> Result, ClientError> { + let session = self.open_session()?; let request = EchoRequest { token: session.token, - sequence: wire_seq, + sequence: session.next_wire_seq, payload: vec![], }; - let packet = encode_echo_request(&request, &session.negotiated.params, hmac_key)?; - let send_result = send(&packet)?; + let bytes = encode_echo_request( + &request, + &session.negotiated.params, + self.config.hmac_key.as_deref(), + )?; + Ok(Some(PreparedProbe { + bytes: bytes.into_boxed_slice(), + seq: session.next_wire_seq, + })) + } - let pending = PendingProbe { - wire_seq, - sent_at: send_result.sent_at, - timeout_at: send_result - .sent_at - .mono - .checked_add(probe_timeout) - .ok_or(ClientError::DurationOverflow)?, - }; - session.pending.commit_insert(pending); - - session.next_wire_seq = session.next_wire_seq.wrapping_add(1); - session.packets_sent = + pub(crate) fn preflight_probe_commit( + &mut self, + prepared: &PreparedProbe, + sent_at: ClientTimestamp, + ) -> Result { + let probe_timeout = self.config.probe_timeout; + let session = self.open_session_mut()?; + if prepared.seq != session.next_wire_seq { + return Err(ClientError::StalePreparedProbe { + prepared_seq: prepared.seq, + next_wire_seq: session.next_wire_seq, + }); + } + session.pending.preflight_insert(prepared.seq)?; + let next_packets_sent = session .packets_sent .checked_add(1) .ok_or(ClientError::CounterOverflow { counter: "packets_sent", })?; + let timeout_at = sent_at + .mono + .checked_add(probe_timeout) + .ok_or(ClientError::DurationOverflow)?; - let schedule_commit = if skip_missed_slots { - schedule.preflight_managed_commit( - scheduled_at_override.expect("managed sends include a scheduled deadline"), - send_result.sent_at.mono, - )? - } else { - schedule.preflight_caller_commit(send_result.sent_at.mono, session.packets_sent)? + Ok(ProbeCommit { + pending: PendingProbe { + wire_seq: prepared.seq, + sent_at, + timeout_at, + }, + next_wire_seq: prepared.seq.wrapping_add(1), + next_packets_sent, + }) + } + + pub(crate) fn commit_probe_sent(&mut self, commit: ProbeCommit, bytes: usize) -> ProbeSent { + let session = match &mut self.state { + MachineState::Open(session) => session, + _ => unreachable!("probe commits are only created for an open session"), }; - let scheduled_at = schedule_commit.scheduled_at; - let timer_error = schedule_commit.timer_error; - schedule.commit(schedule_commit); + let seq = commit.pending.wire_seq; + let sent_at = commit.pending.sent_at; + session.timed_out.remove(seq); + session.completed.remove(seq); + session.pending.commit_insert(commit.pending); + session.next_wire_seq = commit.next_wire_seq; + session.packets_sent = commit.next_packets_sent; - Ok(vec![ClientEvent::EchoSent { - seq: wire_seq, - remote: self.remote, - scheduled_at, - sent_at: send_result.sent_at, - bytes: send_result.bytes, - send_call: send_result.send_call, - timer_error, - }]) + ProbeSent { + seq, + sent_at, + bytes, + } } pub(crate) fn process_received_echo_packet( @@ -830,3 +811,320 @@ fn config_duration_to_ns(field: &str, duration: Duration) -> Result SessionMachine { + let config = ClientConfig { + max_pending_probes, + probe_timeout, + ..ClientConfig::default() + }; + let remote = "127.0.0.1:2112".parse().unwrap(); + let mut machine = SessionMachine::new(config, remote).unwrap(); + let negotiated = NegotiatedParams { + params: machine.requested.clone(), + restrictions: Vec::new(), + }; + machine.state = MachineState::Open(Box::new(ActiveSession { + token: 0x0102_0304_0506_0708, + negotiated, + next_wire_seq: 0, + highest_received_seq: None, + packets_sent: 0, + pending: PendingMap::new(max_pending_probes), + timed_out: TimedOutMap::new(max_pending_probes), + completed: CompletedSet::new(max_pending_probes), + })); + machine + } + + fn active(machine: &SessionMachine) -> &ActiveSession { + match &machine.state { + MachineState::Open(session) => session, + _ => panic!("test machine must be open"), + } + } + + fn active_mut(machine: &mut SessionMachine) -> &mut ActiveSession { + match &mut machine.state { + MachineState::Open(session) => session, + _ => panic!("test machine must be open"), + } + } + + fn timestamp(mono: Instant) -> ClientTimestamp { + ClientTimestamp { + mono, + wall: SystemTime::now(), + } + } + + #[test] + fn dropping_prepared_probe_changes_nothing() { + let machine = open_machine(4, Duration::from_secs(1)); + let prepared = machine.prepare_probe().unwrap().unwrap(); + assert_eq!(prepared.seq, 0); + drop(prepared); + + let session = active(&machine); + assert_eq!(session.next_wire_seq, 0); + assert_eq!(session.packets_sent, 0); + assert_eq!(session.pending.len(), 0); + } + + #[test] + fn repeated_uncommitted_preflight_changes_no_logical_state() { + let mut machine = open_machine(4, Duration::from_secs(1)); + let prepared = machine.prepare_probe().unwrap().unwrap(); + let sent_at = timestamp(Instant::now()); + + let _first_preflight = machine.preflight_probe_commit(&prepared, sent_at).unwrap(); + let _second_preflight = machine.preflight_probe_commit(&prepared, sent_at).unwrap(); + + let session = active(&machine); + assert_eq!(session.next_wire_seq, 0); + assert_eq!(session.packets_sent, 0); + assert_eq!(session.pending.len(), 0); + } + + #[test] + fn stale_prepared_probe_is_rejected_without_changing_state() { + let mut machine = open_machine(4, Duration::from_secs(1)); + let stale = machine.prepare_probe().unwrap().unwrap(); + let accepted = machine.prepare_probe().unwrap().unwrap(); + let sent_at = timestamp(Instant::now()); + let commit = machine.preflight_probe_commit(&accepted, sent_at).unwrap(); + machine.commit_probe_sent(commit, accepted.bytes.len()); + + let capacity = active(&machine).pending.capacity(); + assert!(matches!( + machine.preflight_probe_commit(&stale, sent_at), + Err(ClientError::StalePreparedProbe { + prepared_seq: 0, + next_wire_seq: 1, + }) + )); + + let session = active(&machine); + assert_eq!(session.next_wire_seq, 1); + assert_eq!(session.packets_sent, 1); + assert_eq!(session.pending.len(), 1); + assert_eq!(session.pending.capacity(), capacity); + } + + #[test] + fn simulated_send_error_leaves_machine_and_schedule_unchanged() { + let mut machine = open_machine(4, Duration::from_secs(1)); + let start = Instant::now(); + let mut schedule = ProbeSchedule::new(start, &active(&machine).negotiated).unwrap(); + assert!(schedule.permit_probe_at(start)); + let prepared = machine.prepare_probe().unwrap().unwrap(); + let sent_at = timestamp(start); + let machine_commit = machine.preflight_probe_commit(&prepared, sent_at).unwrap(); + let schedule_commit = schedule + .preflight_caller_commit(sent_at.mono, machine_commit.next_packets_sent) + .unwrap(); + + let _uncommitted_machine = machine_commit; + let _uncommitted_schedule = schedule_commit; + + let session = active(&machine); + assert_eq!(session.next_wire_seq, 0); + assert_eq!(session.packets_sent, 0); + assert_eq!(session.pending.len(), 0); + assert_eq!(schedule.next_send_deadline(), Some(start)); + } + + #[test] + fn short_success_commits_before_reporting_transport_invariant() { + let mut machine = open_machine(4, Duration::from_secs(1)); + let start = Instant::now(); + let mut schedule = ProbeSchedule::new(start, &active(&machine).negotiated).unwrap(); + assert!(schedule.permit_probe_at(start)); + let prepared = machine.prepare_probe().unwrap().unwrap(); + let machine_commit = machine + .preflight_probe_commit(&prepared, timestamp(start)) + .unwrap(); + let schedule_commit = schedule.preflight_managed_commit(start, start).unwrap(); + let expected = prepared.bytes.len(); + let actual = expected - 1; + + machine.commit_probe_sent(machine_commit, actual); + schedule.commit(schedule_commit); + let result = validate_datagram_length(expected, actual); + + assert!(matches!( + result, + Err(ClientError::DatagramLengthMismatch { expected, actual }) + if actual + 1 == expected + )); + let session = active(&machine); + assert_eq!(session.next_wire_seq, 1); + assert_eq!(session.packets_sent, 1); + assert_eq!(session.pending.len(), 1); + assert_eq!( + schedule.next_send_deadline(), + Some(start + machine.config.interval) + ); + } + + #[test] + fn repeated_would_block_style_preflight_commits_once() { + let mut machine = open_machine(4, Duration::from_secs(1)); + let prepared = machine.prepare_probe().unwrap().unwrap(); + let sent_at = timestamp(Instant::now()); + + for _ in 0..3 { + let _would_block_commit = machine.preflight_probe_commit(&prepared, sent_at).unwrap(); + } + let commit = machine.preflight_probe_commit(&prepared, sent_at).unwrap(); + let sent = machine.commit_probe_sent(commit, prepared.bytes.len()); + + assert_eq!(sent.seq, 0); + let session = active(&machine); + assert_eq!(session.next_wire_seq, 1); + assert_eq!(session.packets_sent, 1); + assert_eq!(session.pending.len(), 1); + } + + #[test] + fn probe_commit_does_not_require_presentation_timing() { + let mut machine = open_machine(4, Duration::from_secs(1)); + let prepared = machine.prepare_probe().unwrap().unwrap(); + let sent_at = timestamp(Instant::now()); + let commit = machine.preflight_probe_commit(&prepared, sent_at).unwrap(); + + let sent = machine.commit_probe_sent(commit, prepared.bytes.len()); + + assert_eq!(sent.seq, prepared.seq); + assert_eq!(sent.sent_at, sent_at); + assert_eq!(sent.bytes, prepared.bytes.len()); + } + + #[test] + fn commit_uses_reserved_capacity_and_prevalidated_counter() { + let mut machine = open_machine(2, Duration::from_secs(1)); + active_mut(&mut machine).packets_sent = u64::MAX - 1; + let prepared = machine.prepare_probe().unwrap().unwrap(); + let commit = machine + .preflight_probe_commit(&prepared, timestamp(Instant::now())) + .unwrap(); + let reserved_capacity = active(&machine).pending.capacity(); + + machine.commit_probe_sent(commit, prepared.bytes.len()); + + let session = active(&machine); + assert_eq!(session.pending.capacity(), reserved_capacity); + assert_eq!(session.pending.len(), 1); + assert_eq!(session.packets_sent, u64::MAX); + assert_eq!(session.next_wire_seq, 1); + } + + #[test] + fn pending_capacity_exhaustion_is_detected_before_commit() { + let mut machine = open_machine(1, Duration::from_secs(1)); + let first = machine.prepare_probe().unwrap().unwrap(); + let sent_at = timestamp(Instant::now()); + let commit = machine.preflight_probe_commit(&first, sent_at).unwrap(); + machine.commit_probe_sent(commit, first.bytes.len()); + + let second = machine.prepare_probe().unwrap().unwrap(); + assert!(matches!( + machine.preflight_probe_commit(&second, sent_at), + Err(ClientError::PendingLimitExceeded { limit: 1 }) + )); + assert_eq!(active(&machine).packets_sent, 1); + } + + #[test] + fn pending_sequence_collision_is_detected_before_commit() { + let mut machine = open_machine(2, Duration::from_secs(1)); + let first = machine.prepare_probe().unwrap().unwrap(); + let sent_at = timestamp(Instant::now()); + let commit = machine.preflight_probe_commit(&first, sent_at).unwrap(); + machine.commit_probe_sent(commit, first.bytes.len()); + active_mut(&mut machine).next_wire_seq = 0; + + let reused = machine.prepare_probe().unwrap().unwrap(); + assert!(matches!( + machine.preflight_probe_commit(&reused, sent_at), + Err(ClientError::PendingSequenceCollision { seq: 0 }) + )); + assert_eq!(active(&machine).packets_sent, 1); + } + + #[test] + fn counter_overflow_is_detected_before_commit() { + let mut machine = open_machine(2, Duration::from_secs(1)); + active_mut(&mut machine).packets_sent = u64::MAX; + let prepared = machine.prepare_probe().unwrap().unwrap(); + + assert!(matches!( + machine.preflight_probe_commit(&prepared, timestamp(Instant::now())), + Err(ClientError::CounterOverflow { + counter: "packets_sent" + }) + )); + assert_eq!(active(&machine).pending.len(), 0); + } + + #[test] + fn timeout_overflow_is_detected_before_commit() { + let mut machine = open_machine(2, Duration::MAX); + let prepared = machine.prepare_probe().unwrap().unwrap(); + + assert!(matches!( + machine.preflight_probe_commit(&prepared, timestamp(Instant::now())), + Err(ClientError::DurationOverflow) + )); + assert_eq!(active(&machine).pending.len(), 0); + } + + #[test] + fn wrapping_sequence_from_max_to_zero_remains_valid() { + let mut machine = open_machine(2, Duration::from_secs(1)); + active_mut(&mut machine).next_wire_seq = u32::MAX; + let prepared = machine.prepare_probe().unwrap().unwrap(); + let commit = machine + .preflight_probe_commit(&prepared, timestamp(Instant::now())) + .unwrap(); + machine.commit_probe_sent(commit, prepared.bytes.len()); + + assert_eq!(active(&machine).next_wire_seq, 0); + assert_eq!(machine.prepare_probe().unwrap().unwrap().seq, 0); + } + + #[test] + fn successful_wrapped_reuse_purges_obsolete_history_only_on_commit() { + let mut machine = open_machine(3, Duration::from_secs(1)); + let now = Instant::now(); + let obsolete = PendingProbe { + wire_seq: 0, + sent_at: timestamp(now - Duration::from_secs(1)), + timeout_at: now, + }; + let session = active_mut(&mut machine); + session.next_wire_seq = 0; + session.timed_out.insert(obsolete); + session.completed.insert(0); + + let prepared = machine.prepare_probe().unwrap().unwrap(); + let commit = machine + .preflight_probe_commit(&prepared, timestamp(now)) + .unwrap(); + assert!(active(&machine).timed_out.contains(0)); + assert!(active(&machine).completed.contains(0)); + + machine.commit_probe_sent(commit, prepared.bytes.len()); + let session = active(&machine); + assert!(!session.timed_out.contains(0)); + assert!(!session.completed.contains(0)); + assert!(session.pending.contains(0)); + assert_eq!(session.next_wire_seq, 1); + assert_eq!(session.packets_sent, 1); + } +}