From 7440e7d2174ae7d720cad0254b9421ad6d4f816d Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 3 May 2026 21:26:47 +0700 Subject: [PATCH] test(client): add integration tests for m8 compatibility --- crates/irtt-client/tests/m8_compat.rs | 364 ++++++++++++++++++++++++ crates/irtt-client/tests/support/mod.rs | 359 +++++++++++++++++++++++ 2 files changed, 723 insertions(+) create mode 100644 crates/irtt-client/tests/m8_compat.rs create mode 100644 crates/irtt-client/tests/support/mod.rs diff --git a/crates/irtt-client/tests/m8_compat.rs b/crates/irtt-client/tests/m8_compat.rs new file mode 100644 index 0000000..001b96a --- /dev/null +++ b/crates/irtt-client/tests/m8_compat.rs @@ -0,0 +1,364 @@ +mod support; + +use irtt_client::{ + Client, ClientConfig, ClientError, ClientEvent, OneWayDelaySample, ReceivedStatsSample, + RttSample, ServerTiming, +}; +use irtt_proto::{echo_packet_len, Clock, Params, ReceivedStats, StampAt, TimestampFields}; + +use support::{ + config_for_params, default_params, params_for_modes, run_one_probe, run_one_probe_with_config, + server_fill, standard_timestamps, start_open_server, OneProbeRun, ServerObservation, + RECV_COUNT, RECV_WINDOW, TOKEN, +}; + +struct ReplyView<'a> { + seq: u32, + logical_seq: u64, + rtt: &'a RttSample, + server_timing: Option<&'a ServerTiming>, + one_way: Option<&'a OneWayDelaySample>, + received_stats: Option<&'a ReceivedStatsSample>, + bytes: usize, +} + +#[test] +fn received_stats_modes_drive_negotiated_echo_and_reply_events() { + for (mode, expected_count, expected_window) in [ + (ReceivedStats::None, None, None), + (ReceivedStats::Count, Some(RECV_COUNT), None), + (ReceivedStats::Window, None, Some(RECV_WINDOW)), + (ReceivedStats::Both, Some(RECV_COUNT), Some(RECV_WINDOW)), + ] { + let params = params_for_modes(mode, StampAt::None, Clock::Both); + let run = run_one_probe(params.clone(), TimestampFields::default()); + assert_negotiated_echo_use(&run, ¶ms, false); + + let reply = expect_echo_reply(&run.reply); + assert_eq!(reply.seq, 0); + assert_eq!(reply.logical_seq, 0); + assert_eq!(reply.bytes, echo_packet_len(false, ¶ms)); + assert!(reply.server_timing.is_none()); + assert!(reply.one_way.is_none()); + assert_received_stats(reply.received_stats, expected_count, expected_window); + } +} + +#[test] +fn timestamp_modes_drive_reply_event_timing_fields() { + for mode in [ + StampAt::None, + StampAt::Send, + StampAt::Receive, + StampAt::Both, + StampAt::Midpoint, + ] { + let params = params_for_modes(ReceivedStats::None, mode, Clock::Both); + let run = run_one_probe(params.clone(), standard_timestamps()); + assert_negotiated_echo_use(&run, ¶ms, false); + + let reply = expect_echo_reply(&run.reply); + assert!(reply.received_stats.is_none()); + match mode { + StampAt::None => { + assert!(reply.server_timing.is_none()); + assert!(reply.one_way.is_none()); + assert!(reply.rtt.adjusted.is_none()); + } + StampAt::Send => { + assert_timing_fields(reply.server_timing, false, false, false, false, true, true); + assert_one_way_presence(reply.one_way, false, true); + assert!(reply.rtt.adjusted.is_none()); + } + StampAt::Receive => { + assert_timing_fields(reply.server_timing, true, true, false, false, false, false); + assert_one_way_presence(reply.one_way, true, false); + assert!(reply.rtt.adjusted.is_none()); + } + StampAt::Both => { + assert_timing_fields(reply.server_timing, true, true, false, false, true, true); + assert_one_way_presence(reply.one_way, true, true); + assert!(reply.rtt.adjusted.is_some()); + assert!(reply.rtt.effective <= reply.rtt.raw); + } + StampAt::Midpoint => { + assert_timing_fields(reply.server_timing, false, false, true, true, false, false); + assert_one_way_presence(reply.one_way, true, true); + assert!(reply.rtt.adjusted.is_none()); + } + } + } +} + +#[test] +fn clock_modes_drive_reply_event_clock_fields() { + for clock in [Clock::Wall, Clock::Monotonic, Clock::Both] { + let params = params_for_modes(ReceivedStats::None, StampAt::Both, clock); + let run = run_one_probe(params.clone(), standard_timestamps()); + assert_negotiated_echo_use(&run, ¶ms, false); + + let reply = expect_echo_reply(&run.reply); + assert!(reply.received_stats.is_none()); + match clock { + Clock::Wall => { + assert_timing_fields(reply.server_timing, true, false, false, false, true, false); + assert_one_way_presence(reply.one_way, true, true); + } + Clock::Monotonic => { + assert_timing_fields(reply.server_timing, false, true, false, false, false, true); + assert!(reply.one_way.is_none()); + } + Clock::Both => { + assert_timing_fields(reply.server_timing, true, true, false, false, true, true); + assert_one_way_presence(reply.one_way, true, true); + } + } + assert!(reply.rtt.adjusted.is_some()); + assert!(reply.rtt.effective <= reply.rtt.raw); + } +} + +#[test] +fn rich_mode_exposes_stats_server_timing_and_adjusted_rtt() { + let params = params_for_modes(ReceivedStats::Both, StampAt::Both, Clock::Both); + let run = run_one_probe(params.clone(), standard_timestamps()); + assert_negotiated_echo_use(&run, ¶ms, false); + + let reply = expect_echo_reply(&run.reply); + assert_received_stats(reply.received_stats, Some(RECV_COUNT), Some(RECV_WINDOW)); + assert_timing_fields(reply.server_timing, true, true, false, false, true, true); + assert_one_way_presence(reply.one_way, true, true); + assert!(reply.rtt.adjusted.is_some()); + assert!(reply.rtt.effective <= reply.rtt.raw); +} + +#[test] +fn hmac_rich_mode_uses_negotiated_echo_layout_and_decodes_reply() { + let key = b"compat-secret".to_vec(); + let params = params_for_modes(ReceivedStats::Both, StampAt::Both, Clock::Both); + let config_params = params.clone(); + let config_key = key.clone(); + let run = run_one_probe_with_config(params.clone(), standard_timestamps(), Some(key), |addr| { + ClientConfig { + hmac_key: Some(config_key), + ..config_for_params(addr, &config_params) + } + }); + assert_negotiated_echo_use(&run, ¶ms, true); + + let reply = expect_echo_reply(&run.reply); + assert_eq!(reply.bytes, echo_packet_len(true, ¶ms)); + assert_received_stats(reply.received_stats, Some(RECV_COUNT), Some(RECV_WINDOW)); + assert_timing_fields(reply.server_timing, true, true, false, false, true, true); +} + +#[test] +fn strict_open_rejects_changed_compatibility_params_from_server() { + let mut requested = default_params(); + requested.length = 256; + requested.dscp = 46; + requested.server_fill = server_fill("rand"); + + for returned in changed_compatibility_params(&requested) { + let server = start_open_server(returned, None); + let mut config = config_for_params(server.addr, &requested); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert!(matches!( + client.open(irtt_client::ClientTimestamp::now()), + Err(ClientError::NegotiationRejected { .. }) + )); + server.join(); + } +} + +#[test] +fn loose_open_uses_returned_params_for_echo_layout_and_reply_parsing() { + let mut requested = default_params(); + requested.length = 128; + requested.dscp = 46; + requested.server_fill = server_fill("rand"); + + let mut returned = requested.clone(); + returned.length = 28; + returned.received_stats = ReceivedStats::Count; + returned.stamp_at = StampAt::Receive; + returned.clock = Clock::Wall; + returned.dscp = 8; + returned.server_fill = None; + + let requested_for_config = requested.clone(); + let run = run_one_probe_with_config(returned.clone(), standard_timestamps(), None, |addr| { + ClientConfig { + negotiation_policy: irtt_client::NegotiationPolicy::Loose, + ..config_for_params(addr, &requested_for_config) + } + }); + + assert_eq!(open_params(&run), &requested); + assert_eq!(run.negotiated.params, returned); + assert_echo_uses_params(&run, &returned, false); + + let reply = expect_echo_reply(&run.reply); + assert_received_stats(reply.received_stats, Some(RECV_COUNT), None); + assert_timing_fields(reply.server_timing, true, false, false, false, false, false); + assert_one_way_presence(reply.one_way, true, false); + assert!(reply.rtt.adjusted.is_none()); +} + +fn changed_compatibility_params(requested: &Params) -> Vec { + let mut changed = Vec::new(); + + let mut returned = requested.clone(); + returned.received_stats = ReceivedStats::Count; + changed.push(returned); + + let mut returned = requested.clone(); + returned.stamp_at = StampAt::Midpoint; + changed.push(returned); + + let mut returned = requested.clone(); + returned.clock = Clock::Wall; + changed.push(returned); + + let mut returned = requested.clone(); + returned.server_fill = None; + changed.push(returned); + + let mut returned = requested.clone(); + returned.dscp = 8; + changed.push(returned); + + let mut returned = requested.clone(); + returned.length = 128; + changed.push(returned); + + changed +} + +fn assert_negotiated_echo_use(run: &OneProbeRun, params: &Params, hmac: bool) { + assert_eq!(run.negotiated.params, *params); + assert_eq!(open_params(run), params); + assert_echo_uses_params(run, params, hmac); +} + +fn assert_echo_uses_params(run: &OneProbeRun, params: &Params, hmac: bool) { + let (echo_len, echo_hmac, token, sequence) = echo_observation(run); + assert_eq!(echo_len, echo_packet_len(hmac, params)); + assert_eq!(echo_hmac, hmac); + assert_eq!(token, TOKEN); + assert_eq!(sequence, 0); + + match &run.sent { + ClientEvent::EchoSent { + seq, + logical_seq, + bytes, + .. + } => { + assert_eq!(*seq, 0); + assert_eq!(*logical_seq, 0); + assert_eq!(*bytes, echo_packet_len(hmac, params)); + } + other => panic!("expected EchoSent, got {other:?}"), + } +} + +fn open_params(run: &OneProbeRun) -> &Params { + match &run.observations[0] { + ServerObservation::Open { params, .. } => params, + other => panic!("expected open observation, got {other:?}"), + } +} + +fn echo_observation(run: &OneProbeRun) -> (usize, bool, u64, u32) { + match run.observations[1] { + ServerObservation::Echo { + len, + hmac, + token, + sequence, + } => (len, hmac, token, sequence), + ref other => panic!("expected echo observation, got {other:?}"), + } +} + +fn expect_echo_reply(event: &ClientEvent) -> ReplyView<'_> { + match event { + ClientEvent::EchoReply { + seq, + logical_seq, + rtt, + server_timing, + one_way, + received_stats, + bytes, + .. + } => ReplyView { + seq: *seq, + logical_seq: *logical_seq, + rtt, + server_timing: server_timing.as_ref(), + one_way: one_way.as_ref(), + received_stats: received_stats.as_ref(), + bytes: *bytes, + }, + other => panic!("expected EchoReply, got {other:?}"), + } +} + +fn assert_received_stats( + sample: Option<&ReceivedStatsSample>, + expected_count: Option, + expected_window: Option, +) { + if expected_count.is_none() && expected_window.is_none() { + assert!(sample.is_none()); + } else { + let sample = sample.unwrap(); + assert_eq!(sample.count, expected_count); + assert_eq!(sample.window, expected_window); + } +} + +fn assert_timing_fields( + sample: Option<&ServerTiming>, + recv_wall: bool, + recv_mono: bool, + midpoint_wall: bool, + midpoint_mono: bool, + send_wall: bool, + send_mono: bool, +) { + if !(recv_wall || recv_mono || midpoint_wall || midpoint_mono || send_wall || send_mono) { + assert!(sample.is_none()); + return; + } + + let sample = sample.unwrap(); + assert_eq!(sample.receive_wall_ns.is_some(), recv_wall); + assert_eq!(sample.receive_mono_ns.is_some(), recv_mono); + assert_eq!(sample.midpoint_wall_ns.is_some(), midpoint_wall); + assert_eq!(sample.midpoint_mono_ns.is_some(), midpoint_mono); + assert_eq!(sample.send_wall_ns.is_some(), send_wall); + assert_eq!(sample.send_mono_ns.is_some(), send_mono); + assert_eq!( + sample.processing.is_some(), + (recv_mono && send_mono) || (recv_wall && send_wall) + ); +} + +fn assert_one_way_presence( + sample: Option<&OneWayDelaySample>, + client_to_server: bool, + server_to_client: bool, +) { + if !(client_to_server || server_to_client) { + assert!(sample.is_none()); + return; + } + + let sample = sample.unwrap(); + assert_eq!(sample.client_to_server.is_some(), client_to_server); + assert_eq!(sample.server_to_client.is_some(), server_to_client); +} diff --git a/crates/irtt-client/tests/support/mod.rs b/crates/irtt-client/tests/support/mod.rs new file mode 100644 index 0000000..e143383 --- /dev/null +++ b/crates/irtt-client/tests/support/mod.rs @@ -0,0 +1,359 @@ +use std::{ + net::{SocketAddr, UdpSocket}, + sync::mpsc, + thread::{self, JoinHandle}, + time::{Duration, SystemTime}, +}; + +use irtt_client::{ + Client, ClientConfig, ClientEvent, ClientTimestamp, NegotiatedParams, OpenOutcome, SocketConfig, +}; +use irtt_proto::{ + compute_hmac_in_place, echo_packet_len, flags, verify_hmac, Clock, Params, ReceivedStats, + ServerFill, StampAt, TimestampFields, HMAC_SIZE, MAGIC, PROTOCOL_VERSION, +}; + +const HMAC_OFFSET: usize = 4; +pub const TOKEN: u64 = 0x1234_5678_90ab_cdef; +pub const RECV_COUNT: u32 = 42; +pub const RECV_WINDOW: u64 = 0x07; + +pub struct FakeServer { + pub addr: SocketAddr, + rx: mpsc::Receiver, + done: JoinHandle<()>, +} + +impl FakeServer { + pub fn observations(&self, count: usize) -> Vec { + self.rx.iter().take(count).collect() + } + + pub fn join(self) { + self.done.join().unwrap(); + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ServerObservation { + Open { + params: Params, + hmac: bool, + }, + Echo { + len: usize, + hmac: bool, + token: u64, + sequence: u32, + }, +} + +pub struct OneProbeRun { + pub negotiated: NegotiatedParams, + pub sent: ClientEvent, + pub reply: ClientEvent, + pub observations: Vec, +} + +pub fn default_params() -> Params { + Params { + protocol_version: PROTOCOL_VERSION, + duration_ns: 3_000_000_000, + interval_ns: 1_000_000_000, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Params::default() + } +} + +pub fn params_for_modes(received_stats: ReceivedStats, stamp_at: StampAt, clock: Clock) -> Params { + Params { + received_stats, + stamp_at, + clock, + ..default_params() + } +} + +pub fn config_for_params(addr: SocketAddr, params: &Params) -> ClientConfig { + ClientConfig { + server_addr: addr.to_string(), + duration: if params.duration_ns == 0 { + None + } else { + Some(Duration::from_nanos(params.duration_ns as u64)) + }, + interval: Duration::from_nanos(params.interval_ns as u64), + length: u32::try_from(params.length).unwrap(), + received_stats: params.received_stats, + stamp_at: params.stamp_at, + clock: params.clock, + dscp: u8::try_from(params.dscp).unwrap(), + server_fill: params.server_fill.as_ref().map(|fill| fill.value.clone()), + open_timeouts: vec![Duration::from_millis(200)], + socket_config: SocketConfig { + recv_timeout: Some(Duration::from_millis(500)), + ..Default::default() + }, + ..ClientConfig::default() + } +} + +pub fn standard_timestamps() -> TimestampFields { + TimestampFields { + recv_wall: Some(1), + recv_mono: Some(5_000_000), + midpoint_wall: Some(1), + midpoint_mono: Some(5_000_000), + send_wall: Some(1), + send_mono: Some(5_000_000), + } +} + +pub fn run_one_probe(params: Params, timestamps: TimestampFields) -> OneProbeRun { + let config_params = params.clone(); + run_one_probe_with_config(params, timestamps, None, |addr| { + config_for_params(addr, &config_params) + }) +} + +pub fn run_one_probe_with_config( + server_params: Params, + timestamps: TimestampFields, + hmac_key: Option>, + build_config: F, +) -> OneProbeRun +where + F: FnOnce(SocketAddr) -> ClientConfig, +{ + let server = start_one_probe_server(server_params, timestamps, hmac_key); + let mut client = Client::connect(build_config(server.addr)).unwrap(); + let negotiated = assert_started(client.open(ClientTimestamp::now()).unwrap()); + + let sent_events = client.send_probe().unwrap(); + assert_eq!(sent_events.len(), 1); + + let reply_events = client.recv_once().unwrap(); + assert_eq!(reply_events.len(), 1); + + let observations = server.observations(2); + server.join(); + + OneProbeRun { + negotiated, + sent: sent_events.into_iter().next().unwrap(), + reply: reply_events.into_iter().next().unwrap(), + observations, + } +} + +pub fn start_open_server(params: Params, hmac_key: Option>) -> FakeServer { + start_fake_server(move |socket, tx| { + let (request, peer) = recv_request(&socket); + let (open_params, hmac) = decode_open_request_params(&request, hmac_key.as_deref()); + tx.send(ServerObservation::Open { + params: open_params, + hmac, + }) + .unwrap(); + + let reply = open_reply( + flags::FLAG_OPEN | flags::FLAG_REPLY, + TOKEN, + ¶ms, + hmac_key.as_deref(), + ); + socket.send_to(&reply, peer).unwrap(); + }) +} + +pub fn server_fill(value: &str) -> Option { + Some(ServerFill { + value: value.to_owned(), + }) +} + +fn start_one_probe_server( + params: Params, + timestamps: TimestampFields, + hmac_key: Option>, +) -> FakeServer { + start_fake_server(move |socket, tx| { + let (request, peer) = recv_request(&socket); + let (open_params, hmac) = decode_open_request_params(&request, hmac_key.as_deref()); + tx.send(ServerObservation::Open { + params: open_params, + hmac, + }) + .unwrap(); + + let reply = open_reply( + flags::FLAG_OPEN | flags::FLAG_REPLY, + TOKEN, + ¶ms, + hmac_key.as_deref(), + ); + socket.send_to(&reply, peer).unwrap(); + + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let (request, peer) = recv_request(&socket); + let echo = observe_echo_request(&request, hmac_key.as_deref()); + let sequence = match echo { + ServerObservation::Echo { sequence, .. } => sequence, + ServerObservation::Open { .. } => unreachable!(), + }; + tx.send(echo).unwrap(); + + let timestamps = materialize_wall_timestamps(timestamps); + let reply_packet = + echo_reply_packet(TOKEN, sequence, ¶ms, ×tamps, hmac_key.as_deref()); + socket.send_to(&reply_packet, peer).unwrap(); + }) +} + +fn start_fake_server(handler: F) -> FakeServer +where + F: FnOnce(UdpSocket, mpsc::Sender) + Send + 'static, +{ + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + let addr = socket.local_addr().unwrap(); + let (tx, rx) = mpsc::channel(); + let done = thread::spawn(move || handler(socket, tx)); + FakeServer { addr, rx, done } +} + +fn recv_request(socket: &UdpSocket) -> (Vec, SocketAddr) { + let mut buf = [0_u8; 2048]; + let (size, peer) = socket.recv_from(&mut buf).unwrap(); + (buf[..size].to_vec(), peer) +} + +fn assert_started(outcome: OpenOutcome) -> NegotiatedParams { + match outcome { + OpenOutcome::Started { negotiated, .. } => negotiated, + OpenOutcome::NoTestCompleted { .. } => panic!("expected started outcome"), + } +} + +fn decode_open_request_params(packet: &[u8], hmac_key: Option<&[u8]>) -> (Params, bool) { + assert_eq!(&packet[..3], &MAGIC); + assert!(packet[3] & flags::FLAG_OPEN != 0); + let hmac = packet[3] & flags::FLAG_HMAC != 0; + assert_eq!(hmac, hmac_key.is_some()); + if let Some(key) = hmac_key { + verify_hmac(key, packet, HMAC_OFFSET).unwrap(); + } + let params_offset = 4 + if hmac { HMAC_SIZE } else { 0 }; + (Params::decode(&packet[params_offset..]).unwrap(), hmac) +} + +fn observe_echo_request(packet: &[u8], hmac_key: Option<&[u8]>) -> ServerObservation { + assert_eq!(&packet[..3], &MAGIC); + assert_eq!(packet[3] & flags::FLAG_OPEN, 0); + assert_eq!(packet[3] & flags::FLAG_REPLY, 0); + assert_eq!(packet[3] & flags::FLAG_CLOSE, 0); + let hmac = packet[3] & flags::FLAG_HMAC != 0; + assert_eq!(hmac, hmac_key.is_some()); + if let Some(key) = hmac_key { + verify_hmac(key, packet, HMAC_OFFSET).unwrap(); + } + + let mut pos = 4 + if hmac { HMAC_SIZE } else { 0 }; + let token = u64::from_le_bytes(packet[pos..pos + 8].try_into().unwrap()); + pos += 8; + let sequence = u32::from_le_bytes(packet[pos..pos + 4].try_into().unwrap()); + + ServerObservation::Echo { + len: packet.len(), + hmac, + token, + sequence, + } +} + +fn open_reply(flags: u8, token: u64, params: &Params, hmac_key: Option<&[u8]>) -> Vec { + let mut packet = Vec::new(); + packet.extend_from_slice(&MAGIC); + packet.push(flags | hmac_key.map_or(0, |_| flags::FLAG_HMAC)); + if hmac_key.is_some() { + packet.extend_from_slice(&[0_u8; HMAC_SIZE]); + } + packet.extend_from_slice(&token.to_le_bytes()); + packet.extend_from_slice(¶ms.encode()); + if let Some(key) = hmac_key { + compute_hmac_in_place(key, &mut packet, HMAC_OFFSET).unwrap(); + } + packet +} + +fn echo_reply_packet( + token: u64, + seq: u32, + params: &Params, + timestamps: &TimestampFields, + hmac_key: Option<&[u8]>, +) -> Vec { + let has_hmac = hmac_key.is_some(); + let layout = irtt_proto::PacketLayout::echo(has_hmac, params); + let packet_len = echo_packet_len(has_hmac, params); + let mut packet = Vec::with_capacity(packet_len); + + packet.extend_from_slice(&MAGIC); + packet.push(flags::FLAG_REPLY | hmac_key.map_or(0, |_| flags::FLAG_HMAC)); + if has_hmac { + packet.extend_from_slice(&[0_u8; HMAC_SIZE]); + } + packet.extend_from_slice(&token.to_le_bytes()); + packet.extend_from_slice(&seq.to_le_bytes()); + + if layout.recv_count { + packet.extend_from_slice(&RECV_COUNT.to_le_bytes()); + } + if layout.recv_window { + packet.extend_from_slice(&RECV_WINDOW.to_le_bytes()); + } + if layout.recv_wall { + packet.extend_from_slice(×tamps.recv_wall.unwrap_or(0).to_le_bytes()); + } + if layout.recv_mono { + packet.extend_from_slice(×tamps.recv_mono.unwrap_or(0).to_le_bytes()); + } + if layout.midpoint_wall { + packet.extend_from_slice(×tamps.midpoint_wall.unwrap_or(0).to_le_bytes()); + } + if layout.midpoint_mono { + packet.extend_from_slice(×tamps.midpoint_mono.unwrap_or(0).to_le_bytes()); + } + if layout.send_wall { + packet.extend_from_slice(×tamps.send_wall.unwrap_or(0).to_le_bytes()); + } + if layout.send_mono { + packet.extend_from_slice(×tamps.send_mono.unwrap_or(0).to_le_bytes()); + } + + packet.resize(packet_len, 0); + if let Some(key) = hmac_key { + compute_hmac_in_place(key, &mut packet, HMAC_OFFSET).unwrap(); + } + packet +} + +fn materialize_wall_timestamps(mut timestamps: TimestampFields) -> TimestampFields { + let now_ns = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() as i64; + if timestamps.recv_wall.is_some() { + timestamps.recv_wall = Some(now_ns); + } + if timestamps.midpoint_wall.is_some() { + timestamps.midpoint_wall = Some(now_ns); + } + if timestamps.send_wall.is_some() { + timestamps.send_wall = Some(now_ns); + } + timestamps +}