From 1066d5df72e6bbfcef0407747fc950076b9d58e4 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sat, 15 Aug 2026 18:54:41 +0700 Subject: [PATCH] fix(client): use kernel transmit time for upstream OWD Upstream one-way delay now prefers the correlated Linux TX_SOFTWARE wall timestamp retained by PendingProbe when it is locally plausible (sent_at <= kernel_tx <= received_at, all on the client's own wall clock), falling back to the userspace sent_at.wall sample otherwise. RTT, downstream OWD, IPDV, and timeout/scheduling semantics are unchanged. --- crates/irtt-client/AGENTS.md | 32 +- .../tests/probes_replies/timing_math.rs | 11 +- crates/irtt-client/src/probe.rs | 8 +- crates/irtt-client/src/session/machine.rs | 339 +++++++++++++++++- 4 files changed, 366 insertions(+), 24 deletions(-) diff --git a/crates/irtt-client/AGENTS.md b/crates/irtt-client/AGENTS.md index c96ba5b..cb9f468 100644 --- a/crates/irtt-client/AGENTS.md +++ b/crates/irtt-client/AGENTS.md @@ -100,9 +100,29 @@ alongside the authoritative userspace `sent_at`. It survives pending -> timed-out retention (`SessionMachine::record_kernel_tx_timestamp` updates whichever of `PendingMap`/`TimedOutMap` still holds the probe) and is otherwise bounded by the same limits those maps already enforce — no new -unbounded state. First valid observation wins on duplicates. **It is -currently dormant metadata only**: no RTT, OWD, or IPDV computation reads -it. `sent_at` remains the sole input to all of those, exactly as before this -capability existed. A later change may select it as a preferred send -timestamp; until then, treat any code path that reads -`kernel_tx_timestamp` for a measurement as a bug. +unbounded state. First valid observation wins on duplicates. + +**It 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 + + sent_at.wall <= kernel_tx_timestamp <= reply_received_at.wall + +(inclusive, compared only against the client's own wall clock — never +against the remote server's wall timestamp). Otherwise, including when no +kernel timestamp was observed, the upstream endpoint falls back to +`sent_at.wall`; 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 +downstream (`server_to_client`) one-way delay, exactly as before this +capability existed; `TX_SOFTWARE` is still a software kernel timestamp near +the driver handoff, not a NIC hardware transmit time. IPDV is likewise +unaffected — it is computed from the existing userspace timestamp fields +only. 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 +kernel timestamp that arrives after a reply has already been processed is +simply never observed by that reply. diff --git a/crates/irtt-client/src/client/tests/probes_replies/timing_math.rs b/crates/irtt-client/src/client/tests/probes_replies/timing_math.rs index 0926973..043cffd 100644 --- a/crates/irtt-client/src/client/tests/probes_replies/timing_math.rs +++ b/crates/irtt-client/src/client/tests/probes_replies/timing_math.rs @@ -72,7 +72,7 @@ fn compute_one_way_omits_samples_when_client_wall_time_overflows_i64() { }; assert_eq!( - compute_one_way(&sent_at, &received_at, &ReceiveMeta::default(), &ts), + compute_one_way(&sent_at, &received_at, None, &ReceiveMeta::default(), &ts), None ); } @@ -93,7 +93,8 @@ fn compute_one_way_returns_available_direction_samples() { ..Default::default() }; - let sample = compute_one_way(&sent_at, &received_at, &ReceiveMeta::default(), &ts).unwrap(); + let sample = + compute_one_way(&sent_at, &received_at, None, &ReceiveMeta::default(), &ts).unwrap(); assert_eq!( sample.client_to_server, Some(SignedDuration::from_nanos(15_000_000)) @@ -120,7 +121,8 @@ fn compute_one_way_preserves_negative_client_to_server_delay() { ..Default::default() }; - let sample = compute_one_way(&sent_at, &received_at, &ReceiveMeta::default(), &ts).unwrap(); + let sample = + compute_one_way(&sent_at, &received_at, None, &ReceiveMeta::default(), &ts).unwrap(); assert_eq!( sample.client_to_server, @@ -148,7 +150,8 @@ fn compute_one_way_preserves_negative_server_to_client_delay() { ..Default::default() }; - let sample = compute_one_way(&sent_at, &received_at, &ReceiveMeta::default(), &ts).unwrap(); + let sample = + compute_one_way(&sent_at, &received_at, None, &ReceiveMeta::default(), &ts).unwrap(); assert_eq!( sample.client_to_server, diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index a1f842a..2f73e75 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -12,9 +12,11 @@ pub(crate) struct PendingProbe { pub timeout_at: Instant, /// Observed Linux kernel TX_SOFTWARE wall timestamp for this probe's /// send, when the socket has TX timestamping enabled and a matching - /// `MSG_ERRQUEUE` record has been drained. Dormant metadata only: no - /// measurement in this change reads it. `sent_at` remains authoritative - /// for RTT, timeout, and current one-way delay. + /// `MSG_ERRQUEUE` record has been drained. Optional observed metadata, + /// eligible only for upstream one-way delay after local plausibility + /// validation (see `compute_one_way`'s `preferred_send_wall`). `sent_at` + /// remains authoritative for RTT, timeout, pacing/scheduling semantics, + /// and the userspace fallback for upstream one-way delay. pub kernel_tx_timestamp: Option, } diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index a2f9368..4e36670 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -699,7 +699,13 @@ impl SessionMachine { if let Some(pending) = session.pending.remove(wire_seq) { let rtt = compute_rtt(&pending.sent_at, &now, &reply.timestamps); let server_timing = build_server_timing(&reply.timestamps); - let one_way = compute_one_way(&pending.sent_at, &now, &meta, &reply.timestamps); + let one_way = compute_one_way( + &pending.sent_at, + &now, + pending.kernel_tx_timestamp, + &meta, + &reply.timestamps, + ); let received_stats = build_received_stats(&reply); let is_late = session .highest_received_seq @@ -748,7 +754,13 @@ impl SessionMachine { } else if let Some(timed_out) = session.timed_out.remove(wire_seq) { let rtt = compute_rtt(&timed_out.sent_at, &now, &reply.timestamps); let server_timing = build_server_timing(&reply.timestamps); - let one_way = compute_one_way(&timed_out.sent_at, &now, &meta, &reply.timestamps); + let one_way = compute_one_way( + &timed_out.sent_at, + &now, + timed_out.kernel_tx_timestamp, + &meta, + &reply.timestamps, + ); let received_stats = build_received_stats(&reply); let highest_seen = session.highest_received_seq.unwrap_or(wire_seq); update_highest_received(&mut session.highest_received_seq, wire_seq); @@ -968,23 +980,53 @@ fn build_server_timing(ts: &TimestampFields) -> Option { }) } +/// Selects the client send wall-clock endpoint for upstream one-way delay. +/// +/// Prefers the correlated Linux kernel `TX_SOFTWARE` timestamp — a software +/// timestamp near the driver handoff, not physical NIC departure — when it +/// is locally plausible: it must not precede `sent_at` and must not follow +/// `received_at`, both on the client's own wall clock. Comparing only these +/// two same-client anchors means the check detects local causal-ordering +/// inconsistency (e.g. a backward wall-clock step), not every possible +/// wall-clock discontinuity, and it deliberately does not compare against +/// the remote server's wall clock: cross-host clock 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` 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( + sent_at: SystemTime, + kernel_tx_timestamp: Option, + received_at: SystemTime, +) -> SystemTime { + match kernel_tx_timestamp { + Some(kernel_tx) if sent_at <= kernel_tx && kernel_tx <= received_at => kernel_tx, + _ => sent_at, + } +} + /// Computes both one-way delay directions from wall-clock endpoints. /// -/// The upstream direction always uses the client's send wall time. The -/// downstream direction uses [`ReceiveMeta::preferred_receive_wall`], which -/// prefers a plausible kernel receive timestamp over the userspace receive -/// wall sample; the kernel timestamp is never used for the upstream direction -/// or for RTT. +/// The upstream direction prefers a locally plausible correlated kernel +/// `TX_SOFTWARE` send timestamp over the userspace pre-send sample; see +/// [`preferred_send_wall`]. The downstream direction uses +/// [`ReceiveMeta::preferred_receive_wall`], which prefers a plausible kernel +/// receive timestamp over the userspace receive wall sample. Neither kernel +/// timestamp is ever used for RTT. pub(crate) fn compute_one_way( sent_at: &ClientTimestamp, received_at: &ClientTimestamp, + kernel_tx_timestamp: Option, meta: &ReceiveMeta, ts: &TimestampFields, ) -> Option { let server_recv_wall = ts.recv_wall.or(ts.midpoint_wall); let server_send_wall = ts.send_wall.or(ts.midpoint_wall); - let client_send_ns = unix_epoch_ns_i64(sent_at.wall); + let client_send_wall = preferred_send_wall(sent_at.wall, kernel_tx_timestamp, received_at.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 c2s = server_recv_wall @@ -2531,11 +2573,13 @@ mod tests { } #[test] - fn kernel_tx_timestamp_presence_does_not_change_rtt_or_owd() { + fn kernel_tx_timestamp_changes_only_upstream_one_way_delay() { let mono = Instant::now(); let mut without = machine_with_pending_probe(owd_sent_at(mono)); let mut with = machine_with_pending_probe(owd_sent_at(mono)); - with.record_kernel_tx_timestamp(0, tx_ts(1)); + // Plausible: strictly between sent_at and received_at on the + // client's own wall clock. + with.record_kernel_tx_timestamp(0, owd_wall(2_000_000)); let without_events = process_owd_reply( &mut without, @@ -2558,6 +2602,279 @@ mod tests { [ClientEvent::EchoReply { rtt, one_way, .. }] => (*rtt, *one_way), other => panic!("expected one EchoReply, got {other:?}"), }; - assert_eq!(without_reply, with_reply); + + // RTT is a purely monotonic userspace measurement. + assert_eq!(without_reply.0, with_reply.0, "RTT must be unaffected"); + // Downstream delay is governed solely by the kernel RX selection. + assert_eq!( + without_reply.1.unwrap().server_to_client, + with_reply.1.unwrap().server_to_client, + "downstream delay must be unaffected" + ); + // Upstream delay is the one thing the kernel TX timestamp changes. + assert_ne!( + without_reply.1.unwrap().client_to_server, + with_reply.1.unwrap().client_to_server + ); + assert_eq!( + with_reply.1.unwrap().client_to_server, + Some(SignedDuration::from_nanos(3_000_000)) + ); + } + + // Upstream one-way delay: preferred client send wall selection. + // + // `preferred_send_wall` is exercised directly first (pure plausibility + // logic), then through `compute_one_way`/`process_echo_reply` to pin the + // end-to-end measurement behavior described in the client crate's + // `AGENTS.md`. + + fn send_wall_ms(offset_ms: u64) -> SystemTime { + UNIX_EPOCH + Duration::from_millis(offset_ms) + } + + #[test] + fn preferred_send_wall_without_kernel_timestamp_uses_sent_at() { + let sent_at = send_wall_ms(1_000); + let received_at = send_wall_ms(1_040); + assert_eq!(preferred_send_wall(sent_at, None, received_at), sent_at); + } + + #[test] + fn preferred_send_wall_uses_kernel_timestamp_strictly_between_bounds() { + let sent_at = send_wall_ms(1_000); + let kernel_tx = send_wall_ms(1_005); + let received_at = send_wall_ms(1_040); + assert_eq!( + preferred_send_wall(sent_at, Some(kernel_tx), received_at), + kernel_tx + ); + } + + #[test] + fn preferred_send_wall_accepts_kernel_timestamp_equal_to_sent_at() { + let sent_at = send_wall_ms(1_000); + let received_at = send_wall_ms(1_040); + assert_eq!( + preferred_send_wall(sent_at, Some(sent_at), received_at), + sent_at + ); + } + + #[test] + fn preferred_send_wall_accepts_kernel_timestamp_equal_to_received_at() { + let sent_at = send_wall_ms(1_000); + let received_at = send_wall_ms(1_040); + assert_eq!( + preferred_send_wall(sent_at, Some(received_at), received_at), + received_at + ); + } + + #[test] + fn preferred_send_wall_rejects_kernel_timestamp_earlier_than_sent_at() { + let sent_at = send_wall_ms(1_000); + let kernel_tx = sent_at - Duration::from_nanos(1); + let received_at = send_wall_ms(1_040); + assert_eq!( + preferred_send_wall(sent_at, Some(kernel_tx), received_at), + sent_at + ); + } + + #[test] + fn preferred_send_wall_rejects_kernel_timestamp_later_than_received_at() { + let sent_at = send_wall_ms(1_000); + let received_at = send_wall_ms(1_040); + let kernel_tx = received_at + Duration::from_nanos(1); + assert_eq!( + preferred_send_wall(sent_at, Some(kernel_tx), received_at), + sent_at + ); + } + + #[test] + fn preferred_send_wall_handles_extreme_timestamps_without_panicking() { + let sent_at = send_wall_ms(1_000); + let received_at = send_wall_ms(1_040); + 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)); + + assert_eq!( + preferred_send_wall(sent_at, Some(far_future), received_at), + sent_at + ); + assert_eq!( + preferred_send_wall(sent_at, Some(before_epoch), received_at), + sent_at + ); + assert_eq!( + preferred_send_wall(before_epoch, None, received_at), + before_epoch + ); + } + + fn upstream_probe(sent_wall: SystemTime, mono: Instant) -> SessionMachine { + machine_with_pending_probe(ClientTimestamp { + mono, + wall: sent_wall, + }) + } + + fn upstream_reply( + machine: &mut SessionMachine, + server_recv_wall_ms: u64, + received_wall: SystemTime, + mono: Instant, + ) -> Vec { + let timestamps = TimestampFields { + recv_wall: Some( + i64::try_from(Duration::from_millis(server_recv_wall_ms).as_nanos()).unwrap(), + ), + ..Default::default() + }; + let received_at = ClientTimestamp { + mono: mono + Duration::from_millis(40), + wall: received_wall, + }; + process_owd_reply(machine, timestamps, ReceiveMeta::default(), received_at) + } + + #[test] + fn upstream_one_way_delay_prefers_plausible_kernel_tx_time() { + let mono = Instant::now(); + let mut machine = upstream_probe(send_wall_ms(1_000), mono); + 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 upstream_one_way_delay_falls_back_without_kernel_tx() { + let mono = Instant::now(); + let mut machine = upstream_probe(send_wall_ms(1_000), mono); + + let events = upstream_reply(&mut machine, 1_020, send_wall_ms(1_040), mono); + + assert_eq!( + reply_one_way(&events).unwrap().client_to_server, + Some(SignedDuration::from_nanos(20_000_000)) + ); + } + + #[test] + fn upstream_one_way_delay_falls_back_for_kernel_tx_earlier_than_sent_at() { + let mono = Instant::now(); + let mut machine = upstream_probe(send_wall_ms(1_000), mono); + machine.record_kernel_tx_timestamp(0, send_wall_ms(999)); + + let events = upstream_reply(&mut machine, 1_020, send_wall_ms(1_040), mono); + + assert_eq!( + reply_one_way(&events).unwrap().client_to_server, + Some(SignedDuration::from_nanos(20_000_000)) + ); + } + + #[test] + fn upstream_one_way_delay_falls_back_for_kernel_tx_later_than_received_at() { + let mono = Instant::now(); + let mut machine = upstream_probe(send_wall_ms(1_000), mono); + machine.record_kernel_tx_timestamp(0, send_wall_ms(1_041)); + + let events = upstream_reply(&mut machine, 1_020, send_wall_ms(1_040), mono); + + assert_eq!( + reply_one_way(&events).unwrap().client_to_server, + Some(SignedDuration::from_nanos(20_000_000)) + ); + } + + #[test] + fn upstream_one_way_delay_measurable_late_reply_uses_the_retained_kernel_tx_time() { + let mono = Instant::now(); + let mut machine = machine_with_timed_out_probe(ClientTimestamp { + mono, + wall: send_wall_ms(1_000), + }); + 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!(matches!(events.as_slice(), [ClientEvent::LateReply { .. }])); + assert_eq!( + reply_one_way(&events).unwrap().client_to_server, + Some(SignedDuration::from_nanos(15_000_000)) + ); + } + + #[test] + fn upstream_one_way_delay_untracked_late_reply_stays_unmeasurable() { + let mono = Instant::now(); + let mut machine = open_machine(4, Duration::from_secs(1)); + active_mut(&mut machine).highest_received_seq = Some(5); + + let events = upstream_reply(&mut machine, 1_020, send_wall_ms(1_040), mono); + + match events.as_slice() { + [ClientEvent::LateReply { one_way, .. }] => assert!(one_way.is_none()), + other => panic!("expected an untracked LateReply, got {other:?}"), + } + } + + #[test] + fn upstream_one_way_delay_applies_against_a_server_midpoint_timestamp() { + let mono = Instant::now(); + let mut machine = upstream_probe(send_wall_ms(100), mono); + machine.record_kernel_tx_timestamp(0, send_wall_ms(105)); + + let timestamps = TimestampFields { + midpoint_wall: Some(120_000_000), + ..Default::default() + }; + let received_at = ClientTimestamp { + mono: mono + Duration::from_millis(40), + wall: send_wall_ms(140), + }; + let events = process_owd_reply( + &mut machine, + timestamps, + ReceiveMeta::default(), + received_at, + ); + + assert_eq!( + reply_one_way(&events).unwrap().client_to_server, + Some(SignedDuration::from_nanos(15_000_000)) + ); + } + + #[test] + fn upstream_one_way_delay_remains_unavailable_for_send_only_server_timestamps() { + let mono = Instant::now(); + let mut machine = upstream_probe(send_wall_ms(1_000), mono); + machine.record_kernel_tx_timestamp(0, send_wall_ms(1_005)); + + let timestamps = TimestampFields { + send_wall: Some(1_010_000_000), + ..Default::default() + }; + let received_at = ClientTimestamp { + mono: mono + Duration::from_millis(40), + wall: send_wall_ms(1_040), + }; + let events = process_owd_reply( + &mut machine, + timestamps, + ReceiveMeta::default(), + received_at, + ); + + assert_eq!(reply_one_way(&events).unwrap().client_to_server, None); } }