fix(client): timestamp probes after successful send #106
No files matched your search
@@ -96,33 +96,67 @@ 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 sole input to RTT and the downstream
|
||||||
downstream (`server_to_client`) one-way delay, exactly as before this
|
(`server_to_client`) one-way delay, exactly as before this capability
|
||||||
capability existed; `TX_SOFTWARE` is still a software kernel timestamp near
|
existed; `TX_SOFTWARE` is still a software kernel timestamp near the driver
|
||||||
the driver handoff, not a NIC hardware transmit time. IPDV is likewise
|
handoff, not a NIC hardware transmit time. IPDV is likewise unaffected — it
|
||||||
unaffected — it is computed from the existing userspace timestamp fields
|
is computed from the existing userspace timestamp fields only. A normal
|
||||||
only. A normal reply and a measurable late reply (one whose `PendingProbe`
|
reply and a measurable late reply (one whose `PendingProbe` is still
|
||||||
is still retained in `TimedOutMap`) apply the identical selection policy; an
|
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.
|
||||||
@@ -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;
|
||||||
|
|
||||||
|
|||||||
@@ -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),
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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,
|
||||||
sent_at,
|
|
||||||
timeout_at,
|
timeout_at,
|
||||||
kernel_tx_timestamp: None,
|
tx_not_before_wall: send_anchor.wall,
|
||||||
},
|
|
||||||
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))
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
Reference in new issue
Block a user
When probe send-call durations vary, moving public
sent_atfrom before the send to after it changes send-direction IPDV:ReplySample::from_reply_partscopies this value intoclient_send_mono/client_send_wall_ns, andsend_ipdv_nscompares those fields across probes. The commit description acknowledges this semantic shift, so retaining “IPDV is likewise unaffected” as repository guidance establishes a false invariant; document the shift here, or preserve a separate pre-send timestamp if IPDV truly must remain unchanged.AGENTS.md reference: crates/irtt-client/AGENTS.md:L153-L157
Useful? React with 👍 / 👎.
Fixed — AGENTS.md now documents that the upstream IPDV component (
send_ipdv_ns, fed byReplySample.client_send_mono/client_send_wall_nsfromsent_at) shifts endpoint along withsent_atitself, rather than claiming IPDV is unaffected. Pushed ine7b0632.🤖 Addressed by Claude Code
sent_atis not an input to the downstream (server_to_client) one-way delay:compute_one_wayderives that direction frommeta.preferred_receive_wall(received_at.wall)and the server send timestamp. Retaining this statement in repository guidance establishes a false timing invariant and can cause future changes or tests to couple downstream delay to the probe's send endpoint; documentsent_atas the RTT and upstream-fallback endpoint instead.Useful? React with 👍 / 👎.
Fixed — reworded to state sent_at is the RTT endpoint and the upstream (client_to_server) OWD userspace fallback only; downstream (server_to_client) OWD is called out explicitly as unaffected, derived solely from ReceiveMeta::preferred_receive_wall and the negotiated server send timestamp. Pushed in
aac5247.🤖 Addressed by Claude Code