From dc5b19059ef23e10a90cce6aef27024080dc96c2 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Fri, 2 Oct 2026 08:33:49 +0700 Subject: [PATCH] tests(client): cleanup unnecessary tests and hooks --- Cargo.lock | 1 - crates/irtt-client/Cargo.toml | 1 - crates/irtt-client/src/async_client.rs | 14 +- crates/irtt-client/src/async_client/tests.rs | 139 +- crates/irtt-client/src/client/tests/close.rs | 14 - crates/irtt-client/src/client/tests/config.rs | 19 - crates/irtt-client/src/client/tests/dscp.rs | 21 +- crates/irtt-client/src/client/tests/hmac.rs | 18 - crates/irtt-client/src/client/tests/mod.rs | 2 +- .../src/client/tests/negotiation.rs | 125 -- .../irtt-client/src/client/tests/no_test.rs | 16 - crates/irtt-client/src/client/tests/open.rs | 14 - crates/irtt-client/src/client/tests/ttl.rs | 6 +- crates/irtt-client/src/managed/blocking.rs | 18 +- .../irtt-client/src/managed/blocking_tests.rs | 12 +- crates/irtt-client/src/managed/model.rs | 196 --- crates/irtt-client/src/managed/schedule.rs | 20 +- crates/irtt-client/src/managed/task.rs | 98 +- crates/irtt-client/src/managed/tests.rs | 158 +-- crates/irtt-client/src/metadata.rs | 44 - crates/irtt-client/src/probe.rs | 121 +- crates/irtt-client/src/receive.rs | 42 +- crates/irtt-client/src/receive/linux.rs | 18 +- .../src/receive/linux/error_queue.rs | 13 +- crates/irtt-client/src/session/machine.rs | 1209 +---------------- crates/irtt-client/src/socket.rs | 44 +- crates/irtt-client/src/socket_options.rs | 360 +---- 27 files changed, 118 insertions(+), 2625 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 08db319..c046d8e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -720,7 +720,6 @@ dependencies = [ "irtt-proto", "irtt-server", "nix 0.31.3", - "proptest", "socket2", "thiserror 2.0.21", "tokio", diff --git a/crates/irtt-client/Cargo.toml b/crates/irtt-client/Cargo.toml index a775f17..d171d8a 100644 --- a/crates/irtt-client/Cargo.toml +++ b/crates/irtt-client/Cargo.toml @@ -47,5 +47,4 @@ doc-scrape-examples = true # Keep the packaged client testable against the latest published server while # workspace tests continue to exercise the local source through this path. irtt-server = { version = "0.5.1", path = "../irtt-server" } -proptest = "1.11.0" tokio = { version = "1.53.1", default-features = false, features = ["rt", "sync"] } diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index f4999ca..e6b607b 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -85,7 +85,7 @@ impl AsyncOpenCleanup { /// Owned state for one asynchronous open transaction. /// /// This state deliberately contains no reference to [`AsyncClient`], allowing -/// a future managed target to store the client and opening state side by side. +/// a managed target to store the client and opening state side by side. #[derive(Debug)] pub(crate) struct AsyncOpenState { attempt: usize, @@ -567,7 +567,7 @@ impl AsyncClient { }; if self.machine.is_peer_closed() { self.prepared_probe = None; - if self.clear_peer_close_dscp().is_ok() { + if clear_dscp_on_tokio_socket(&self.socket, self.remote).is_ok() { self.applied_traffic_class = None; } } @@ -607,7 +607,7 @@ impl AsyncClient { return Poll::Ready(Err(ClientError::Socket(error))); } } - if let Err(error) = self.clear_close_dscp() { + if let Err(error) = clear_dscp_on_tokio_socket(&self.socket, self.remote) { return Poll::Ready(Err(error)); } let mut rollback = @@ -798,14 +798,6 @@ impl AsyncClient { } } - fn clear_peer_close_dscp(&self) -> Result<(), ClientError> { - clear_dscp_on_tokio_socket(&self.socket, self.remote) - } - - fn clear_close_dscp(&self) -> Result<(), ClientError> { - clear_dscp_on_tokio_socket(&self.socket, self.remote) - } - fn restore_dscp_best_effort(&self, previous_traffic_class: Option) { let _ = match previous_traffic_class { Some(traffic_class) => { diff --git a/crates/irtt-client/src/async_client/tests.rs b/crates/irtt-client/src/async_client/tests.rs index 761b358..4c90633 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}; mod in_tree_server; use super::*; -use crate::{socket_options::tokio_socket_traffic_class, Client, RunMode, SocketConfig}; +use crate::{Client, RunMode, SocketConfig}; use in_tree_server::InTreeServer; use irtt_server::ServerConfig; @@ -68,10 +68,6 @@ async fn opened_client(addr: SocketAddr, dscp: u8) -> AsyncClient { client } -fn socket_dscp(client: &AsyncClient) -> u32 { - tokio_socket_traffic_class(&client.socket, client.remote).unwrap() >> 2 -} - fn start_server(handler: F) -> TestServer where F: FnOnce(UdpSocket, mpsc::Sender>) + Send + 'static, @@ -382,85 +378,24 @@ fn recv_after_local_close_and_no_test_fails_on_first_poll() { } #[test] -fn blocking_and_async_hmac_dscp_lifecycle_are_semantically_equivalent() { - let key = b"async-equivalence-key".to_vec(); - let blocking_server = InTreeServer::start(ServerConfig::default().with_hmac_key(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(); - drop(blocking_server); - - let async_server = InTreeServer::start(ServerConfig::default().with_hmac_key(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)) +fn authenticated_async_lifecycle_applies_negotiated_dscp() { + let key = b"async-lifecycle-key".to_vec(); + let server = InTreeServer::start(ServerConfig::default().with_hmac_key(key.clone())); + runtime().block_on(async { + let mut client = AsyncClient::connect(config(server.addr, Some(key), 46)) .await .unwrap(); let opened = client.open().await.unwrap(); - assert_eq!(socket_dscp(&client), 46); - 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!(open_negotiated(&opened).params.dscp, 184); + #[cfg(not(any( + target_os = "fuchsia", target_os = "redox", target_os = "solaris", + target_os = "illumos", target_os = "haiku", + )))] + assert_eq!(socket2::SockRef::from(&client.socket).tos_v4().unwrap() >> 2, 46); + client.send_probe().await.unwrap(); + assert!(matches!(client.recv().await.unwrap().as_slice(), [ClientEvent::EchoReply { seq: 0, .. }])); + assert!(matches!(client.close().await.unwrap().as_slice(), [ClientEvent::SessionClosed { token, .. }] if *token == open_token(&opened))); }); - drop(async_server); - - assert_eq!( - open_negotiated(&blocking_open), - open_negotiated(&async_open) - ); - assert_eq!( - open_negotiated(&async_open).params.dscp, - 184, - "negotiated Params::dscp is the raw wire byte for codepoint 46" - ); - assert_matching_event_shape(&blocking_sent[0], &async_sent[0]); - assert_matching_event_shape(&blocking_reply[0], &async_reply[0]); - assert!(matches!( - blocking_close.as_slice(), - [ClientEvent::SessionClosed { token, .. }] if *token == open_token(&blocking_open) - )); - assert!(matches!( - async_close.as_slice(), - [ClientEvent::SessionClosed { token, .. }] if *token == open_token(&async_open) - )); - assert_matching_event_shape(&blocking_close[0], &async_close[0]); -} - -#[test] -fn blocking_and_async_no_test_are_semantically_equivalent() { - let blocking_no_test_server = InTreeServer::start(ServerConfig::default()); - 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.negotiated_params().is_none()); - drop(blocking_no_test_server); - - let async_no_test_server = InTreeServer::start(ServerConfig::default()); - 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.negotiated_params().is_none()); - opened - }); - drop(async_no_test_server); - 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) - ); } #[test] @@ -724,50 +659,6 @@ fn blocking_and_async_complete_every_caller_paced_probe() { } } -#[test] -fn blocking_and_async_reject_probes_before_open_and_after_close() { - let blocking_server = InTreeServer::start(ServerConfig::default()); - let mut blocking = Client::connect(config(blocking_server.addr, None, 0)).unwrap(); - let blocking_before = blocking.send_probe().unwrap_err(); - blocking.open().unwrap(); - let blocking_reopen = blocking.open().unwrap_err(); - blocking.close().unwrap(); - let blocking_after_send = blocking.send_probe().unwrap_err(); - let blocking_after_close = blocking.close().unwrap_err(); - drop(blocking_server); - - let async_server = InTreeServer::start(ServerConfig::default()); - let (async_before, async_reopen, async_after_send, async_after_close) = - runtime().block_on(async { - let mut client = AsyncClient::connect(config(async_server.addr, None, 0)) - .await - .unwrap(); - let before = client.send_probe().await.unwrap_err(); - client.open().await.unwrap(); - let reopen = client.open().await.unwrap_err(); - client.close().await.unwrap(); - let after_send = client.send_probe().await.unwrap_err(); - let after_close = client.close().await.unwrap_err(); - (before, reopen, after_send, after_close) - }); - drop(async_server); - - assert_eq!(error_name(&blocking_before), "not open"); - assert_eq!(error_name(&blocking_reopen), "already open"); - assert_eq!(error_name(&blocking_after_send), "already closed"); - assert_eq!(error_name(&blocking_after_close), "already closed"); - assert_eq!(error_name(&blocking_before), error_name(&async_before)); - assert_eq!(error_name(&blocking_reopen), error_name(&async_reopen)); - assert_eq!( - error_name(&blocking_after_send), - error_name(&async_after_send) - ); - assert_eq!( - error_name(&blocking_after_close), - error_name(&async_after_close) - ); -} - #[test] fn ignored_open_traffic_uses_one_request_and_deadline_per_attempt() { let server = start_server(move |socket, tx| { diff --git a/crates/irtt-client/src/client/tests/close.rs b/crates/irtt-client/src/client/tests/close.rs index e3c0c35..21ae27b 100644 --- a/crates/irtt-client/src/client/tests/close.rs +++ b/crates/irtt-client/src/client/tests/close.rs @@ -93,20 +93,6 @@ fn recv_available_stops_after_peer_close() { server.join(); } -#[test] -fn normal_echo_reply_does_not_close_session() { - let params = default_params(); - let (mut client, server) = open_client_with_echo_server(¶ms); - client.send_probe().unwrap(); - - let events = client.recv_once().unwrap(); - assert!(matches!(events.as_slice(), [ClientEvent::EchoReply { .. }])); - assert!(client.send_probe().is_ok()); - - client.close().unwrap(); - server.join(); -} - #[test] fn close_flagged_duplicate_emits_duplicate_then_closes() { let params = default_params(); diff --git a/crates/irtt-client/src/client/tests/config.rs b/crates/irtt-client/src/client/tests/config.rs index 07fe6d4..f706c29 100644 --- a/crates/irtt-client/src/client/tests/config.rs +++ b/crates/irtt-client/src/client/tests/config.rs @@ -1,24 +1,5 @@ use super::*; -#[test] -fn client_config_default() { - let config = ClientConfig::default(); - assert_eq!(config.duration, Some(Duration::from_secs(3))); - assert_eq!(config.interval, Duration::from_secs(1)); - assert_eq!(config.length, 0); - assert_eq!(config.received_stats, ReceivedStats::Both); - assert_eq!(config.stamp_at, StampAt::Both); - assert_eq!(config.clock, Clock::Both); - assert_eq!(config.dscp, 0); - assert_eq!(config.hmac_key, None); - assert_eq!(config.server_fill, None); - assert_eq!(config.open_timeouts, DEFAULT_OPEN_TIMEOUTS); - assert_eq!(config.run_mode, RunMode::Normal); - assert_eq!(config.negotiation_policy, NegotiationPolicy::Strict); - assert_eq!(config.probe_timeout, Duration::from_secs(4)); - assert_eq!(config.max_pending_probes, 4096); -} - #[test] fn params_from_config_maps_compatibility_fields() { let config = ClientConfig { diff --git a/crates/irtt-client/src/client/tests/dscp.rs b/crates/irtt-client/src/client/tests/dscp.rs index cbc43d0..d4e0891 100644 --- a/crates/irtt-client/src/client/tests/dscp.rs +++ b/crates/irtt-client/src/client/tests/dscp.rs @@ -1,5 +1,4 @@ use super::*; -use crate::socket_options::socket_traffic_class; #[test] #[cfg(not(any( @@ -25,20 +24,17 @@ fn normal_open_applies_negotiated_traffic_class_after_open_and_close_clears_it() let mut client = Client::connect(config).unwrap(); assert_eq!( - socket_traffic_class(&client.socket, client.remote).unwrap() & 0xfc, + socket2::SockRef::from(&client.socket).tos_v4().unwrap() & 0xfc, 0 ); assert_open_started(client.open().unwrap()); assert_eq!( - socket_traffic_class(&client.socket, client.remote).unwrap() & 0xfc, + socket2::SockRef::from(&client.socket).tos_v4().unwrap() & 0xfc, 184 ); client.close().unwrap(); - assert_eq!( - socket_traffic_class(&client.socket, client.remote).unwrap(), - 0 - ); + assert_eq!(socket2::SockRef::from(&client.socket).tos_v4().unwrap(), 0); server.join(); } @@ -69,10 +65,7 @@ fn normal_open_uses_negotiated_traffic_class_not_requested_dscp() { negotiated: 0, }] ); - assert_eq!( - socket_traffic_class(&client.socket, client.remote).unwrap(), - 0 - ); + assert_eq!(socket2::SockRef::from(&client.socket).tos_v4().unwrap(), 0); server.join(); } @@ -128,10 +121,6 @@ fn authenticated_peer_close_clears_negotiated_traffic_class() { ] )); - assert_eq!(client.applied_traffic_class, None); - assert_eq!( - socket_traffic_class(&client.socket, client.remote).unwrap(), - 0 - ); + assert_eq!(socket2::SockRef::from(&client.socket).tos_v4().unwrap(), 0); server.join(); } diff --git a/crates/irtt-client/src/client/tests/hmac.rs b/crates/irtt-client/src/client/tests/hmac.rs index 95cfd81..0a4c6dd 100644 --- a/crates/irtt-client/src/client/tests/hmac.rs +++ b/crates/irtt-client/src/client/tests/hmac.rs @@ -1,23 +1,5 @@ use super::*; -#[test] -fn hmac_open_success() { - let key = b"secret".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key.clone()); - let params = params_from_config(&config).unwrap(); - let server = start_fake_server(move |socket, tx| { - let (request, peer) = recv_request(&socket, &tx); - verify_packet_hmac(&key, &request).unwrap(); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&key)); - socket.send_to(&reply, peer).unwrap(); - }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - assert_open_started(client.open().unwrap()); - server.join(); -} - #[test] fn hmac_open_ignores_missing_hmac_before_valid_reply() { let key = b"secret".to_vec(); diff --git a/crates/irtt-client/src/client/tests/mod.rs b/crates/irtt-client/src/client/tests/mod.rs index bb13b07..3a1f0d8 100644 --- a/crates/irtt-client/src/client/tests/mod.rs +++ b/crates/irtt-client/src/client/tests/mod.rs @@ -1,5 +1,5 @@ use super::*; -use crate::config::{NegotiationPolicy, DEFAULT_OPEN_TIMEOUTS}; +use crate::config::NegotiationPolicy; use crate::{ session::machine::{ compute_one_way, compute_rtt, params_from_config, sequence_is_after, sequence_is_before, diff --git a/crates/irtt-client/src/client/tests/negotiation.rs b/crates/irtt-client/src/client/tests/negotiation.rs index 49bd347..fa48adb 100644 --- a/crates/irtt-client/src/client/tests/negotiation.rs +++ b/crates/irtt-client/src/client/tests/negotiation.rs @@ -17,38 +17,6 @@ fn rejection_reason(requested: &Params, returned: &Params, policy: NegotiationPo } } -#[test] -fn strict_negotiation_accepts_identical_params() { - let config = ClientConfig::default(); - let params = params_from_config(&config).unwrap(); - let negotiated = assert_negotiates(¶ms, ¶ms, NegotiationPolicy::Strict); - assert_eq!(negotiated.params, params); - assert!(negotiated.restrictions.is_empty()); -} - -#[test] -fn strict_negotiation_rejects_changed_params() { - let config = ClientConfig { - dscp: 46, - ..ClientConfig::default() - }; - let requested = params_from_config(&config).unwrap(); - assert_eq!( - requested.dscp, 184, - "codepoint 46 must become raw wire byte 184" - ); - let mut returned = requested.clone(); - returned.dscp = 0; - assert_eq!( - rejection_reason(&requested, &returned, NegotiationPolicy::Strict), - NegotiationRestriction::DscpChanged { - requested: 46, - negotiated: 0, - } - .message() - ); -} - #[test] fn loose_negotiation_accepts_server_restricted_params() { let config = ClientConfig::default(); @@ -88,79 +56,6 @@ fn loose_negotiation_rejects_non_positive_returned_interval() { } } -#[test] -fn loose_negotiation_accepts_documented_duration_combinations() { - let finite_config = ClientConfig::default(); - let requested_finite = params_from_config(&finite_config).unwrap(); - - let returned_same_finite = requested_finite.clone(); - assert!(assert_negotiates( - &requested_finite, - &returned_same_finite, - NegotiationPolicy::Loose - ) - .restrictions - .is_empty()); - - let mut returned_shorter_finite = requested_finite.clone(); - returned_shorter_finite.duration_ns /= 2; - assert_eq!( - assert_negotiates( - &requested_finite, - &returned_shorter_finite, - NegotiationPolicy::Loose - ) - .restrictions, - vec![NegotiationRestriction::DurationReduced { - requested_ns: requested_finite.duration_ns, - negotiated_ns: returned_shorter_finite.duration_ns, - }] - ); - - let continuous_config = ClientConfig { - duration: None, - ..ClientConfig::default() - }; - let requested_continuous = params_from_config(&continuous_config).unwrap(); - - let returned_continuous = requested_continuous.clone(); - assert!(assert_negotiates( - &requested_continuous, - &returned_continuous, - NegotiationPolicy::Loose - ) - .restrictions - .is_empty()); - - let mut returned_finite = requested_continuous.clone(); - returned_finite.duration_ns = 1_000_000_000; - assert_eq!( - assert_negotiates( - &requested_continuous, - &returned_finite, - NegotiationPolicy::Loose - ) - .restrictions, - vec![NegotiationRestriction::DurationReduced { - requested_ns: requested_continuous.duration_ns, - negotiated_ns: returned_finite.duration_ns, - }] - ); -} - -#[test] -fn loose_negotiation_rejects_finite_request_returned_continuous() { - let config = ClientConfig::default(); - let requested = params_from_config(&config).unwrap(); - let mut returned = requested.clone(); - returned.duration_ns = 0; - - assert_eq!( - rejection_reason(&requested, &returned, NegotiationPolicy::Loose), - "server returned continuous duration for finite request" - ); -} - #[test] fn loose_negotiation_rejects_negative_returned_length() { let config = ClientConfig::default(); @@ -238,26 +133,6 @@ fn loose_negotiation_records_dscp_disabled_by_server() { ); } -#[test] -fn strict_negotiation_rejects_dscp_disabled_by_server_as_specific_restriction() { - let config = ClientConfig { - dscp: 46, - ..ClientConfig::default() - }; - let requested = params_from_config(&config).unwrap(); - let mut returned = requested.clone(); - returned.dscp = 0; - - assert_eq!( - rejection_reason(&requested, &returned, NegotiationPolicy::Strict), - NegotiationRestriction::DscpChanged { - requested: 46, - negotiated: 0, - } - .message() - ); -} - #[test] fn negotiation_rejects_unsupported_dscp_changes() { let config = ClientConfig { diff --git a/crates/irtt-client/src/client/tests/no_test.rs b/crates/irtt-client/src/client/tests/no_test.rs index 313e897..00bc9c8 100644 --- a/crates/irtt-client/src/client/tests/no_test.rs +++ b/crates/irtt-client/src/client/tests/no_test.rs @@ -113,19 +113,3 @@ fn open_fails_after_no_test_completed() { assert!(matches!(client.open(), Err(ClientError::AlreadyCompleted))); server.join(); } - -#[test] -fn no_test_success_validates_params() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - let params = params_from_config(&config).unwrap(); - let server = no_test_server(params.clone(), 0); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); - - let negotiated = assert_no_test_completed(client.open().unwrap()); - assert_eq!(negotiated.params, params); - - assert_eq!(client.applied_traffic_class, None); - server.join(); -} diff --git a/crates/irtt-client/src/client/tests/open.rs b/crates/irtt-client/src/client/tests/open.rs index 36e2002..9e99995 100644 --- a/crates/irtt-client/src/client/tests/open.rs +++ b/crates/irtt-client/src/client/tests/open.rs @@ -136,20 +136,6 @@ fn post_token_negotiation_failure_sends_cleanup_close() { server.join(); } -#[test] -fn successful_open_handshake() { - let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - let params = params_from_config(&config).unwrap(); - let server = open_success_server(params.clone()); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - - let negotiated = assert_open_started(client.open().unwrap()); - assert_eq!(negotiated.params, params); - - assert_eq!(client.applied_traffic_class, Some(0)); - server.join(); -} - #[test] fn successful_open_restores_configured_receive_timeout() { let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); diff --git a/crates/irtt-client/src/client/tests/ttl.rs b/crates/irtt-client/src/client/tests/ttl.rs index 19a806b..16d89ca 100644 --- a/crates/irtt-client/src/client/tests/ttl.rs +++ b/crates/irtt-client/src/client/tests/ttl.rs @@ -1,5 +1,3 @@ -use crate::socket_options::socket_ttl; - use super::*; #[test] @@ -27,13 +25,13 @@ fn connect_applies_configured_ipv4_ttl_before_open_and_close_preserves_it() { config.socket_config.ttl = Some(64); let mut client = Client::connect(config).unwrap(); - assert_eq!(socket_ttl(&client.socket, client.remote).unwrap(), 64); + assert_eq!(socket2::SockRef::from(&client.socket).ttl_v4().unwrap(), 64); client.open().unwrap(); server.join(); client.close().unwrap(); - assert_eq!(socket_ttl(&client.socket, client.remote).unwrap(), 64); + assert_eq!(socket2::SockRef::from(&client.socket).ttl_v4().unwrap(), 64); } #[test] diff --git a/crates/irtt-client/src/managed/blocking.rs b/crates/irtt-client/src/managed/blocking.rs index 4febf65..192cfe8 100644 --- a/crates/irtt-client/src/managed/blocking.rs +++ b/crates/irtt-client/src/managed/blocking.rs @@ -157,22 +157,14 @@ impl BlockingManagedClient { } fn join_worker(&mut self) -> Result { - join_worker( - self.worker - .take() - .expect("blocking managed worker is joined at most once"), - ) + self.worker + .take() + .expect("blocking managed worker is joined at most once") + .join() + .map_err(|_| BlockingManagedJoinError::WorkerPanicked) } } -pub(super) fn join_worker( - worker: JoinHandle, -) -> Result { - worker - .join() - .map_err(|_| BlockingManagedJoinError::WorkerPanicked) -} - impl Drop for BlockingManagedClient { fn drop(&mut self) { if self.worker.is_some() { diff --git a/crates/irtt-client/src/managed/blocking_tests.rs b/crates/irtt-client/src/managed/blocking_tests.rs index 743eae8..a5ea559 100644 --- a/crates/irtt-client/src/managed/blocking_tests.rs +++ b/crates/irtt-client/src/managed/blocking_tests.rs @@ -12,8 +12,7 @@ use irtt_proto::{ use tokio::{runtime::Builder, time::timeout}; use super::{ - blocking::{join_worker, WorkerRuntime}, - BlockingManagedClient, BlockingManagedJoinError, ManagedClientConfig, ManagedCompletionPolicy, + blocking::WorkerRuntime, BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedPacing, ManagedTargetConfig, ManagedTargetEndReason, }; @@ -309,15 +308,6 @@ fn dropping_owner_stops_and_joins_worker() { server.finish(); } -#[test] -fn worker_panic_is_a_join_error() { - let worker = thread::spawn(|| -> super::ManagedOutcome { panic!("test worker panic") }); - assert_eq!( - join_worker(worker), - Err(BlockingManagedJoinError::WorkerPanicked) - ); -} - #[test] fn dropping_worker_runtime_inside_async_context_does_not_panic() { test_runtime().block_on(async { diff --git a/crates/irtt-client/src/managed/model.rs b/crates/irtt-client/src/managed/model.rs index 56e259e..1fa046e 100644 --- a/crates/irtt-client/src/managed/model.rs +++ b/crates/irtt-client/src/managed/model.rs @@ -449,199 +449,3 @@ pub(crate) fn classify_client_error( message: Arc::from(error.to_string()), } } - -#[cfg(test)] -mod tests { - use std::{collections::TryReserveError, io}; - - use super::*; - - fn io_error() -> io::Error { - io::Error::other("boom") - } - - fn try_reserve_error() -> TryReserveError { - Vec::::new().try_reserve(usize::MAX).unwrap_err() - } - - fn classification_cases() -> Vec<(ClientError, ManagedTargetFailureKind)> { - vec![ - ( - ClientError::Resolve { - addr: "example.invalid:2112".to_owned(), - }, - ManagedTargetFailureKind::Resolve, - ), - ( - ClientError::Socket(io_error()), - ManagedTargetFailureKind::Socket, - ), - ( - ClientError::DatagramLengthMismatch { - expected: 64, - actual: 32, - }, - ManagedTargetFailureKind::Socket, - ), - ( - ClientError::SocketOption { - operation: "set read timeout", - remote: "127.0.0.1:2112".parse().unwrap(), - source: io_error(), - }, - ManagedTargetFailureKind::SocketOption, - ), - ( - ClientError::ReadTimeoutRestore { source: io_error() }, - ManagedTargetFailureKind::SocketOption, - ), - ( - ClientError::Protocol(irtt_proto::ProtoError::BadMagic), - ManagedTargetFailureKind::Protocol, - ), - ( - ClientError::ProtocolVersionMismatch { - requested: 1, - received: 2, - }, - ManagedTargetFailureKind::Protocol, - ), - (ClientError::ZeroToken, ManagedTargetFailureKind::Protocol), - ( - ClientError::UnexpectedNoTestReply, - ManagedTargetFailureKind::Protocol, - ), - ( - ClientError::NonZeroNoTestToken { token: 7 }, - ManagedTargetFailureKind::Protocol, - ), - ( - ClientError::ServerRejected, - ManagedTargetFailureKind::Protocol, - ), - ( - ClientError::NegotiationRejected { - reason: "duration".to_owned(), - }, - ManagedTargetFailureKind::Protocol, - ), - (ClientError::OpenTimeout, ManagedTargetFailureKind::Timeout), - ( - ClientError::InvalidConfig { - reason: "interval".to_owned(), - }, - ManagedTargetFailureKind::Configuration, - ), - ( - ClientError::OpenTimeoutTooSmall { - timeout: Duration::from_millis(1), - minimum: Duration::from_millis(10), - }, - ManagedTargetFailureKind::Configuration, - ), - ( - ClientError::NoOpenTimeouts, - ManagedTargetFailureKind::Configuration, - ), - ( - ClientError::AllocationFailed { - operation: "pending probes", - source: try_reserve_error(), - }, - ManagedTargetFailureKind::ResourceExhausted, - ), - ( - ClientError::CounterOverflow { - counter: "packets_sent", - }, - ManagedTargetFailureKind::ResourceExhausted, - ), - ( - ClientError::DurationOverflow, - ManagedTargetFailureKind::ResourceExhausted, - ), - ( - ClientError::PendingLimitExceeded { limit: 8 }, - ManagedTargetFailureKind::ResourceExhausted, - ), - ( - ClientError::NoTokioRuntime, - ManagedTargetFailureKind::InvalidState, - ), - (ClientError::NotOpen, ManagedTargetFailureKind::InvalidState), - ( - ClientError::AlreadyOpen, - ManagedTargetFailureKind::InvalidState, - ), - ( - ClientError::AlreadyCompleted, - ManagedTargetFailureKind::InvalidState, - ), - ( - ClientError::AlreadyClosed, - ManagedTargetFailureKind::InvalidState, - ), - ( - ClientError::StalePreparedProbe { - prepared_seq: 3, - next_wire_seq: 4, - }, - ManagedTargetFailureKind::InvalidState, - ), - ( - ClientError::PendingSequenceCollision { seq: 9 }, - ManagedTargetFailureKind::InvalidState, - ), - ] - } - - #[test] - fn every_client_error_classifies_to_its_documented_kind() { - for (error, expected) in classification_cases() { - let failure = classify_client_error(ManagedTargetFailurePhase::Opening, &error); - assert_eq!( - failure.kind, expected, - "unexpected classification for {error:?}" - ); - assert_eq!(failure.phase, ManagedTargetFailurePhase::Opening); - assert_eq!(&*failure.message, error.to_string()); - } - } - - #[test] - fn classification_preserves_the_reported_phase() { - for phase in [ - ManagedTargetFailurePhase::Connecting, - ManagedTargetFailurePhase::Opening, - ManagedTargetFailurePhase::Sending, - ManagedTargetFailurePhase::Receiving, - ManagedTargetFailurePhase::Timing, - ManagedTargetFailurePhase::Closing, - ] { - let failure = classify_client_error(phase, &ClientError::OpenTimeout); - assert_eq!(failure.phase, phase); - } - } - - #[test] - fn phase_and_kind_display_without_debug_spelling() { - assert_eq!( - ManagedTargetFailurePhase::Connecting.to_string(), - "connecting" - ); - assert_eq!(ManagedTargetFailurePhase::Closing.to_string(), "closing"); - assert_eq!( - ManagedTargetFailureKind::SocketOption.to_string(), - "socket option" - ); - assert_eq!( - ManagedTargetFailureKind::ResourceExhausted.to_string(), - "resource exhausted" - ); - assert_eq!( - ManagedTargetFailureKind::InvalidState.to_string(), - "invalid state" - ); - assert_eq!(ManagedTargetFailureKind::Other.to_string(), "other"); - } -} diff --git a/crates/irtt-client/src/managed/schedule.rs b/crates/irtt-client/src/managed/schedule.rs index b51e8ae..854314d 100644 --- a/crates/irtt-client/src/managed/schedule.rs +++ b/crates/irtt-client/src/managed/schedule.rs @@ -70,7 +70,14 @@ impl ProbeSchedule { ) -> Result { let (scheduled_at, next_send_at) = advance_cadence(scheduled_at, self.interval, permission_at)?; - Ok(self.finish_commit(scheduled_at, next_send_at)) + Ok(ScheduleCommit { + scheduled_at, + next_send_at: if self.end_at.is_some_and(|end| next_send_at >= end) { + None + } else { + Some(next_send_at) + }, + }) } pub(crate) fn commit(&mut self, commit: ScheduleCommit) { @@ -90,17 +97,6 @@ impl ProbeSchedule { pub(crate) fn is_finished(&self) -> bool { self.next_send_at.is_none() } - - fn finish_commit(&self, scheduled_at: Instant, next_send_at: Instant) -> ScheduleCommit { - ScheduleCommit { - scheduled_at, - next_send_at: if self.end_at.is_some_and(|end| next_send_at >= end) { - None - } else { - Some(next_send_at) - }, - } - } } pub(crate) fn advance_cadence( diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index 2eb4d1e..1a09358 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -181,6 +181,7 @@ pub struct ManagedClientHandle { update_submission_hook: Option>, } +// Pauses between admission and enqueue to exercise stop/seal linearization. #[cfg(test)] #[derive(Default)] struct UpdateSubmissionTestHook { @@ -402,10 +403,9 @@ impl Future for ManagedStopReceipt { type ConnectFuture = Pin> + Send + 'static>>; type WakeFuture = Pin>> + Send + 'static>>; +// Snapshots at publication verify status is visible before the corresponding event. #[cfg(test)] type EventObservations = Arc)>>>; -#[cfg(test)] -type StaggerObservations = Arc>>; fn arm_wake(mut receiver: watch::Receiver<()>) -> WakeFuture { Box::pin(async move { @@ -525,17 +525,7 @@ impl TargetRuntime { /// The client owns `packets_sent`; this runtime only mirrors it so a /// terminal outcome can report it after the client is gone. fn sync_packets_sent(&mut self, client: &AsyncClient) { - self.record_packets_sent(client.packets_sent()); - } - - /// Adopt an already-read authoritative sent count. - fn record_packets_sent(&mut self, packets_sent: u64) { - self.counters.packets_sent = packets_sent; - } - - /// Account for one client event against this target's durable counters. - fn observe(&mut self, event: &ClientEvent) { - self.counters.observe(event); + self.counters.packets_sent = client.packets_sent(); } /// Build this target's durable outcome. @@ -723,21 +713,17 @@ pub struct ManagedClientTask { #[cfg(test)] event_observations: Option, #[cfg(test)] - stagger_observations: Option, - #[cfg(test)] drain_test_hook: DrainTestHook, + // Counts work that cannot be observed through lossy presentation events. #[cfg(test)] timeout_inspections: usize, - #[cfg(test)] - deadline_inspections: usize, } +// Holds queued replies past the drain deadline to exercise the final receive budget. #[cfg(test)] #[derive(Default)] struct DrainTestHook { defer_work_until_deadline: bool, - fail_receive: bool, - fail_close: bool, } impl ManagedClientTask { @@ -837,7 +823,7 @@ impl ManagedClientTask { fn publish_client_events(&mut self, index: usize, events: Vec) { for event in events { - self.targets[index].observe(&event); + self.targets[index].counters.observe(&event); self.publish_event(ManagedEvent::Client { target: self.targets[index].instance.clone(), event, @@ -1421,10 +1407,6 @@ impl ManagedClientTask { return true; } - #[cfg(test)] - let injected_receive_failure = self.take_drain_receive_failure(); - #[cfg(not(test))] - let injected_receive_failure = None; let mut retained_state_changed = false; let received = match client.poll_recv(cx) { Poll::Pending => false, @@ -1441,13 +1423,7 @@ impl ManagedClientTask { return self.begin_close(index, client, primary_end, cleanup_failure, now); } }; - if let Some(error) = injected_receive_failure { - self.targets[index].sync_packets_sent(&client); - cleanup_failure.get_or_insert_with(|| { - classify_client_error(ManagedTargetFailurePhase::Receiving, &error) - }); - return self.begin_close(index, client, primary_end, cleanup_failure, now); - } + if client.is_peer_closed() { self.targets[index].sync_packets_sent(&client); self.finish_target(index, ManagedTargetEndReason::PeerClosed, cleanup_failure); @@ -1525,17 +1501,7 @@ impl ManagedClientTask { Poll::Ready(Ok(events)) => { self.targets[index].sync_packets_sent(&client); self.publish_client_events(index, events); - #[cfg(test)] - let injected_close_failure = self.take_drain_close_failure().map(|error| { - classify_client_error(ManagedTargetFailurePhase::Closing, &error) - }); - #[cfg(not(test))] - let injected_close_failure = None; - self.finish_target( - index, - primary_end, - cleanup_failure.or(injected_close_failure), - ); + self.finish_target(index, primary_end, cleanup_failure); false } Poll::Ready(Err(error)) => { @@ -1616,26 +1582,6 @@ impl ManagedClientTask { true } - #[cfg(test)] - fn take_drain_receive_failure(&mut self) -> Option { - if !mem::take(&mut self.drain_test_hook.fail_receive) { - return None; - } - Some(ClientError::Socket(std::io::Error::other( - "injected drain receive failure", - ))) - } - - #[cfg(test)] - fn take_drain_close_failure(&mut self) -> Option { - if !mem::take(&mut self.drain_test_hook.fail_close) { - return None; - } - Some(ClientError::Socket(std::io::Error::other( - "injected drain close failure", - ))) - } - fn active_count(&self) -> usize { self.targets .iter() @@ -1643,7 +1589,7 @@ impl ManagedClientTask { .count() } - fn active_stagger_spacing(&self) -> Option<(usize, Duration)> { + fn active_stagger_spacing(&self) -> Option { let mut active = 0; let mut minimum: Option = None; for target in &self.targets { @@ -1661,7 +1607,7 @@ impl ManagedClientTask { .interval(); minimum = minimum.into_iter().chain(Some(interval)).min(); } - minimum.map(|interval| (active, stagger_spacing(interval, active))) + minimum.map(|interval| stagger_spacing(interval, active)) } fn stagger_target_added(&mut self, now: Instant) { @@ -1672,7 +1618,7 @@ impl ManagedClientTask { let candidate = self .last_stagger_send .zip(self.active_stagger_spacing()) - .and_then(|(last, (_, spacing))| last.checked_add(spacing)) + .and_then(|(last, spacing)| last.checked_add(spacing)) .filter(|gate| *gate > now); self.send_gate = candidate.map(|candidate| existing.min(candidate)); } @@ -1708,21 +1654,17 @@ impl ManagedClientTask { fn record_stagger_acceptance( &mut self, result: SendResult, - stagger_spacing: Option<(usize, Duration)>, + stagger_spacing: Option, accepted_at: Instant, ) { if !result.accepted() { return; } - let Some((_active, spacing)) = stagger_spacing else { + let Some(spacing) = stagger_spacing else { return; }; self.last_stagger_send = Some(accepted_at); self.send_gate = self.next_stagger_gate(spacing, accepted_at); - #[cfg(test)] - if let Some(observations) = &self.stagger_observations { - observations.lock().unwrap().push((_active, spacing)); - } } fn poll_one_send( @@ -1730,7 +1672,7 @@ impl ManagedClientTask { index: usize, cx: &mut Context<'_>, now: Instant, - stagger_spacing: Option<(usize, Duration)>, + stagger_spacing: Option, ) -> SendResult { if !self.targets[index].desired || self.targets[index].retirement.is_some() { self.targets[index].send_waiting = false; @@ -1832,7 +1774,7 @@ impl ManagedClientTask { } else { None }; - self.targets[index].record_packets_sent(after); + self.targets[index].counters.packets_sent = after; match result { Poll::Pending => { self.targets[index].send_waiting = true; @@ -2109,11 +2051,7 @@ impl ManagedClientTask { .all(|target| matches!(target.state, TargetState::Terminal)) } - fn next_deadline(&mut self) -> Option { - #[cfg(test)] - { - self.deadline_inspections += self.targets.len(); - } + fn next_deadline(&self) -> Option { let mut non_send_deadline = None; let mut send_deadline = None; for target in &self.targets { @@ -2541,13 +2479,9 @@ fn build_task( #[cfg(test)] event_observations: None, #[cfg(test)] - stagger_observations: None, - #[cfg(test)] drain_test_hook: DrainTestHook::default(), #[cfg(test)] timeout_inspections: 0, - #[cfg(test)] - deadline_inspections: 0, }; let handle = ManagedClientHandle { stop, diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index 2579356..3326bf1 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -16,7 +16,7 @@ use irtt_proto::{ use tokio::runtime::{Builder, Runtime}; use super::*; -use crate::{socket::resolution_call_counts, ClientAuthConfig, NegotiationPolicy, RunMode}; +use crate::{ClientAuthConfig, NegotiationPolicy, RunMode}; const TOKEN: u64 = 0x1234_5678_90ab_cdef; @@ -388,44 +388,6 @@ fn assert_rotated_without_catchup(sends: &[(TargetInstance, u32, Instant, Instan } } -fn run_negotiated_stagger_case( - requested: Duration, - negotiated: &[Duration], -) -> (Vec<(usize, Duration)>, Vec) { - let servers = negotiated - .iter() - .map(|interval| start_server_negotiating(ServerBehavior::Echo, None, Some(*interval))) - .collect::>(); - let mut managed = config(ManagedPacing::Staggered); - managed.client.duration = Some(Duration::from_millis(190)); - managed.client.interval = requested; - managed.client.negotiation_policy = NegotiationPolicy::Loose; - let targets = servers - .iter() - .enumerate() - .map(|(index, server)| target(&format!("target-{index}"), server.addr)) - .collect(); - let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); - let observations = Arc::new(Mutex::new(Vec::new())); - task.stagger_observations = Some(Arc::clone(&observations)); - let outcome = runtime().block_on(task); - let packets_sent = (0..servers.len()) - .map(|index| { - outcome - .recent_target_outcomes - .iter() - .find(|target| target.target.id.as_ref() == format!("target-{index}")) - .unwrap() - .packets_sent - }) - .collect(); - for server in servers { - server.finish(); - } - let spacings = observations.lock().unwrap().clone(); - (spacings, packets_sent) -} - fn run_deferred_drain_burst(packet_count: usize) -> ManagedOutcome { let server = start_server(ServerBehavior::DeferredBurst(packet_count), None); let mut managed = config(ManagedPacing::Staggered); @@ -459,13 +421,11 @@ fn run_deferred_drain_burst(packet_count: usize) -> ManagedOutcome { #[test] fn construction_is_runtime_and_io_free() { - let before = resolution_call_counts(); let (task, handle) = ManagedClient::task( config(ManagedPacing::Staggered), vec![ManagedTargetConfig::new("dns", "no-such-host.invalid")], ) .unwrap(); - assert_eq!(resolution_call_counts(), before); assert_eq!(handle.status().lifecycle, ManagedLifecycle::NotStarted); drop(task); } @@ -1400,7 +1360,6 @@ fn first_poll_without_runtime_fails_durably() { #[test] fn pre_poll_stop() { - let before = resolution_call_counts(); let (task, handle) = ManagedClient::task( config(ManagedPacing::Staggered), vec![ManagedTargetConfig::new("one", "no-such-host.invalid")], @@ -1413,7 +1372,6 @@ fn pre_poll_stop() { }; assert!(poll_once(receipt.as_mut()).is_ready()); assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested); - assert_eq!(resolution_call_counts(), before); } #[test] @@ -1604,45 +1562,6 @@ fn staggered_pacing_fairness() { second.finish(); } -#[test] -fn staggered_pacing_uses_all_active_negotiated_intervals() { - let requested = Duration::from_millis(120); - - let (single, single_packets) = - run_negotiated_stagger_case(requested, &[Duration::from_millis(40)]); - assert!(single - .iter() - .all(|entry| *entry == (1, Duration::from_millis(40)))); - assert!(single_packets[0] >= 4); - - let (different, different_packets) = run_negotiated_stagger_case( - requested, - &[Duration::from_millis(30), Duration::from_millis(90)], - ); - let while_both_active = different - .iter() - .filter(|(active, _)| *active == 2) - .collect::>(); - assert!(!while_both_active.is_empty()); - assert!(while_both_active - .iter() - .all(|(_, spacing)| *spacing == Duration::from_millis(15))); - assert!(different_packets[0] > different_packets[1]); - - let (equal, _) = run_negotiated_stagger_case( - requested, - &[Duration::from_millis(60), Duration::from_millis(60)], - ); - let while_both_active = equal - .iter() - .filter(|(active, _)| *active == 2) - .collect::>(); - assert!(!while_both_active.is_empty()); - assert!(while_both_active - .iter() - .all(|(_, spacing)| *spacing == Duration::from_millis(30))); -} - #[test] fn stagger_gate_tracks_active_membership() { let first = @@ -1690,7 +1609,7 @@ fn stagger_gate_tracks_active_membership() { let accepted_at = Instant::now(); task.record_stagger_acceptance( SendResult::Failed { accepted: true }, - Some((2, Duration::from_millis(50))), + Some(Duration::from_millis(50)), accepted_at, ); assert_eq!( @@ -1752,11 +1671,7 @@ fn the_stagger_gate_advances_by_one_spacing_per_send_without_accumulating_latene let start = Instant::now(); let lateness = Duration::from_millis(3); - task.record_stagger_acceptance( - SendResult::Ready { accepted: true }, - Some((1, spacing)), - start, - ); + task.record_stagger_acceptance(SendResult::Ready { accepted: true }, Some(spacing), start); assert_eq!(task.send_gate, Some(start + spacing), "first send anchors"); // Every later send is observed `lateness` after the gate that released it. @@ -1766,7 +1681,7 @@ fn the_stagger_gate_advances_by_one_spacing_per_send_without_accumulating_latene let released_at = start + spacing * slot; task.record_stagger_acceptance( SendResult::Ready { accepted: true }, - Some((1, spacing)), + Some(spacing), released_at + lateness, ); assert_eq!( @@ -1782,7 +1697,7 @@ fn the_stagger_gate_advances_by_one_spacing_per_send_without_accumulating_latene let resumed_at = stalled_gate + spacing; task.record_stagger_acceptance( SendResult::Ready { accepted: true }, - Some((1, spacing)), + Some(spacing), resumed_at, ); assert_eq!(task.send_gate, Some(resumed_at + spacing)); @@ -1926,69 +1841,6 @@ fn pending_limit_failure_drains_reply_and_closes_session() { assert!(has_close(&records)); } -#[test] -fn drain_failures_preserve_primary_outcome_and_first_cleanup_failure() { - let completed_server = start_server(ServerBehavior::Echo, None); - let (mut completed_task, _) = ManagedClient::task( - config(ManagedPacing::Staggered), - vec![target("completed", completed_server.addr)], - ) - .unwrap(); - completed_task.drain_test_hook.fail_receive = true; - completed_task.drain_test_hook.fail_close = true; - let completed = runtime().block_on(completed_task); - let completed_target = &completed.recent_target_outcomes[0]; - assert!(matches!( - completed_target.end_reason, - ManagedTargetEndReason::TestComplete - )); - assert_eq!(completed.failed_target_outcomes, 0); - assert!(matches!( - completed_target.cleanup_failure, - Some(ManagedTargetFailure { - phase: ManagedTargetFailurePhase::Receiving, - kind: ManagedTargetFailureKind::Socket, - .. - }) - )); - completed_server.finish(); - - let stopped_server = start_server(ServerBehavior::Echo, None); - let mut managed = config(ManagedPacing::Staggered); - managed.completion = ManagedCompletionPolicy::ExplicitStop; - managed.client.duration = None; - let (mut stopped_task, handle) = - ManagedClient::task(managed, vec![target("stopped", stopped_server.addr)]).unwrap(); - stopped_task.drain_test_hook.fail_receive = true; - let seen = Arc::clone(&stopped_server.probe_seen); - let stopper = thread::spawn(move || { - let (flag, ready) = &*seen; - let guard = flag.lock().unwrap(); - let (guard, timeout) = ready - .wait_timeout_while(guard, Duration::from_secs(2), |seen| !*seen) - .unwrap(); - assert!(*guard && !timeout.timed_out()); - drop(handle.stop()); - }); - let stopped = runtime().block_on(stopped_task); - stopper.join().unwrap(); - let stopped_target = &stopped.recent_target_outcomes[0]; - assert!(matches!( - stopped_target.end_reason, - ManagedTargetEndReason::Stopped - )); - assert_eq!(stopped.failed_target_outcomes, 0); - assert!(matches!( - stopped_target.cleanup_failure, - Some(ManagedTargetFailure { - phase: ManagedTargetFailurePhase::Receiving, - kind: ManagedTargetFailureKind::Socket, - .. - }) - )); - stopped_server.finish(); -} - #[test] fn final_drain_overflow_is_rejected_and_preserves_success() { let mut invalid = config(ManagedPacing::Staggered); diff --git a/crates/irtt-client/src/metadata.rs b/crates/irtt-client/src/metadata.rs index 0f18c3c..4a0f371 100644 --- a/crates/irtt-client/src/metadata.rs +++ b/crates/irtt-client/src/metadata.rs @@ -135,37 +135,6 @@ mod tests { ); } - #[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(); @@ -180,19 +149,6 @@ mod tests { ); } - #[test] - fn metadata_observed_traffic_class_zero_preserves_observed_zero() { - let packet_meta = PacketMeta::from(ReceiveMeta { - traffic_class: Some(0), - kernel_rx_timestamp: None, - }); - - assert_eq!(packet_meta.traffic_class, Some(0)); - assert_eq!(packet_meta.dscp, Some(0)); - assert_eq!(packet_meta.ecn, Some(0)); - assert_eq!(packet_meta.kernel_rx_timestamp, None); - } - #[test] fn metadata_observed_traffic_class_derives_dscp_and_ecn() { let packet_meta = PacketMeta::from(ReceiveMeta { diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index 1cd6856..4080139 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -1,5 +1,5 @@ use std::{ - collections::{HashMap, HashSet, TryReserveError, VecDeque}, + collections::{HashMap, HashSet, VecDeque}, time::{Instant, SystemTime}, }; @@ -85,7 +85,12 @@ impl PendingMap { }); } if self.map.len() == self.map.capacity() { - self.map.try_reserve(1).map_err(pending_allocation_failed)?; + self.map + .try_reserve(1) + .map_err(|source| ClientError::AllocationFailed { + operation: "pending probe storage", + source, + })?; } Ok(()) } @@ -186,6 +191,12 @@ impl PendingMap { ExpiredBatch { probes, more_due } } + pub fn is_empty(&self) -> bool { + self.map.is_empty() + } + + // Verify repeated or abandoned send preflights do not create pending probes. + #[cfg(test)] pub fn len(&self) -> usize { self.map.len() } @@ -234,34 +245,11 @@ impl PendingMap { } } + // Check sequence-reuse and kernel timestamp retention without consuming a probe. #[cfg(test)] pub fn contains(&self, wire_seq: u32) -> bool { self.map.contains_key(&wire_seq) } - - #[cfg(test)] - pub fn capacity(&self) -> usize { - self.map.capacity() - } - - #[cfg(test)] - pub fn linkage(&self) -> (Option, Option) { - (self.first, self.last) - } - - #[cfg(test)] - pub fn entry_links(&self, wire_seq: u32) -> Option<(Option, Option)> { - self.map - .get(&wire_seq) - .map(|entry| (entry.previous, entry.next)) - } -} - -fn pending_allocation_failed(source: TryReserveError) -> ClientError { - ClientError::AllocationFailed { - operation: "pending probe storage", - source, - } } #[derive(Debug)] @@ -312,31 +300,17 @@ impl TimedOutMap { self.map.get_mut(&wire_seq) } - pub fn clear(&mut self) { - self.map.clear(); - self.insertion_order.clear(); - } - - #[cfg(test)] - pub fn len(&self) -> usize { - self.map.len() - } - #[cfg(feature = "tokio")] pub fn latest_timeout_deadline(&self) -> Option { self.map.values().map(|probe| probe.timeout_at).max() } + // Check sequence-reuse and kernel timestamp retention without consuming a probe. #[cfg(test)] pub fn contains(&self, wire_seq: u32) -> bool { self.map.contains_key(&wire_seq) } - #[cfg(test)] - fn insertion_order_len(&self) -> usize { - self.insertion_order.len() - } - fn evict_oldest(&mut self) { while let Some(oldest_key) = self.insertion_order.pop_front() { if self.map.remove(&oldest_key).is_some() { @@ -438,23 +412,6 @@ mod tests { )); } - #[test] - fn pending_map_preflight_reserves_without_changing_contents() { - let mut map = PendingMap::new(2); - let initial_capacity = map.capacity(); - - map.preflight_insert(7).unwrap(); - let reserved_capacity = map.capacity(); - assert!(reserved_capacity >= initial_capacity); - assert_eq!(map.len(), 0); - assert!(!map.contains(7)); - - map.preflight_insert(7).unwrap(); - assert_eq!(map.capacity(), reserved_capacity); - assert_eq!(map.len(), 0); - assert!(!map.contains(7)); - } - #[test] fn pending_map_preflight_rejects_sequence_collision() { let mut map = PendingMap::new(2); @@ -466,21 +423,7 @@ mod tests { map.preflight_insert(9), Err(ClientError::PendingSequenceCollision { seq: 9 }) )); - assert_eq!(map.len(), 1); - } - - #[test] - fn pending_map_allocation_failure_maps_to_dedicated_error() { - let mut map = HashMap::::new(); - let source = map.try_reserve(usize::MAX).unwrap_err(); - - assert!(matches!( - pending_allocation_failed(source), - ClientError::AllocationFailed { - operation: "pending probe storage", - .. - } - )); + assert_eq!(map.map.len(), 1); } #[test] @@ -499,10 +442,6 @@ mod tests { map.latest_timeout_deadline(), Some(now + Duration::from_secs(3)) ); - assert_eq!(map.linkage(), (Some(1), Some(3))); - assert_eq!(map.entry_links(1), Some((None, Some(2)))); - assert_eq!(map.entry_links(2), Some((Some(1), Some(3)))); - assert_eq!(map.entry_links(3), Some((Some(2), None))); assert!(map.remove(2).is_some()); assert_eq!( @@ -513,9 +452,6 @@ mod tests { map.latest_timeout_deadline(), Some(now + Duration::from_secs(3)) ); - assert_eq!(map.linkage(), (Some(1), Some(3))); - assert_eq!(map.entry_links(1), Some((None, Some(3)))); - assert_eq!(map.entry_links(3), Some((Some(1), None))); assert!(map.remove(1).is_some()); assert_eq!( @@ -526,13 +462,10 @@ mod tests { map.latest_timeout_deadline(), Some(now + Duration::from_secs(3)) ); - assert_eq!(map.linkage(), (Some(3), Some(3))); - assert_eq!(map.entry_links(3), Some((None, None))); assert!(map.remove(3).is_some()); assert_eq!(map.next_timeout_deadline(), None); assert_eq!(map.latest_timeout_deadline(), None); - assert_eq!(map.linkage(), (None, None)); } #[test] @@ -554,7 +487,6 @@ mod tests { let empty = bounded.drain_expired_bounded(now + Duration::from_secs(4), 0); assert!(empty.probes.is_empty()); assert!(empty.more_due); - assert_eq!(bounded.linkage(), (Some(1), Some(4))); let mut actual = Vec::new(); loop { @@ -573,7 +505,6 @@ mod tests { assert_eq!(actual, expected); assert_eq!(actual, vec![1, 2, 3, 4]); - assert_eq!(bounded.linkage(), (None, None)); } #[test] @@ -585,8 +516,6 @@ mod tests { insert(&mut map, pending(7, now + Duration::from_secs(1))); assert!(map.drain_expired_bounded(now, usize::MAX).probes.is_empty()); - assert_eq!(map.linkage(), (Some(7), Some(7))); - assert_eq!(map.entry_links(7), Some((None, None))); assert_eq!( map.drain_expired_bounded(now + Duration::from_secs(1), usize::MAX) .probes @@ -595,7 +524,6 @@ mod tests { .collect::>(), vec![7] ); - assert_eq!(map.linkage(), (None, None)); } #[test] @@ -603,13 +531,12 @@ mod tests { let mut map = PendingMap::new(2); let now = Instant::now(); map.preflight_insert(7).unwrap(); - let capacity = map.capacity(); + let capacity = map.map.capacity(); map.commit_insert(pending(7, now)); - assert_eq!(map.capacity(), capacity); - assert_eq!(map.len(), 1); - assert_eq!(map.linkage(), (Some(7), Some(7))); + assert_eq!(map.map.capacity(), capacity); + assert_eq!(map.map.len(), 1); } #[test] @@ -624,8 +551,6 @@ mod tests { } assert!(map.remove(2).is_some()); - assert_eq!(map.entry_links(1), Some((None, Some(3)))); - assert_eq!(map.entry_links(3), Some((Some(1), Some(4)))); let expired = map.drain_expired_bounded(now + Duration::from_secs(3), usize::MAX); assert_eq!( expired @@ -635,9 +560,7 @@ mod tests { .collect::>(), vec![1, 3] ); - assert_eq!(map.linkage(), (Some(4), Some(4))); assert!(map.remove(4).is_some()); - assert_eq!(map.linkage(), (None, None)); } #[test] @@ -660,7 +583,7 @@ mod tests { map.insert(pending(1, now)); map.insert(pending(2, now)); - assert_eq!(map.len(), 2); + assert_eq!(map.map.len(), 2); assert!(map.remove(0).is_none()); assert!(map.remove(1).is_some()); assert!(map.remove(2).is_some()); @@ -674,8 +597,8 @@ mod tests { for i in 0..20 { map.insert(pending(i, now)); assert!(map.remove(i).is_some()); - assert_eq!(map.len(), 0); - assert_eq!(map.insertion_order_len(), 0); + assert_eq!(map.map.len(), 0); + assert_eq!(map.insertion_order.len(), 0); } } } diff --git a/crates/irtt-client/src/receive.rs b/crates/irtt-client/src/receive.rs index d9d7e00..f05aeec 100644 --- a/crates/irtt-client/src/receive.rs +++ b/crates/irtt-client/src/receive.rs @@ -112,11 +112,11 @@ pub(crate) fn drain_tx_timestamps( linux::drain_tx_timestamps(socket, on_timestamp) } -#[cfg(test)] +#[cfg(all(test, feature = "tokio"))] mod tests { use std::net::UdpSocket; - use crate::{metadata::ReceiveMeta, receive::recv_datagram, timing::ClientTimestamp}; + use crate::metadata::ReceiveMeta; fn connected_loopback_pair() -> (UdpSocket, UdpSocket) { let a = UdpSocket::bind("127.0.0.1:0").unwrap(); @@ -126,44 +126,6 @@ mod tests { (a, b) } - #[test] - fn fallback_receive_returns_length() { - let (sender, receiver) = connected_loopback_pair(); - sender.send(b"hello").unwrap(); - - let mut buf = [0_u8; 16]; - let datagram = recv_datagram(&receiver, &mut buf).unwrap(); - - assert_eq!(datagram.len, 5); - assert_eq!(&buf[..datagram.len], b"hello"); - } - - #[test] - fn fallback_receive_returns_default_metadata() { - let (sender, receiver) = connected_loopback_pair(); - sender.send(b"meta").unwrap(); - - let mut buf = [0_u8; 16]; - let datagram = recv_datagram(&receiver, &mut buf).unwrap(); - - assert_eq!(datagram.meta, ReceiveMeta::default()); - } - - #[test] - fn fallback_receive_captures_timestamp_after_successful_receive() { - let (sender, receiver) = connected_loopback_pair(); - sender.send(b"time").unwrap(); - - let before = ClientTimestamp::now(); - let mut buf = [0_u8; 16]; - let datagram = recv_datagram(&receiver, &mut buf).unwrap(); - let after = ClientTimestamp::now(); - - assert!(datagram.received_at.mono >= before.mono); - assert!(datagram.received_at.mono <= after.mono); - } - - #[cfg(feature = "tokio")] #[test] fn tokio_try_receive_preserves_fallback_metadata_and_would_block() { let runtime = tokio::runtime::Builder::new_current_thread() diff --git a/crates/irtt-client/src/receive/linux.rs b/crates/irtt-client/src/receive/linux.rs index 53c9a53..0311f31 100644 --- a/crates/irtt-client/src/receive/linux.rs +++ b/crates/irtt-client/src/receive/linux.rs @@ -21,8 +21,8 @@ //! session so later IDs cannot be misattributed. It performs a small bounded, //! nonblocking read of //! `MSG_ERRQUEUE` so the adapter can opportunistically collect timestamps -//! without ever waiting for one. The observed kernel TX wall time is retained -//! as metadata only; it is not consumed by any measurement in this change. +//! without ever waiting for one. A plausible kernel TX wall time is used only +//! for upstream one-way delay; RTT retains the userspace send instant. use std::{ io::{self, IoSliceMut}, @@ -161,12 +161,12 @@ fn recv_datagram_fd( buf: &mut [u8], flags: MsgFlags, ) -> Result { - let mut control = ControlBuffer::new(); + let mut control = ControlBuffer([0; CONTROL_LEN]); let mut iov = [IoSliceMut::new(buf)]; // The socket is connected, so the source address is not needed; `()` skips // copying it out. - let msg = recvmsg::<()>(socket_fd, &mut iov, Some(control.as_mut_slice()), flags)?; + let msg = recvmsg::<()>(socket_fd, &mut iov, Some(&mut control.0), flags)?; let received_at = ClientTimestamp::now(); Ok(ReceivedDatagram { @@ -216,16 +216,6 @@ fn system_time_from_timespec(timespec: TimeSpec) -> Option { #[repr(align(8))] struct ControlBuffer([u8; CONTROL_LEN]); -impl ControlBuffer { - fn new() -> Self { - Self([0; CONTROL_LEN]) - } - - fn as_mut_slice(&mut self) -> &mut [u8] { - &mut self.0 - } -} - #[cfg(test)] mod tests { use std::{ diff --git a/crates/irtt-client/src/receive/linux/error_queue.rs b/crates/irtt-client/src/receive/linux/error_queue.rs index 8d16e65..f2d3703 100644 --- a/crates/irtt-client/src/receive/linux/error_queue.rs +++ b/crates/irtt-client/src/receive/linux/error_queue.rs @@ -103,10 +103,7 @@ pub(crate) fn classify(cmsgs: impl Iterator) -> Erro for cmsg in cmsgs { match cmsg { - ControlMessageOwned::Ipv4RecvErr(err, _) => { - extended_error = Some((err.ee_errno, err.ee_origin, err.ee_info, err.ee_data)); - } - ControlMessageOwned::Ipv6RecvErr(err, _) => { + ControlMessageOwned::Ipv4RecvErr(err, _) | ControlMessageOwned::Ipv6RecvErr(err, _) => { extended_error = Some((err.ee_errno, err.ee_origin, err.ee_info, err.ee_data)); } ControlMessageOwned::ScmTimestampsns(observed) => { @@ -145,12 +142,6 @@ pub(crate) fn classify(cmsgs: impl Iterator) -> Erro #[repr(align(8))] struct ErrorQueueControlBuffer([u8; ERROR_QUEUE_CONTROL_LEN]); -impl ErrorQueueControlBuffer { - fn new() -> Self { - Self([0; ERROR_QUEUE_CONTROL_LEN]) - } -} - /// Nonblocking drain of a single record from `fd`'s `MSG_ERRQUEUE`. /// /// Returns `Ok(None)` when the queue is empty (`EAGAIN`/`EWOULDBLOCK`) — the @@ -171,7 +162,7 @@ impl ErrorQueueControlBuffer { pub(crate) fn try_recv_error_queue_record(fd: RawFd) -> io::Result> { let mut payload: [u8; 0] = []; let mut iov = [IoSliceMut::new(&mut payload)]; - let mut control = ErrorQueueControlBuffer::new(); + let mut control = ErrorQueueControlBuffer([0; ERROR_QUEUE_CONTROL_LEN]); match recvmsg::<()>( fd, diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 8ff6326..45d252d 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -84,27 +84,16 @@ impl PreparedOpenAcceptance { pub(crate) fn normal_negotiated(&self) -> Option<&NegotiatedParams> { match &self.next_state { MachineState::Open(session) => Some(&session.negotiated), - MachineState::NoTestCompleted => None, - MachineState::Connected | MachineState::Closed { .. } => { - unreachable!("open acceptance only prepares open or no-test state") - } + _ => None, } } pub(crate) fn cleanup_close_packet(&self) -> Option<&[u8]> { match &self.next_state { MachineState::Open(session) => Some(&session.local_close_packet), - MachineState::NoTestCompleted => None, - MachineState::Connected | MachineState::Closed { .. } => { - unreachable!("open acceptance only prepares open or no-test state") - } + _ => None, } } - - #[cfg(test)] - pub(crate) fn has_prepared_active_session(&self) -> bool { - matches!(self.next_state, MachineState::Open(_)) - } } #[derive(Debug)] @@ -559,19 +548,11 @@ impl SessionMachine { pub(crate) fn pending_is_empty(&self) -> bool { match &self.state { - MachineState::Open(session) => session.pending.len() == 0, + MachineState::Open(session) => session.pending.is_empty(), _ => true, } } - #[cfg(test)] - pub(crate) fn is_terminal(&self) -> bool { - matches!( - self.state, - MachineState::NoTestCompleted | MachineState::Closed { .. } - ) - } - pub(crate) fn is_peer_closed(&self) -> bool { matches!( self.state, @@ -582,7 +563,6 @@ impl SessionMachine { ) } - #[cfg(any(feature = "tokio", test))] pub(crate) fn packets_sent(&self) -> u64 { match &self.state { MachineState::Open(session) => session.packets_sent, @@ -594,9 +574,7 @@ impl SessionMachine { pub(crate) fn next_probe_timeout_deadline(&self) -> Option { match &self.state { MachineState::Open(session) => session.pending.next_timeout_deadline(), - MachineState::Connected - | MachineState::NoTestCompleted - | MachineState::Closed { .. } => None, + _ => None, } } @@ -609,17 +587,11 @@ impl SessionMachine { .into_iter() .chain(session.timed_out.latest_timeout_deadline()) .max(), - MachineState::Connected - | MachineState::NoTestCompleted - | MachineState::Closed { .. } => None, + _ => None, } } - #[cfg(test)] - pub(crate) fn is_open(&self) -> bool { - matches!(self.state, MachineState::Open(_)) - } - + #[cfg(feature = "tokio")] pub(crate) fn ensure_open(&self) -> Result<(), ClientError> { self.open_session().map(|_| ()) } @@ -856,35 +828,20 @@ impl SessionMachine { }; if should_close { - self.close_from_peer(token, now, &mut events); + let packets_sent = self.packets_sent(); + self.state = MachineState::Closed { + source: CloseSource::Peer, + packets_sent, + }; + events.push(ClientEvent::SessionClosed { + remote: self.remote, + token, + at: now, + }); } Ok(events) } - fn close_from_peer(&mut self, token: u64, now: ClientTimestamp, events: &mut Vec) { - self.transition_to_closed(CloseSource::Peer); - events.push(ClientEvent::SessionClosed { - remote: self.remote, - token, - at: now, - }); - } - - fn transition_to_closed(&mut self, source: CloseSource) { - let packets_sent = match &mut self.state { - MachineState::Open(session) => { - session.timed_out.clear(); - session.packets_sent - } - MachineState::Closed { packets_sent, .. } => *packets_sent, - MachineState::Connected | MachineState::NoTestCompleted => 0, - }; - self.state = MachineState::Closed { - source, - packets_sent, - }; - } - fn open_session(&self) -> Result<&ActiveSession, ClientError> { match &self.state { MachineState::Open(session) => Ok(session), @@ -965,7 +922,7 @@ pub(crate) fn sequence_is_after(candidate: u32, current: u32) -> bool { } pub(crate) fn sequence_is_before(candidate: u32, current: u32) -> bool { - current != candidate && current.wrapping_sub(candidate) < (1 << 31) + sequence_is_after(current, candidate) } pub(crate) fn compute_rtt( @@ -1189,29 +1146,13 @@ mod tests { }; let remote = "127.0.0.1:2112".parse().unwrap(); let mut machine = SessionMachine::new(config, remote).unwrap(); - let negotiated = NegotiatedParams { - params: machine.requested.clone(), - restrictions: Vec::new(), - }; - machine.state = MachineState::Open(Box::new(ActiveSession { - token: 0x0102_0304_0506_0708, - negotiated, - local_close_packet: encode_request( - RequestToEncode::Close { - token: 0x0102_0304_0506_0708, - }, - None, + let prepared = machine + .prepare_open_acceptance( + normal_open_reply(&machine, 0x0102_0304_0506_0708), + timestamp(Instant::now()), ) - .unwrap() - .into_boxed_slice(), - next_wire_seq: 0, - highest_received_seq: None, - packets_sent: 0, - pending: PendingMap::new(max_pending_probes), - timed_out: TimedOutMap::new(max_pending_probes), - completed: CompletedSet::new(max_pending_probes), - kernel_tx_correlation_valid: true, - })); + .unwrap(); + machine.commit_open(prepared); machine } @@ -1255,19 +1196,6 @@ mod tests { irtt_proto::encode_open_reply(reply, machine.config.hmac_key.as_deref()) } - #[test] - fn prepared_open_request_is_exact_and_inert_when_dropped() { - let machine = connected_machine(ClientConfig::default()); - let first = machine.prepare_open_request().unwrap(); - let second = machine.prepare_open_request().unwrap(); - - assert_eq!(first.bytes, second.bytes); - assert!(!first.bytes.is_empty()); - drop(first); - drop(second); - assert!(matches!(machine.state, MachineState::Connected)); - } - #[test] fn malformed_wrong_direction_and_invalid_flags_are_ignored() { let machine = connected_machine(ClientConfig::default()); @@ -1465,244 +1393,6 @@ mod tests { assert!(failure.cleanup_close.is_none()); } - #[test] - fn dropping_prepared_acceptance_leaves_connected_state() { - let machine = connected_machine(ClientConfig::default()); - let prepared = machine - .prepare_open_acceptance( - normal_open_reply(&machine, 0x1020_3040_5060_7080), - timestamp(Instant::now()), - ) - .unwrap(); - - assert!(prepared.has_prepared_active_session()); - assert!(prepared.cleanup_close_packet().is_some()); - drop(prepared); - assert!(matches!(machine.state, MachineState::Connected)); - } - - #[test] - fn commit_open_assigns_prebuilt_state_once() { - let mut machine = connected_machine(ClientConfig::default()); - let token = 0x1020_3040_5060_7080; - let prepared = machine - .prepare_open_acceptance( - normal_open_reply(&machine, token), - timestamp(Instant::now()), - ) - .unwrap(); - - let expected_close = prepared.cleanup_close_packet().unwrap().to_vec(); - let outcome = machine.commit_open(prepared); - - assert!(matches!( - outcome, - OpenOutcome::Started { - token: outcome_token, - .. - } if outcome_token == token - )); - assert!(machine.is_open()); - assert_eq!(active(&machine).local_close_packet.as_ref(), expected_close); - assert!(matches!( - machine.prepare_open_request(), - Err(ClientError::AlreadyOpen) - )); - } - - #[test] - fn no_test_preparation_completes_without_active_session() { - let config = ClientConfig { - run_mode: RunMode::NoTest, - ..ClientConfig::default() - }; - let mut machine = connected_machine(config); - let reply = OpenReply { - flags: flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE, - token: 0, - params: machine.requested.clone(), - }; - let prepared = machine - .prepare_open_acceptance(reply, timestamp(Instant::now())) - .unwrap(); - - assert!(prepared.normal_negotiated().is_none()); - assert!(prepared.cleanup_close_packet().is_none()); - assert!(matches!(machine.state, MachineState::Connected)); - let outcome = machine.commit_open(prepared); - assert!(matches!(outcome, OpenOutcome::NoTestCompleted { .. })); - assert!(matches!(machine.state, MachineState::NoTestCompleted)); - } - - #[test] - fn cleanup_close_is_preencoded_with_token_and_hmac() { - let key = b"cleanup-key".to_vec(); - let machine = connected_machine(ClientConfig { - hmac_key: Some(key.clone()), - ..ClientConfig::default() - }); - let token = 0x1020_3040_5060_7080; - let prepared = machine - .prepare_open_acceptance( - normal_open_reply(&machine, token), - timestamp(Instant::now()), - ) - .unwrap(); - let cleanup = prepared.cleanup_close_packet().unwrap(); - - assert_eq!(cleanup[3], flags::FLAG_CLOSE | flags::FLAG_HMAC); - irtt_proto::verify_hmac(&key, cleanup, 4).unwrap(); - assert_eq!( - u64::from_le_bytes( - cleanup[4 + irtt_proto::HMAC_SIZE..12 + irtt_proto::HMAC_SIZE] - .try_into() - .unwrap() - ), - token - ); - assert!(matches!(machine.state, MachineState::Connected)); - } - - #[test] - fn prepare_close_is_stable_inert_and_does_not_determine_event_timestamp() { - let mut machine = open_machine(4, Duration::from_secs(1)); - active_mut(&mut machine).packets_sent = 7; - - let first_bytes = { - let prepared = machine.prepare_close().unwrap(); - assert_eq!( - prepared.bytes, - encode_request( - RequestToEncode::Close { - token: 0x0102_0304_0506_0708, - }, - None, - ) - .unwrap() - ); - assert_eq!(prepared.commit.packets_sent, 7); - assert_eq!(prepared.commit.token, 0x0102_0304_0506_0708); - prepared.bytes.to_vec() - }; - assert!(machine.is_open()); - assert_eq!(active(&machine).packets_sent, 7); - - let prepared = machine.prepare_close().unwrap(); - assert_eq!(prepared.bytes, first_bytes); - assert_eq!(prepared.commit.packets_sent, 7); - assert_eq!(prepared.commit.token, 0x0102_0304_0506_0708); - assert!(machine.is_open()); - } - - #[test] - fn local_close_commit_uses_supplied_exact_timestamp() { - let mut machine = open_machine(4, Duration::from_secs(1)); - let remote = machine.remote; - active_mut(&mut machine).packets_sent = 7; - let prepared = machine.prepare_close().unwrap(); - let sent_at = ClientTimestamp { - wall: UNIX_EPOCH + Duration::from_secs(1_234), - mono: Instant::now() + Duration::from_secs(5), - }; - let event = machine.commit_local_close(prepared.commit, sent_at); - - assert!(matches!( - event, - ClientEvent::SessionClosed { - remote: event_remote, - token: 0x0102_0304_0506_0708, - at, - } if event_remote == remote - && at == sent_at - )); - assert!(matches!( - machine.state, - MachineState::Closed { - source: CloseSource::Local, - packets_sent: 7, - } - )); - assert!(matches!( - machine.prepare_close(), - Err(ClientError::AlreadyClosed) - )); - } - - #[test] - fn prepared_close_reuses_hmac_packet_built_during_open_acceptance() { - let key = b"close-key".to_vec(); - let mut machine = connected_machine(ClientConfig { - hmac_key: Some(key.clone()), - ..ClientConfig::default() - }); - let token = 0x1020_3040_5060_7080; - let prepared_open = machine - .prepare_open_acceptance( - normal_open_reply(&machine, token), - timestamp(Instant::now()), - ) - .unwrap(); - let opening_close = prepared_open.cleanup_close_packet().unwrap().to_vec(); - machine.commit_open(prepared_open); - - let prepared_close = machine.prepare_close().unwrap(); - assert_eq!(prepared_close.bytes, opening_close); - assert_eq!( - prepared_close.bytes[3], - flags::FLAG_CLOSE | flags::FLAG_HMAC - ); - irtt_proto::verify_hmac(&key, prepared_close.bytes, 4).unwrap(); - assert_eq!( - u64::from_le_bytes( - prepared_close.bytes[4 + irtt_proto::HMAC_SIZE..12 + irtt_proto::HMAC_SIZE] - .try_into() - .unwrap() - ), - token - ); - } - - #[test] - fn close_preparation_preserves_non_open_errors_and_peer_source() { - let connected = connected_machine(ClientConfig::default()); - assert!(matches!( - connected.prepare_close(), - Err(ClientError::NotOpen) - )); - - let mut no_test = connected_machine(ClientConfig { - run_mode: RunMode::NoTest, - ..ClientConfig::default() - }); - let reply = OpenReply { - flags: flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE, - token: 0, - params: no_test.requested.clone(), - }; - let prepared = no_test - .prepare_open_acceptance(reply, timestamp(Instant::now())) - .unwrap(); - no_test.commit_open(prepared); - assert!(matches!( - no_test.prepare_close(), - Err(ClientError::AlreadyCompleted) - )); - - let mut peer_closed = open_machine(4, Duration::from_secs(1)); - peer_closed.transition_to_closed(CloseSource::Peer); - assert!(matches!( - peer_closed.state, - MachineState::Closed { - source: CloseSource::Peer, - .. - } - )); - assert!(matches!( - peer_closed.prepare_close(), - Err(ClientError::AlreadyClosed) - )); - } - #[test] fn uncommitted_probe_preparation_changes_no_authoritative_state() { let mut machine = open_machine(4, Duration::from_secs(1)); @@ -1737,7 +1427,6 @@ mod tests { let commit = machine.finalize_probe_commit(preflight, sent_at).unwrap(); machine.commit_probe_sent(commit, sent_at, accepted.bytes.len()); - let capacity = active(&machine).pending.capacity(); assert!(matches!( machine.preflight_probe_commit(&stale), Err(ClientError::StalePreparedProbe { @@ -1750,7 +1439,6 @@ mod tests { assert_eq!(session.next_wire_seq, 1); assert_eq!(session.packets_sent, 1); assert_eq!(session.pending.len(), 1); - assert_eq!(session.pending.capacity(), capacity); } #[test] @@ -1774,51 +1462,6 @@ mod tests { assert_eq!(session.pending.len(), 1); } - #[test] - fn probe_commit_does_not_require_presentation_timing() { - let mut machine = open_machine(4, Duration::from_secs(1)); - let prepared = machine.prepare_probe().unwrap(); - let send_anchor = timestamp(Instant::now()); - let preflight = machine.preflight_probe_commit(&prepared).unwrap(); - let commit = machine - .finalize_probe_commit(preflight, send_anchor) - .unwrap(); - // Fallible work (timeout deadline arithmetic, tx lower bound) is - // finalized from the pre-send anchor, not any post-send sample. - assert_eq!(commit.timeout_at, send_anchor.mono + Duration::from_secs(1)); - assert_eq!(commit.tx_not_before_wall, send_anchor.wall); - - // The post-send measurement `sent_at` is supplied afterward and can - // legitimately differ from the pre-send anchor. - let sent_at = timestamp(Instant::now()); - let sent = machine.commit_probe_sent(commit, sent_at, prepared.bytes.len()); - - assert_eq!(sent.seq, prepared.seq); - assert_eq!(sent.sent_at, sent_at); - assert_eq!(sent.bytes, prepared.bytes.len()); - } - - #[test] - fn commit_uses_reserved_capacity_and_prevalidated_counter() { - let mut machine = open_machine(2, Duration::from_secs(1)); - active_mut(&mut machine).packets_sent = u64::MAX - 1; - let prepared = machine.prepare_probe().unwrap(); - let preflight = machine.preflight_probe_commit(&prepared).unwrap(); - let reserved_capacity = active(&machine).pending.capacity(); - assert_eq!(preflight.next_packets_sent, u64::MAX); - let commit = machine - .finalize_probe_commit(preflight, timestamp(Instant::now())) - .unwrap(); - - machine.commit_probe_sent(commit, timestamp(Instant::now()), prepared.bytes.len()); - - let session = active(&machine); - assert_eq!(session.pending.capacity(), reserved_capacity); - assert_eq!(session.pending.len(), 1); - assert_eq!(session.packets_sent, u64::MAX); - assert_eq!(session.next_wire_seq, 1); - } - #[test] fn pending_capacity_exhaustion_is_detected_before_commit() { let mut machine = open_machine(1, Duration::from_secs(1)); @@ -1909,7 +1552,6 @@ mod tests { ] )); assert_eq!(active(&machine).pending.len(), 0); - assert_eq!(active(&machine).timed_out.len(), 3); } #[test] @@ -2581,21 +2223,6 @@ mod tests { assert!(active_mut(&mut machine).timed_out.get_mut(first).is_none()); } - #[test] - fn kernel_tx_timestamp_recording_is_a_no_op_once_closed() { - let mut machine = open_machine(4, Duration::from_secs(1)); - let now = Instant::now(); - let seq = send_probe(&mut machine, timestamp(now)); - let prepared = machine.prepare_close().unwrap(); - let commit = prepared.commit; - machine.commit_local_close(commit, timestamp(now)); - assert!(machine.is_terminal()); - - // Recording after close must not panic (there is no open session to - // index into) and must remain inert. - machine.record_kernel_tx_timestamp(seq, tx_ts(7)); - } - #[test] fn kernel_tx_timestamp_ids_wrap_like_wire_seq() { let mut machine = open_machine(4, Duration::from_secs(1)); @@ -2685,27 +2312,6 @@ mod tests { UNIX_EPOCH + Duration::from_millis(offset_ms) } - #[test] - fn preferred_send_wall_without_kernel_timestamp_uses_sent_at() { - let anchor = send_wall_ms(1_000); - let received_at = send_wall_ms(1_040); - assert_eq!( - preferred_send_wall(anchor, anchor, None, received_at), - anchor - ); - } - - #[test] - fn preferred_send_wall_uses_kernel_timestamp_strictly_between_bounds() { - let anchor = 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(anchor, anchor, Some(kernel_tx), received_at), - kernel_tx - ); - } - #[test] fn preferred_send_wall_accepts_kernel_timestamp_equal_to_lower_bound() { let anchor = send_wall_ms(1_000); @@ -2726,93 +2332,6 @@ mod tests { ); } - #[test] - fn preferred_send_wall_rejects_kernel_timestamp_earlier_than_lower_bound() { - let anchor = send_wall_ms(1_000); - let kernel_tx = anchor - Duration::from_nanos(1); - let received_at = send_wall_ms(1_040); - assert_eq!( - preferred_send_wall(anchor, anchor, Some(kernel_tx), received_at), - anchor - ); - } - - #[test] - fn preferred_send_wall_rejects_kernel_timestamp_later_than_received_at() { - let anchor = 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(anchor, anchor, Some(kernel_tx), received_at), - anchor - ); - } - - #[test] - fn preferred_send_wall_handles_extreme_timestamps_without_panicking() { - let anchor = 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(anchor, anchor, Some(far_future), received_at), - anchor - ); - assert_eq!( - preferred_send_wall(anchor, anchor, Some(before_epoch), received_at), - anchor - ); - assert_eq!( - preferred_send_wall(before_epoch, before_epoch, None, received_at), - before_epoch - ); - } - - // Pre-/post-send decoupling (F-04): the plausibility lower bound is the - // pre-send anchor, but the fallback endpoint is the post-send sample, - // and the two can legitimately differ. - - #[test] - fn preferred_send_wall_accepts_kernel_timestamp_before_post_send_sample() { - let tx_not_before_wall = send_wall_ms(1_000); - let post_send_sent_at = send_wall_ms(1_008); - let kernel_tx = send_wall_ms(1_005); - let received_at = send_wall_ms(1_040); - - // The kernel timestamp precedes the post-send sample but follows - // the pre-send anchor: it must still be accepted. - assert_eq!( - preferred_send_wall( - tx_not_before_wall, - post_send_sent_at, - Some(kernel_tx), - received_at - ), - kernel_tx - ); - } - - #[test] - fn preferred_send_wall_rejected_kernel_timestamp_falls_back_to_post_send_sample() { - let tx_not_before_wall = send_wall_ms(1_000); - let post_send_sent_at = send_wall_ms(1_008); - let kernel_tx = send_wall_ms(999); - let received_at = send_wall_ms(1_040); - - // Below the pre-send lower bound: rejected, and the fallback is the - // post-send sample, not the pre-send anchor. - assert_eq!( - preferred_send_wall( - tx_not_before_wall, - post_send_sent_at, - Some(kernel_tx), - received_at - ), - post_send_sent_at - ); - } - fn upstream_probe(sent_wall: SystemTime, mono: Instant) -> SessionMachine { machine_with_pending_probe(ClientTimestamp { mono, @@ -3096,686 +2615,4 @@ mod tests { Some(SignedDuration::from_nanos(12_000_000)) ); } - - /// Model-based tests for `SessionMachine`'s open-session operation - /// contract: probe sends, reply classification, timeouts, and close. - /// - /// ## Reference-model boundary - /// - /// The open handshake itself (accept/reject, run-mode, version and - /// token checks) is deliberately out of scope here: it already has - /// focused deterministic coverage above in `mod tests`, and every - /// generated case here starts from an already-`Open` session built the - /// same way `open_machine` above builds one, by constructing - /// `MachineState::Open` directly. What *is* generated and checked - /// against an independent reference model is the combinatorial space - /// that only exists once a session is open: interleaved probe sends, - /// valid/duplicate/late/unknown/wrong-token replies, peer and local - /// close, bounded and unbounded timeout polling, and `u32` wire-sequence - /// wraparound. - /// - /// The reference [`Model`] tracks only state the public contract - /// documents: open/closed(local)/closed(peer) state, the next wire - /// sequence and total sent-packet counters (mirroring - /// `SessionMachine::packets_sent`), which sequences are pending, timed - /// out, or completed (this drives the exact - /// `EchoReply`/`LateReply`/`DuplicateReply`/`Warning` classification - /// documented on `ClientEvent`), each pending probe's timeout deadline - /// (documented on `ClientEvent::EchoLoss::timeout_at`), and the highest - /// accepted in-window sequence used for late-reply classification. It - /// deliberately does *not* model the bounded FIFO eviction that - /// `PendingMap`/`TimedOutMap`/`CompletedSet` apply once - /// `max_pending_probes` is exceeded: that is a private capacity policy - /// with its own focused unit tests in `probe.rs`, not part of this - /// contract. Every generated case here uses a `max_pending_probes` far - /// larger than the generated operation count, so capacity/eviction - /// behavior never confounds the classification assertions made here. - /// - /// `u32` sequence wraparound is exercised by seeding a session's - /// starting wire sequence near `u32::MAX` through direct construction, - /// the same way `open_machine` above builds an open session without - /// running the handshake. The repository's `AGENTS.md` testing policy - /// names "sequence wrap" as a sanctioned reason to construct state - /// directly rather than drive billions of real sends. - mod model_properties { - use std::collections::{HashSet, VecDeque}; - - use proptest::prelude::*; - use proptest::test_runner::TestCaseError; - - use super::*; - use irtt_proto::{encode_echo_reply, ReceivedStats, StampAt}; - - const TOKEN: u64 = 0x1122_3344_5566_7788; - const WRONG_TOKEN: u64 = 0x8877_6655_4433_2211; - const PROBE_TIMEOUT: Duration = Duration::from_millis(30); - /// Far larger than the generated operation count (`1..=OP_LIMIT`), - /// so pending/timed-out/completed bounded eviction never triggers; - /// see the module doc comment. - const MAX_PENDING_PROBES: usize = 4096; - const OP_LIMIT: usize = 50; - const CASES: u32 = 256; - - fn model_config() -> ClientConfig { - ClientConfig { - received_stats: ReceivedStats::None, - stamp_at: StampAt::None, - clock: Clock::Wall, - probe_timeout: PROBE_TIMEOUT, - max_pending_probes: MAX_PENDING_PROBES, - run_mode: RunMode::Normal, - ..ClientConfig::default() - } - } - - /// Builds an already-`Open` session directly, the same way `mod - /// tests`'s `open_machine` does, starting its wire sequence at - /// `start_seq` instead of always zero. See the module doc comment - /// for why this bypasses the open handshake and for the sequence - /// seeding rationale. - fn open_machine_for_model(start_seq: u32) -> (SessionMachine, Params) { - let config = model_config(); - let remote = "127.0.0.1:2112".parse().unwrap(); - let mut machine = SessionMachine::new(config, remote).unwrap(); - let params = machine.requested.clone(); - let negotiated = NegotiatedParams { - params: params.clone(), - restrictions: Vec::new(), - }; - machine.state = MachineState::Open(Box::new(ActiveSession { - token: TOKEN, - negotiated, - local_close_packet: encode_request(RequestToEncode::Close { token: TOKEN }, None) - .unwrap() - .into_boxed_slice(), - next_wire_seq: start_seq, - highest_received_seq: None, - packets_sent: 0, - pending: PendingMap::new(MAX_PENDING_PROBES), - timed_out: TimedOutMap::new(MAX_PENDING_PROBES), - completed: CompletedSet::new(MAX_PENDING_PROBES), - kernel_tx_correlation_valid: true, - })); - (machine, params) - } - - fn build_echo_reply(params: &Params, token: u64, seq: u32, close: bool) -> Vec { - let mut flags_value = flags::FLAG_REPLY; - if close { - flags_value |= flags::FLAG_CLOSE; - } - let reply = EchoReply { - flags: flags_value, - token, - sequence: seq, - recv_count: None, - recv_window: None, - timestamps: TimestampFields::default(), - payload: Vec::new(), - }; - encode_echo_reply(&reply, params, None).unwrap() - } - - /// Independently-authored mirror of the sequence-space half-window - /// "after" comparison (RFC 1982 style), used only to predict - /// expected late/in-window classification. Deliberately not a call - /// into `super::sequence_is_after`/`sequence_is_before`, so a bug in - /// the production comparator is not automatically invisible to this - /// model. - fn seq_after(candidate: u32, current: u32) -> bool { - candidate != current && candidate.wrapping_sub(current) < (1 << 31) - } - - fn seq_before(candidate: u32, current: u32) -> bool { - seq_after(current, candidate) - } - - #[derive(Debug, Clone, Copy, PartialEq, Eq)] - enum ModelState { - Open, - ClosedLocal, - ClosedPeer, - } - - #[derive(Debug, Clone, Copy, PartialEq, Eq)] - enum ExpectedEvent { - EchoReply(u32), - LateReply(u32), - DuplicateReply(u32), - WarnWrongToken, - WarnUntracked, - SessionClosed, - } - - /// The independent reference model. See the module doc comment for - /// its deliberate boundary against `SessionMachine`'s private state. - struct Model { - state: ModelState, - next_wire_seq: u32, - packets_sent: u64, - pending: VecDeque<(u32, Instant)>, - timed_out: HashSet, - completed: HashSet, - highest_received_seq: Option, - } - - impl Model { - fn new(start_seq: u32) -> Self { - Self { - state: ModelState::Open, - next_wire_seq: start_seq, - packets_sent: 0, - pending: VecDeque::new(), - timed_out: HashSet::new(), - completed: HashSet::new(), - highest_received_seq: None, - } - } - - fn pending_remove(&mut self, seq: u32) -> bool { - if let Some(pos) = self.pending.iter().position(|&(s, _)| s == seq) { - self.pending.remove(pos); - true - } else { - false - } - } - - fn sorted_pending(&self) -> Vec { - let mut v: Vec = self.pending.iter().map(|&(s, _)| s).collect(); - v.sort_unstable(); - v - } - - fn sorted_timed_out(&self) -> Vec { - let mut v: Vec = self.timed_out.iter().copied().collect(); - v.sort_unstable(); - v - } - - fn sorted_completed(&self) -> Vec { - let mut v: Vec = self.completed.iter().copied().collect(); - v.sort_unstable(); - v - } - - /// Resolves a generated [`ReplyTarget`] to a concrete sequence - /// number against the model's *current* membership. When the - /// requested bucket is empty this falls back to the raw index as - /// a literal (likely-unknown) sequence, so every `ReplyTarget` - /// variant remains a valid, meaningful generator input - /// regardless of what has happened so far. - fn resolve(&self, target: ReplyTarget) -> u32 { - fn pick(v: &[u32], i: u32) -> u32 { - if v.is_empty() { - i - } else { - v[(i as usize) % v.len()] - } - } - match target { - ReplyTarget::Pending(i) => pick(&self.sorted_pending(), i), - ReplyTarget::TimedOut(i) => pick(&self.sorted_timed_out(), i), - ReplyTarget::Completed(i) => pick(&self.sorted_completed(), i), - ReplyTarget::Unknown(seq) => seq, - } - } - - fn update_highest(&mut self, seq: u32) { - self.highest_received_seq = Some(match self.highest_received_seq { - None => seq, - Some(h) if seq_after(seq, h) => seq, - Some(h) => h, - }); - } - - fn apply_send(&mut self, seq: u32, timeout_at: Instant) { - assert_eq!( - seq, self.next_wire_seq, - "model/machine wire sequence desync" - ); - self.pending.push_back((seq, timeout_at)); - self.next_wire_seq = self.next_wire_seq.wrapping_add(1); - self.packets_sent += 1; - } - - fn apply_poll(&mut self, now: Instant, limit: usize) -> (Vec, bool) { - let mut expired = Vec::new(); - while expired.len() < limit { - let Some(&(seq, timeout_at)) = self.pending.front() else { - break; - }; - if timeout_at > now { - break; - } - self.pending.pop_front(); - self.timed_out.insert(seq); - expired.push(seq); - } - let more_due = self - .pending - .front() - .is_some_and(|&(_, timeout_at)| timeout_at <= now); - (expired, more_due) - } - - /// Mirrors `SessionMachine::process_echo_reply`'s classification - /// exactly (see that function's branches), mutating this model - /// the same way. Returns the event kinds a correct - /// implementation must produce, in order. - fn reply(&mut self, seq: u32, wrong_token: bool, close: bool) -> Vec { - if wrong_token { - // A wrong-token reply is rejected before the close flag - // is ever inspected, so `close` has no effect here. - return vec![ExpectedEvent::WarnWrongToken]; - } - let mut events = Vec::new(); - if self.pending_remove(seq) { - let is_late = self - .highest_received_seq - .is_some_and(|h| seq_before(seq, h)); - self.update_highest(seq); - self.completed.insert(seq); - events.push(if is_late { - ExpectedEvent::LateReply(seq) - } else { - ExpectedEvent::EchoReply(seq) - }); - } else if self.completed.contains(&seq) { - self.update_highest(seq); - events.push(ExpectedEvent::DuplicateReply(seq)); - } else if self.timed_out.remove(&seq) { - self.update_highest(seq); - self.completed.insert(seq); - events.push(ExpectedEvent::LateReply(seq)); - } else if self - .highest_received_seq - .is_some_and(|h| seq_before(seq, h)) - { - events.push(ExpectedEvent::LateReply(seq)); - } else { - events.push(ExpectedEvent::WarnUntracked); - } - if close { - self.state = ModelState::ClosedPeer; - events.push(ExpectedEvent::SessionClosed); - } - events - } - } - - fn classify_real(events: &[ClientEvent]) -> Vec { - events - .iter() - .map(|event| match event { - ClientEvent::EchoReply { seq, .. } => ExpectedEvent::EchoReply(*seq), - ClientEvent::LateReply { seq, .. } => ExpectedEvent::LateReply(*seq), - ClientEvent::DuplicateReply { seq, .. } => ExpectedEvent::DuplicateReply(*seq), - ClientEvent::Warning { - kind: WarningKind::WrongToken, - .. - } => ExpectedEvent::WarnWrongToken, - ClientEvent::Warning { - kind: WarningKind::UntrackedReply, - .. - } => ExpectedEvent::WarnUntracked, - ClientEvent::SessionClosed { .. } => ExpectedEvent::SessionClosed, - other => panic!("unexpected event from an open session: {other:?}"), - }) - .collect() - } - - fn is_already_closed(result: &Result) -> bool { - matches!(result, Err(ClientError::AlreadyClosed)) - } - - #[derive(Debug, Clone, Copy, PartialEq, Eq)] - struct Observable { - is_open: bool, - is_terminal: bool, - is_peer_closed: bool, - packets_sent: u64, - pending_is_empty: bool, - } - - fn observe(machine: &SessionMachine) -> Observable { - Observable { - is_open: machine.is_open(), - is_terminal: machine.is_terminal(), - is_peer_closed: machine.is_peer_closed(), - packets_sent: machine.packets_sent(), - pending_is_empty: machine.pending_is_empty(), - } - } - - fn check_generic_invariants( - machine: &SessionMachine, - model: &Model, - ) -> Result<(), TestCaseError> { - prop_assert_eq!(machine.is_open(), model.state == ModelState::Open); - prop_assert_eq!( - machine.is_terminal(), - matches!( - model.state, - ModelState::ClosedLocal | ModelState::ClosedPeer - ) - ); - prop_assert_eq!( - machine.is_peer_closed(), - model.state == ModelState::ClosedPeer - ); - prop_assert_eq!(machine.packets_sent(), model.packets_sent); - if model.state == ModelState::Open { - prop_assert_eq!(machine.pending_is_empty(), model.pending.is_empty()); - } else { - prop_assert!(machine.pending_is_empty()); - } - Ok(()) - } - - /// Drives the full real send transaction: prepare, preflight, - /// finalize, commit. Mirrors how `Client`/`AsyncClient` drive - /// `SessionMachine` around a real socket send, using `now` for both - /// the pre-send anchor and the post-send `sent_at` sample since no - /// real send elapses time here. - fn do_send( - machine: &mut SessionMachine, - now: ClientTimestamp, - ) -> Result { - let prepared = machine.prepare_probe()?; - let preflight = machine.preflight_probe_commit(&prepared)?; - let commit = machine.finalize_probe_commit(preflight, now)?; - let bytes = prepared.bytes.len(); - Ok(machine.commit_probe_sent(commit, now, bytes)) - } - - #[derive(Debug, Clone, Copy)] - enum ReplyTarget { - Pending(u32), - TimedOut(u32), - Completed(u32), - Unknown(u32), - } - - #[derive(Debug, Clone, Copy)] - enum Op { - Send, - StaleProbe, - PrepareAbandon, - Reply { - target: ReplyTarget, - wrong_token: bool, - close: bool, - }, - PollTimeouts, - PollTimeoutsBounded(u8), - AdvanceClock(u16), - LocalClose, - ReopenAttempt, - } - - fn op_strategy() -> impl Strategy { - let reply_target = prop_oneof![ - any::().prop_map(ReplyTarget::Pending), - any::().prop_map(ReplyTarget::TimedOut), - any::().prop_map(ReplyTarget::Completed), - any::().prop_map(ReplyTarget::Unknown), - ]; - prop_oneof![ - 4 => Just(Op::Send), - 1 => Just(Op::StaleProbe), - 1 => Just(Op::PrepareAbandon), - 5 => (reply_target, any::(), any::()).prop_map( - |(target, wrong_token, close)| Op::Reply { - target, - wrong_token, - close, - } - ), - 2 => Just(Op::PollTimeouts), - 1 => (0u8..=6).prop_map(Op::PollTimeoutsBounded), - 4 => (0u16..=120).prop_map(Op::AdvanceClock), - 1 => Just(Op::LocalClose), - 1 => Just(Op::ReopenAttempt), - ] - } - - /// Biased toward `0` (the common case) and toward the last few - /// values before `u32::MAX` (so a handful of sends from here cross - /// the wraparound boundary), with a uniformly random third case for - /// general coverage. - fn start_seq_strategy() -> impl Strategy { - prop_oneof![ - 2 => Just(0u32), - 3 => (u32::MAX - 4)..=u32::MAX, - 2 => any::(), - ] - } - - fn apply_op( - machine: &mut SessionMachine, - model: &mut Model, - params: &Params, - now: &mut ClientTimestamp, - op: Op, - ) -> Result<(), TestCaseError> { - match op { - Op::Send => { - let result = do_send(machine, *now); - match model.state { - ModelState::Open => { - let sent = result.expect( - "open session send should succeed within chosen capacity bounds", - ); - prop_assert_eq!(sent.seq, model.next_wire_seq); - let timeout_at = now - .mono - .checked_add(PROBE_TIMEOUT) - .expect("no overflow at test timescales"); - model.apply_send(sent.seq, timeout_at); - } - _ => prop_assert!( - is_already_closed(&result), - "expected AlreadyClosed, got {:?}", - result - ), - } - } - Op::StaleProbe => { - if model.state != ModelState::Open { - let result = machine.prepare_probe(); - prop_assert!( - is_already_closed(&result), - "expected AlreadyClosed, got {:?}", - result - ); - } else { - let prepared_a = machine.prepare_probe().unwrap(); - prop_assert_eq!(prepared_a.seq, model.next_wire_seq); - - let sent = do_send(machine, *now).expect( - "intervening send should succeed within chosen capacity bounds", - ); - prop_assert_eq!(sent.seq, prepared_a.seq); - let timeout_at = now - .mono - .checked_add(PROBE_TIMEOUT) - .expect("no overflow at test timescales"); - model.apply_send(sent.seq, timeout_at); - - match machine.preflight_probe_commit(&prepared_a) { - Err(ClientError::StalePreparedProbe { - prepared_seq, - next_wire_seq, - }) => { - prop_assert_eq!(prepared_seq, prepared_a.seq); - prop_assert_eq!(next_wire_seq, model.next_wire_seq); - } - other => { - prop_assert!(false, "expected StalePreparedProbe, got {:?}", other) - } - } - } - } - Op::PrepareAbandon => { - let before = observe(machine); - let result = machine.prepare_probe(); - match model.state { - ModelState::Open => { - let prepared = result.unwrap(); - prop_assert_eq!(prepared.seq, model.next_wire_seq); - prop_assert!(!prepared.bytes.is_empty()); - } - _ => prop_assert!( - is_already_closed(&result), - "expected AlreadyClosed, got {:?}", - result - ), - } - prop_assert_eq!(observe(machine), before); - } - Op::Reply { - target, - wrong_token, - close, - } => { - let seq = model.resolve(target); - let token = if wrong_token { WRONG_TOKEN } else { TOKEN }; - let packet = build_echo_reply(params, token, seq, close); - let result = - machine.process_received_echo_packet(&packet, *now, ReceiveMeta::default()); - match model.state { - ModelState::Open => { - let events = result.expect("open session processes echo packets"); - let expected = model.reply(seq, wrong_token, close); - prop_assert_eq!(classify_real(&events), expected); - } - _ => prop_assert!( - is_already_closed(&result), - "expected AlreadyClosed, got {:?}", - result - ), - } - } - Op::PollTimeouts => { - let result = machine.poll_timeouts_at(now.mono); - match model.state { - ModelState::Open => { - let events = result.expect("open session polls succeed"); - let (expired, _more_due) = model.apply_poll(now.mono, usize::MAX); - let real: Vec = events - .iter() - .map(|event| match event { - ClientEvent::EchoLoss { seq, .. } => *seq, - other => { - panic!("unexpected event from poll_timeouts_at: {other:?}") - } - }) - .collect(); - prop_assert_eq!(real, expired); - } - _ => prop_assert!( - is_already_closed(&result), - "expected AlreadyClosed, got {:?}", - result - ), - } - } - Op::PollTimeoutsBounded(limit) => { - let result = machine.poll_timeouts_bounded_at(now.mono, limit as usize); - match model.state { - ModelState::Open => { - let batch = result.expect("open session polls succeed"); - let (expired, more_due) = model.apply_poll(now.mono, limit as usize); - let real: Vec = batch - .events - .iter() - .map(|event| match event { - ClientEvent::EchoLoss { seq, .. } => *seq, - other => panic!( - "unexpected event from poll_timeouts_bounded_at: {other:?}" - ), - }) - .collect(); - prop_assert_eq!(real, expired); - prop_assert_eq!(batch.more_due, more_due); - } - _ => prop_assert!( - is_already_closed(&result), - "expected AlreadyClosed, got {:?}", - result - ), - } - } - Op::AdvanceClock(millis) => { - let delta = Duration::from_millis(u64::from(millis)); - now.mono += delta; - now.wall += delta; - } - Op::LocalClose => match model.state { - ModelState::Open => { - let PreparedClose { commit, .. } = machine - .prepare_close() - .expect("open session can prepare close"); - let event = machine.commit_local_close(commit, *now); - let is_session_closed = matches!(event, ClientEvent::SessionClosed { .. }); - prop_assert!(is_session_closed); - model.state = ModelState::ClosedLocal; - } - _ => { - let result = machine.prepare_close(); - prop_assert!( - is_already_closed(&result), - "expected AlreadyClosed, got {:?}", - result - ); - } - }, - Op::ReopenAttempt => { - let before = observe(machine); - let result = machine.prepare_open_request(); - match model.state { - ModelState::Open => { - prop_assert!(matches!(result, Err(ClientError::AlreadyOpen))); - } - ModelState::ClosedLocal | ModelState::ClosedPeer => { - prop_assert!(is_already_closed(&result)); - } - } - prop_assert_eq!(observe(machine), before); - } - } - check_generic_invariants(machine, model) - } - - proptest! { - #![proptest_config(ProptestConfig::with_cases(CASES))] - - /// Drives a generated sequence of operations through both a real, - /// already-`Open` `SessionMachine` and an independent reference - /// [`Model`], asserting after every operation that the observable - /// state (open/terminal/peer-closed/packets-sent/pending-empty) - /// matches, every error returned matches the model's - /// expectation, and every event/classification produced for a - /// probe send, a reply, or a timeout poll matches the model's - /// prediction. No operation should ever panic. - #[test] - fn open_session_matches_reference_model( - start_seq in start_seq_strategy(), - ops in proptest::collection::vec(op_strategy(), 1..=OP_LIMIT), - ) { - let (mut machine, params) = open_machine_for_model(start_seq); - let mut model = Model::new(start_seq); - let mut now = ClientTimestamp { - mono: Instant::now(), - wall: SystemTime::now(), - }; - - check_generic_invariants(&machine, &model)?; - for op in ops { - apply_op(&mut machine, &mut model, ¶ms, &mut now, op)?; - } - } - } - } } diff --git a/crates/irtt-client/src/socket.rs b/crates/irtt-client/src/socket.rs index 9a15b1a..1f32c29 100644 --- a/crates/irtt-client/src/socket.rs +++ b/crates/irtt-client/src/socket.rs @@ -36,9 +36,6 @@ pub(crate) fn validate_open_timeouts(timeouts: &[Duration]) -> Result<(), Client } pub(crate) fn resolve_remote(config: &ClientConfig) -> Result { - #[cfg(all(test, feature = "tokio"))] - SYNC_RESOLVER_CALLS.with(|calls| calls.set(calls.get() + 1)); - let addr = normalize_server_addr(&config.server_addr); let mut addrs = addr .to_socket_addrs() @@ -57,9 +54,6 @@ pub(crate) async fn resolve_remote_tokio(config: &ClientConfig) -> Result bool { && (!config.socket_config.ipv6_only || remote.is_ipv6()) } -#[cfg(all(test, feature = "tokio"))] -std::thread_local! { - static SYNC_RESOLVER_CALLS: std::cell::Cell = const { std::cell::Cell::new(0) }; - static TOKIO_DNS_LOOKUPS: std::cell::Cell = const { std::cell::Cell::new(0) }; -} - -#[cfg(all(test, feature = "tokio"))] -pub(crate) fn resolution_call_counts() -> (usize, usize) { - ( - SYNC_RESOLVER_CALLS.with(std::cell::Cell::get), - TOKIO_DNS_LOOKUPS.with(std::cell::Cell::get), - ) -} - pub(crate) fn normalize_server_addr(addr: &str) -> String { if addr.parse::().is_ok() { return addr.to_owned(); @@ -280,7 +260,7 @@ mod tests { #[cfg(feature = "tokio")] #[test] - fn tokio_resolver_filters_families_without_using_blocking_resolution() { + fn tokio_resolver_filters_address_families() { let mut config = ClientConfig { server_addr: "localhost:2112".to_owned(), ..ClientConfig::default() @@ -288,13 +268,9 @@ mod tests { config.socket_config.ipv4_only = true; tokio_runtime().block_on(async { - let before = resolution_call_counts(); let remote = resolve_remote_tokio(&config).await.unwrap(); - let after = resolution_call_counts(); assert!(remote.is_ipv4()); - assert_eq!(after.0, before.0); - assert_eq!(after.1, before.1 + 1); }); } @@ -313,24 +289,6 @@ mod tests { }); } - #[cfg(feature = "tokio")] - #[test] - fn tokio_resolver_literal_path_bypasses_all_name_resolution() { - let config = ClientConfig { - server_addr: "127.0.0.1:2112".to_owned(), - ..ClientConfig::default() - }; - - tokio_runtime().block_on(async { - let before = resolution_call_counts(); - let remote = resolve_remote_tokio(&config).await.unwrap(); - let after = resolution_call_counts(); - - assert_eq!(remote, SocketAddr::from(([127, 0, 0, 1], 2112))); - assert_eq!(after, before); - }); - } - #[cfg(feature = "tokio")] #[test] fn tokio_resolver_rejects_disallowed_literal_family() { diff --git a/crates/irtt-client/src/socket_options.rs b/crates/irtt-client/src/socket_options.rs index 74f6e38..74e1100 100644 --- a/crates/irtt-client/src/socket_options.rs +++ b/crates/irtt-client/src/socket_options.rs @@ -140,52 +140,15 @@ pub(crate) fn apply_ttl_to_socket( ttl: u32, ) -> Result<(), ClientError> { validate_ttl(ttl)?; - set_socket_ttl(socket, remote, ttl).map_err(|source| ClientError::SocketOption { - operation: "set TTL/hop limit", - remote, - source, - }) -} - -#[cfg(test)] -pub(crate) fn socket_traffic_class( - socket: &UdpSocket, - remote: SocketAddr, -) -> Result { - get_socket_traffic_class(socket, remote).map_err(|source| ClientError::SocketOption { - operation: "read DSCP socket option", - remote, - source, - }) -} - -#[cfg(all(test, feature = "tokio"))] -pub(crate) fn tokio_socket_traffic_class( - socket: &tokio::net::UdpSocket, - remote: SocketAddr, -) -> Result { - get_socket_traffic_class_ref(SockRef::from(socket), remote).map_err(|source| { + set_socket_ttl_ref(SockRef::from(socket), remote, ttl).map_err(|source| { ClientError::SocketOption { - operation: "read DSCP socket option", + operation: "set TTL/hop limit", remote, source, } }) } -#[cfg(test)] -pub(crate) fn socket_ttl(socket: &UdpSocket, remote: SocketAddr) -> Result { - get_socket_ttl(socket, remote).map_err(|source| ClientError::SocketOption { - operation: "read TTL/hop limit socket option", - remote, - source, - }) -} - -fn set_socket_ttl(socket: &UdpSocket, remote: SocketAddr, ttl: u32) -> io::Result<()> { - set_socket_ttl_ref(SockRef::from(socket), remote, ttl) -} - fn set_socket_ttl_ref(socket: SockRef<'_>, remote: SocketAddr, ttl: u32) -> io::Result<()> { if remote.is_ipv4() { socket.set_ttl_v4(ttl) @@ -194,23 +157,13 @@ fn set_socket_ttl_ref(socket: SockRef<'_>, remote: SocketAddr, ttl: u32) -> io:: } } -#[cfg(test)] -fn get_socket_ttl(socket: &UdpSocket, remote: SocketAddr) -> io::Result { - let socket = SockRef::from(socket); - if remote.is_ipv4() { - socket.ttl_v4() - } else { - socket.unicast_hops_v6() - } -} - fn set_socket_traffic_class( socket: &UdpSocket, remote: SocketAddr, traffic_class: u32, operation: &'static str, ) -> Result<(), ClientError> { - set_socket_traffic_class_io(socket, remote, traffic_class).map_err(|source| { + set_socket_traffic_class_ref(SockRef::from(socket), remote, traffic_class).map_err(|source| { ClientError::SocketOption { operation, remote, @@ -219,14 +172,6 @@ fn set_socket_traffic_class( }) } -fn set_socket_traffic_class_io( - socket: &UdpSocket, - remote: SocketAddr, - traffic_class: u32, -) -> io::Result<()> { - set_socket_traffic_class_ref(SockRef::from(socket), remote, traffic_class) -} - #[cfg(feature = "tokio")] fn set_tokio_socket_traffic_class( socket: &tokio::net::UdpSocket, @@ -335,302 +280,3 @@ fn unsupported_traffic_class(traffic_class: u32, feature: &'static str) -> io::R )) } } - -#[cfg(all( - test, - not(any( - target_os = "fuchsia", - target_os = "redox", - target_os = "solaris", - target_os = "illumos", - target_os = "haiku", - )) -))] -fn get_ipv4_traffic_class(socket: SockRef<'_>) -> io::Result { - socket.tos_v4() -} - -#[cfg(all( - test, - any( - target_os = "fuchsia", - target_os = "redox", - target_os = "solaris", - target_os = "illumos", - target_os = "haiku", - ) -))] -fn get_ipv4_traffic_class(_socket: SockRef<'_>) -> io::Result { - unsupported_readback("IPv4 DSCP socket option readback") -} - -#[cfg(all( - test, - any( - target_os = "android", - target_os = "dragonfly", - target_os = "freebsd", - target_os = "fuchsia", - target_os = "linux", - target_os = "macos", - target_os = "netbsd", - target_os = "openbsd", - target_os = "cygwin", - ) -))] -fn get_ipv6_traffic_class(socket: SockRef<'_>) -> io::Result { - socket.tclass_v6() -} - -#[cfg(all( - test, - not(any( - target_os = "android", - target_os = "dragonfly", - target_os = "freebsd", - target_os = "fuchsia", - target_os = "linux", - target_os = "macos", - target_os = "netbsd", - target_os = "openbsd", - target_os = "cygwin", - )) -))] -fn get_ipv6_traffic_class(_socket: SockRef<'_>) -> io::Result { - unsupported_readback("IPv6 DSCP socket option readback") -} - -#[cfg(test)] -fn get_socket_traffic_class(socket: &UdpSocket, remote: SocketAddr) -> io::Result { - get_socket_traffic_class_ref(SockRef::from(socket), remote) -} - -#[cfg(test)] -fn get_socket_traffic_class_ref(socket: SockRef<'_>, remote: SocketAddr) -> io::Result { - if remote.is_ipv4() { - get_ipv4_traffic_class(socket) - } else { - get_ipv6_traffic_class(socket) - } -} - -#[cfg(all( - test, - any( - target_os = "fuchsia", - target_os = "redox", - target_os = "solaris", - target_os = "illumos", - target_os = "haiku", - not(any( - target_os = "android", - target_os = "dragonfly", - target_os = "freebsd", - target_os = "fuchsia", - target_os = "linux", - target_os = "macos", - target_os = "netbsd", - target_os = "openbsd", - target_os = "cygwin", - )), - ) -))] -fn unsupported_readback(feature: &'static str) -> io::Result { - Err(io::Error::new( - io::ErrorKind::Unsupported, - format!("{feature} is unsupported on this target"), - )) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn converts_dscp_codepoint_to_traffic_class_byte() { - for (dscp, traffic_class) in [(0, 0), (1, 4), (46, 184), (63, 252)] { - let value = dscp_codepoint_to_traffic_class(dscp).unwrap(); - assert_eq!(value, traffic_class); - assert_eq!(value & 0b11, 0); - } - } - - #[test] - fn rejects_out_of_range_dscp_codepoint() { - assert!(matches!( - dscp_codepoint_to_traffic_class(64), - Err(ClientError::InvalidConfig { .. }) - )); - } - - #[test] - fn validates_ttl_range() { - assert!(validate_ttl(1).is_ok()); - assert!(validate_ttl(64).is_ok()); - assert!(validate_ttl(255).is_ok()); - assert!(matches!( - validate_ttl(0), - Err(ClientError::InvalidConfig { .. }) - )); - assert!(matches!( - validate_ttl(256), - Err(ClientError::InvalidConfig { .. }) - )); - } - - #[test] - fn ipv4_socket_option_sets_ttl() { - let remote = SocketAddr::from(([127, 0, 0, 1], 9)); - let socket = UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap(); - socket.connect(remote).unwrap(); - - apply_ttl_to_socket(&socket, remote, 64).unwrap(); - assert_eq!(socket_ttl(&socket, remote).unwrap(), 64); - - apply_ttl_to_socket(&socket, remote, 1).unwrap(); - assert_eq!(socket_ttl(&socket, remote).unwrap(), 1); - } - - #[test] - fn ipv6_socket_option_sets_unicast_hop_limit() { - let remote = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], 9)); - let Some(socket) = bind_connected_ipv6_loopback(remote) else { - return; - }; - - apply_ttl_to_socket(&socket, remote, 64).unwrap(); - let ttl = match socket_ttl(&socket, remote) { - Ok(ttl) => ttl, - Err(error) if is_unsupported_socket_readback(&error) => { - eprintln!("skipping IPv6 hop-limit readback test: {error}"); - return; - } - Err(error) => panic!("{error}"), - }; - assert_eq!(ttl, 64); - - apply_ttl_to_socket(&socket, remote, 1).unwrap(); - assert_eq!(socket_ttl(&socket, remote).unwrap(), 1); - } - - #[test] - #[cfg(not(any( - target_os = "fuchsia", - target_os = "redox", - target_os = "solaris", - target_os = "illumos", - target_os = "haiku", - )))] - fn ipv4_socket_option_sets_and_clears_traffic_class() { - let remote = SocketAddr::from(([127, 0, 0, 1], 9)); - let socket = UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap(); - socket.connect(remote).unwrap(); - - // 184 is the raw wire TOS/Traffic Class byte for DSCP codepoint 46 - // (EF); applying it directly must not shift it again. - apply_traffic_class_to_socket(&socket, remote, 184).unwrap(); - assert_eq!(socket_traffic_class(&socket, remote).unwrap(), 184); - - clear_dscp_on_socket(&socket, remote).unwrap(); - assert_eq!(socket_traffic_class(&socket, remote).unwrap(), 0); - } - - #[cfg(all( - feature = "tokio", - not(any( - target_os = "fuchsia", - target_os = "redox", - target_os = "solaris", - target_os = "illumos", - target_os = "haiku", - )) - ))] - #[test] - fn tokio_ipv4_socket_option_sets_and_clears_traffic_class() { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_io() - .build() - .unwrap(); - - runtime.block_on(async { - let remote = SocketAddr::from(([127, 0, 0, 1], 9)); - let socket = UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap(); - socket.connect(remote).unwrap(); - socket.set_nonblocking(true).unwrap(); - let socket = tokio::net::UdpSocket::from_std(socket).unwrap(); - - apply_traffic_class_to_tokio_socket(&socket, remote, 184).unwrap(); - assert_eq!(tokio_socket_traffic_class(&socket, remote).unwrap(), 184); - - clear_dscp_on_tokio_socket(&socket, remote).unwrap(); - assert_eq!(tokio_socket_traffic_class(&socket, remote).unwrap(), 0); - }); - } - - #[test] - #[cfg(any( - target_os = "android", - target_os = "dragonfly", - target_os = "freebsd", - target_os = "fuchsia", - target_os = "linux", - target_os = "macos", - target_os = "netbsd", - target_os = "openbsd", - target_os = "cygwin", - ))] - fn ipv6_socket_option_sets_and_clears_traffic_class() { - let remote = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], 9)); - let Some(socket) = bind_connected_ipv6_loopback(remote) else { - return; - }; - - apply_traffic_class_to_socket(&socket, remote, 184).unwrap(); - let traffic_class = match socket_traffic_class(&socket, remote) { - Ok(traffic_class) => traffic_class, - Err(error) if is_unsupported_socket_readback(&error) => { - eprintln!("skipping IPv6 traffic-class readback test: {error}"); - return; - } - Err(error) => panic!("{error}"), - }; - assert_eq!(traffic_class, 184); - - clear_dscp_on_socket(&socket, remote).unwrap(); - assert_eq!(socket_traffic_class(&socket, remote).unwrap(), 0); - } - - fn bind_connected_ipv6_loopback(remote: SocketAddr) -> Option { - let socket = match UdpSocket::bind(SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], 0))) { - Ok(socket) => socket, - Err(error) if is_unavailable_ipv6_loopback(&error) => { - eprintln!("skipping IPv6 socket option test: IPv6 loopback unavailable: {error}"); - return None; - } - Err(error) => panic!("{error}"), - }; - match socket.connect(remote) { - Ok(()) => Some(socket), - Err(error) if is_unavailable_ipv6_loopback(&error) => { - eprintln!("skipping IPv6 socket option test: IPv6 loopback unavailable: {error}"); - None - } - Err(error) => panic!("{error}"), - } - } - - fn is_unavailable_ipv6_loopback(error: &io::Error) -> bool { - matches!( - error.kind(), - io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported - ) - } - - fn is_unsupported_socket_readback(error: &ClientError) -> bool { - matches!( - error, - ClientError::SocketOption { source, .. } - if matches!(source.kind(), io::ErrorKind::Unsupported) - ) - } -}