Merge pull request #100 from Lochnair/fix/client-kernel-rx-owd

fix(client): use kernel receive time for downstream OWD

crates/irtt-client/src/client/tests/probes_replies/timing_math.rs

@@ -1,4 +1,5 @@
 use super::*;
+use crate::metadata::ReceiveMeta;
 
 #[test]
 fn server_processing_greater_than_raw_does_not_underflow() {
@@ -70,7 +71,10 @@ fn compute_one_way_omits_samples_when_client_wall_time_overflows_i64() {
         ..Default::default()
     };
 
-    assert_eq!(compute_one_way(&sent_at, &received_at, &ts), None);
+    assert_eq!(
+        compute_one_way(&sent_at, &received_at, &ReceiveMeta::default(), &ts),
+        None
+    );
 }
 
 #[test]
@@ -89,7 +93,7 @@ fn compute_one_way_returns_available_direction_samples() {
         ..Default::default()
     };
 
-    let sample = compute_one_way(&sent_at, &received_at, &ts).unwrap();
+    let sample = compute_one_way(&sent_at, &received_at, &ReceiveMeta::default(), &ts).unwrap();
     assert_eq!(
         sample.client_to_server,
         Some(SignedDuration::from_nanos(15_000_000))
@@ -116,7 +120,7 @@ fn compute_one_way_preserves_negative_client_to_server_delay() {
         ..Default::default()
     };
 
-    let sample = compute_one_way(&sent_at, &received_at, &ts).unwrap();
+    let sample = compute_one_way(&sent_at, &received_at, &ReceiveMeta::default(), &ts).unwrap();
 
     assert_eq!(
         sample.client_to_server,
@@ -144,7 +148,7 @@ fn compute_one_way_preserves_negative_server_to_client_delay() {
         ..Default::default()
     };
 
-    let sample = compute_one_way(&sent_at, &received_at, &ts).unwrap();
+    let sample = compute_one_way(&sent_at, &received_at, &ReceiveMeta::default(), &ts).unwrap();
 
     assert_eq!(
         sample.client_to_server,

crates/irtt-client/src/event.rs

@@ -348,7 +348,11 @@ pub struct OneWayDelaySample {
     /// Server-to-client delay.
     ///
     /// This is computed from the server send wall time and the client receive
-    /// wall time when both are available.
+    /// wall time when both are available. The client receive endpoint is the
+    /// kernel receive timestamp reported with the datagram when one is
+    /// available and plausible for this receive, and the userspace receive wall
+    /// time otherwise. RTT is unaffected and remains a userspace monotonic
+    /// measurement.
     pub server_to_client: Option<SignedDuration>,
 }
 
@@ -377,5 +381,11 @@ pub struct PacketMeta {
     pub traffic_class: Option<u8>,
     pub dscp: Option<u8>,
     pub ecn: Option<u8>,
+    /// Kernel software receive timestamp reported with the datagram, where the
+    /// platform provides one.
+    ///
+    /// This is the raw observed value, reported whether or not it was accepted
+    /// as the downstream one-way delay endpoint. It is a kernel software
+    /// timestamp, not a NIC hardware ingress time.
     pub kernel_rx_timestamp: Option<std::time::SystemTime>,
 }

crates/irtt-client/src/metadata.rs

@@ -1,13 +1,51 @@
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
 
 use crate::event::PacketMeta;
 
+/// Largest lag behind the paired userspace receive sample that a kernel
+/// receive timestamp may show and still be used as a measurement endpoint.
+///
+/// This is a sanity guard, not an expected kernel-to-userspace wakeup
+/// latency: ordinary wakeup latency is orders of magnitude smaller. A kernel
+/// timestamp this far behind the userspace sample most likely reflects a
+/// realtime clock discontinuity or otherwise unusable metadata, and falling
+/// back is preferable to injecting a bad cross-host timing sample.
+const MAX_KERNEL_RX_LAG: Duration = Duration::from_secs(1);
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 pub(crate) struct ReceiveMeta {
     pub(crate) traffic_class: Option<u8>,
     pub(crate) kernel_rx_timestamp: Option<SystemTime>,
 }
 
+impl ReceiveMeta {
+    /// Receive wall-clock endpoint to use for downstream one-way delay.
+    ///
+    /// Prefers an observed kernel receive timestamp, which is sampled earlier
+    /// than the userspace receive instant and therefore excludes socket wakeup
+    /// latency from the measured server-to-client delay. The kernel value is
+    /// only used when it is plausible for the datagram that produced
+    /// `userspace_wall`: it cannot be later than the userspace sample that
+    /// observed the datagram, and it may not lag it by more than
+    /// [`MAX_KERNEL_RX_LAG`]. Anything else falls back to `userspace_wall`.
+    ///
+    /// Rejecting a kernel timestamp here never discards it as observed
+    /// metadata: [`PacketMeta::kernel_rx_timestamp`] still reports the raw
+    /// value.
+    pub(crate) fn preferred_receive_wall(&self, userspace_wall: SystemTime) -> SystemTime {
+        let Some(kernel_wall) = self.kernel_rx_timestamp else {
+            return userspace_wall;
+        };
+        // `duration_since` fails exactly when the kernel timestamp is later
+        // than the userspace sample, which cannot happen for a datagram that
+        // sample observed.
+        match userspace_wall.duration_since(kernel_wall) {
+            Ok(lag) if lag <= MAX_KERNEL_RX_LAG => kernel_wall,
+            _ => userspace_wall,
+        }
+    }
+}
+
 impl From<ReceiveMeta> for PacketMeta {
     fn from(meta: ReceiveMeta) -> Self {
         Self {
@@ -21,9 +59,127 @@ impl From<ReceiveMeta> for PacketMeta {
 
 #[cfg(test)]
 mod tests {
-    use super::ReceiveMeta;
+    use std::time::{Duration, SystemTime};
+
+    use super::{ReceiveMeta, MAX_KERNEL_RX_LAG};
     use crate::event::PacketMeta;
 
+    fn userspace_wall() -> SystemTime {
+        SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000)
+    }
+
+    fn meta_with_kernel_rx(kernel_rx_timestamp: SystemTime) -> ReceiveMeta {
+        ReceiveMeta {
+            traffic_class: None,
+            kernel_rx_timestamp: Some(kernel_rx_timestamp),
+        }
+    }
+
+    #[test]
+    fn preferred_receive_wall_without_kernel_timestamp_uses_userspace_sample() {
+        let userspace = userspace_wall();
+
+        assert_eq!(
+            ReceiveMeta::default().preferred_receive_wall(userspace),
+            userspace
+        );
+    }
+
+    #[test]
+    fn preferred_receive_wall_uses_plausible_earlier_kernel_timestamp() {
+        let userspace = userspace_wall();
+        let kernel = userspace - Duration::from_micros(80);
+
+        assert_eq!(
+            meta_with_kernel_rx(kernel).preferred_receive_wall(userspace),
+            kernel
+        );
+    }
+
+    #[test]
+    fn preferred_receive_wall_accepts_kernel_timestamp_equal_to_userspace_sample() {
+        let userspace = userspace_wall();
+
+        assert_eq!(
+            meta_with_kernel_rx(userspace).preferred_receive_wall(userspace),
+            userspace
+        );
+    }
+
+    #[test]
+    fn preferred_receive_wall_rejects_kernel_timestamp_later_than_userspace_sample() {
+        let userspace = userspace_wall();
+        let kernel = userspace + Duration::from_nanos(1);
+
+        assert_eq!(
+            meta_with_kernel_rx(kernel).preferred_receive_wall(userspace),
+            userspace
+        );
+    }
+
+    #[test]
+    fn preferred_receive_wall_accepts_kernel_timestamp_exactly_at_the_lag_bound() {
+        // The bound is inclusive: a lag of exactly MAX_KERNEL_RX_LAG is
+        // still accepted, one nanosecond more is not.
+        let userspace = userspace_wall();
+        let at_bound = userspace - MAX_KERNEL_RX_LAG;
+        let past_bound = userspace - MAX_KERNEL_RX_LAG - Duration::from_nanos(1);
+
+        assert_eq!(
+            meta_with_kernel_rx(at_bound).preferred_receive_wall(userspace),
+            at_bound
+        );
+        assert_eq!(
+            meta_with_kernel_rx(past_bound).preferred_receive_wall(userspace),
+            userspace
+        );
+    }
+
+    #[test]
+    fn preferred_receive_wall_rejects_implausibly_old_kernel_timestamp() {
+        let userspace = userspace_wall();
+        let kernel = SystemTime::UNIX_EPOCH;
+
+        assert_eq!(
+            meta_with_kernel_rx(kernel).preferred_receive_wall(userspace),
+            userspace
+        );
+    }
+
+    #[test]
+    fn preferred_receive_wall_handles_extreme_timestamps_without_panicking() {
+        let userspace = userspace_wall();
+        let far_future = SystemTime::UNIX_EPOCH + Duration::from_secs(u64::from(u32::MAX)) * 4;
+        let before_epoch = SystemTime::UNIX_EPOCH - Duration::from_secs(u64::from(u32::MAX));
+
+        assert_eq!(
+            meta_with_kernel_rx(far_future).preferred_receive_wall(userspace),
+            userspace
+        );
+        assert_eq!(
+            meta_with_kernel_rx(before_epoch).preferred_receive_wall(userspace),
+            userspace
+        );
+        assert_eq!(
+            ReceiveMeta::default().preferred_receive_wall(before_epoch),
+            before_epoch
+        );
+    }
+
+    #[test]
+    fn metadata_preserves_kernel_timestamp_rejected_for_measurement() {
+        let userspace = userspace_wall();
+        let rejected = userspace + Duration::from_secs(30);
+        let meta = meta_with_kernel_rx(rejected);
+
+        assert_eq!(meta.preferred_receive_wall(userspace), userspace);
+        assert_eq!(
+            PacketMeta::from(meta).kernel_rx_timestamp,
+            Some(rejected),
+            "observed metadata must survive rejection as a measurement endpoint"
+        );
+    }
+
     #[test]
     fn metadata_observed_traffic_class_zero_preserves_observed_zero() {
         let packet_meta = PacketMeta::from(ReceiveMeta {

crates/irtt-client/src/session/machine.rs

@@ -669,7 +669,7 @@ 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, &reply.timestamps);
+                let one_way = compute_one_way(&pending.sent_at, &now, &meta, &reply.timestamps);
                 let received_stats = build_received_stats(&reply);
                 let is_late = session
                     .highest_received_seq
@@ -718,7 +718,7 @@ 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, &reply.timestamps);
+                let one_way = compute_one_way(&timed_out.sent_at, &now, &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);
@@ -938,16 +938,24 @@ fn build_server_timing(ts: &TimestampFields) -> Option<ServerTiming> {
     })
 }
 
+/// 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.
 pub(crate) fn compute_one_way(
     sent_at: &ClientTimestamp,
     received_at: &ClientTimestamp,
+    meta: &ReceiveMeta,
     ts: &TimestampFields,
 ) -> Option<OneWayDelaySample> {
     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_recv_ns = unix_epoch_ns_i64(received_at.wall);
+    let client_recv_ns = unix_epoch_ns_i64(meta.preferred_receive_wall(received_at.wall));
 
     let c2s = server_recv_wall
         .zip(client_send_ns)
@@ -1842,4 +1850,371 @@ mod tests {
         assert_eq!(session.next_wire_seq, 1);
         assert_eq!(session.packets_sent, 1);
     }
+
+    // Downstream one-way delay endpoint selection.
+    //
+    // All values are anchored at a fixed wall-clock base so the expected
+    // delays are exact: the client sends at the base, the server receives 5 ms
+    // later and sends 10 ms later, userspace observes the reply at 30 ms
+    // (20 ms downstream) and the kernel observed it at 25 ms (15 ms
+    // downstream).
+    const OWD_BASE_WALL_NS: i64 = 10_000_000_000;
+    const OWD_SERVER_RECV_WALL_NS: i64 = OWD_BASE_WALL_NS + 5_000_000;
+    const OWD_SERVER_SEND_WALL_NS: i64 = OWD_BASE_WALL_NS + 10_000_000;
+
+    fn owd_wall(offset_ns: u64) -> SystemTime {
+        UNIX_EPOCH + Duration::from_nanos(u64::try_from(OWD_BASE_WALL_NS).unwrap() + offset_ns)
+    }
+
+    fn owd_sent_at(mono: Instant) -> ClientTimestamp {
+        ClientTimestamp {
+            mono,
+            wall: owd_wall(0),
+        }
+    }
+
+    fn owd_received_at(mono: Instant) -> ClientTimestamp {
+        ClientTimestamp {
+            mono: mono + Duration::from_millis(40),
+            wall: owd_wall(30_000_000),
+        }
+    }
+
+    fn owd_timestamps() -> TimestampFields {
+        TimestampFields {
+            recv_wall: Some(OWD_SERVER_RECV_WALL_NS),
+            send_wall: Some(OWD_SERVER_SEND_WALL_NS),
+            ..Default::default()
+        }
+    }
+
+    fn owd_reply(timestamps: TimestampFields) -> EchoReply {
+        EchoReply {
+            flags: flags::FLAG_REPLY,
+            token: 0x0102_0304_0506_0708,
+            sequence: 0,
+            recv_count: None,
+            recv_window: None,
+            timestamps,
+            payload: Vec::new(),
+        }
+    }
+
+    fn kernel_rx_meta(offset_ns: u64) -> ReceiveMeta {
+        ReceiveMeta {
+            traffic_class: None,
+            kernel_rx_timestamp: Some(owd_wall(offset_ns)),
+        }
+    }
+
+    /// Machine with one outstanding probe for sequence 0.
+    fn machine_with_pending_probe(sent_at: ClientTimestamp) -> SessionMachine {
+        let mut machine = open_machine(4, Duration::from_secs(1));
+        let session = active_mut(&mut machine);
+        session.pending.preflight_insert(0).unwrap();
+        session.pending.commit_insert(PendingProbe {
+            wire_seq: 0,
+            sent_at,
+            timeout_at: sent_at.mono + Duration::from_secs(1),
+        });
+        session.next_wire_seq = 1;
+        machine
+    }
+
+    /// Machine whose probe for sequence 0 already timed out, so a reply for it
+    /// is a measurable late reply.
+    fn machine_with_timed_out_probe(sent_at: ClientTimestamp) -> SessionMachine {
+        let mut machine = open_machine(4, Duration::from_secs(1));
+        let session = active_mut(&mut machine);
+        session.timed_out.insert(PendingProbe {
+            wire_seq: 0,
+            sent_at,
+            timeout_at: sent_at.mono + Duration::from_secs(1),
+        });
+        session.next_wire_seq = 1;
+        machine
+    }
+
+    fn reply_one_way(events: &[ClientEvent]) -> Option<OneWayDelaySample> {
+        match events {
+            [ClientEvent::EchoReply { one_way, .. } | ClientEvent::LateReply { one_way, .. }] => {
+                *one_way
+            }
+            other => panic!("expected a single measurable reply event, got {other:?}"),
+        }
+    }
+
+    fn reply_rtt(events: &[ClientEvent]) -> Option<RttSample> {
+        match events {
+            [ClientEvent::EchoReply { rtt, .. }] => Some(*rtt),
+            [ClientEvent::LateReply { rtt, .. }] => *rtt,
+            other => panic!("expected a single measurable reply event, got {other:?}"),
+        }
+    }
+
+    fn process_owd_reply(
+        machine: &mut SessionMachine,
+        timestamps: TimestampFields,
+        meta: ReceiveMeta,
+        received_at: ClientTimestamp,
+    ) -> Vec<ClientEvent> {
+        machine
+            .process_echo_reply(owd_reply(timestamps), 64, received_at, meta)
+            .unwrap()
+    }
+
+    #[test]
+    fn downstream_one_way_delay_prefers_valid_kernel_receive_time() {
+        let mono = Instant::now();
+        let mut machine = machine_with_pending_probe(owd_sent_at(mono));
+
+        let events = process_owd_reply(
+            &mut machine,
+            owd_timestamps(),
+            kernel_rx_meta(25_000_000),
+            owd_received_at(mono),
+        );
+
+        let one_way = reply_one_way(&events).unwrap();
+        assert_eq!(
+            one_way.server_to_client,
+            Some(SignedDuration::from_nanos(15_000_000))
+        );
+        assert_eq!(
+            one_way.client_to_server,
+            Some(SignedDuration::from_nanos(5_000_000))
+        );
+    }
+
+    #[test]
+    fn downstream_one_way_delay_uses_userspace_receive_time_without_kernel_metadata() {
+        let mono = Instant::now();
+        let mut machine = machine_with_pending_probe(owd_sent_at(mono));
+
+        let events = process_owd_reply(
+            &mut machine,
+            owd_timestamps(),
+            ReceiveMeta::default(),
+            owd_received_at(mono),
+        );
+
+        let one_way = reply_one_way(&events).unwrap();
+        assert_eq!(
+            one_way.server_to_client,
+            Some(SignedDuration::from_nanos(20_000_000))
+        );
+        assert_eq!(
+            one_way.client_to_server,
+            Some(SignedDuration::from_nanos(5_000_000))
+        );
+    }
+
+    #[test]
+    fn downstream_one_way_delay_falls_back_for_implausible_kernel_receive_time() {
+        let mono = Instant::now();
+        // Later than the userspace sample that observed the datagram.
+        let mut future = machine_with_pending_probe(owd_sent_at(mono));
+        let future_events = process_owd_reply(
+            &mut future,
+            owd_timestamps(),
+            kernel_rx_meta(35_000_000),
+            owd_received_at(mono),
+        );
+
+        // Lagging the userspace sample by far more than MAX_KERNEL_RX_LAG.
+        let mut stale = machine_with_pending_probe(owd_sent_at(mono));
+        let stale_meta = ReceiveMeta {
+            traffic_class: None,
+            kernel_rx_timestamp: Some(UNIX_EPOCH),
+        };
+        let stale_events = process_owd_reply(
+            &mut stale,
+            owd_timestamps(),
+            stale_meta,
+            owd_received_at(mono),
+        );
+
+        for events in [&future_events, &stale_events] {
+            assert_eq!(
+                reply_one_way(events).unwrap().server_to_client,
+                Some(SignedDuration::from_nanos(20_000_000))
+            );
+        }
+    }
+
+    #[test]
+    fn measurable_late_reply_uses_the_same_receive_wall_selection() {
+        let mono = Instant::now();
+        let mut kernel = machine_with_timed_out_probe(owd_sent_at(mono));
+        let kernel_events = process_owd_reply(
+            &mut kernel,
+            owd_timestamps(),
+            kernel_rx_meta(25_000_000),
+            owd_received_at(mono),
+        );
+
+        let mut userspace = machine_with_timed_out_probe(owd_sent_at(mono));
+        let userspace_events = process_owd_reply(
+            &mut userspace,
+            owd_timestamps(),
+            ReceiveMeta::default(),
+            owd_received_at(mono),
+        );
+
+        assert!(matches!(
+            kernel_events.as_slice(),
+            [ClientEvent::LateReply { .. }]
+        ));
+        assert_eq!(
+            reply_one_way(&kernel_events).unwrap().server_to_client,
+            Some(SignedDuration::from_nanos(15_000_000))
+        );
+        assert_eq!(
+            reply_one_way(&userspace_events).unwrap().server_to_client,
+            Some(SignedDuration::from_nanos(20_000_000))
+        );
+    }
+
+    #[test]
+    fn untracked_late_reply_reports_no_one_way_delay_with_kernel_metadata() {
+        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 = process_owd_reply(
+            &mut machine,
+            owd_timestamps(),
+            kernel_rx_meta(25_000_000),
+            owd_received_at(mono),
+        );
+
+        match events.as_slice() {
+            [ClientEvent::LateReply {
+                sent_at,
+                rtt,
+                one_way,
+                ..
+            }] => {
+                assert!(sent_at.is_none());
+                assert!(rtt.is_none());
+                assert!(one_way.is_none());
+            }
+            other => panic!("expected an untracked LateReply, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn kernel_receive_time_changes_no_measurement_other_than_downstream_delay() {
+        let mono = Instant::now();
+        let timestamps = TimestampFields {
+            recv_mono: Some(1_000_000),
+            send_mono: Some(3_000_000),
+            ..owd_timestamps()
+        };
+
+        let mut kernel = machine_with_pending_probe(owd_sent_at(mono));
+        let kernel_events = process_owd_reply(
+            &mut kernel,
+            timestamps.clone(),
+            kernel_rx_meta(25_000_000),
+            owd_received_at(mono),
+        );
+
+        let mut userspace = machine_with_pending_probe(owd_sent_at(mono));
+        let userspace_events = process_owd_reply(
+            &mut userspace,
+            timestamps,
+            ReceiveMeta::default(),
+            owd_received_at(mono),
+        );
+
+        // RTT stays a purely monotonic userspace measurement.
+        assert_eq!(
+            reply_rtt(&kernel_events),
+            reply_rtt(&userspace_events),
+            "kernel receive metadata must not affect RTT"
+        );
+        assert_eq!(
+            reply_rtt(&kernel_events).unwrap().raw,
+            Duration::from_millis(40)
+        );
+
+        // Upstream delay uses the client send wall time only.
+        assert_eq!(
+            reply_one_way(&kernel_events).unwrap().client_to_server,
+            reply_one_way(&userspace_events).unwrap().client_to_server
+        );
+        assert_ne!(
+            reply_one_way(&kernel_events).unwrap().server_to_client,
+            reply_one_way(&userspace_events).unwrap().server_to_client
+        );
+    }
+
+    #[test]
+    fn kernel_receive_time_applies_against_a_server_midpoint_timestamp() {
+        let mono = Instant::now();
+        let midpoint = TimestampFields {
+            midpoint_wall: Some(OWD_SERVER_SEND_WALL_NS),
+            ..Default::default()
+        };
+        let mut machine = machine_with_pending_probe(owd_sent_at(mono));
+
+        let events = process_owd_reply(
+            &mut machine,
+            midpoint,
+            kernel_rx_meta(25_000_000),
+            owd_received_at(mono),
+        );
+
+        let one_way = reply_one_way(&events).unwrap();
+        assert_eq!(
+            one_way.server_to_client,
+            Some(SignedDuration::from_nanos(15_000_000))
+        );
+        assert_eq!(
+            one_way.client_to_server,
+            Some(SignedDuration::from_nanos(10_000_000))
+        );
+    }
+
+    #[test]
+    fn kernel_receive_time_does_not_create_missing_cross_instant_directions() {
+        let mono = Instant::now();
+        let receive_only = TimestampFields {
+            recv_wall: Some(OWD_SERVER_RECV_WALL_NS),
+            ..Default::default()
+        };
+        let send_only = TimestampFields {
+            send_wall: Some(OWD_SERVER_SEND_WALL_NS),
+            ..Default::default()
+        };
+
+        let mut receive_machine = machine_with_pending_probe(owd_sent_at(mono));
+        let receive_events = process_owd_reply(
+            &mut receive_machine,
+            receive_only,
+            kernel_rx_meta(25_000_000),
+            owd_received_at(mono),
+        );
+        let receive_sample = reply_one_way(&receive_events).unwrap();
+        assert_eq!(
+            receive_sample.client_to_server,
+            Some(SignedDuration::from_nanos(5_000_000))
+        );
+        assert_eq!(receive_sample.server_to_client, None);
+
+        let mut send_machine = machine_with_pending_probe(owd_sent_at(mono));
+        let send_events = process_owd_reply(
+            &mut send_machine,
+            send_only,
+            kernel_rx_meta(25_000_000),
+            owd_received_at(mono),
+        );
+        let send_sample = reply_one_way(&send_events).unwrap();
+        assert_eq!(send_sample.client_to_server, None);
+        assert_eq!(
+            send_sample.server_to_client,
+            Some(SignedDuration::from_nanos(15_000_000))
+        );
+    }
 }