From e861de5247a341a13fbeac3b130637bfd289d873 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 15 Sep 2026 13:03:36 +0700 Subject: [PATCH] fix(client): invalidate TX correlation after send failures --- crates/irtt-client/AGENTS.md | 5 ++- crates/irtt-client/src/async_client.rs | 10 ++++-- crates/irtt-client/src/client.rs | 15 ++++++--- crates/irtt-client/src/probe.rs | 3 +- crates/irtt-client/src/receive/linux.rs | 4 ++- crates/irtt-client/src/session/machine.rs | 38 +++++++++++++++++++++++ 6 files changed, 66 insertions(+), 9 deletions(-) diff --git a/crates/irtt-client/AGENTS.md b/crates/irtt-client/AGENTS.md index abc0c7a..f72ba5a 100644 --- a/crates/irtt-client/AGENTS.md +++ b/crates/irtt-client/AGENTS.md @@ -39,7 +39,10 @@ assumption that failed or nonblocking sends cannot cause that desynchrony. The effect is confined to the optional timing enhancement. A kernel timestamp whose ID cannot be matched to a pending or retained timed-out probe is discarded, as is any implausible timestamp; `sent_at` remains the ordinary -userspace fallback. This is safe because: +userspace fallback. **After every failed probe submission, disable kernel-ID +correlation for the rest of that session.** A gap means an otherwise plausible +later ID can name the wrong probe, so subsequent error-queue timestamps must +be discarded rather than attached by ID. This is safe because: - **Normal successful sends track together.** `wire_seq` starts at 0 on every successful Open (`SessionMachine::commit_open` builds a fresh diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index 2c2bfa6..c531c47 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -608,8 +608,14 @@ impl AsyncClient { // No sent_at measurement, no machine commit, no schedule // commit: the transaction is fully discarded and retried // with a fresh send_anchor on the next loop iteration. - Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue, - Err(error) => return Poll::Ready(Err(ClientError::Socket(error))), + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + self.machine.invalidate_kernel_tx_correlation(); + continue; + } + Err(error) => { + self.machine.invalidate_kernel_tx_correlation(); + return Poll::Ready(Err(ClientError::Socket(error))); + } }; #[cfg(not(test))] diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index ceee7a9..2d18a5d 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -167,9 +167,9 @@ impl Client { /// unrelated datagrams are ignored until the current attempt's absolute /// deadline, so one attempt may consume several datagrams without /// retransmitting. Silence or ignored traffic eventually produces - /// [`ClientError::OpenTimeout`]. A malformed or incompatible reply that is - /// recognized as coming from the connected peer is terminal; when HMAC is - /// configured, that recognition additionally requires authentication. + /// [`ClientError::OpenTimeout`]. A structurally recognizable malformed or + /// incompatible Open reply from the connected peer is terminal; when HMAC + /// is configured, that recognition additionally requires authentication. /// /// When a trusted reply allocates a token but later negotiation or socket /// preparation fails, the client sends a best-effort cleanup close and @@ -524,11 +524,18 @@ impl Client { .unwrap_or_else(Instant::now); #[cfg(test)] if fail_send { + runtime.invalidate_kernel_tx_correlation(); return Err(ClientError::Socket(io::Error::other( "injected probe send failure", ))); } - let bytes = socket.send(&prepared.bytes)?; + let bytes = match socket.send(&prepared.bytes) { + Ok(bytes) => bytes, + Err(error) => { + runtime.invalidate_kernel_tx_correlation(); + return Err(ClientError::Socket(error)); + } + }; #[cfg(not(test))] let send_finished_at = Instant::now(); #[cfg(test)] diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index 6a5d58c..4d035cf 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -39,7 +39,8 @@ pub(crate) struct PendingProbe { /// eligible only for upstream one-way delay after local plausibility /// validation against `tx_not_before_wall` (see `compute_one_way`'s /// `preferred_send_wall`). `sent_at` remains the userspace fallback for - /// upstream one-way delay when this is absent or implausible. + /// upstream one-way delay when this is absent, implausible, or kernel-ID + /// correlation was invalidated by a failed probe submission. pub kernel_tx_timestamp: Option, } diff --git a/crates/irtt-client/src/receive/linux.rs b/crates/irtt-client/src/receive/linux.rs index 893e116..53c9a53 100644 --- a/crates/irtt-client/src/receive/linux.rs +++ b/crates/irtt-client/src/receive/linux.rs @@ -17,7 +17,9 @@ //! A kernel ID can theoretically be consumed by a send that later fails, //! desynchronizing that correlation. [`drain_tx_timestamps`] discards //! unmatched or implausible records; the userspace `sent_at` timestamp remains -//! the fallback. It performs a small bounded, nonblocking read of +//! the fallback. After a probe send failure, correlation is disabled for that +//! session so later IDs cannot be misattributed. It performs a small bounded, +//! nonblocking read of //! `MSG_ERRQUEUE` so the adapter can opportunistically collect timestamps //! without ever waiting for one. The observed kernel TX wall time is retained //! as metadata only; it is not consumed by any measurement in this change. diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 3c925f5..4398d9d 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -60,6 +60,7 @@ struct ActiveSession { pending: PendingMap, timed_out: TimedOutMap, completed: CompletedSet, + kernel_tx_correlation_valid: bool, } #[derive(Debug)] @@ -423,6 +424,18 @@ impl SessionMachine { } } + /// Disable Linux kernel TX timestamp correlation after a probe send fails. + /// + /// A failed submission can consume a kernel `SOF_TIMESTAMPING_OPT_ID` + /// without advancing the wire sequence. Existing associated timestamps + /// remain valid observations; later error-queue timestamps cannot be + /// associated safely and are ignored for the rest of this session. + pub(crate) fn invalidate_kernel_tx_correlation(&mut self) { + if let MachineState::Open(session) = &mut self.state { + session.kernel_tx_correlation_valid = false; + } + } + /// Record an observed Linux kernel TX timestamp for `wire_seq`, the /// automatic `SOF_TIMESTAMPING_OPT_ID` the kernel assigned when the /// datagram was submitted. @@ -443,6 +456,9 @@ impl SessionMachine { let MachineState::Open(session) = &mut self.state else { return; }; + if !session.kernel_tx_correlation_valid { + return; + } if let Some(probe) = session.pending.get_mut(wire_seq) { probe.kernel_tx_timestamp.get_or_insert(timestamp); return; @@ -636,6 +652,7 @@ impl SessionMachine { pending: PendingMap::new(self.config.max_pending_probes), timed_out: TimedOutMap::new(self.config.max_pending_probes), completed: CompletedSet::new(self.config.max_pending_probes), + kernel_tx_correlation_valid: true, })); let event = ClientEvent::SessionStarted { @@ -1231,6 +1248,7 @@ mod tests { pending: PendingMap::new(max_pending_probes), timed_out: TimedOutMap::new(max_pending_probes), completed: CompletedSet::new(max_pending_probes), + kernel_tx_correlation_valid: true, })); machine } @@ -2873,6 +2891,25 @@ mod tests { ); } + #[test] + fn invalidated_kernel_tx_correlation_uses_userspace_send_time() { + let mono = Instant::now(); + let mut machine = upstream_probe(send_wall_ms(1_000), mono); + + // A failed send can consume a kernel ID without advancing wire_seq. + // A later timestamp whose ID happens to match a pending wire sequence + // must not be associated after that gap. + machine.invalidate_kernel_tx_correlation(); + machine.record_kernel_tx_timestamp(0, send_wall_ms(1_005)); + + let events = upstream_reply(&mut machine, 1_020, send_wall_ms(1_040), mono); + + assert_eq!( + reply_one_way(&events).unwrap().client_to_server, + Some(SignedDuration::from_nanos(20_000_000)) + ); + } + #[test] fn upstream_one_way_delay_falls_back_without_kernel_tx() { let mono = Instant::now(); @@ -3195,6 +3232,7 @@ mod tests { pending: PendingMap::new(MAX_PENDING_PROBES), timed_out: TimedOutMap::new(MAX_PENDING_PROBES), completed: CompletedSet::new(MAX_PENDING_PROBES), + kernel_tx_correlation_valid: true, })); (machine, params) }