From b1e32d543cb59a1fc334eab3f034180c5ff41658 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 15:31:03 +0700 Subject: [PATCH] test(client): verify async lifecycle equivalence --- crates/irtt-client/src/async_client/tests.rs | 312 ++++++++++++++++++- crates/irtt-client/src/lib.rs | 2 + 2 files changed, 313 insertions(+), 1 deletion(-) diff --git a/crates/irtt-client/src/async_client/tests.rs b/crates/irtt-client/src/async_client/tests.rs index f2dee54..d4f8189 100644 --- a/crates/irtt-client/src/async_client/tests.rs +++ b/crates/irtt-client/src/async_client/tests.rs @@ -17,7 +17,7 @@ use tokio::runtime::{Builder, Runtime}; use super::*; use crate::{ - socket::resolution_call_counts, socket_options::tokio_socket_traffic_class, RunMode, + socket::resolution_call_counts, socket_options::tokio_socket_traffic_class, Client, RunMode, SocketConfig, WarningKind, }; @@ -192,6 +192,87 @@ fn start_peer_close_server() -> TestServer { }) } +fn start_echo_close_server(key: Option>) -> TestServer { + start_server(move |socket, tx| { + let (open_packet, peer) = recv_packet(&socket, &tx); + let request = decode_open_request(&open_packet, key.as_deref()).unwrap(); + send_open_reply( + &socket, + peer, + request.params.clone(), + key.as_deref(), + flags::FLAG_OPEN | flags::FLAG_REPLY, + TOKEN, + ); + + let (probe_packet, _) = recv_packet(&socket, &tx); + let probe = decode_echo_request(&probe_packet, &request.params, key.as_deref()).unwrap(); + socket + .send_to( + &echo_reply( + &request.params, + probe.sequence, + TOKEN, + flags::FLAG_REPLY, + key.as_deref(), + ), + peer, + ) + .unwrap(); + + let (close_packet, _) = recv_packet(&socket, &tx); + assert_eq!( + decode_close_request(&close_packet, key.as_deref()) + .unwrap() + .token, + TOKEN + ); + }) +} + +fn start_timeout_server() -> TestServer { + start_server(move |socket, tx| { + let (open_packet, peer) = recv_packet(&socket, &tx); + let request = decode_open_request(&open_packet, None).unwrap(); + send_open_reply( + &socket, + peer, + request.params, + None, + flags::FLAG_OPEN | flags::FLAG_REPLY, + TOKEN, + ); + let _ = recv_packet(&socket, &tx); + }) +} + +fn start_filtered_open_server(key: Vec) -> TestServer { + start_server(move |socket, tx| { + let (open_packet, peer) = recv_packet(&socket, &tx); + let request = decode_open_request(&open_packet, Some(&key)).unwrap(); + socket.send_to(&[0_u8], peer).unwrap(); + let mut bad_hmac = encode_open_reply( + &OpenReply { + flags: flags::FLAG_OPEN | flags::FLAG_REPLY, + token: TOKEN, + params: request.params.clone(), + }, + Some(&key), + ) + .unwrap(); + bad_hmac[4] ^= 0xff; + socket.send_to(&bad_hmac, peer).unwrap(); + send_open_reply( + &socket, + peer, + request.params, + Some(&key), + flags::FLAG_OPEN | flags::FLAG_REPLY, + TOKEN, + ); + }) +} + fn poll_once(future: Pin<&mut F>) -> Poll { let mut context = Context::from_waker(Waker::noop()); Future::poll(future, &mut context) @@ -1190,3 +1271,232 @@ fn close_timestamp_is_sampled_after_success_and_short_send_commits() { }); assert_eq!(server.finish().len(), 1); } + +#[test] +fn blocking_and_async_hmac_dscp_lifecycle_are_semantically_equivalent() { + let key = b"async-equivalence-key".to_vec(); + let blocking_server = start_echo_close_server(Some(key.clone())); + let mut blocking = + Client::connect(config(blocking_server.addr, Some(key.clone()), 46)).unwrap(); + let blocking_open = blocking.open().unwrap(); + let blocking_sent = blocking.send_probe().unwrap(); + let blocking_reply = blocking.recv_once().unwrap(); + let blocking_close = blocking.close().unwrap(); + assert_eq!(blocking_server.finish().len(), 3); + + let async_server = start_echo_close_server(Some(key.clone())); + let (async_open, async_sent, async_reply, async_close) = runtime().block_on(async { + let mut client = AsyncClient::connect(config(async_server.addr, Some(key), 46)) + .await + .unwrap(); + let opened = client.open().await.unwrap(); + assert_eq!( + tokio_socket_traffic_class(&client.socket, client.remote).unwrap() & 0xfc, + 46 << 2 + ); + let sent = client.send_probe().await.unwrap(); + let reply = client.recv().await.unwrap(); + let closed = client.close().await.unwrap(); + (opened, sent, reply, closed) + }); + assert_eq!(async_server.finish().len(), 3); + + assert_eq!( + open_negotiated(&blocking_open), + open_negotiated(&async_open) + ); + assert_eq!(open_negotiated(&async_open).params.dscp, 46); + assert_matching_event_shape(&blocking_sent[0], &async_sent[0]); + assert_matching_event_shape(&blocking_reply[0], &async_reply[0]); + assert_matching_event_shape(&blocking_close[0], &async_close[0]); +} + +#[test] +fn blocking_and_async_no_test_timeout_and_peer_close_are_equivalent() { + let blocking_no_test_server = start_no_test_server(); + let mut blocking_no_test_config = config(blocking_no_test_server.addr, None, 0); + blocking_no_test_config.run_mode = RunMode::NoTest; + let mut blocking_no_test = Client::connect(blocking_no_test_config).unwrap(); + let blocking_no_test_open = blocking_no_test.open().unwrap(); + assert!(blocking_no_test.is_run_complete()); + assert_eq!(blocking_no_test_server.finish().len(), 1); + + let async_no_test_server = start_no_test_server(); + let mut async_no_test_config = config(async_no_test_server.addr, None, 0); + async_no_test_config.run_mode = RunMode::NoTest; + let async_no_test_open = runtime().block_on(async { + let mut client = AsyncClient::connect(async_no_test_config).await.unwrap(); + let opened = client.open().await.unwrap(); + assert!(client.is_run_complete()); + opened + }); + assert_eq!(async_no_test_server.finish().len(), 1); + assert!(matches!( + blocking_no_test_open, + OpenOutcome::NoTestCompleted { .. } + )); + assert!(matches!( + async_no_test_open, + OpenOutcome::NoTestCompleted { .. } + )); + assert_eq!( + open_negotiated(&blocking_no_test_open), + open_negotiated(&async_no_test_open) + ); + + let blocking_timeout_server = start_timeout_server(); + let mut blocking_timeout_config = config(blocking_timeout_server.addr, None, 0); + blocking_timeout_config.probe_timeout = Duration::from_millis(5); + let mut blocking_timeout = Client::connect(blocking_timeout_config).unwrap(); + blocking_timeout.open().unwrap(); + blocking_timeout.send_probe().unwrap(); + thread::sleep(Duration::from_millis(10)); + let blocking_loss = blocking_timeout.poll_timeouts().unwrap(); + assert_eq!(blocking_timeout_server.finish().len(), 2); + + let async_timeout_server = start_timeout_server(); + let mut async_timeout_config = config(async_timeout_server.addr, None, 0); + async_timeout_config.probe_timeout = Duration::from_millis(5); + let async_loss = runtime().block_on(async { + let mut client = AsyncClient::connect(async_timeout_config).await.unwrap(); + client.open().await.unwrap(); + client.send_probe().await.unwrap(); + time::sleep(Duration::from_millis(10)).await; + client.poll_timeouts().unwrap() + }); + assert_eq!(async_timeout_server.finish().len(), 2); + assert_matching_event_shape(&blocking_loss[0], &async_loss[0]); + + let blocking_peer_close_server = start_peer_close_server(); + let mut blocking_peer_close = + Client::connect(config(blocking_peer_close_server.addr, None, 0)).unwrap(); + blocking_peer_close.open().unwrap(); + blocking_peer_close.send_probe().unwrap(); + let blocking_peer_close_events = blocking_peer_close.recv_once().unwrap(); + assert!(blocking_peer_close.is_peer_closed()); + assert_eq!(blocking_peer_close_server.finish().len(), 2); + + let async_peer_close_server = start_peer_close_server(); + let async_peer_close_events = runtime().block_on(async { + let mut client = AsyncClient::connect(config(async_peer_close_server.addr, None, 0)) + .await + .unwrap(); + client.open().await.unwrap(); + client.send_probe().await.unwrap(); + let events = client.recv().await.unwrap(); + assert!(client.is_peer_closed()); + events + }); + assert_eq!(async_peer_close_server.finish().len(), 2); + assert_eq!( + blocking_peer_close_events.len(), + async_peer_close_events.len() + ); + for (blocking, asynchronous) in blocking_peer_close_events + .iter() + .zip(&async_peer_close_events) + { + assert_matching_event_shape(blocking, asynchronous); + } +} + +#[test] +fn blocking_and_async_filtered_open_and_finite_completion_are_equivalent() { + let key = b"filtered-open-equivalence".to_vec(); + let blocking_server = start_filtered_open_server(key.clone()); + let mut blocking = Client::connect(config(blocking_server.addr, Some(key.clone()), 0)).unwrap(); + let blocking_open = blocking.open().unwrap(); + assert_eq!(blocking_server.finish().len(), 1); + + let async_server = start_filtered_open_server(key.clone()); + let async_open = runtime().block_on(async { + let mut client = AsyncClient::connect(config(async_server.addr, Some(key), 0)) + .await + .unwrap(); + client.open().await.unwrap() + }); + assert_eq!(async_server.finish().len(), 1); + assert_eq!( + open_negotiated(&blocking_open), + open_negotiated(&async_open) + ); + + let blocking_finite_server = start_open_server(None, 0); + let mut blocking_config = config(blocking_finite_server.addr, None, 0); + blocking_config.interval = Duration::from_millis(5); + blocking_config.duration = Some(Duration::from_millis(20)); + let mut blocking = Client::connect(blocking_config).unwrap(); + blocking.open().unwrap(); + thread::sleep(Duration::from_millis(25)); + assert!(blocking.send_probe().unwrap().is_empty()); + assert!(blocking.is_run_complete()); + assert_eq!(blocking_finite_server.finish().len(), 1); + + let async_finite_server = start_open_server(None, 0); + let mut async_config = config(async_finite_server.addr, None, 0); + async_config.interval = Duration::from_millis(5); + async_config.duration = Some(Duration::from_millis(20)); + runtime().block_on(async { + let mut client = AsyncClient::connect(async_config).await.unwrap(); + client.open().await.unwrap(); + time::sleep(Duration::from_millis(25)).await; + assert!(client.send_probe().await.unwrap().is_empty()); + assert!(client.is_run_complete()); + }); + assert_eq!(async_finite_server.finish().len(), 1); +} + +fn open_negotiated(outcome: &OpenOutcome) -> &crate::NegotiatedParams { + match outcome { + OpenOutcome::Started { negotiated, .. } + | OpenOutcome::NoTestCompleted { negotiated, .. } => negotiated, + } +} + +fn assert_matching_event_shape(blocking: &ClientEvent, asynchronous: &ClientEvent) { + match (blocking, asynchronous) { + ( + ClientEvent::EchoSent { + seq: left_seq, + bytes: left_bytes, + .. + }, + ClientEvent::EchoSent { + seq: right_seq, + bytes: right_bytes, + .. + }, + ) => { + assert_eq!(left_seq, right_seq); + assert_eq!(left_bytes, right_bytes); + } + ( + ClientEvent::EchoReply { + seq: left_seq, + bytes: left_bytes, + .. + }, + ClientEvent::EchoReply { + seq: right_seq, + bytes: right_bytes, + .. + }, + ) => { + assert_eq!(left_seq, right_seq); + assert_eq!(left_bytes, right_bytes); + } + ( + ClientEvent::SessionClosed { + token: left_token, .. + }, + ClientEvent::SessionClosed { + token: right_token, .. + }, + ) => assert_eq!(left_token, right_token), + ( + ClientEvent::EchoLoss { seq: left_seq, .. }, + ClientEvent::EchoLoss { seq: right_seq, .. }, + ) => assert_eq!(left_seq, right_seq), + pair => panic!("event shapes differ: {pair:?}"), + } +} diff --git a/crates/irtt-client/src/lib.rs b/crates/irtt-client/src/lib.rs index 4ecbc75..e0971c3 100644 --- a/crates/irtt-client/src/lib.rs +++ b/crates/irtt-client/src/lib.rs @@ -3,6 +3,8 @@ //! `irtt-client` opens IRTT-compatible sessions, sends echo probes, receives //! replies, classifies loss/late/duplicate packets, and emits [`ClientEvent`] //! values for callers to consume directly or aggregate with `irtt-stats`. +//! With the `tokio` feature enabled, `AsyncClient` provides the same low-level +//! lifecycle for callers that own a Tokio runtime and drive readiness directly. //! //! Timing values intentionally preserve signed measurement semantics. When //! server timing is available, [`RttSample::effective`] is adjusted for server