Merge pull request #106 from Lochnair/fix/client-post-send-timestamp

fix(client): timestamp probes after successful send
This commit was merged in pull request #106.
This commit is contained in:
Nils Andreas Svee authored and GitHub committed 2026-08-15 20:02:51 +07:00
commit 6d7adf52a7
11 files changed
+549 -129

No files matched your search

+64 -22
View File
@@ -96,33 +96,75 @@ socket.
## Retained state ## Retained state
`PendingProbe` carries an optional `kernel_tx_timestamp: Option<SystemTime>` `PendingProbe` carries an optional `kernel_tx_timestamp: Option<SystemTime>`
alongside the authoritative userspace `sent_at`. It survives pending -> alongside the userspace `sent_at`. It survives pending -> timed-out retention
timed-out retention (`SessionMachine::record_kernel_tx_timestamp` updates (`SessionMachine::record_kernel_tx_timestamp` updates whichever of
whichever of `PendingMap`/`TimedOutMap` still holds the probe) and is `PendingMap`/`TimedOutMap` still holds the probe) and is otherwise bounded by
otherwise bounded by the same limits those maps already enforce — no new the same limits those maps already enforce — no new unbounded state. First
unbounded state. First valid observation wins on duplicates. valid observation wins on duplicates.
**It is eligible for exactly one measurement: the upstream (`client_to_server`) ### Pre-send anchor vs. post-send `sent_at` (F-04)
one-way delay send endpoint.** `compute_one_way`'s private `preferred_send_wall`
helper prefers it over the userspace `sent_at.wall` sample when
sent_at.wall <= kernel_tx_timestamp <= reply_received_at.wall The probe send transaction splits into two timestamps around the socket
send, not one:
- A **private pre-send anchor**, sampled immediately before the fallible
commit preflight (`SessionMachine::finalize_probe_commit`) that must
happen before the socket send: timeout deadline `checked_add` arithmetic
(so `DurationOverflow` still guarantees nothing was transmitted) and the
kernel TX lower-plausibility bound. Its monotonic half becomes
`PendingProbe::timeout_at`'s basis; its wall half becomes
`PendingProbe::tx_not_before_wall`. It is never exposed publicly.
- The **public `sent_at`**, a paired wall/monotonic `ClientTimestamp`
captured immediately after the socket send returns successfully
(`SessionMachine::commit_probe_sent`, which is infallible — all fallible
work already happened against the pre-send anchor). This is the RTT
endpoint, the userspace upstream-OWD fallback endpoint, `timer_error`'s
basis, and the value surfaced on `EchoSent`/`EchoReply`/`EchoLoss`/
`LateReply`.
So `timeout_at` is deliberately **not** derived from `sent_at.mono` — it
remains anchored to before the send, on purpose, so the existing pre-send
timeout/overflow/no-send contract is unchanged. Only the *measurement*
`sent_at` moved to after the send. A `WouldBlock` `try_send` (Tokio) never
samples a `sent_at`, never calls `finalize_probe_commit`'s matching
`commit_probe_sent`, and never advances the machine or schedule; each retry
gets its own fresh pre-send anchor, and only a successful attempt's anchor
and `sent_at` are ever retained.
**The kernel TX timestamp is eligible for exactly one measurement: the
upstream (`client_to_server`) one-way delay send endpoint.**
`compute_one_way`'s private `preferred_send_wall` helper prefers it over the
userspace `sent_at.wall` sample when
tx_not_before_wall <= kernel_tx_timestamp <= reply_received_at.wall
(inclusive, compared only against the client's own wall clock — never (inclusive, compared only against the client's own wall clock — never
against the remote server's wall timestamp). Otherwise, including when no against the remote server's wall timestamp). The lower bound is the
kernel timestamp was observed, the upstream endpoint falls back to pre-send `tx_not_before_wall`, not the post-send `sent_at.wall`: a
`sent_at.wall`; a rejected observation is never erased from `PendingProbe`, legitimate `TX_SOFTWARE` timestamp can be generated while the send syscall
only skipped for that measurement. There is no maximum-lag bound: unlike the is still in flight, so it can legitimately precede the post-send `sent_at`
kernel receive side, legitimate client send-to-reply time can exceed a sample without being implausible. Otherwise, including when no kernel
second. timestamp was observed, the upstream endpoint falls back to `sent_at.wall`
(the post-send sample); a rejected observation is never erased from
`PendingProbe`, only skipped for that measurement. There is no
maximum-lag bound: unlike the kernel receive side, legitimate client
send-to-reply time can exceed a second.
`sent_at` remains the sole input to RTT, timeout/scheduling, and the `sent_at` remains the RTT endpoint and the userspace fallback for the
downstream (`server_to_client`) one-way delay, exactly as before this upstream (`client_to_server`) one-way delay, exactly as before this
capability existed; `TX_SOFTWARE` is still a software kernel timestamp near capability existed. It is not an input to the downstream
the driver handoff, not a NIC hardware transmit time. IPDV is likewise (`server_to_client`) one-way delay, which is derived solely from
unaffected — it is computed from the existing userspace timestamp fields `ReceiveMeta::preferred_receive_wall` and the negotiated server send
only. A normal reply and a measurable late reply (one whose `PendingProbe` timestamp — unaffected by this capability or by F-04. `TX_SOFTWARE` is
is still retained in `TimedOutMap`) apply the identical selection policy; an still a software kernel timestamp near the driver handoff, not a NIC
hardware transmit time. `irtt-stats`'s `ReplySample`
copies public `sent_at` straight into `client_send_mono`/`client_send_wall_ns`,
which `send_ipdv_ns` compares across probes for the upstream IPDV component —
so that component shifts endpoint along with `sent_at` itself (pre-send vs.
post-send) exactly as RTT and the upstream OWD fallback do; it is not a
separate invariant. A normal
reply and a measurable late reply (one whose `PendingProbe` is still
retained in `TimedOutMap`) apply the identical selection policy; an
untracked late reply has no retained probe and stays unmeasurable, and a untracked late reply has no retained probe and stays unmeasurable, and a
kernel timestamp that arrives after a reply has already been processed is kernel timestamp that arrives after a reply has already been processed is
simply never observed by that reply. simply never observed by that reply.
+24 -5
View File
@@ -528,15 +528,20 @@ impl AsyncClient {
let expected_bytes = prepared.bytes.len(); let expected_bytes = prepared.bytes.len();
let scheduled_at = schedule_commit.scheduled_at; let scheduled_at = schedule_commit.scheduled_at;
// Private pre-send anchor: finalizes all fallible commit work
// (timeout deadline arithmetic, kernel TX lower bound) before
// the send attempt. A WouldBlock retry samples a fresh one on
// the next loop iteration; only the successful attempt's anchor
// is retained (via `machine_commit`/`commit_probe_sent` below).
#[cfg(not(test))] #[cfg(not(test))]
let sent_at = ClientTimestamp::now(); let send_anchor = ClientTimestamp::now();
#[cfg(test)] #[cfg(test)]
let sent_at = test_timestamps let send_anchor = test_timestamps
.map(|timestamps| timestamps.sent_at) .map(|timestamps| timestamps.send_anchor)
.unwrap_or_else(ClientTimestamp::now); .unwrap_or_else(ClientTimestamp::now);
let machine_commit = match self let machine_commit = match self
.machine .machine
.finalize_probe_commit(machine_preflight, sent_at) .finalize_probe_commit(machine_preflight, send_anchor)
{ {
Ok(commit) => commit, Ok(commit) => commit,
Err(error) => return Poll::Ready(Err(error)), Err(error) => return Poll::Ready(Err(error)),
@@ -553,6 +558,9 @@ impl AsyncClient {
let send_result = self.socket.try_send(&prepared.bytes); let send_result = self.socket.try_send(&prepared.bytes);
let bytes = match send_result { let bytes = match send_result {
Ok(bytes) => bytes, Ok(bytes) => bytes,
// 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) if error.kind() == io::ErrorKind::WouldBlock => continue,
Err(error) => return Poll::Ready(Err(ClientError::Socket(error))), Err(error) => return Poll::Ready(Err(ClientError::Socket(error))),
}; };
@@ -563,7 +571,18 @@ impl AsyncClient {
let send_finished_at = test_timestamps let send_finished_at = test_timestamps
.map(|timestamps| timestamps.send_finished_at) .map(|timestamps| timestamps.send_finished_at)
.unwrap_or_else(Instant::now); .unwrap_or_else(Instant::now);
let sent = self.machine.commit_probe_sent(machine_commit, bytes); // Public post-send measurement, captured only after the
// successful send above. The subsequent machine/schedule commit
// is infallible.
#[cfg(not(test))]
let sent_at = ClientTimestamp::now();
#[cfg(test)]
let sent_at = test_timestamps
.map(|timestamps| timestamps.sent_at)
.unwrap_or_else(ClientTimestamp::now);
let sent = self
.machine
.commit_probe_sent(machine_commit, sent_at, bytes);
schedule.commit(schedule_commit); schedule.commit(schedule_commit);
self.prepared_probe = None; self.prepared_probe = None;
+8 -1
View File
@@ -689,12 +689,14 @@ fn probe_false_readiness_resamples_all_timestamps() {
client.test_hooks.probe_timestamps.borrow_mut().extend([ client.test_hooks.probe_timestamps.borrow_mut().extend([
ProbeSendTimestamps { ProbeSendTimestamps {
permission_at: scheduled_at + Duration::from_millis(1), permission_at: scheduled_at + Duration::from_millis(1),
send_anchor: first_sent,
sent_at: first_sent, sent_at: first_sent,
send_call_start: scheduled_at + Duration::from_millis(4), send_call_start: scheduled_at + Duration::from_millis(4),
send_finished_at: scheduled_at + Duration::from_millis(6), send_finished_at: scheduled_at + Duration::from_millis(6),
}, },
ProbeSendTimestamps { ProbeSendTimestamps {
permission_at: scheduled_at + Duration::from_millis(8), permission_at: scheduled_at + Duration::from_millis(8),
send_anchor: second_sent,
sent_at: second_sent, sent_at: second_sent,
send_call_start: scheduled_at + Duration::from_millis(12), send_call_start: scheduled_at + Duration::from_millis(12),
send_finished_at: scheduled_at + Duration::from_millis(17), send_finished_at: scheduled_at + Duration::from_millis(17),
@@ -741,6 +743,7 @@ fn public_timeout_polling_remains_exhaustive() {
wall: SystemTime::UNIX_EPOCH, wall: SystemTime::UNIX_EPOCH,
}, },
timeout_at: now - Duration::from_secs(1), timeout_at: now - Duration::from_secs(1),
tx_not_before_wall: SystemTime::UNIX_EPOCH,
kernel_tx_timestamp: None, kernel_tx_timestamp: None,
}); });
} }
@@ -803,7 +806,7 @@ fn stale_retained_probe_is_rejected_before_send() {
.unwrap(); .unwrap();
client client
.machine .machine
.commit_probe_sent(commit, advanced.bytes.len()); .commit_probe_sent(commit, ClientTimestamp::now(), advanced.bytes.len());
let attempts = client.test_hooks.send_attempts.get(); let attempts = client.test_hooks.send_attempts.get();
assert!(matches!( assert!(matches!(
@@ -832,6 +835,10 @@ fn finite_duration_boundary_clears_preparation_without_send() {
.borrow_mut() .borrow_mut()
.push_back(ProbeSendTimestamps { .push_back(ProbeSendTimestamps {
permission_at: start + Duration::from_millis(20), permission_at: start + Duration::from_millis(20),
send_anchor: ClientTimestamp {
wall: SystemTime::UNIX_EPOCH,
mono: start + Duration::from_millis(21),
},
sent_at: ClientTimestamp { sent_at: ClientTimestamp {
wall: SystemTime::UNIX_EPOCH, wall: SystemTime::UNIX_EPOCH,
mono: start + Duration::from_millis(21), mono: start + Duration::from_millis(21),
+23 -6
View File
@@ -75,9 +75,12 @@ struct ClientTestHooks {
#[derive(Debug, Clone, Copy)] #[derive(Debug, Clone, Copy)]
pub(crate) struct ProbeSendTimestamps { pub(crate) struct ProbeSendTimestamps {
pub(crate) permission_at: Instant, pub(crate) permission_at: Instant,
pub(crate) sent_at: ClientTimestamp, /// Private pre-send anchor: timeout deadline / kernel TX lower bound.
pub(crate) send_anchor: ClientTimestamp,
pub(crate) send_call_start: Instant, pub(crate) send_call_start: Instant,
pub(crate) send_finished_at: Instant, pub(crate) send_finished_at: Instant,
/// Public post-send measurement: RTT / OWD fallback / timer_error.
pub(crate) sent_at: ClientTimestamp,
} }
#[cfg(test)] #[cfg(test)]
@@ -410,13 +413,17 @@ impl Client {
let scheduled_at = schedule_commit.scheduled_at; let scheduled_at = schedule_commit.scheduled_at;
#[cfg(test)] #[cfg(test)]
let reported_bytes = self.probe_reported_len.take(); let reported_bytes = self.probe_reported_len.take();
// All fallible probe-commit preflight (timeout deadline arithmetic,
// capacity/collision checks already done above) is finalized from
// this PRIVATE pre-send anchor, before the socket send. It is not
// the public measurement `sent_at` captured further below.
#[cfg(not(test))] #[cfg(not(test))]
let sent_at = ClientTimestamp::now(); let send_anchor = ClientTimestamp::now();
#[cfg(test)] #[cfg(test)]
let sent_at = test_timestamps let send_anchor = test_timestamps
.map(|timestamps| timestamps.sent_at) .map(|timestamps| timestamps.send_anchor)
.unwrap_or_else(ClientTimestamp::now); .unwrap_or_else(ClientTimestamp::now);
let machine_commit = runtime.finalize_probe_commit(machine_preflight, sent_at)?; let machine_commit = runtime.finalize_probe_commit(machine_preflight, send_anchor)?;
#[cfg(not(test))] #[cfg(not(test))]
let send_call_start = Instant::now(); let send_call_start = Instant::now();
#[cfg(test)] #[cfg(test)]
@@ -436,7 +443,17 @@ impl Client {
let send_finished_at = test_timestamps let send_finished_at = test_timestamps
.map(|timestamps| timestamps.send_finished_at) .map(|timestamps| timestamps.send_finished_at)
.unwrap_or_else(Instant::now); .unwrap_or_else(Instant::now);
let sent = runtime.commit_probe_sent(machine_commit, bytes); // The public measurement timestamp: paired wall/monotonic sample
// captured immediately after the successful socket send completed.
// This, not `send_anchor` above, is the RTT/OWD-fallback/timer_error
// endpoint. The infallible machine/schedule commit below never fails.
#[cfg(not(test))]
let sent_at = ClientTimestamp::now();
#[cfg(test)]
let sent_at = test_timestamps
.map(|timestamps| timestamps.sent_at)
.unwrap_or_else(ClientTimestamp::now);
let sent = runtime.commit_probe_sent(machine_commit, sent_at, bytes);
schedule.commit(schedule_commit); schedule.commit(schedule_commit);
let send_call = send_finished_at.saturating_duration_since(send_call_start); let send_call = send_finished_at.saturating_duration_since(send_call_start);
@@ -3,6 +3,7 @@ use super::*;
fn probe_timestamps(permission_at: Instant, sent_at: ClientTimestamp) -> ProbeSendTimestamps { fn probe_timestamps(permission_at: Instant, sent_at: ClientTimestamp) -> ProbeSendTimestamps {
ProbeSendTimestamps { ProbeSendTimestamps {
permission_at, permission_at,
send_anchor: sent_at,
sent_at, sent_at,
send_call_start: sent_at.mono, send_call_start: sent_at.mono,
send_finished_at: sent_at.mono, send_finished_at: sent_at.mono,
@@ -176,12 +177,21 @@ fn blocking_probe_separates_permission_committed_send_and_send_call_timing() {
.unwrap(), .unwrap(),
); );
let permission_at = opened_mono + Duration::from_millis(500); let permission_at = opened_mono + Duration::from_millis(500);
let sent_at = ClientTimestamp { // Distinct pre-send anchor and post-send `sent_at`: the anchor drives
// the operational timeout deadline, while `sent_at` (captured after the
// send call) is the public measurement/timer_error endpoint. They are
// deliberately decoupled by this PR.
let send_anchor = ClientTimestamp {
wall: SystemTime::now(), wall: SystemTime::now(),
mono: opened_mono + Duration::from_millis(525), mono: opened_mono + Duration::from_millis(525),
}; };
let sent_at = ClientTimestamp {
wall: SystemTime::now(),
mono: opened_mono + Duration::from_millis(534),
};
let timestamps = ProbeSendTimestamps { let timestamps = ProbeSendTimestamps {
permission_at, permission_at,
send_anchor,
sent_at, sent_at,
send_call_start: opened_mono + Duration::from_millis(526), send_call_start: opened_mono + Duration::from_millis(526),
send_finished_at: opened_mono + Duration::from_millis(533), send_finished_at: opened_mono + Duration::from_millis(533),
@@ -200,11 +210,14 @@ fn blocking_probe_separates_permission_committed_send_and_send_call_timing() {
}] if *scheduled_at == opened_mono }] if *scheduled_at == opened_mono
&& *event_sent_at == sent_at && *event_sent_at == sent_at
&& *send_call == Duration::from_millis(7) && *send_call == Duration::from_millis(7)
&& *timer_error == Duration::from_millis(525) && *timer_error == Duration::from_millis(534)
)); ));
assert_eq!(client.next_send_deadline(), Some(opened_mono + interval)); assert_eq!(client.next_send_deadline(), Some(opened_mono + interval));
assert_eq!(client.runtime.packets_sent(), 1); assert_eq!(client.runtime.packets_sent(), 1);
let timeout_at = sent_at.mono + client.probe_timeout(); // The timeout deadline is anchored to the pre-send `send_anchor`, not
// to the post-send `sent_at` captured 9ms later.
let timeout_at = send_anchor.mono + client.probe_timeout();
assert_ne!(timeout_at, sent_at.mono + client.probe_timeout());
assert!(client assert!(client
.poll_timeouts_at(timeout_at - Duration::from_nanos(1)) .poll_timeouts_at(timeout_at - Duration::from_nanos(1))
.unwrap() .unwrap()
@@ -236,6 +249,7 @@ fn blocking_send_call_excludes_wrapped_history_cleanup() {
client.runtime.seed_wrapped_probe_history_for_test(sent_at); client.runtime.seed_wrapped_probe_history_for_test(sent_at);
let timestamps = ProbeSendTimestamps { let timestamps = ProbeSendTimestamps {
permission_at: scheduled_at, permission_at: scheduled_at,
send_anchor: sent_at,
sent_at, sent_at,
send_call_start: scheduled_at + Duration::from_millis(6), send_call_start: scheduled_at + Duration::from_millis(6),
send_finished_at: scheduled_at + Duration::from_millis(13), send_finished_at: scheduled_at + Duration::from_millis(13),
@@ -343,6 +357,7 @@ fn managed_probe_skips_missed_schedule_slots() {
}; };
let timestamps = ProbeSendTimestamps { let timestamps = ProbeSendTimestamps {
permission_at, permission_at,
send_anchor: delayed_send,
sent_at: delayed_send, sent_at: delayed_send,
send_call_start: first_deadline + Duration::from_millis(49), send_call_start: first_deadline + Duration::from_millis(49),
send_finished_at: first_deadline + Duration::from_millis(55), send_finished_at: first_deadline + Duration::from_millis(55),
@@ -72,7 +72,14 @@ fn compute_one_way_omits_samples_when_client_wall_time_overflows_i64() {
}; };
assert_eq!( assert_eq!(
compute_one_way(&sent_at, &received_at, None, &ReceiveMeta::default(), &ts), compute_one_way(
&sent_at,
sent_at.wall,
&received_at,
None,
&ReceiveMeta::default(),
&ts
),
None None
); );
} }
@@ -93,8 +100,15 @@ fn compute_one_way_returns_available_direction_samples() {
..Default::default() ..Default::default()
}; };
let sample = let sample = compute_one_way(
compute_one_way(&sent_at, &received_at, None, &ReceiveMeta::default(), &ts).unwrap(); &sent_at,
sent_at.wall,
&received_at,
None,
&ReceiveMeta::default(),
&ts,
)
.unwrap();
assert_eq!( assert_eq!(
sample.client_to_server, sample.client_to_server,
Some(SignedDuration::from_nanos(15_000_000)) Some(SignedDuration::from_nanos(15_000_000))
@@ -121,8 +135,15 @@ fn compute_one_way_preserves_negative_client_to_server_delay() {
..Default::default() ..Default::default()
}; };
let sample = let sample = compute_one_way(
compute_one_way(&sent_at, &received_at, None, &ReceiveMeta::default(), &ts).unwrap(); &sent_at,
sent_at.wall,
&received_at,
None,
&ReceiveMeta::default(),
&ts,
)
.unwrap();
assert_eq!( assert_eq!(
sample.client_to_server, sample.client_to_server,
@@ -150,8 +171,15 @@ fn compute_one_way_preserves_negative_server_to_client_delay() {
..Default::default() ..Default::default()
}; };
let sample = let sample = compute_one_way(
compute_one_way(&sent_at, &received_at, None, &ReceiveMeta::default(), &ts).unwrap(); &sent_at,
sent_at.wall,
&received_at,
None,
&ReceiveMeta::default(),
&ts,
)
.unwrap();
assert_eq!( assert_eq!(
sample.client_to_server, sample.client_to_server,
+15 -7
View File
@@ -94,16 +94,19 @@ pub enum ClientEvent {
remote: SocketAddr, remote: SocketAddr,
/// Monotonic deadline at which the probe was scheduled to be sent. /// Monotonic deadline at which the probe was scheduled to be sent.
scheduled_at: Instant, scheduled_at: Instant,
/// Client wall/monotonic timestamp captured after probe preparation, /// Paired client wall/monotonic timestamp captured immediately after
/// immediately before timeout finalization and the socket send. /// the successful UDP send completed. This is the RTT and userspace
/// upstream one-way-delay fallback endpoint. The operational probe
/// timeout deadline is computed from an earlier, private, pre-send
/// timestamp instead — see the client crate's `AGENTS.md`.
sent_at: ClientTimestamp, sent_at: ClientTimestamp,
/// Number of bytes passed to the UDP socket. /// Number of bytes passed to the UDP socket.
bytes: usize, bytes: usize,
/// Elapsed time spent in the socket send call, excluding protocol and /// Elapsed time spent in the socket send call, excluding protocol and
/// scheduler commit work. /// scheduler commit work.
send_call: Duration, send_call: Duration,
/// Absolute difference between the scheduled send time and actual send /// Absolute difference between the scheduled send time and the
/// timestamp. /// post-send `sent_at` timestamp above.
timer_error: Duration, timer_error: Duration,
}, },
@@ -117,7 +120,8 @@ pub enum ClientEvent {
seq: u32, seq: u32,
/// Resolved remote socket address. /// Resolved remote socket address.
remote: SocketAddr, remote: SocketAddr,
/// Client timestamp recorded when the matching request was sent. /// Client timestamp recorded immediately after the matching request
/// finished sending (see `ClientEvent::EchoSent::sent_at`).
sent_at: ClientTimestamp, sent_at: ClientTimestamp,
/// Client timestamp recorded when the reply was received. /// Client timestamp recorded when the reply was received.
received_at: ClientTimestamp, received_at: ClientTimestamp,
@@ -143,7 +147,10 @@ pub enum ClientEvent {
EchoLoss { EchoLoss {
/// Wire sequence number that timed out. /// Wire sequence number that timed out.
seq: u32, seq: u32,
/// Client timestamp recorded when the request was sent. /// Client timestamp recorded immediately after the request finished
/// sending (see `ClientEvent::EchoSent::sent_at`). The timeout
/// deadline below is anchored to an earlier, private, pre-send
/// timestamp, not to this one.
sent_at: ClientTimestamp, sent_at: ClientTimestamp,
/// Monotonic deadline at which the probe was declared lost. /// Monotonic deadline at which the probe was declared lost.
timeout_at: std::time::Instant, timeout_at: std::time::Instant,
@@ -178,7 +185,8 @@ pub enum ClientEvent {
highest_seen: u32, highest_seen: u32,
/// Resolved remote socket address. /// Resolved remote socket address.
remote: SocketAddr, remote: SocketAddr,
/// Original client send timestamp, when still retained. /// Original post-send client timestamp, when still retained (see
/// `ClientEvent::EchoSent::sent_at`).
sent_at: Option<ClientTimestamp>, sent_at: Option<ClientTimestamp>,
/// Client timestamp recorded when the late reply was received. /// Client timestamp recorded when the late reply was received.
received_at: ClientTimestamp, received_at: ClientTimestamp,
+1
View File
@@ -354,6 +354,7 @@ fn timeout_probe(seq: u32, sent_at: Instant, timeout_at: Instant) -> PendingProb
wall: SystemTime::UNIX_EPOCH, wall: SystemTime::UNIX_EPOCH,
}, },
timeout_at, timeout_at,
tx_not_before_wall: SystemTime::UNIX_EPOCH,
kernel_tx_timestamp: None, kernel_tx_timestamp: None,
} }
} }
+29 -4
View File
@@ -8,15 +8,38 @@ use crate::{error::ClientError, timing::ClientTimestamp};
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub(crate) struct PendingProbe { pub(crate) struct PendingProbe {
pub wire_seq: u32, pub wire_seq: u32,
/// Paired client wall/monotonic timestamp captured immediately after the
/// successful UDP send completed. This is the RTT endpoint (via
/// `compute_rtt`), the userspace fallback endpoint for upstream one-way
/// delay, `timer_error`'s basis, and the value surfaced on the public
/// `EchoSent`/`EchoReply`/`EchoLoss`/`LateReply` events. It is *not* the
/// input to `timeout_at` below — see that field's doc.
pub sent_at: ClientTimestamp, pub sent_at: ClientTimestamp,
/// Operational timeout deadline, computed from the pre-send send
/// anchor's monotonic timestamp plus the configured probe timeout
/// (`SessionMachine::finalize_probe_commit`), not from `sent_at.mono`
/// above. Timeout semantics deliberately stay anchored to before the
/// socket send so that fallible deadline arithmetic (and its overflow
/// check) still runs, and a `PendingProbe` is only ever created, before
/// the datagram is transmitted.
pub timeout_at: Instant, pub timeout_at: Instant,
/// Local wall-clock lower bound for validating an asynchronous Linux
/// kernel `TX_SOFTWARE` timestamp, sampled at the same pre-send send
/// anchor as `timeout_at` — i.e. before the socket send, not from
/// `sent_at.wall` above. A legitimate `TX_SOFTWARE` timestamp can be
/// generated during the send path and therefore observed earlier than
/// the post-send `sent_at.wall` sample; using this earlier pre-send
/// bound instead of `sent_at.wall` avoids rejecting such a timestamp
/// merely because userspace samples `sent_at` after `send()` returns.
/// See `compute_one_way`'s `preferred_send_wall`.
pub tx_not_before_wall: SystemTime,
/// Observed Linux kernel TX_SOFTWARE wall timestamp for this probe's /// Observed Linux kernel TX_SOFTWARE wall timestamp for this probe's
/// send, when the socket has TX timestamping enabled and a matching /// send, when the socket has TX timestamping enabled and a matching
/// `MSG_ERRQUEUE` record has been drained. Optional observed metadata, /// `MSG_ERRQUEUE` record has been drained. Optional observed metadata,
/// eligible only for upstream one-way delay after local plausibility /// eligible only for upstream one-way delay after local plausibility
/// validation (see `compute_one_way`'s `preferred_send_wall`). `sent_at` /// validation against `tx_not_before_wall` (see `compute_one_way`'s
/// remains authoritative for RTT, timeout, pacing/scheduling semantics, /// `preferred_send_wall`). `sent_at` remains the userspace fallback for
/// and the userspace fallback for upstream one-way delay. /// upstream one-way delay when this is absent or implausible.
pub kernel_tx_timestamp: Option<SystemTime>, pub kernel_tx_timestamp: Option<SystemTime>,
} }
@@ -386,10 +409,12 @@ mod tests {
} }
fn pending(seq: u32, timeout_at: Instant) -> PendingProbe { fn pending(seq: u32, timeout_at: Instant) -> PendingProbe {
let sent_at = ts(timeout_at - Duration::from_secs(1));
PendingProbe { PendingProbe {
wire_seq: seq, wire_seq: seq,
sent_at: ts(timeout_at - Duration::from_secs(1)), sent_at,
timeout_at, timeout_at,
tx_not_before_wall: sent_at.wall,
kernel_tx_timestamp: None, kernel_tx_timestamp: None,
} }
} }
+293 -73
View File
@@ -152,7 +152,13 @@ pub(crate) struct ProbeCommitPreflight {
#[derive(Debug)] #[derive(Debug)]
pub(crate) struct ProbeCommit { pub(crate) struct ProbeCommit {
pending: PendingProbe, wire_seq: u32,
/// Operational timeout deadline, derived from the pre-send send anchor.
/// See [`PendingProbe::timeout_at`].
timeout_at: Instant,
/// Pre-send wall-clock lower bound for kernel TX plausibility. See
/// [`PendingProbe::tx_not_before_wall`].
tx_not_before_wall: SystemTime,
next_wire_seq: u32, next_wire_seq: u32,
next_packets_sent: u64, next_packets_sent: u64,
} }
@@ -351,38 +357,62 @@ impl SessionMachine {
}) })
} }
/// Finalizes all fallible probe-commit preflight work that must happen
/// before the socket send: the timeout deadline arithmetic (which can
/// overflow) and capturing the pre-send wall-clock lower bound used to
/// validate a later asynchronous kernel TX timestamp.
///
/// `send_anchor` is a *private* pre-send timestamp, not the public
/// measurement `sent_at` — it is sampled immediately before this call,
/// before the socket send. Its `mono` field anchors the operational
/// timeout deadline and its `wall` field becomes
/// [`PendingProbe::tx_not_before_wall`]. See the client crate's
/// `AGENTS.md` for why timeout semantics deliberately stay pre-send
/// anchored while the public `sent_at` measurement moves after the send.
pub(crate) fn finalize_probe_commit( pub(crate) fn finalize_probe_commit(
&self, &self,
preflight: ProbeCommitPreflight, preflight: ProbeCommitPreflight,
sent_at: ClientTimestamp, send_anchor: ClientTimestamp,
) -> Result<ProbeCommit, ClientError> { ) -> Result<ProbeCommit, ClientError> {
let timeout_at = sent_at let timeout_at = send_anchor
.mono .mono
.checked_add(self.config.probe_timeout) .checked_add(self.config.probe_timeout)
.ok_or(ClientError::DurationOverflow)?; .ok_or(ClientError::DurationOverflow)?;
Ok(ProbeCommit { Ok(ProbeCommit {
pending: PendingProbe { wire_seq: preflight.seq,
wire_seq: preflight.seq, timeout_at,
sent_at, tx_not_before_wall: send_anchor.wall,
timeout_at,
kernel_tx_timestamp: None,
},
next_wire_seq: preflight.next_wire_seq, next_wire_seq: preflight.next_wire_seq,
next_packets_sent: preflight.next_packets_sent, next_packets_sent: preflight.next_packets_sent,
}) })
} }
pub(crate) fn commit_probe_sent(&mut self, commit: ProbeCommit, bytes: usize) -> ProbeSent { /// Infallibly commits a probe as sent, using the post-send measurement
/// `sent_at` captured immediately after the successful socket send
/// returned. All fallible work already happened in
/// [`Self::finalize_probe_commit`]; nothing here can fail.
pub(crate) fn commit_probe_sent(
&mut self,
commit: ProbeCommit,
sent_at: ClientTimestamp,
bytes: usize,
) -> ProbeSent {
let session = match &mut self.state { let session = match &mut self.state {
MachineState::Open(session) => session, MachineState::Open(session) => session,
_ => unreachable!("probe commits are only created for an open session"), _ => unreachable!("probe commits are only created for an open session"),
}; };
let seq = commit.pending.wire_seq; let seq = commit.wire_seq;
let sent_at = commit.pending.sent_at; let pending = PendingProbe {
wire_seq: seq,
sent_at,
timeout_at: commit.timeout_at,
tx_not_before_wall: commit.tx_not_before_wall,
kernel_tx_timestamp: None,
};
session.timed_out.remove(seq); session.timed_out.remove(seq);
session.completed.remove(seq); session.completed.remove(seq);
session.pending.commit_insert(commit.pending); session.pending.commit_insert(pending);
session.next_wire_seq = commit.next_wire_seq; session.next_wire_seq = commit.next_wire_seq;
session.packets_sent = commit.next_packets_sent; session.packets_sent = commit.next_packets_sent;
@@ -701,6 +731,7 @@ impl SessionMachine {
let server_timing = build_server_timing(&reply.timestamps); let server_timing = build_server_timing(&reply.timestamps);
let one_way = compute_one_way( let one_way = compute_one_way(
&pending.sent_at, &pending.sent_at,
pending.tx_not_before_wall,
&now, &now,
pending.kernel_tx_timestamp, pending.kernel_tx_timestamp,
&meta, &meta,
@@ -756,6 +787,7 @@ impl SessionMachine {
let server_timing = build_server_timing(&reply.timestamps); let server_timing = build_server_timing(&reply.timestamps);
let one_way = compute_one_way( let one_way = compute_one_way(
&timed_out.sent_at, &timed_out.sent_at,
timed_out.tx_not_before_wall,
&now, &now,
timed_out.kernel_tx_timestamp, timed_out.kernel_tx_timestamp,
&meta, &meta,
@@ -984,39 +1016,49 @@ fn build_server_timing(ts: &TimestampFields) -> Option<ServerTiming> {
/// ///
/// Prefers the correlated Linux kernel `TX_SOFTWARE` timestamp — a software /// Prefers the correlated Linux kernel `TX_SOFTWARE` timestamp — a software
/// timestamp near the driver handoff, not physical NIC departure — when it /// timestamp near the driver handoff, not physical NIC departure — when it
/// is locally plausible: it must not precede `sent_at` and must not follow /// is locally plausible: it must not precede `tx_not_before_wall` and must
/// `received_at`, both on the client's own wall clock. Comparing only these /// not follow `received_at`, both on the client's own wall clock.
/// two same-client anchors means the check detects local causal-ordering /// `tx_not_before_wall` is the pre-send send-anchor sample, not the
/// inconsistency (e.g. a backward wall-clock step), not every possible /// post-send `sent_at_wall` fallback below — a legitimate `TX_SOFTWARE`
/// wall-clock discontinuity, and it deliberately does not compare against /// timestamp can be generated while the send syscall is still in flight and
/// the remote server's wall clock: cross-host clock offset is a /// therefore legitimately precede the post-send `sent_at` sample, so the
/// synchronization property of the resulting delay, not evidence that this /// lower bound must stay anchored before the send rather than after it.
/// endpoint is invalid. There is no maximum lag bound — unlike scheduler /// Comparing only these two same-client anchors means the check detects
/// wakeup delay on the receive side, legitimate send-to-receive time can /// local causal-ordering inconsistency (e.g. a backward wall-clock step),
/// exceed a second. Falls back to `sent_at` when the kernel timestamp is /// not every possible wall-clock discontinuity, and it deliberately does
/// absent or fails this check; the raw observation on `PendingProbe` is /// not compare against the remote server's wall clock: cross-host clock
/// never mutated by a fallback here. /// offset is a synchronization property of the resulting delay, not
/// evidence that this endpoint is invalid. There is no maximum lag bound —
/// unlike scheduler wakeup delay on the receive side, legitimate
/// send-to-receive time can exceed a second. Falls back to `sent_at_wall`
/// (the post-send measurement) when the kernel timestamp is absent or fails
/// this check; the raw observation on `PendingProbe` is never mutated by a
/// fallback here.
fn preferred_send_wall( fn preferred_send_wall(
sent_at: SystemTime, tx_not_before_wall: SystemTime,
sent_at_wall: SystemTime,
kernel_tx_timestamp: Option<SystemTime>, kernel_tx_timestamp: Option<SystemTime>,
received_at: SystemTime, received_at: SystemTime,
) -> SystemTime { ) -> SystemTime {
match kernel_tx_timestamp { match kernel_tx_timestamp {
Some(kernel_tx) if sent_at <= kernel_tx && kernel_tx <= received_at => kernel_tx, Some(kernel_tx) if tx_not_before_wall <= kernel_tx && kernel_tx <= received_at => kernel_tx,
_ => sent_at, _ => sent_at_wall,
} }
} }
/// Computes both one-way delay directions from wall-clock endpoints. /// Computes both one-way delay directions from wall-clock endpoints.
/// ///
/// The upstream direction prefers a locally plausible correlated kernel /// The upstream direction prefers a locally plausible correlated kernel
/// `TX_SOFTWARE` send timestamp over the userspace pre-send sample; see /// `TX_SOFTWARE` send timestamp over the userspace post-send `sent_at`
/// [`preferred_send_wall`]. The downstream direction uses /// sample; see [`preferred_send_wall`]. `tx_not_before_wall` is the private
/// pre-send lower bound used only for that plausibility check — it is never
/// itself a candidate endpoint. The downstream direction uses
/// [`ReceiveMeta::preferred_receive_wall`], which prefers a plausible kernel /// [`ReceiveMeta::preferred_receive_wall`], which prefers a plausible kernel
/// receive timestamp over the userspace receive wall sample. Neither kernel /// receive timestamp over the userspace receive wall sample. Neither kernel
/// timestamp is ever used for RTT. /// timestamp is ever used for RTT.
pub(crate) fn compute_one_way( pub(crate) fn compute_one_way(
sent_at: &ClientTimestamp, sent_at: &ClientTimestamp,
tx_not_before_wall: SystemTime,
received_at: &ClientTimestamp, received_at: &ClientTimestamp,
kernel_tx_timestamp: Option<SystemTime>, kernel_tx_timestamp: Option<SystemTime>,
meta: &ReceiveMeta, meta: &ReceiveMeta,
@@ -1025,7 +1067,12 @@ pub(crate) fn compute_one_way(
let server_recv_wall = ts.recv_wall.or(ts.midpoint_wall); let server_recv_wall = ts.recv_wall.or(ts.midpoint_wall);
let server_send_wall = ts.send_wall.or(ts.midpoint_wall); let server_send_wall = ts.send_wall.or(ts.midpoint_wall);
let client_send_wall = preferred_send_wall(sent_at.wall, kernel_tx_timestamp, received_at.wall); let client_send_wall = preferred_send_wall(
tx_not_before_wall,
sent_at.wall,
kernel_tx_timestamp,
received_at.wall,
);
let client_send_ns = unix_epoch_ns_i64(client_send_wall); let client_send_ns = unix_epoch_ns_i64(client_send_wall);
let client_recv_ns = unix_epoch_ns_i64(meta.preferred_receive_wall(received_at.wall)); let client_recv_ns = unix_epoch_ns_i64(meta.preferred_receive_wall(received_at.wall));
@@ -1125,6 +1172,7 @@ impl SessionMachine {
wire_seq: 0, wire_seq: 0,
sent_at: now, sent_at: now,
timeout_at: now.mono, timeout_at: now.mono,
tx_not_before_wall: now.wall,
kernel_tx_timestamp: None, kernel_tx_timestamp: None,
}); });
session.completed.insert(0); session.completed.insert(0);
@@ -1707,7 +1755,7 @@ mod tests {
let sent_at = timestamp(Instant::now()); let sent_at = timestamp(Instant::now());
let preflight = machine.preflight_probe_commit(&accepted).unwrap(); let preflight = machine.preflight_probe_commit(&accepted).unwrap();
let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap(); let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap();
machine.commit_probe_sent(commit, accepted.bytes.len()); machine.commit_probe_sent(commit, sent_at, accepted.bytes.len());
let capacity = active(&machine).pending.capacity(); let capacity = active(&machine).pending.capacity();
assert!(matches!( assert!(matches!(
@@ -1737,7 +1785,7 @@ mod tests {
} }
let preflight = machine.preflight_probe_commit(&prepared).unwrap(); let preflight = machine.preflight_probe_commit(&prepared).unwrap();
let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap(); let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap();
let sent = machine.commit_probe_sent(commit, prepared.bytes.len()); let sent = machine.commit_probe_sent(commit, sent_at, prepared.bytes.len());
assert_eq!(sent.seq, 0); assert_eq!(sent.seq, 0);
let session = active(&machine); let session = active(&machine);
@@ -1750,16 +1798,20 @@ mod tests {
fn probe_commit_does_not_require_presentation_timing() { fn probe_commit_does_not_require_presentation_timing() {
let mut machine = open_machine(4, Duration::from_secs(1)); let mut machine = open_machine(4, Duration::from_secs(1));
let prepared = machine.prepare_probe().unwrap().unwrap(); let prepared = machine.prepare_probe().unwrap().unwrap();
let sent_at = timestamp(Instant::now()); let send_anchor = timestamp(Instant::now());
let preflight = machine.preflight_probe_commit(&prepared).unwrap(); let preflight = machine.preflight_probe_commit(&prepared).unwrap();
let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap(); let commit = machine
assert_eq!(commit.pending.sent_at, sent_at); .finalize_probe_commit(preflight, send_anchor)
assert_eq!( .unwrap();
commit.pending.timeout_at, // Fallible work (timeout deadline arithmetic, tx lower bound) is
sent_at.mono + Duration::from_secs(1) // finalized from the pre-send anchor, not any post-send sample.
); assert_eq!(commit.timeout_at, send_anchor.mono + Duration::from_secs(1));
assert_eq!(commit.tx_not_before_wall, send_anchor.wall);
let sent = machine.commit_probe_sent(commit, prepared.bytes.len()); // The post-send measurement `sent_at` is supplied afterward and can
// legitimately differ from the pre-send anchor.
let sent_at = timestamp(Instant::now());
let sent = machine.commit_probe_sent(commit, sent_at, prepared.bytes.len());
assert_eq!(sent.seq, prepared.seq); assert_eq!(sent.seq, prepared.seq);
assert_eq!(sent.sent_at, sent_at); assert_eq!(sent.sent_at, sent_at);
@@ -1778,7 +1830,7 @@ mod tests {
.finalize_probe_commit(preflight, timestamp(Instant::now())) .finalize_probe_commit(preflight, timestamp(Instant::now()))
.unwrap(); .unwrap();
machine.commit_probe_sent(commit, prepared.bytes.len()); machine.commit_probe_sent(commit, timestamp(Instant::now()), prepared.bytes.len());
let session = active(&machine); let session = active(&machine);
assert_eq!(session.pending.capacity(), reserved_capacity); assert_eq!(session.pending.capacity(), reserved_capacity);
@@ -1794,7 +1846,7 @@ mod tests {
let sent_at = timestamp(Instant::now()); let sent_at = timestamp(Instant::now());
let preflight = machine.preflight_probe_commit(&first).unwrap(); let preflight = machine.preflight_probe_commit(&first).unwrap();
let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap(); let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap();
machine.commit_probe_sent(commit, first.bytes.len()); machine.commit_probe_sent(commit, sent_at, first.bytes.len());
let second = machine.prepare_probe().unwrap().unwrap(); let second = machine.prepare_probe().unwrap().unwrap();
assert!(matches!( assert!(matches!(
@@ -1811,7 +1863,7 @@ mod tests {
let sent_at = timestamp(Instant::now()); let sent_at = timestamp(Instant::now());
let preflight = machine.preflight_probe_commit(&first).unwrap(); let preflight = machine.preflight_probe_commit(&first).unwrap();
let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap(); let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap();
machine.commit_probe_sent(commit, first.bytes.len()); machine.commit_probe_sent(commit, sent_at, first.bytes.len());
active_mut(&mut machine).next_wire_seq = 0; active_mut(&mut machine).next_wire_seq = 0;
let reused = machine.prepare_probe().unwrap().unwrap(); let reused = machine.prepare_probe().unwrap().unwrap();
@@ -1862,6 +1914,7 @@ mod tests {
wire_seq: seq, wire_seq: seq,
sent_at, sent_at,
timeout_at: sent_at.mono + Duration::from_secs(1), timeout_at: sent_at.mono + Duration::from_secs(1),
tx_not_before_wall: sent_at.wall,
kernel_tx_timestamp: None, kernel_tx_timestamp: None,
}); });
} }
@@ -1888,7 +1941,7 @@ mod tests {
let commit = machine let commit = machine
.finalize_probe_commit(preflight, timestamp(Instant::now())) .finalize_probe_commit(preflight, timestamp(Instant::now()))
.unwrap(); .unwrap();
machine.commit_probe_sent(commit, prepared.bytes.len()); machine.commit_probe_sent(commit, timestamp(Instant::now()), prepared.bytes.len());
assert_eq!(active(&machine).next_wire_seq, 0); assert_eq!(active(&machine).next_wire_seq, 0);
assert_eq!(machine.prepare_probe().unwrap().unwrap().seq, 0); assert_eq!(machine.prepare_probe().unwrap().unwrap().seq, 0);
@@ -1898,10 +1951,12 @@ mod tests {
fn successful_wrapped_reuse_purges_obsolete_history_only_on_commit() { fn successful_wrapped_reuse_purges_obsolete_history_only_on_commit() {
let mut machine = open_machine(3, Duration::from_secs(1)); let mut machine = open_machine(3, Duration::from_secs(1));
let now = Instant::now(); let now = Instant::now();
let obsolete_sent_at = timestamp(now - Duration::from_secs(1));
let obsolete = PendingProbe { let obsolete = PendingProbe {
wire_seq: 0, wire_seq: 0,
sent_at: timestamp(now - Duration::from_secs(1)), sent_at: obsolete_sent_at,
timeout_at: now, timeout_at: now,
tx_not_before_wall: obsolete_sent_at.wall,
kernel_tx_timestamp: None, kernel_tx_timestamp: None,
}; };
let session = active_mut(&mut machine); let session = active_mut(&mut machine);
@@ -1917,7 +1972,7 @@ mod tests {
assert!(active(&machine).timed_out.contains(0)); assert!(active(&machine).timed_out.contains(0));
assert!(active(&machine).completed.contains(0)); assert!(active(&machine).completed.contains(0));
machine.commit_probe_sent(commit, prepared.bytes.len()); machine.commit_probe_sent(commit, timestamp(now), prepared.bytes.len());
let session = active(&machine); let session = active(&machine);
assert!(!session.timed_out.contains(0)); assert!(!session.timed_out.contains(0));
assert!(!session.completed.contains(0)); assert!(!session.completed.contains(0));
@@ -1982,8 +2037,21 @@ mod tests {
} }
} }
/// Machine with one outstanding probe for sequence 0. /// Machine with one outstanding probe for sequence 0. The pre-send
/// `tx_not_before_wall` bound defaults to `sent_at.wall`, i.e. this
/// helper does not exercise the pre-/post-send distinction; use
/// [`machine_with_pending_probe_anchored`] where that distinction
/// matters.
fn machine_with_pending_probe(sent_at: ClientTimestamp) -> SessionMachine { fn machine_with_pending_probe(sent_at: ClientTimestamp) -> SessionMachine {
machine_with_pending_probe_anchored(sent_at.wall, sent_at)
}
/// Machine with one outstanding probe for sequence 0, with an explicit
/// pre-send `tx_not_before_wall` bound independent of `sent_at.wall`.
fn machine_with_pending_probe_anchored(
tx_not_before_wall: SystemTime,
sent_at: ClientTimestamp,
) -> SessionMachine {
let mut machine = open_machine(4, Duration::from_secs(1)); let mut machine = open_machine(4, Duration::from_secs(1));
let session = active_mut(&mut machine); let session = active_mut(&mut machine);
session.pending.preflight_insert(0).unwrap(); session.pending.preflight_insert(0).unwrap();
@@ -1991,6 +2059,7 @@ mod tests {
wire_seq: 0, wire_seq: 0,
sent_at, sent_at,
timeout_at: sent_at.mono + Duration::from_secs(1), timeout_at: sent_at.mono + Duration::from_secs(1),
tx_not_before_wall,
kernel_tx_timestamp: None, kernel_tx_timestamp: None,
}); });
session.next_wire_seq = 1; session.next_wire_seq = 1;
@@ -2006,6 +2075,7 @@ mod tests {
wire_seq: 0, wire_seq: 0,
sent_at, sent_at,
timeout_at: sent_at.mono + Duration::from_secs(1), timeout_at: sent_at.mono + Duration::from_secs(1),
tx_not_before_wall: sent_at.wall,
kernel_tx_timestamp: None, kernel_tx_timestamp: None,
}); });
session.next_wire_seq = 1; session.next_wire_seq = 1;
@@ -2310,7 +2380,9 @@ mod tests {
let prepared = machine.prepare_probe().unwrap().unwrap(); let prepared = machine.prepare_probe().unwrap().unwrap();
let preflight = machine.preflight_probe_commit(&prepared).unwrap(); let preflight = machine.preflight_probe_commit(&prepared).unwrap();
let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap(); let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap();
machine.commit_probe_sent(commit, prepared.bytes.len()).seq machine
.commit_probe_sent(commit, sent_at, prepared.bytes.len())
.seq
} }
#[test] #[test]
@@ -2635,85 +2707,132 @@ mod tests {
#[test] #[test]
fn preferred_send_wall_without_kernel_timestamp_uses_sent_at() { fn preferred_send_wall_without_kernel_timestamp_uses_sent_at() {
let sent_at = send_wall_ms(1_000); let anchor = send_wall_ms(1_000);
let received_at = send_wall_ms(1_040); let received_at = send_wall_ms(1_040);
assert_eq!(preferred_send_wall(sent_at, None, received_at), sent_at); assert_eq!(
preferred_send_wall(anchor, anchor, None, received_at),
anchor
);
} }
#[test] #[test]
fn preferred_send_wall_uses_kernel_timestamp_strictly_between_bounds() { fn preferred_send_wall_uses_kernel_timestamp_strictly_between_bounds() {
let sent_at = send_wall_ms(1_000); let anchor = send_wall_ms(1_000);
let kernel_tx = send_wall_ms(1_005); let kernel_tx = send_wall_ms(1_005);
let received_at = send_wall_ms(1_040); let received_at = send_wall_ms(1_040);
assert_eq!( assert_eq!(
preferred_send_wall(sent_at, Some(kernel_tx), received_at), preferred_send_wall(anchor, anchor, Some(kernel_tx), received_at),
kernel_tx kernel_tx
); );
} }
#[test] #[test]
fn preferred_send_wall_accepts_kernel_timestamp_equal_to_sent_at() { fn preferred_send_wall_accepts_kernel_timestamp_equal_to_lower_bound() {
let sent_at = send_wall_ms(1_000); let anchor = send_wall_ms(1_000);
let received_at = send_wall_ms(1_040); let received_at = send_wall_ms(1_040);
assert_eq!( assert_eq!(
preferred_send_wall(sent_at, Some(sent_at), received_at), preferred_send_wall(anchor, anchor, Some(anchor), received_at),
sent_at anchor
); );
} }
#[test] #[test]
fn preferred_send_wall_accepts_kernel_timestamp_equal_to_received_at() { fn preferred_send_wall_accepts_kernel_timestamp_equal_to_received_at() {
let sent_at = send_wall_ms(1_000); let anchor = send_wall_ms(1_000);
let received_at = send_wall_ms(1_040); let received_at = send_wall_ms(1_040);
assert_eq!( assert_eq!(
preferred_send_wall(sent_at, Some(received_at), received_at), preferred_send_wall(anchor, anchor, Some(received_at), received_at),
received_at received_at
); );
} }
#[test] #[test]
fn preferred_send_wall_rejects_kernel_timestamp_earlier_than_sent_at() { fn preferred_send_wall_rejects_kernel_timestamp_earlier_than_lower_bound() {
let sent_at = send_wall_ms(1_000); let anchor = send_wall_ms(1_000);
let kernel_tx = sent_at - Duration::from_nanos(1); let kernel_tx = anchor - Duration::from_nanos(1);
let received_at = send_wall_ms(1_040); let received_at = send_wall_ms(1_040);
assert_eq!( assert_eq!(
preferred_send_wall(sent_at, Some(kernel_tx), received_at), preferred_send_wall(anchor, anchor, Some(kernel_tx), received_at),
sent_at anchor
); );
} }
#[test] #[test]
fn preferred_send_wall_rejects_kernel_timestamp_later_than_received_at() { fn preferred_send_wall_rejects_kernel_timestamp_later_than_received_at() {
let sent_at = send_wall_ms(1_000); let anchor = send_wall_ms(1_000);
let received_at = send_wall_ms(1_040); let received_at = send_wall_ms(1_040);
let kernel_tx = received_at + Duration::from_nanos(1); let kernel_tx = received_at + Duration::from_nanos(1);
assert_eq!( assert_eq!(
preferred_send_wall(sent_at, Some(kernel_tx), received_at), preferred_send_wall(anchor, anchor, Some(kernel_tx), received_at),
sent_at anchor
); );
} }
#[test] #[test]
fn preferred_send_wall_handles_extreme_timestamps_without_panicking() { fn preferred_send_wall_handles_extreme_timestamps_without_panicking() {
let sent_at = send_wall_ms(1_000); let anchor = send_wall_ms(1_000);
let received_at = send_wall_ms(1_040); let received_at = send_wall_ms(1_040);
let far_future = UNIX_EPOCH + Duration::from_secs(u64::from(u32::MAX)) * 4; let far_future = UNIX_EPOCH + Duration::from_secs(u64::from(u32::MAX)) * 4;
let before_epoch = UNIX_EPOCH - Duration::from_secs(u64::from(u32::MAX)); let before_epoch = UNIX_EPOCH - Duration::from_secs(u64::from(u32::MAX));
assert_eq!( assert_eq!(
preferred_send_wall(sent_at, Some(far_future), received_at), preferred_send_wall(anchor, anchor, Some(far_future), received_at),
sent_at anchor
); );
assert_eq!( assert_eq!(
preferred_send_wall(sent_at, Some(before_epoch), received_at), preferred_send_wall(anchor, anchor, Some(before_epoch), received_at),
sent_at anchor
); );
assert_eq!( assert_eq!(
preferred_send_wall(before_epoch, None, received_at), preferred_send_wall(before_epoch, before_epoch, None, received_at),
before_epoch before_epoch
); );
} }
// Pre-/post-send decoupling (F-04): the plausibility lower bound is the
// pre-send anchor, but the fallback endpoint is the post-send sample,
// and the two can legitimately differ.
#[test]
fn preferred_send_wall_accepts_kernel_timestamp_before_post_send_sample() {
let tx_not_before_wall = send_wall_ms(1_000);
let post_send_sent_at = send_wall_ms(1_008);
let kernel_tx = send_wall_ms(1_005);
let received_at = send_wall_ms(1_040);
// The kernel timestamp precedes the post-send sample but follows
// the pre-send anchor: it must still be accepted.
assert_eq!(
preferred_send_wall(
tx_not_before_wall,
post_send_sent_at,
Some(kernel_tx),
received_at
),
kernel_tx
);
}
#[test]
fn preferred_send_wall_rejected_kernel_timestamp_falls_back_to_post_send_sample() {
let tx_not_before_wall = send_wall_ms(1_000);
let post_send_sent_at = send_wall_ms(1_008);
let kernel_tx = send_wall_ms(999);
let received_at = send_wall_ms(1_040);
// Below the pre-send lower bound: rejected, and the fallback is the
// post-send sample, not the pre-send anchor.
assert_eq!(
preferred_send_wall(
tx_not_before_wall,
post_send_sent_at,
Some(kernel_tx),
received_at
),
post_send_sent_at
);
}
fn upstream_probe(sent_wall: SystemTime, mono: Instant) -> SessionMachine { fn upstream_probe(sent_wall: SystemTime, mono: Instant) -> SessionMachine {
machine_with_pending_probe(ClientTimestamp { machine_with_pending_probe(ClientTimestamp {
mono, mono,
@@ -2877,4 +2996,105 @@ mod tests {
assert_eq!(reply_one_way(&events).unwrap().client_to_server, None); assert_eq!(reply_one_way(&events).unwrap().client_to_server, None);
} }
// F-04: post-send `sent_at` measurement timing.
//
// `sent_at` moved from immediately before timeout finalization/socket
// send to immediately after a successful send. These tests pin the
// resulting end-to-end behavior: raw RTT and the userspace upstream OWD
// fallback both shrink because they no longer include the send-call
// interval, while kernel TX plausibility keeps a separate pre-send
// lower bound so a legitimate `TX_SOFTWARE` timestamp generated during
// the send path is not rejected merely for preceding the post-send
// sample.
#[test]
fn raw_rtt_uses_post_send_sent_at_not_pre_send_anchor() {
let base = Instant::now();
// A pre-send anchor at base+100ms is deliberately not used here:
// raw RTT must be measured from the post-send `sent_at` at
// base+110ms, giving 50ms, not the 60ms a pre-send anchor would
// have produced.
let sent_at = ClientTimestamp {
mono: base + Duration::from_millis(110),
wall: SystemTime::now(),
};
let received_at = ClientTimestamp {
mono: base + Duration::from_millis(160),
wall: SystemTime::now(),
};
let rtt = compute_rtt(&sent_at, &received_at, &TimestampFields::default());
assert_eq!(rtt.raw, Duration::from_millis(50));
assert_eq!(rtt.effective, SignedDuration::from_duration(rtt.raw));
}
#[test]
fn userspace_upstream_owd_fallback_uses_post_send_sent_at() {
let mono = Instant::now();
let mut machine = machine_with_pending_probe_anchored(
send_wall_ms(1_000), // pre-send anchor
ClientTimestamp {
mono,
wall: send_wall_ms(1_008), // post-send sent_at
},
);
let events = upstream_reply(&mut machine, 1_020, send_wall_ms(1_040), mono);
// Old behavior would have used the pre-send 1000ms sample, giving
// 20ms. The new fallback uses the post-send 1008ms sample.
assert_eq!(
reply_one_way(&events).unwrap().client_to_server,
Some(SignedDuration::from_nanos(12_000_000))
);
}
#[test]
fn kernel_tx_accepted_even_though_earlier_than_post_send_sent_at() {
let mono = Instant::now();
let mut machine = machine_with_pending_probe_anchored(
send_wall_ms(1_000), // pre-send anchor / tx_not_before_wall
ClientTimestamp {
mono,
wall: send_wall_ms(1_008), // post-send sent_at
},
);
// The kernel TX timestamp is after the pre-send anchor and before
// the reply, but before the post-send `sent_at` sample. It must
// remain accepted: rejecting it here would be the regression this
// test guards against.
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(15_000_000))
);
}
#[test]
fn kernel_tx_below_pre_send_anchor_falls_back_to_post_send_not_pre_send() {
let mono = Instant::now();
let mut machine = machine_with_pending_probe_anchored(
send_wall_ms(1_000), // pre-send anchor / tx_not_before_wall
ClientTimestamp {
mono,
wall: send_wall_ms(1_008), // post-send sent_at
},
);
// Earlier than even the pre-send anchor: implausible, rejected.
machine.record_kernel_tx_timestamp(0, send_wall_ms(999));
let events = upstream_reply(&mut machine, 1_020, send_wall_ms(1_040), mono);
// Fallback is the post-send 1008ms sample (12ms), not the pre-send
// 1000ms anchor (which would have given 20ms).
assert_eq!(
reply_one_way(&events).unwrap().client_to_server,
Some(SignedDuration::from_nanos(12_000_000))
);
}
} }
+39 -1
View File
@@ -110,7 +110,20 @@ pub(crate) fn normalize_event(event: &ClientEvent) -> Option<StatsEvent> {
at: received_at.mono, at: received_at.mono,
bytes: *bytes, bytes: *bytes,
}), }),
ClientEvent::EchoLoss { timeout_at, .. } => Some(StatsEvent::Loss { at: *timeout_at }), ClientEvent::EchoLoss {
sent_at,
timeout_at,
..
} => Some(StatsEvent::Loss {
// `timeout_at` is anchored to the pre-send probe timestamp (see
// irtt-client's AGENTS.md), so it can precede this same probe's
// own post-send `sent_at` when a probe_timeout is small relative
// to the send-call duration. Rolling-window eviction assumes
// events are pushed in non-decreasing `at()` order, so clamp to
// `sent_at` to preserve that invariant without changing the
// public `timeout_at` value on the event itself.
at: (*timeout_at).max(sent_at.mono),
}),
ClientEvent::Warning { at, .. } => Some(StatsEvent::Warning { at: at.mono }), ClientEvent::Warning { at, .. } => Some(StatsEvent::Warning { at: at.mono }),
ClientEvent::SessionStarted { .. } ClientEvent::SessionStarted { .. }
| ClientEvent::NoTestCompleted { .. } | ClientEvent::NoTestCompleted { .. }
@@ -219,4 +232,29 @@ mod tests {
let now = SystemTime::now(); let now = SystemTime::now();
assert!(system_time_ns(now).is_some()); assert!(system_time_ns(now).is_some());
} }
#[test]
fn loss_windowing_timestamp_is_clamped_to_its_own_sent_at() {
let base = Instant::now();
let sent_at = ClientTimestamp {
mono: base + Duration::from_millis(50),
wall: UNIX_EPOCH + Duration::from_millis(50),
};
// The pre-send-anchored `timeout_at` can precede this same probe's
// own post-send `sent_at` when probe_timeout is small relative to
// the send-call duration (see irtt-client's AGENTS.md). Rolling-
// window eviction assumes non-decreasing event timestamps, so the
// windowing `at()` must not regress behind the probe's own Sent
// event even though the public `timeout_at` field legitimately can.
let timeout_at = base + Duration::from_millis(10);
let event = ClientEvent::EchoLoss {
seq: 0,
sent_at,
timeout_at,
};
let normalized = normalize_event(&event).expect("loss event normalizes");
assert_eq!(normalized.at(), sent_at.mono);
}
} }