fix(client): invalidate TX correlation after send failures

This commit is contained in:
Nils Andreas Svee committed 2026-09-15 13:03:36 +07:00
1 parent 188d6e5354
commit e861de5247
6 files changed
+66 -9

No files matched your search

+4 -1
View File
@@ -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 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 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 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 - **Normal successful sends track together.** `wire_seq` starts at 0 on every
successful Open (`SessionMachine::commit_open` builds a fresh successful Open (`SessionMachine::commit_open` builds a fresh
+8 -2
View File
@@ -608,8 +608,14 @@ impl AsyncClient {
// No sent_at measurement, no machine commit, no schedule // No sent_at measurement, no machine commit, no schedule
// commit: the transaction is fully discarded and retried // commit: the transaction is fully discarded and retried
// with a fresh send_anchor on the next loop iteration. // with a fresh send_anchor on the next loop iteration.
Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue, Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
Err(error) => return Poll::Ready(Err(ClientError::Socket(error))), 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))] #[cfg(not(test))]
+11 -4
View File
@@ -167,9 +167,9 @@ impl Client {
/// unrelated datagrams are ignored until the current attempt's absolute /// unrelated datagrams are ignored until the current attempt's absolute
/// deadline, so one attempt may consume several datagrams without /// deadline, so one attempt may consume several datagrams without
/// retransmitting. Silence or ignored traffic eventually produces /// retransmitting. Silence or ignored traffic eventually produces
/// [`ClientError::OpenTimeout`]. A malformed or incompatible reply that is /// [`ClientError::OpenTimeout`]. A structurally recognizable malformed or
/// recognized as coming from the connected peer is terminal; when HMAC is /// incompatible Open reply from the connected peer is terminal; when HMAC
/// configured, that recognition additionally requires authentication. /// is configured, that recognition additionally requires authentication.
/// ///
/// When a trusted reply allocates a token but later negotiation or socket /// When a trusted reply allocates a token but later negotiation or socket
/// preparation fails, the client sends a best-effort cleanup close and /// preparation fails, the client sends a best-effort cleanup close and
@@ -524,11 +524,18 @@ impl Client {
.unwrap_or_else(Instant::now); .unwrap_or_else(Instant::now);
#[cfg(test)] #[cfg(test)]
if fail_send { if fail_send {
runtime.invalidate_kernel_tx_correlation();
return Err(ClientError::Socket(io::Error::other( return Err(ClientError::Socket(io::Error::other(
"injected probe send failure", "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))] #[cfg(not(test))]
let send_finished_at = Instant::now(); let send_finished_at = Instant::now();
#[cfg(test)] #[cfg(test)]
+2 -1
View File
@@ -39,7 +39,8 @@ pub(crate) struct PendingProbe {
/// eligible only for upstream one-way delay after local plausibility /// eligible only for upstream one-way delay after local plausibility
/// validation against `tx_not_before_wall` (see `compute_one_way`'s /// validation against `tx_not_before_wall` (see `compute_one_way`'s
/// `preferred_send_wall`). `sent_at` remains the userspace fallback for /// `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<SystemTime>, pub kernel_tx_timestamp: Option<SystemTime>,
} }
+3 -1
View File
@@ -17,7 +17,9 @@
//! A kernel ID can theoretically be consumed by a send that later fails, //! A kernel ID can theoretically be consumed by a send that later fails,
//! desynchronizing that correlation. [`drain_tx_timestamps`] discards //! desynchronizing that correlation. [`drain_tx_timestamps`] discards
//! unmatched or implausible records; the userspace `sent_at` timestamp remains //! 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 //! `MSG_ERRQUEUE` so the adapter can opportunistically collect timestamps
//! without ever waiting for one. The observed kernel TX wall time is retained //! 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. //! as metadata only; it is not consumed by any measurement in this change.
+38
View File
@@ -60,6 +60,7 @@ struct ActiveSession {
pending: PendingMap, pending: PendingMap,
timed_out: TimedOutMap, timed_out: TimedOutMap,
completed: CompletedSet, completed: CompletedSet,
kernel_tx_correlation_valid: bool,
} }
#[derive(Debug)] #[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 /// Record an observed Linux kernel TX timestamp for `wire_seq`, the
/// automatic `SOF_TIMESTAMPING_OPT_ID` the kernel assigned when the /// automatic `SOF_TIMESTAMPING_OPT_ID` the kernel assigned when the
/// datagram was submitted. /// datagram was submitted.
@@ -443,6 +456,9 @@ impl SessionMachine {
let MachineState::Open(session) = &mut self.state else { let MachineState::Open(session) = &mut self.state else {
return; return;
}; };
if !session.kernel_tx_correlation_valid {
return;
}
if let Some(probe) = session.pending.get_mut(wire_seq) { if let Some(probe) = session.pending.get_mut(wire_seq) {
probe.kernel_tx_timestamp.get_or_insert(timestamp); probe.kernel_tx_timestamp.get_or_insert(timestamp);
return; return;
@@ -636,6 +652,7 @@ impl SessionMachine {
pending: PendingMap::new(self.config.max_pending_probes), pending: PendingMap::new(self.config.max_pending_probes),
timed_out: TimedOutMap::new(self.config.max_pending_probes), timed_out: TimedOutMap::new(self.config.max_pending_probes),
completed: CompletedSet::new(self.config.max_pending_probes), completed: CompletedSet::new(self.config.max_pending_probes),
kernel_tx_correlation_valid: true,
})); }));
let event = ClientEvent::SessionStarted { let event = ClientEvent::SessionStarted {
@@ -1231,6 +1248,7 @@ mod tests {
pending: PendingMap::new(max_pending_probes), pending: PendingMap::new(max_pending_probes),
timed_out: TimedOutMap::new(max_pending_probes), timed_out: TimedOutMap::new(max_pending_probes),
completed: CompletedSet::new(max_pending_probes), completed: CompletedSet::new(max_pending_probes),
kernel_tx_correlation_valid: true,
})); }));
machine 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] #[test]
fn upstream_one_way_delay_falls_back_without_kernel_tx() { fn upstream_one_way_delay_falls_back_without_kernel_tx() {
let mono = Instant::now(); let mono = Instant::now();
@@ -3195,6 +3232,7 @@ mod tests {
pending: PendingMap::new(MAX_PENDING_PROBES), pending: PendingMap::new(MAX_PENDING_PROBES),
timed_out: TimedOutMap::new(MAX_PENDING_PROBES), timed_out: TimedOutMap::new(MAX_PENDING_PROBES),
completed: CompletedSet::new(MAX_PENDING_PROBES), completed: CompletedSet::new(MAX_PENDING_PROBES),
kernel_tx_correlation_valid: true,
})); }));
(machine, params) (machine, params)
} }