feat(client): expose probe send commitment through typed receipts

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<Vec<ClientEvent>, 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<SendReceipt, SendProbeError> {
         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<Result<Vec<ClientEvent>, ClientError>> {
+    ) -> Poll<Result<SendReceipt, SendProbeError>> {
         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));
         }
     }
 

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<Vec<ClientEvent>, 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<SendReceipt, SendProbeError> {
         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(())

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;

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,

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<SendReceipt> 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<ClientError>,
+    },
+}
+
+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(),
+        })
+    }
+}