diff --git a/crates/irtt-client/AGENTS.md b/crates/irtt-client/AGENTS.md index f171d74..15947e9 100644 --- a/crates/irtt-client/AGENTS.md +++ b/crates/irtt-client/AGENTS.md @@ -2,6 +2,42 @@ Repository-wide guidance in the root `AGENTS.md` also applies here. +## Probe-send commitment API + +`Client::send_probe` and `AsyncClient::send_probe` return +`Result`; the Tokio poll driver returns the same +result inside `Poll`. A receipt represents an Echo datagram accepted by the +socket and already committed in session state: sequence and sent count advanced, +and the probe entered pending tracking. It contains `seq`, `remote`, the public +post-send `sent_at`, accepted `bytes`, and `send_call`, without schedule metadata. +`ClientEvent::from(receipt)` constructs `EchoSent` with no schedule annotations. + +Classify every failure explicitly: `NotCommitted(ClientError)` means no datagram +was accepted and no probe committed; `AfterCommit { receipt, source }` preserves +the authoritative accepted send when length validation or TX error-queue +processing fails. There is no blanket `From` conversion. Do not +roll back a committed probe or classify a post-send error with an indiscriminate +`?`. + +All fallible state/capacity/counter/allocation and timeout preparation precedes +submission. Immediately after `send`/`try_send` returns `Ok(bytes)`, capture the +public post-send timestamp, commit infallibly, and construct the receipt before +fallible post-send processing. No suspension, readiness wait, fallible allocation, +or fallible state operation may intervene between acceptance and commitment; +the Tokio poll must never return `Pending` after acceptance. + +The managed driver consumes receipts for cadence/stagger acceptance and constructs +`EchoSent` with its scheduled timestamp and timer error. A post-commit failure +advances cadence and publishes that send before failure cleanup; the underlying +cause remains a sending failure. `NotCommitted` and `Pending` advance neither +cadence nor stagger gating. Counters are cumulative accounting, never evidence +of one operation's commitment. + +Tokio retains one prepared probe across cancellation/readiness false positives +and `WouldBlock`, samples a fresh private anchor on each retry, and clears the +prepared probe after commitment even when post-send processing fails. Failed +submissions still invalidate kernel TX ID correlation as documented below. + ## Linux kernel TX timestamp capture On Linux, with the `ancillary` feature, the client best-effort upgrades its @@ -84,7 +120,9 @@ backlog. `Client`/`AsyncClient` call it after a successful probe send, before processing an inbound reply, and before evaluating timeouts; it never waits for a timestamp; an absent timestamp after all of those opportunities simply stays absent. A genuine `SocketError` record stops the drain and is -surfaced through the ordinary `ClientError::Socket` path. +surfaced as `ClientError::Socket`. During post-send processing it is wrapped in +`SendProbeError::AfterCommit` with the accepted send's receipt; receive/timeout +operations retain the ordinary `ClientError` path. The Tokio adapter uses the same raw nonblocking `recvmsg(MSG_ERRQUEUE | MSG_DONTWAIT)` rather than `try_io`/`poll_recv_ready`: error-queue readiness diff --git a/crates/irtt-client/README.md b/crates/irtt-client/README.md index fe4bc92..bbb1bc1 100644 --- a/crates/irtt-client/README.md +++ b/crates/irtt-client/README.md @@ -26,8 +26,14 @@ binaries are built on. negotiated parameters, pending-probe status, and timeout deadlines; callers own cadence and run duration. The managed driver provides fixed cadence, missed-slot handling, finite runs, and staggered or burst pacing. Low-level -`EchoSent` events leave `scheduled_at` and `timer_error` absent; the managed -driver supplies those schedule measurements. +sends return `Result` instead of event lists. +A `SendReceipt` carries the committed sequence, remote, post-send timestamp, +accepted bytes, and send-call duration. `NotCommitted(ClientError)` guarantees +no accepted datagram or committed probe; `AfterCommit { receipt, source }` +preserves the accepted send when subsequent validation or TX processing fails. +Use `ClientEvent::from(receipt)` for an `EchoSent` event with `scheduled_at` and +`timer_error` absent; the managed driver supplies those schedule measurements +and publishes committed sends even when post-send processing fails. Tokio stays optional: the default build has no runtime and no Tokio dependency at all. diff --git a/crates/irtt-client/examples/blocking_client.rs b/crates/irtt-client/examples/blocking_client.rs index b8002ba..befa7e2 100644 --- a/crates/irtt-client/examples/blocking_client.rs +++ b/crates/irtt-client/examples/blocking_client.rs @@ -83,13 +83,16 @@ fn main() { } if now < end && now >= next_send { match client.send_probe() { - Ok(events) => { - events.iter().for_each(print_event); + Ok(receipt) => { + print_event(&receipt.into()); while next_send <= Instant::now() { next_send += interval; } } Err(err) => { + if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err { + print_event(&(*receipt).into()); + } eprintln!("send failed: {err}"); break; } diff --git a/crates/irtt-client/examples/tokio_client.rs b/crates/irtt-client/examples/tokio_client.rs index cdef84d..d5124ff 100644 --- a/crates/irtt-client/examples/tokio_client.rs +++ b/crates/irtt-client/examples/tokio_client.rs @@ -121,13 +121,16 @@ async fn run() { if Instant::now() < end && Instant::now() >= next_send { match client.send_probe().await { - Ok(events) => { - events.iter().for_each(print_event); + Ok(receipt) => { + print_event(&receipt.into()); while next_send <= Instant::now() { next_send += interval; } } Err(err) => { + if let irtt_client::SendProbeError::AfterCommit { receipt, .. } = &err { + print_event(&(*receipt).into()); + } eprintln!("send failed: {err}"); break; } diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index c2bf37e..3d3e759 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -10,7 +10,7 @@ use std::{ use tokio::time; use crate::{ - client::{echo_sent_event, validate_datagram_length}, + client::validate_datagram_length, config::ClientConfig, error::ClientError, event::{ClientEvent, OpenOutcome}, @@ -22,6 +22,7 @@ use crate::{ socket::{connect_tokio_udp_socket, resolve_remote_tokio}, socket_options::{apply_traffic_class_to_tokio_socket, clear_dscp_on_tokio_socket}, timing::ClientTimestamp, + SendProbeError, SendReceipt, }; /// Maximum amount of immediately available opening work performed by one poll. @@ -253,8 +254,13 @@ impl AsyncClient { /// /// A prepared packet is retained across readiness false positives, /// cancellation, and socket errors so a later call retries the same logical - /// probe without advancing state before kernel acceptance. - pub async fn send_probe(&mut self) -> Result, ClientError> { + /// probe without advancing state before kernel acceptance. Cancellation or + /// `WouldBlock` before submission is uncommitted. After acceptance there is + /// no suspension before infallible commitment, and the prepared probe is + /// cleared even if post-send processing returns [`SendProbeError::AfterCommit`]. + /// That failure retains the authoritative receipt; earlier failures return + /// [`SendProbeError::NotCommitted`]. + pub async fn send_probe(&mut self) -> Result { poll_fn(|cx| self.poll_send_probe(cx)).await } @@ -436,14 +442,14 @@ impl AsyncClient { pub(crate) fn poll_send_probe( &mut self, cx: &mut Context<'_>, - ) -> Poll, ClientError>> { + ) -> Poll> { if let Err(error) = self.machine.ensure_open() { - return Poll::Ready(Err(error)); + return Poll::Ready(Err(SendProbeError::NotCommitted(error))); } if self.prepared_probe.is_none() { match self.machine.prepare_probe() { Ok(prepared) => self.prepared_probe = Some(prepared), - Err(error) => return Poll::Ready(Err(error)), + Err(error) => return Poll::Ready(Err(SendProbeError::NotCommitted(error))), } } loop { @@ -451,7 +457,9 @@ impl AsyncClient { Poll::Pending => return Poll::Pending, Poll::Ready(Ok(())) => {} Poll::Ready(Err(error)) => { - return Poll::Ready(Err(ClientError::Socket(error))); + return Poll::Ready(Err(SendProbeError::NotCommitted(ClientError::Socket( + error, + )))); } } @@ -461,15 +469,8 @@ impl AsyncClient { .expect("prepared probe was retained across readiness"); let machine_preflight = match self.machine.preflight_probe_commit(prepared) { Ok(preflight) => preflight, - Err(error) => return Poll::Ready(Err(error)), + Err(error) => return Poll::Ready(Err(SendProbeError::NotCommitted(error))), }; - let mut events = Vec::new(); - if let Err(source) = events.try_reserve(1) { - return Poll::Ready(Err(ClientError::AllocationFailed { - operation: "probe event result", - source, - })); - } let expected_bytes = prepared.bytes.len(); // Private pre-send anchor: finalizes all fallible commit work @@ -483,7 +484,7 @@ impl AsyncClient { .finalize_probe_commit(machine_preflight, send_anchor) { Ok(commit) => commit, - Err(error) => return Poll::Ready(Err(error)), + Err(error) => return Poll::Ready(Err(SendProbeError::NotCommitted(error))), }; let send_call_start = Instant::now(); let send_result = self.socket.try_send(&prepared.bytes); @@ -497,7 +498,9 @@ impl AsyncClient { } Err(error) => { self.machine.invalidate_kernel_tx_correlation(); - return Poll::Ready(Err(ClientError::Socket(error))); + return Poll::Ready(Err(SendProbeError::NotCommitted(ClientError::Socket( + error, + )))); } }; let send_finished_at = Instant::now(); @@ -512,14 +515,26 @@ impl AsyncClient { self.prepared_probe = None; let send_call = send_finished_at.saturating_duration_since(send_call_start); - if let Err(error) = validate_datagram_length(expected_bytes, bytes) { - return Poll::Ready(Err(error)); + let receipt = SendReceipt { + seq: sent.seq, + remote: self.remote, + sent_at: sent.sent_at, + bytes: sent.bytes, + send_call, + }; + if let Err(source) = validate_datagram_length(expected_bytes, bytes) { + return Poll::Ready(Err(SendProbeError::AfterCommit { + receipt, + source: Box::new(source), + })); } - if let Err(error) = self.drain_tx_timestamps() { - return Poll::Ready(Err(error)); + if let Err(source) = self.drain_tx_timestamps() { + return Poll::Ready(Err(SendProbeError::AfterCommit { + receipt, + source: Box::new(source), + })); } - events.push(echo_sent_event(self.remote, sent, send_call)); - return Poll::Ready(Ok(events)); + return Poll::Ready(Ok(receipt)); } } diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 2706c97..27b5d2a 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -10,12 +10,13 @@ use crate::{ event::{ClientEvent, OpenOutcome}, receive::{drain_tx_timestamps, recv_datagram, try_enable_tx_timestamping, ReceivedDatagram}, session::machine::{ - recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, ProbeSent, - SessionMachine, MAX_OPEN_PACKET_SIZE, + recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, SessionMachine, + MAX_OPEN_PACKET_SIZE, }, socket::{connect_udp_socket, resolve_remote}, socket_options::{apply_traffic_class_to_socket, clear_dscp_on_socket}, timing::ClientTimestamp, + SendProbeError, SendReceipt, }; #[derive(Debug)] @@ -193,18 +194,19 @@ impl Client { /// Send one echo probe now. The caller owns pacing and run duration. /// /// The negotiated interval and duration do not gate low-level sends. - pub fn send_probe(&mut self) -> Result, ClientError> { + /// A receipt means socket acceptance and infallible session commitment have + /// completed. Failures before acceptance return [`SendProbeError::NotCommitted`]; + /// later validation or TX processing failures preserve the receipt in + /// [`SendProbeError::AfterCommit`]. + pub fn send_probe(&mut self) -> Result { let remote = self.remote; let (runtime, socket) = (&mut self.runtime, &self.socket); - let prepared = runtime.prepare_probe()?; - let machine_preflight = runtime.preflight_probe_commit(&prepared)?; - let mut events = Vec::new(); - events - .try_reserve(1) - .map_err(|source| ClientError::AllocationFailed { - operation: "probe event result", - source, - })?; + let prepared = runtime + .prepare_probe() + .map_err(SendProbeError::NotCommitted)?; + let machine_preflight = runtime + .preflight_probe_commit(&prepared) + .map_err(SendProbeError::NotCommitted)?; let expected_bytes = prepared.bytes.len(); // All fallible probe-commit preflight (timeout deadline arithmetic, @@ -212,14 +214,16 @@ impl Client { // this PRIVATE pre-send anchor, before the socket send. It is not // the public measurement `sent_at` captured further below. let send_anchor = ClientTimestamp::now(); - let machine_commit = runtime.finalize_probe_commit(machine_preflight, send_anchor)?; + let machine_commit = runtime + .finalize_probe_commit(machine_preflight, send_anchor) + .map_err(SendProbeError::NotCommitted)?; let send_call_start = Instant::now(); let bytes = match socket.send(&prepared.bytes) { Ok(bytes) => bytes, Err(error) => { runtime.invalidate_kernel_tx_correlation(); - return Err(ClientError::Socket(error)); + return Err(SendProbeError::NotCommitted(ClientError::Socket(error))); } }; let send_finished_at = Instant::now(); @@ -231,11 +235,26 @@ impl Client { let sent_at = ClientTimestamp::now(); let sent = runtime.commit_probe_sent(machine_commit, sent_at, bytes); let send_call = send_finished_at.saturating_duration_since(send_call_start); - validate_datagram_length(expected_bytes, bytes)?; - self.drain_tx_timestamps()?; + let receipt = SendReceipt { + seq: sent.seq, + remote, + sent_at: sent.sent_at, + bytes: sent.bytes, + send_call, + }; + validate_datagram_length(expected_bytes, bytes).map_err(|source| { + SendProbeError::AfterCommit { + receipt, + source: Box::new(source), + } + })?; + self.drain_tx_timestamps() + .map_err(|source| SendProbeError::AfterCommit { + receipt, + source: Box::new(source), + })?; - events.push(echo_sent_event(remote, sent, send_call)); - Ok(events) + Ok(receipt) } /// Receive and classify at most one datagram from the socket. @@ -604,22 +623,6 @@ impl Client { } } -pub(crate) fn echo_sent_event( - remote: SocketAddr, - sent: ProbeSent, - send_call: Duration, -) -> ClientEvent { - ClientEvent::EchoSent { - seq: sent.seq, - remote, - scheduled_at: None, - sent_at: sent.sent_at, - bytes: sent.bytes, - send_call, - timer_error: None, - } -} - pub(crate) fn validate_datagram_length(expected: usize, actual: usize) -> Result<(), ClientError> { if actual == expected { Ok(()) diff --git a/crates/irtt-client/src/lib.rs b/crates/irtt-client/src/lib.rs index 1115b3d..b9b1d05 100644 --- a/crates/irtt-client/src/lib.rs +++ b/crates/irtt-client/src/lib.rs @@ -19,7 +19,11 @@ //! `tokio` feature, `AsyncClient` provides the corresponding low-level Tokio //! adapter for a caller that owns a runtime and drives socket readiness //! itself. Both adapters send one probe whenever called; callers own cadence -//! and run duration, using `negotiation()` after Open. [`managed`] owns +//! and run duration, using `negotiation()` after Open. Sends return a +//! [`SendReceipt`] after socket acceptance and session commitment; +//! [`SendProbeError`] distinguishes uncommitted failures from post-commit +//! failures carrying that receipt. Convert a receipt to [`ClientEvent`] for +//! presentation or statistics. [`managed`] owns //! negotiated-interval scheduling, finite runs, and staggered/burst pacing: //! `ManagedClientTask` / //! `ManagedClientHandle` (`tokio` feature) run and control one or more @@ -54,6 +58,7 @@ pub mod managed; mod metadata; mod probe; mod receive; +mod send; mod session; mod socket; mod socket_options; @@ -73,5 +78,6 @@ pub use event::{ ClientEvent, NoTestCompleted, OneWayDelaySample, OpenOutcome, PacketMeta, ReceivedStatsSample, RttSample, ServerTiming, SessionStarted, SignedDuration, WarningKind, }; +pub use send::{SendProbeError, SendReceipt}; pub use session::{AcceptedSessionParameters, NegotiationChange, NegotiationResult}; pub use timing::ClientTimestamp; diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index b91267e..f79fb63 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -21,7 +21,7 @@ use crate::{ async_client::{AsyncClient, AsyncOpenState}, session::machine::SessionMachine, socket_options::validate_ttl, - ClientConfig, ClientError, ClientEvent, OpenOutcome, + ClientConfig, ClientError, ClientEvent, OpenOutcome, SendProbeError, }; use super::schedule::{instant_abs_diff, ProbeSchedule}; @@ -1623,10 +1623,13 @@ impl ManagedClientTask { } }; let scheduled_at = commit.scheduled_at; - let before = client.packets_sent(); - let mut result = client.poll_send_probe(cx); - let after = client.packets_sent(); - let accepted = after > before; + let result = client.poll_send_probe(cx); + let receipt = match &result { + Poll::Ready(Ok(receipt)) + | Poll::Ready(Err(SendProbeError::AfterCommit { receipt, .. })) => Some(*receipt), + Poll::Pending | Poll::Ready(Err(SendProbeError::NotCommitted(_))) => None, + }; + let accepted = receipt.is_some(); let send_result = match &result { Poll::Pending => SendResult::Pending, Poll::Ready(Ok(_)) => SendResult::Ready { accepted }, @@ -1639,20 +1642,6 @@ impl ManagedClientTask { .as_mut() .unwrap() .commit(commit); - if let Poll::Ready(Ok(events)) = &mut result { - for event in events { - if let ClientEvent::EchoSent { - scheduled_at: intended, - sent_at, - timer_error, - .. - } = event - { - *intended = Some(scheduled_at); - *timer_error = Some(instant_abs_diff(sent_at.mono, scheduled_at)); - } - } - } } let schedule_error = if accepted { self.targets[index] @@ -1664,17 +1653,32 @@ impl ManagedClientTask { } else { None }; - self.targets[index].counters.packets_sent = after; + self.targets[index].sync_packets_sent(&client); + // A committed send must reach the lifecycle/accounting stream even + // when post-send processing fails. Publish before failure cleanup. + let sent_event = receipt.map(|receipt| { + let mut event = ClientEvent::from(receipt); + if let ClientEvent::EchoSent { + scheduled_at: intended, + timer_error, + .. + } = &mut event + { + *intended = Some(scheduled_at); + *timer_error = Some(instant_abs_diff(receipt.sent_at.mono, scheduled_at)); + } + event + }); match result { Poll::Pending => { self.targets[index].send_waiting = true; self.targets[index].state = TargetState::Active { client }; SendResult::Pending } - Poll::Ready(Ok(events)) => { + Poll::Ready(Ok(_)) => { self.targets[index].send_waiting = false; self.targets[index].state = TargetState::Active { client }; - self.publish_client_events(index, events); + self.publish_client_events(index, vec![sent_event.unwrap()]); if let Some(error) = schedule_error { let TargetState::Active { client } = mem::replace(&mut self.targets[index].state, TargetState::Terminal) @@ -1696,6 +1700,13 @@ impl ManagedClientTask { } Poll::Ready(Err(error)) => { + if let Some(event) = sent_event { + self.publish_client_events(index, vec![event]); + } + let error = match error { + SendProbeError::NotCommitted(source) => source, + SendProbeError::AfterCommit { source, .. } => *source, + }; self.targets[index].send_waiting = false; self.begin_open_session_failure( index, diff --git a/crates/irtt-client/src/send.rs b/crates/irtt-client/src/send.rs new file mode 100644 index 0000000..830e442 --- /dev/null +++ b/crates/irtt-client/src/send.rs @@ -0,0 +1,80 @@ +use std::{error::Error, fmt, net::SocketAddr, time::Duration}; + +use crate::{ClientError, ClientEvent, ClientTimestamp}; + +/// An Echo datagram accepted by the UDP socket and committed in session state. +/// +/// The wire sequence and sent count have advanced, and the probe is pending. +/// This receipt remains authoritative even if subsequent validation or kernel +/// TX timestamp processing fails. It carries transport facts only; managed +/// scheduling annotations belong on the corresponding [`ClientEvent::EchoSent`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SendReceipt { + /// Committed wire sequence number. + pub seq: u32, + /// Resolved remote socket address. + pub remote: SocketAddr, + /// Public measurement timestamp sampled immediately after socket acceptance. + /// The private pre-send anchor still determines the timeout deadline. + pub sent_at: ClientTimestamp, + /// Number of bytes the socket reported accepting. + pub bytes: usize, + /// Time spent in the successful socket send call. + pub send_call: Duration, +} + +impl From for ClientEvent { + /// Construct the committed send event without scheduling annotations. + fn from(receipt: SendReceipt) -> Self { + Self::EchoSent { + seq: receipt.seq, + remote: receipt.remote, + scheduled_at: None, + sent_at: receipt.sent_at, + bytes: receipt.bytes, + send_call: receipt.send_call, + timer_error: None, + } + } +} + +/// Probe-send failure with an explicit socket-acceptance commitment boundary. +/// +/// There is deliberately no blanket conversion from [`ClientError`]: each send +/// failure must be classified at the point where its commitment is known. +#[derive(Debug)] +pub enum SendProbeError { + /// No datagram was accepted and no probe state or sequence was committed. + NotCommitted(ClientError), + /// The probe was committed, then post-send processing failed. + /// The pending probe and advanced sequence/count remain committed. + AfterCommit { + /// Authoritative receipt for the accepted send. + receipt: SendReceipt, + /// Operational cause of the post-send failure. + source: Box, + }, +} + +impl fmt::Display for SendProbeError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::NotCommitted(source) => write!(f, "probe send was not committed: {source}"), + Self::AfterCommit { source, .. } => { + write!( + f, + "probe send was committed, but post-send processing failed: {source}" + ) + } + } + } +} + +impl Error for SendProbeError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + Some(match self { + Self::NotCommitted(source) => source, + Self::AfterCommit { source, .. } => source.as_ref(), + }) + } +} diff --git a/crates/irtt-client/tests/client/close.rs b/crates/irtt-client/tests/client/close.rs index 5c5317b..fe115b7 100644 --- a/crates/irtt-client/tests/client/close.rs +++ b/crates/irtt-client/tests/client/close.rs @@ -60,7 +60,9 @@ fn send_probe_fails_after_close() { client.close().unwrap(); assert!(matches!( client.send_probe(), - Err(ClientError::AlreadyClosed) + Err(irtt_client::SendProbeError::NotCommitted( + ClientError::AlreadyClosed + )) )); server.join(); } @@ -206,9 +208,7 @@ fn close_flagged_retained_timeout_emits_late_then_closes() { .unwrap(); assert_open_started(client.open().unwrap()); let sent = client.send_probe().unwrap(); - let ClientEvent::EchoSent { sent_at, .. } = &sent[0] else { - panic!("expected EchoSent"); - }; + let sent_at = sent.sent_at; assert!(matches!( client .poll_timeouts_at(sent_at.mono + client.probe_timeout()) @@ -426,10 +426,7 @@ fn wrong_token_close_flag_does_not_close_session() { }] )); assert!(!client.is_peer_closed()); - assert!(matches!( - client.send_probe().unwrap().as_slice(), - [ClientEvent::EchoSent { seq: 1, .. }] - )); + assert_eq!(client.send_probe().unwrap().seq, 1); client.close().unwrap(); server.join(); } @@ -498,7 +495,9 @@ fn close_flagged_echo_reply_emits_reply_then_closes_without_sending_close() { assert!(matches!( client.send_probe(), - Err(ClientError::AlreadyClosed) + Err(irtt_client::SendProbeError::NotCommitted( + ClientError::AlreadyClosed + )) )); let first = server.rx.recv_timeout(Duration::from_millis(100)).unwrap(); diff --git a/crates/irtt-client/tests/client/no_test.rs b/crates/irtt-client/tests/client/no_test.rs index 98b88d6..ab7c9f7 100644 --- a/crates/irtt-client/tests/client/no_test.rs +++ b/crates/irtt-client/tests/client/no_test.rs @@ -101,7 +101,9 @@ fn send_probe_fails_after_no_test_completed() { assert_no_test_completed(client.open().unwrap()); assert!(matches!( client.send_probe(), - Err(ClientError::AlreadyCompleted) + Err(irtt_client::SendProbeError::NotCommitted( + ClientError::AlreadyCompleted + )) )); server.join(); } diff --git a/crates/irtt-client/tests/client/probes_replies/timeouts.rs b/crates/irtt-client/tests/client/probes_replies/timeouts.rs index edcf79e..cd7af57 100644 --- a/crates/irtt-client/tests/client/probes_replies/timeouts.rs +++ b/crates/irtt-client/tests/client/probes_replies/timeouts.rs @@ -26,7 +26,9 @@ fn probe_timeout_overflow_transmits_no_echo() { assert_open_started(client.open().unwrap()); assert!(matches!( client.send_probe(), - Err(ClientError::DurationOverflow) + Err(irtt_client::SendProbeError::NotCommitted( + ClientError::DurationOverflow + )) )); assert!(!client.has_pending_probes()); client.close().unwrap(); @@ -181,7 +183,9 @@ fn pending_full_does_not_send_packet() { assert!(matches!( client.send_probe(), - Err(ClientError::PendingLimitExceeded { limit: 2 }) + Err(irtt_client::SendProbeError::NotCommitted( + ClientError::PendingLimitExceeded { limit: 2 } + )) )); thread::sleep(Duration::from_millis(30)); diff --git a/crates/irtt-client/tests/linux_transport.rs b/crates/irtt-client/tests/linux_transport.rs index c6abee7..cdbc68e 100644 --- a/crates/irtt-client/tests/linux_transport.rs +++ b/crates/irtt-client/tests/linux_transport.rs @@ -261,10 +261,23 @@ fn a_closed_udp_peer_surfaces_a_socket_error() { close_tx.send(()).unwrap(); task.join().unwrap(); // Refusal can arrive during send's TX drain or the following receive. - let result = client.send_probe().and_then(|_| client.recv_once()); + // Either way this first Echo was accepted and must stay pending. + let (receipt, error) = match client.send_probe() { + Ok(receipt) => (receipt, client.recv_once().unwrap_err()), + Err(irtt_client::SendProbeError::AfterCommit { receipt, source }) => (receipt, *source), + Err(error) => panic!("first Echo should be accepted: {error:?}"), + }; + assert_eq!(receipt.seq, 0); + assert_eq!(receipt.remote, addr); + assert!(client.has_pending_probes()); assert!( - matches!(result, Err(irtt_client::ClientError::Socket(ref error)) + matches!(error, irtt_client::ClientError::Socket(ref error) if error.kind() == std::io::ErrorKind::ConnectionRefused), - "{result:?}" + "{error:?}" ); + assert!(matches!( + client.poll_timeouts_at(receipt.sent_at.mono + client.probe_timeout()) + .unwrap().as_slice(), + [ClientEvent::EchoLoss { seq: 0, sent_at, .. }] if *sent_at == receipt.sent_at + )); } diff --git a/crates/irtt-integration-tests/tests/async_client.rs b/crates/irtt-integration-tests/tests/async_client.rs index 949a626..5ea93e3 100644 --- a/crates/irtt-integration-tests/tests/async_client.rs +++ b/crates/irtt-integration-tests/tests/async_client.rs @@ -72,14 +72,10 @@ async fn exercise_client_server() { OpenOutcome::NoTestCompleted(_) => panic!("normal client unexpectedly ran no-test"), } - assert!(matches!( - client.send_probe().await.unwrap().as_slice(), - [ClientEvent::EchoSent { - seq: 0, - remote, - .. - }] if *remote == server_addr - )); + let receipt = client.send_probe().await.unwrap(); + assert_eq!(receipt.seq, 0); + assert_eq!(receipt.remote, server_addr); + assert!(client.has_pending_probes()); let events = client.recv().await.unwrap(); let [ClientEvent::EchoReply { seq: 0, @@ -97,6 +93,7 @@ async fn exercise_client_server() { panic!("expected a measured reply, got {events:?}"); }; assert_eq!(*remote, server_addr); + assert_eq!(*sent_at, receipt.sent_at); assert_eq!(rtt.raw, received_at.mono.duration_since(sent_at.mono)); // Verify the measurement endpoint through public reply metadata and real // OS timestamps. This also covers the userspace fallback on other targets. diff --git a/crates/irtt-integration-tests/tests/async_client_lifecycle.rs b/crates/irtt-integration-tests/tests/async_client_lifecycle.rs index dbd874a..bcf537a 100644 --- a/crates/irtt-integration-tests/tests/async_client_lifecycle.rs +++ b/crates/irtt-integration-tests/tests/async_client_lifecycle.rs @@ -17,7 +17,7 @@ use tokio::runtime::{Builder, Runtime}; mod support; use irtt_client::{AsyncClient, ClientConfig, ClientError, ClientEvent, OpenOutcome}; -use irtt_client::{Client, RunMode}; +use irtt_client::{Client, RunMode, SendProbeError}; use irtt_server::ServerConfig; use std::{ io, @@ -274,6 +274,11 @@ fn recv_before_open_fails_on_first_poll_without_socket_readiness() { poll_recv_once(&mut client), Poll::Ready(Err(ClientError::NotOpen)) )); + let mut send = Box::pin(client.send_probe()); + assert!(matches!( + poll_once(send.as_mut()), + Poll::Ready(Err(SendProbeError::NotCommitted(ClientError::NotOpen))) + )); }); } @@ -376,6 +381,13 @@ fn recv_after_local_close_and_no_test_fails_on_first_poll() { poll_recv_once(&mut client), Poll::Ready(Err(ClientError::AlreadyClosed)) )); + let mut send = Box::pin(client.send_probe()); + assert!(matches!( + poll_once(send.as_mut()), + Poll::Ready(Err(SendProbeError::NotCommitted( + ClientError::AlreadyClosed + ))) + )); }); assert_eq!(close_server.finish().len(), 2); @@ -391,6 +403,13 @@ fn recv_after_local_close_and_no_test_fails_on_first_poll() { poll_recv_once(&mut client), Poll::Ready(Err(ClientError::AlreadyCompleted)) )); + let mut send = Box::pin(client.send_probe()); + assert!(matches!( + poll_once(send.as_mut()), + Poll::Ready(Err(SendProbeError::NotCommitted( + ClientError::AlreadyCompleted + ))) + )); }); assert_eq!(no_test_server.finish().len(), 1); } @@ -556,7 +575,7 @@ fn blocking_and_async_roll_back_a_failed_open_and_accept_a_retry() { assert!(!blocking.has_pending_probes()); let blocking_second = blocking.open().unwrap(); assert!(!blocking.has_pending_probes()); - let blocking_sent = blocking.send_probe().unwrap(); + let blocking_sent = [ClientEvent::from(blocking.send_probe().unwrap())]; let blocking_reply = blocking.recv_once().unwrap(); let blocking_close = blocking.close().unwrap(); assert_eq!(blocking_server.finish().len(), 4); @@ -572,7 +591,7 @@ fn blocking_and_async_roll_back_a_failed_open_and_accept_a_retry() { assert!(!client.has_pending_probes()); let second = client.open().await.unwrap(); assert!(!client.has_pending_probes()); - let sent = client.send_probe().await.unwrap(); + let sent = [ClientEvent::from(client.send_probe().await.unwrap())]; let reply = client.recv().await.unwrap(); let closed = client.close().await.unwrap(); (first, second, sent, reply, closed) @@ -624,7 +643,7 @@ fn blocking_and_async_complete_every_caller_paced_probe() { let mut blocking_sent = Vec::new(); let mut blocking_replies = Vec::new(); for _ in 0..PROBES { - blocking_sent.extend(blocking.send_probe().unwrap()); + blocking_sent.push(ClientEvent::from(blocking.send_probe().unwrap())); assert!(blocking.has_pending_probes()); assert!(blocking.next_probe_timeout_deadline().is_some()); blocking_replies.extend(blocking.recv_once().unwrap()); @@ -644,7 +663,7 @@ fn blocking_and_async_complete_every_caller_paced_probe() { let mut sent = Vec::new(); let mut replies = Vec::new(); for _ in 0..PROBES { - sent.extend(client.send_probe().await.unwrap()); + sent.push(ClientEvent::from(client.send_probe().await.unwrap())); assert!(client.has_pending_probes()); assert!(client.next_probe_timeout_deadline().is_some()); replies.extend(client.recv().await.unwrap()); diff --git a/crates/irtt-integration-tests/tests/hmac.rs b/crates/irtt-integration-tests/tests/hmac.rs index a4e512e..0ad37b6 100644 --- a/crates/irtt-integration-tests/tests/hmac.rs +++ b/crates/irtt-integration-tests/tests/hmac.rs @@ -109,7 +109,9 @@ fn hmac_close_success_sends_authenticated_close_and_closes_session() { )); assert!(matches!( client.send_probe(), - Err(ClientError::AlreadyClosed) + Err(irtt_client::SendProbeError::NotCommitted( + ClientError::AlreadyClosed + )) )); let observations = server.observations(2); @@ -175,7 +177,7 @@ fn bad_hmac_echo_reply_is_rejected_without_echo_reply_event() { assert_started(outcome, ¶ms); let sent = client.send_probe().unwrap(); - assert_eq!(sent.len(), 1); + assert_eq!(sent.seq, 0); let events = client.recv_once().unwrap(); assert!(matches!( events.as_slice(), @@ -292,7 +294,7 @@ fn backend_hmac_correct_key_succeeds() { assert!(matches!(outcome, OpenOutcome::Started(_))); let sent = client.send_probe().unwrap(); - assert_eq!(sent.len(), 1); + assert_eq!(sent.seq, 0); let events = client.recv_once().unwrap(); assert_eq!(events.len(), 1); diff --git a/crates/irtt-integration-tests/tests/protocol_options/backend_smoke.rs b/crates/irtt-integration-tests/tests/protocol_options/backend_smoke.rs index 3b5c195..a69ec07 100644 --- a/crates/irtt-integration-tests/tests/protocol_options/backend_smoke.rs +++ b/crates/irtt-integration-tests/tests/protocol_options/backend_smoke.rs @@ -17,7 +17,7 @@ fn backend_basic_open_echo_close() { assert!(matches!(outcome, irtt_client::OpenOutcome::Started(_))); let sent = client.send_probe().unwrap(); - assert_eq!(sent.len(), 1); + assert_eq!(sent.seq, 0); let events = client.recv_once().unwrap(); assert_eq!(events.len(), 1); diff --git a/crates/irtt-integration-tests/tests/support/mod.rs b/crates/irtt-integration-tests/tests/support/mod.rs index f5b38b0..d53e61c 100644 --- a/crates/irtt-integration-tests/tests/support/mod.rs +++ b/crates/irtt-integration-tests/tests/support/mod.rs @@ -243,8 +243,7 @@ where .unwrap(); let negotiated = assert_started(client.open().unwrap()); - let sent_events = client.send_probe().unwrap(); - assert_eq!(sent_events.len(), 1); + let receipt = client.send_probe().unwrap(); let reply_events = client.recv_once().unwrap(); assert_eq!(reply_events.len(), 1); @@ -254,7 +253,7 @@ where OneProbeRun { negotiated, - sent: sent_events.into_iter().next().unwrap(), + sent: receipt.into(), reply: reply_events.into_iter().next().unwrap(), observations, }