From 5d791d1fcdc4b3f7d57298992a437103b33111ac Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Fri, 1 May 2026 10:05:29 +0700 Subject: [PATCH] feat(client): emit EchoSent event with scheduling metadata --- crates/irtt-cli/src/lib.rs | 25 ++++++++++ crates/irtt-client/src/client.rs | 78 ++++++++++++++++++++++++++++++-- crates/irtt-client/src/event.rs | 16 ++++++- 3 files changed, 115 insertions(+), 4 deletions(-) diff --git a/crates/irtt-cli/src/lib.rs b/crates/irtt-cli/src/lib.rs index 7864a08..d784875 100644 --- a/crates/irtt-cli/src/lib.rs +++ b/crates/irtt-cli/src/lib.rs @@ -152,6 +152,9 @@ pub fn parse_length(input: &str) -> Result { } pub fn format_event(event: &ClientEvent, mode: OutputMode) -> Option { + if matches!(event, ClientEvent::EchoSent { .. }) { + return None; + } match mode { OutputMode::RttUs => format_rtt_us(event), OutputMode::Machine => Some(format_machine(event)), @@ -197,6 +200,7 @@ fn format_machine(event: &ClientEvent) -> String { write_token(&mut out, *token); write_wall(&mut out, "at_ns", at.wall); } + ClientEvent::EchoSent { .. } => {} ClientEvent::EchoReply { seq, logical_seq, @@ -300,12 +304,14 @@ fn format_simple(event: &ClientEvent) -> String { ClientEvent::SessionClosed { remote, token, .. } => { format!("session closed remote={remote} token={token:#x}") } + ClientEvent::EchoSent { .. } => String::new(), ClientEvent::EchoReply { seq, logical_seq, remote, rtt, server_timing, + bytes: _, .. } => { let mut out = format!( @@ -667,6 +673,25 @@ mod tests { ); } + #[test] + fn echo_sent_is_not_formatted_for_stream_outputs() { + let ts = test_timestamp(Duration::from_secs(1)); + let event = ClientEvent::EchoSent { + seq: 1, + logical_seq: 2, + remote: test_remote(), + scheduled_at: ts.mono, + sent_at: ts, + bytes: 64, + send_call: Duration::from_micros(10), + timer_error: Duration::ZERO, + }; + + assert!(format_event(&event, OutputMode::RttUs).is_none()); + assert!(format_event(&event, OutputMode::Machine).is_none()); + assert!(format_event(&event, OutputMode::Simple).is_none()); + } + #[test] fn warning_and_loss_variants_format_without_panicking() { let ts = test_timestamp(Duration::from_secs(1)); diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index a2785d4..aa229e1 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -196,6 +196,7 @@ impl Client { let wire_seq = session.next_wire_seq; let logical_seq = session.next_logical_seq; + let scheduled_at = session.next_send_at; let request = EchoRequest { token, @@ -205,7 +206,10 @@ impl Client { }; let packet = encode_echo_request(&request, self.config.hmac_key.as_deref())?; let sent_at = override_ts.unwrap_or_else(ClientTimestamp::now); - self.socket.send(&packet)?; + let send_call_start = Instant::now(); + let bytes = self.socket.send(&packet)?; + let send_call = send_call_start.elapsed(); + let timer_error = instant_abs_diff(sent_at.mono, scheduled_at); let session = self .session @@ -242,7 +246,16 @@ impl Client { } } - Ok(vec![]) + Ok(vec![ClientEvent::EchoSent { + seq: wire_seq, + logical_seq, + remote: self.remote, + scheduled_at, + sent_at, + bytes, + send_call, + timer_error, + }]) } fn recv_once_inner( @@ -606,6 +619,12 @@ fn update_highest_received(session: &mut ActiveSession, wire_seq: u32) { ); } +fn instant_abs_diff(left: Instant, right: Instant) -> Duration { + left.checked_duration_since(right) + .or_else(|| right.checked_duration_since(left)) + .unwrap_or(Duration::ZERO) +} + fn compute_rtt( sent_at: &ClientTimestamp, received_at: &ClientTimestamp, @@ -1541,7 +1560,23 @@ mod tests { }; let mut client = Client::connect(config).unwrap(); assert_open_started(client.open(ClientTimestamp::now()).unwrap()); - client.send_probe().unwrap(); + let events = client.send_probe().unwrap(); + assert_eq!(events.len(), 1); + match &events[0] { + ClientEvent::EchoSent { + seq, + logical_seq, + remote, + bytes, + .. + } => { + assert_eq!(*seq, 0); + assert_eq!(*logical_seq, 0); + assert_eq!(*remote, server.addr); + assert_eq!(*bytes, echo_packet_len(false, ¶ms)); + } + other => panic!("expected EchoSent, got {other:?}"), + } thread::sleep(Duration::from_millis(30)); let packets: Vec<_> = server.rx.try_iter().collect(); let echo_reqs: Vec<_> = packets @@ -1559,6 +1594,43 @@ mod tests { server.join(); } + #[test] + fn echo_sent_reports_schedule_and_timer_error() { + let params = default_params(); + let server = silent_open_server(params); + let config = ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + let start = ClientTimestamp { + mono: Instant::now(), + wall: SystemTime::now(), + }; + assert_open_started(client.open(start).unwrap()); + + let events = client.send_probe_at(start).unwrap(); + match &events[0] { + ClientEvent::EchoSent { + scheduled_at, + sent_at, + timer_error, + .. + } => { + assert_eq!(*scheduled_at, start.mono); + assert_eq!(*sent_at, start); + assert_eq!(*timer_error, Duration::ZERO); + } + other => panic!("expected EchoSent, got {other:?}"), + } + + client.close(ClientTimestamp::now()).unwrap(); + server.join(); + } + #[test] fn send_probe_starts_seq_at_zero_and_increments() { let params = default_params(); diff --git a/crates/irtt-client/src/event.rs b/crates/irtt-client/src/event.rs index 9b20b12..8f43da3 100644 --- a/crates/irtt-client/src/event.rs +++ b/crates/irtt-client/src/event.rs @@ -1,4 +1,7 @@ -use std::{net::SocketAddr, time::Duration}; +use std::{ + net::SocketAddr, + time::{Duration, Instant}, +}; use crate::{session::NegotiatedParams, timing::ClientTimestamp}; @@ -36,6 +39,17 @@ pub enum ClientEvent { at: ClientTimestamp, }, + EchoSent { + seq: u32, + logical_seq: u64, + remote: SocketAddr, + scheduled_at: Instant, + sent_at: ClientTimestamp, + bytes: usize, + send_call: Duration, + timer_error: Duration, + }, + EchoReply { seq: u32, logical_seq: u64,