From 20960e1b8dea114e4e47295270260bcd7790d6cd Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 4 Oct 2026 17:30:18 +0700 Subject: [PATCH] refactor(client): make configuration reusable across endpoints --- crates/irtt-app/src/shared/client/config.rs | 46 +++-- crates/irtt-app/src/shared/client/mod.rs | 2 +- crates/irtt-client/README.md | 32 +++ .../irtt-client/examples/blocking_client.rs | 28 +-- crates/irtt-client/examples/managed_client.rs | 14 +- crates/irtt-client/examples/tokio_client.rs | 25 ++- crates/irtt-client/src/async_client.rs | 25 ++- crates/irtt-client/src/client.rs | 58 ++++-- crates/irtt-client/src/config.rs | 186 +++++++++++------- crates/irtt-client/src/lib.rs | 5 +- crates/irtt-client/src/managed/model.rs | 2 + crates/irtt-client/src/managed/task.rs | 29 ++- crates/irtt-client/src/session/machine.rs | 77 ++++---- crates/irtt-client/src/session/negotiation.rs | 4 +- crates/irtt-client/src/socket.rs | 70 +++---- crates/irtt-client/tests/client/close.rs | 94 ++++----- crates/irtt-client/tests/client/config.rs | 61 ++++-- crates/irtt-client/tests/client/hmac.rs | 15 +- crates/irtt-client/tests/client/no_test.rs | 49 +++-- crates/irtt-client/tests/client/open.rs | 89 +++++---- .../client/probes_replies/packet_length.rs | 42 ++-- .../probes_replies/reply_classification.rs | 26 +-- .../client/probes_replies/reply_decode.rs | 26 +-- .../client/probes_replies/reply_metadata.rs | 13 +- .../tests/client/probes_replies/timeouts.rs | 42 ++-- crates/irtt-client/tests/client/support.rs | 21 +- crates/irtt-client/tests/client/ttl.rs | 12 +- crates/irtt-client/tests/linux_transport.rs | 45 +++-- .../tests/async_client.rs | 32 +-- .../tests/async_client_lifecycle.rs | 107 ++++++---- .../tests/blocking_managed.rs | 17 +- .../tests/fill_negotiation.rs | 32 ++- .../tests/fill_wire_bytes.rs | 33 ++-- crates/irtt-integration-tests/tests/hmac.rs | 69 ++++--- .../tests/large_packets.rs | 15 +- .../tests/managed_behavior.rs | 26 ++- .../tests/managed_lifecycle.rs | 24 ++- .../tests/negotiation_policy.rs | 69 ++++--- .../tests/protocol_options.rs | 23 ++- .../tests/protocol_options/backend_smoke.rs | 24 ++- .../tests/support/mod.rs | 58 +++--- .../tests/traffic_class.rs | 34 ++-- .../tests/wildcard_source.rs | 34 ++-- 43 files changed, 1012 insertions(+), 723 deletions(-) diff --git a/crates/irtt-app/src/shared/client/config.rs b/crates/irtt-app/src/shared/client/config.rs index 311f0eb..35301e6 100644 --- a/crates/irtt-app/src/shared/client/config.rs +++ b/crates/irtt-app/src/shared/client/config.rs @@ -1,41 +1,45 @@ use std::time::Duration; -use irtt_client::{Authentication, ClientConfig, HmacKey, NegotiationPolicy, SocketConfig}; +use irtt_client::{ + Authentication, ClientConfig, HmacKey, NegotiationPolicy, OpenPolicy, SessionRequest, + SocketConfig, +}; use super::args::CommonClientArgs; -pub const DEFAULT_RECV_TIMEOUT: Duration = Duration::from_millis(20); - impl CommonClientArgs { /// Build the shared client configuration template for a managed run. /// - /// The template carries no server address: the managed driver replaces it - /// per target from that target's configuration, so deriving one here from - /// an arbitrary target would be dead data. + /// The config is endpoint-independent; managed targets own their addresses. pub fn to_client_config(&self, duration: Duration) -> ClientConfig { ClientConfig { - duration: (!duration.is_zero()).then_some(duration), - interval: self.interval, - length: self.length, - received_stats: self.stats.into(), - stamp_at: self.tstamp.into(), - clock: self.clock.into(), - dscp: self.dscp, + open: OpenPolicy { + negotiation: if self.loose { + NegotiationPolicy::Loose + } else { + NegotiationPolicy::Strict + }, + ..Default::default() + }, + request: SessionRequest { + duration: (!duration.is_zero()).then_some(duration), + interval: self.interval, + length: self.length, + received_stats: self.stats.into(), + stamp_at: self.tstamp.into(), + clock: self.clock.into(), + dscp: self.dscp, + server_fill: self.server_fill.clone(), + ..Default::default() + }, auth: self .hmac .as_ref() .map_or(Authentication::Unauthenticated, |key| { Authentication::Hmac(HmacKey::new(key.as_bytes())) }), - server_fill: self.server_fill.clone(), - negotiation_policy: if self.loose { - NegotiationPolicy::Loose - } else { - NegotiationPolicy::Strict - }, - socket_config: SocketConfig { + socket: SocketConfig { ttl: self.ttl, - recv_timeout: Some(DEFAULT_RECV_TIMEOUT), ..SocketConfig::default() }, ..ClientConfig::default() diff --git a/crates/irtt-app/src/shared/client/mod.rs b/crates/irtt-app/src/shared/client/mod.rs index 27a2cf4..e72a847 100644 --- a/crates/irtt-app/src/shared/client/mod.rs +++ b/crates/irtt-app/src/shared/client/mod.rs @@ -8,7 +8,7 @@ pub use args::{ parse_dscp, parse_duration, parse_length, parse_server_fill, parse_test_duration, parse_ttl, ClockArg, CommonClientArgs, ReceivedStatsArg, TimestampArg, }; -pub use config::{expected_probe_count, DEFAULT_RECV_TIMEOUT}; +pub use config::expected_probe_count; pub use prepare::{ prepare_managed_run, ManagedRunSetup, TargetSelection, MANAGED_EVENT_CAPACITY, STDIN_MAX_DESIRED_TARGETS, STDIN_MAX_LIVE_TARGET_GENERATIONS, STDIN_OUTCOME_HISTORY_LIMIT, diff --git a/crates/irtt-client/README.md b/crates/irtt-client/README.md index 7e496fd..fe4bc92 100644 --- a/crates/irtt-client/README.md +++ b/crates/irtt-client/README.md @@ -34,6 +34,38 @@ dependency at all. See `examples/` in the repository for runnable examples of each tier. +## Reusable configuration + +`ClientConfig` contains no endpoint. The low-level APIs take it separately: +`Client::connect(endpoint, config)` and +`AsyncClient::connect(endpoint, config).await`, accepting `String` or `&str` +endpoints. An omitted port defaults to 2112. + +Configuration is grouped by ownership: + +- `address_family`: `AddressFamily::{Any, Ipv4, Ipv6}`, defaulting to `Any`. + It filters remote resolution; `Ipv6` also sets `IPV6_V6ONLY`. +- `socket`: `SocketConfig` for local bind, device/routing options, and TTL. +- `request`: `SessionRequest` for duration, interval, length, received stats, + timestamps, clock, DSCP, server fill, and run mode. +- `open`: `OpenPolicy` for attempt `timeouts` and `negotiation` policy. +- `auth`, `probe_timeout`, and `max_pending_probes`: authentication and local + probe tracking policy. + +Clone one config to connect to multiple endpoints. Managed targets retain their +own ID, endpoint, and authentication inheritance/override; shared +`ManagedClientConfig.client` is reusable without endpoint mutation. + +Blocking receive policy belongs to `Client`: +`client.set_recv_timeout(Some(duration))?` sets it before or after opening, +and `client.set_recv_timeout(None)?` restores the default indefinite wait. +Open attempts use only `open.timeouts`, restoring the adapter setting afterward. +Async and managed clients have no blocking receive-timeout setting. + +This breaks the former flat config, endpoint-in-config, socket family flags, +and `SocketConfig.recv_timeout` APIs. Use the grouped fields and adapter method +directly; defaults and protocol/runtime behavior are preserved. + ## Authentication `ClientConfig.auth` is concrete: `Authentication::Unauthenticated` or diff --git a/crates/irtt-client/examples/blocking_client.rs b/crates/irtt-client/examples/blocking_client.rs index 055c648..f248463 100644 --- a/crates/irtt-client/examples/blocking_client.rs +++ b/crates/irtt-client/examples/blocking_client.rs @@ -12,32 +12,33 @@ //! ``` //! //! Without a reachable server this exits quickly with an open-timeout error -//! instead of hanging, because the example shortens `open_timeouts` for a +//! instead of hanging, because the example shortens `open.timeouts` for a //! fast demonstration; a real caller would normally keep the default retry //! schedule. use std::time::{Duration, Instant}; -use irtt_client::{Client, ClientConfig, ClientEvent, OpenOutcome}; +use irtt_client::{Client, ClientConfig, ClientEvent, OpenOutcome, OpenPolicy, SessionRequest}; const RUN_DURATION: Duration = Duration::from_secs(2); fn main() { let config = ClientConfig { - server_addr: "127.0.0.1:2112".to_owned(), - duration: Some(RUN_DURATION), - interval: Duration::from_millis(200), - // A single short attempt keeps this example fast when no server is - // reachable; production callers should generally keep the default. - open_timeouts: vec![Duration::from_millis(300)], - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(100)), + open: OpenPolicy { + // A single short attempt keeps this example fast when no server is + // reachable; production callers should generally keep the default. + timeouts: vec![Duration::from_millis(300)], + ..Default::default() + }, + request: SessionRequest { + duration: Some(RUN_DURATION), + interval: Duration::from_millis(200), ..Default::default() }, ..Default::default() }; - let mut client = match Client::connect(config) { + let mut client = match Client::connect("127.0.0.1:2112", config) { Ok(client) => client, Err(err) => { eprintln!("failed to prepare client socket: {err}"); @@ -45,6 +46,11 @@ fn main() { } }; + if let Err(err) = client.set_recv_timeout(Some(Duration::from_millis(100))) { + eprintln!("failed to configure receive timeout: {err}"); + return; + } + let negotiated = match client.open() { Ok(OpenOutcome::Started { event, negotiated, .. diff --git a/crates/irtt-client/examples/managed_client.rs b/crates/irtt-client/examples/managed_client.rs index 1377442..76e2fdc 100644 --- a/crates/irtt-client/examples/managed_client.rs +++ b/crates/irtt-client/examples/managed_client.rs @@ -22,14 +22,20 @@ use irtt_client::managed::{ BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedEvent, ManagedTargetConfig, }; -use irtt_client::ClientConfig; +use irtt_client::{ClientConfig, OpenPolicy, SessionRequest}; fn main() { let config = ManagedClientConfig { client: ClientConfig { - duration: None, - interval: Duration::from_millis(200), - open_timeouts: vec![Duration::from_millis(300)], + open: OpenPolicy { + timeouts: vec![Duration::from_millis(300)], + ..Default::default() + }, + request: SessionRequest { + duration: None, + interval: Duration::from_millis(200), + ..Default::default() + }, ..Default::default() }, completion: ManagedCompletionPolicy::ExplicitStop, diff --git a/crates/irtt-client/examples/tokio_client.rs b/crates/irtt-client/examples/tokio_client.rs index 74e8ec9..4f71cc3 100644 --- a/crates/irtt-client/examples/tokio_client.rs +++ b/crates/irtt-client/examples/tokio_client.rs @@ -13,12 +13,14 @@ //! ``` //! //! Without a reachable server this exits quickly with an open-timeout error -//! instead of hanging, because the example shortens `open_timeouts` for a +//! instead of hanging, because the example shortens `open.timeouts` for a //! fast demonstration. use std::time::{Duration, Instant}; -use irtt_client::{AsyncClient, ClientConfig, ClientEvent, OpenOutcome}; +use irtt_client::{ + AsyncClient, ClientConfig, ClientEvent, OpenOutcome, OpenPolicy, SessionRequest, +}; /// Upper bound on how long the receive wait can go without also calling /// `poll_timeouts()`. `AsyncClient` does not expose the earliest pending @@ -41,16 +43,21 @@ fn main() { async fn run() { let config = ClientConfig { - server_addr: "127.0.0.1:2112".to_owned(), - duration: Some(Duration::from_secs(2)), - interval: Duration::from_millis(200), - // A single short attempt keeps this example fast when no server is - // reachable; production callers should generally keep the default. - open_timeouts: vec![Duration::from_millis(300)], + open: OpenPolicy { + // A single short attempt keeps this example fast when no server is + // reachable; production callers should generally keep the default. + timeouts: vec![Duration::from_millis(300)], + ..Default::default() + }, + request: SessionRequest { + duration: Some(Duration::from_secs(2)), + interval: Duration::from_millis(200), + ..Default::default() + }, ..Default::default() }; - let mut client = match AsyncClient::connect(config).await { + let mut client = match AsyncClient::connect("127.0.0.1:2112", config).await { Ok(client) => client, Err(err) => { eprintln!("failed to prepare client socket: {err}"); diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index 85d086a..585a476 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -19,7 +19,7 @@ use crate::{ recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, PreparedOpenRequest, PreparedProbe, SessionMachine, TimeoutBatch, MAX_OPEN_PACKET_SIZE, }, - socket::{connect_tokio_udp_socket, resolve_remote_tokio, validate_open_timeouts}, + socket::{connect_tokio_udp_socket, resolve_remote_tokio}, socket_options::{apply_traffic_class_to_tokio_socket, clear_dscp_on_tokio_socket}, timing::ClientTimestamp, }; @@ -170,7 +170,7 @@ impl AsyncOpenState { /// use irtt_client::{AsyncClient, ClientConfig}; /// /// # async fn run() -> Result<(), Box> { -/// let mut client = AsyncClient::connect(ClientConfig::default()).await?; +/// let mut client = AsyncClient::connect("127.0.0.1:2112", ClientConfig::default()).await?; /// let outcome = client.open().await?; /// println!("opened: {outcome:?}"); /// let sent = client.send_probe().await?; @@ -199,19 +199,26 @@ pub struct AsyncClient { } impl AsyncClient { - /// Resolve the configured server and construct one connected Tokio UDP + /// Resolve the endpoint and construct one connected Tokio UDP /// socket. /// /// Polling this future without a current Tokio runtime returns /// [`ClientError::NoTokioRuntime`]. A runtime without enabled I/O or time /// drivers is outside this type's runtime contract. - pub async fn connect(config: ClientConfig) -> Result { + /// + /// The endpoint accepts a name or address; an omitted port defaults to 2112. + /// IPv6 literals may be bracketed or unbracketed with the default port. + /// The reusable config contains no endpoint identity. + pub async fn connect( + endpoint: impl AsRef, + config: ClientConfig, + ) -> Result { tokio::runtime::Handle::try_current().map_err(|_| ClientError::NoTokioRuntime)?; - validate_open_timeouts(&config.open_timeouts)?; - let remote = resolve_remote_tokio(&config).await?; + config.open.validate()?; + let remote = resolve_remote_tokio(endpoint.as_ref(), config.address_family).await?; let machine = SessionMachine::new(config.clone(), remote)?; let prepared_open = machine.prepare_open_request()?; - let socket = connect_tokio_udp_socket(&config.socket_config, remote)?; + let socket = connect_tokio_udp_socket(&config.socket, remote, config.address_family)?; Ok(Self { socket, @@ -307,7 +314,7 @@ impl AsyncClient { return self.poll_open_cleanup(state, cx); } - let attempt_count = self.machine.config().open_timeouts.len(); + let attempt_count = self.machine.config().open.timeouts.len(); if state.attempt >= attempt_count { return Poll::Ready(Err(ClientError::OpenTimeout)); } @@ -316,7 +323,7 @@ impl AsyncClient { return Poll::Ready(Err(ClientError::OpenTimeout)); } if state.deadline.is_none() { - let timeout = self.machine.config().open_timeouts[state.attempt]; + let timeout = self.machine.config().open.timeouts[state.attempt]; if let Err(error) = state.start_attempt(timeout) { return Poll::Ready(Err(error)); } diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 063e08b..f9e8f41 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -13,7 +13,7 @@ use crate::{ recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, ProbeSent, SessionMachine, MAX_OPEN_PACKET_SIZE, }, - socket::{connect_udp_socket, resolve_remote, validate_open_timeouts}, + socket::{connect_udp_socket, resolve_remote}, socket_options::{apply_traffic_class_to_socket, clear_dscp_on_socket}, timing::ClientTimestamp, }; @@ -50,7 +50,7 @@ struct PreparedClientOpenFailure { /// use irtt_client::{Client, ClientConfig}; /// /// # fn run() -> Result<(), Box> { -/// let mut client = Client::connect(ClientConfig::default())?; +/// let mut client = Client::connect("127.0.0.1:2112", ClientConfig::default())?; /// let outcome = client.open()?; /// println!("opened: {outcome:?}"); /// let sent = client.send_probe()?; @@ -70,6 +70,7 @@ pub struct Client { socket: UdpSocket, remote: SocketAddr, recv_buffer: Vec, + recv_timeout: Option, applied_traffic_class: Option, /// Whether the socket's `SO_TIMESTAMPING` flags carry `TX_SOFTWARE` + /// `OPT_ID` + `OPT_TSONLY`. Always `false` off Linux, without the @@ -81,32 +82,53 @@ pub struct Client { } impl Client { - /// Resolve the configured server and create a connected UDP socket. + /// Resolve the endpoint and create a connected UDP socket. /// /// This validates local configuration and prepares the open request, but it /// does not contact the server. Call [`open`](Self::open) to perform the /// IRTT open exchange. - pub fn connect(config: ClientConfig) -> Result { - validate_open_timeouts(&config.open_timeouts)?; - let remote = resolve_remote(&config)?; + /// + /// The endpoint accepts a name or address; an omitted port defaults to 2112. + /// IPv6 literals may be bracketed or unbracketed with the default port. + /// The reusable config contains no endpoint identity. + pub fn connect(endpoint: impl AsRef, config: ClientConfig) -> Result { + config.open.validate()?; + let remote = resolve_remote(endpoint.as_ref(), config.address_family)?; let runtime = SessionMachine::new(config.clone(), remote)?; - let socket = connect_udp_socket(&config.socket_config, remote)?; + let socket = connect_udp_socket(&config.socket, remote, config.address_family)?; Ok(Self { runtime, socket, remote, recv_buffer: vec![0_u8; recv_buffer_size(false, None)?], + recv_timeout: None, applied_traffic_class: None, tx_timestamping_enabled: false, }) } + /// Set the blocking receive timeout used outside open attempts. + /// + /// `None` (the default) lets `recv_once()` block indefinitely. A nonzero + /// timeout bounds one logical receive, including retries after EINTR. + /// `recv_available()` retains its bounded drain behavior. Open attempts + /// use only [`OpenPolicy::timeouts`](crate::OpenPolicy::timeouts) and + /// restore this setting on success or failure. + /// + /// A zero duration is rejected by the socket; the previous setting is + /// retained when applying the new timeout fails. + pub fn set_recv_timeout(&mut self, timeout: Option) -> Result<(), ClientError> { + self.socket.set_read_timeout(timeout)?; + self.recv_timeout = timeout; + Ok(()) + } + /// Perform the IRTT open exchange. /// /// On success, returns the negotiated open outcome and transitions the /// client into either an open probe session or completed no-test state. - /// Open attempts use [`ClientConfig::open_timeouts`]. Malformed or + /// Open attempts use [`OpenPolicy::timeouts`](crate::OpenPolicy::timeouts). Malformed or /// unrelated datagrams are ignored until the current attempt's absolute /// deadline, so one attempt may consume several datagrams without /// retransmitting. Silence or ignored traffic eventually produces @@ -121,9 +143,7 @@ impl Client { pub fn open(&mut self) -> Result { let result = self.open_transaction(); if result.is_err() { - let _ = self - .socket - .set_read_timeout(self.runtime.config().socket_config.recv_timeout); + let _ = self.socket.set_read_timeout(self.recv_timeout); } result } @@ -225,8 +245,7 @@ impl Client { /// Requires an open session; lifecycle errors are returned before socket I/O. pub fn recv_once(&mut self) -> Result, ClientError> { self.runtime.ensure_open()?; - self.socket - .set_read_timeout(self.runtime.config().socket_config.recv_timeout)?; + self.socket.set_read_timeout(self.recv_timeout)?; self.recv_once_inner() } @@ -267,7 +286,7 @@ impl Client { fn recv_datagram_retrying_interrupted( &mut self, ) -> Result, ClientError> { - let configured_timeout = self.runtime.config().socket_config.recv_timeout; + let configured_timeout = self.recv_timeout; let deadline = match configured_timeout { Some(timeout) => Some( Instant::now() @@ -322,8 +341,7 @@ impl Client { /// Requires an open session; lifecycle errors are returned before socket I/O. pub fn recv_available(&mut self, budget: RecvBudget) -> Result, ClientError> { self.runtime.ensure_open()?; - self.socket - .set_read_timeout(self.runtime.config().socket_config.recv_timeout)?; + self.socket.set_read_timeout(self.recv_timeout)?; let mut all_events = Vec::new(); for _ in 0..budget.max_packets { let events = self.recv_once_inner()?; @@ -383,10 +401,10 @@ impl Client { fn open_transaction(&mut self) -> Result { let request = self.runtime.prepare_open_request()?; let mut buf = [0_u8; MAX_OPEN_PACKET_SIZE]; - let attempt_count = self.runtime.config().open_timeouts.len(); + let attempt_count = self.runtime.config().open.timeouts.len(); for attempt in 0..attempt_count { - let timeout = self.runtime.config().open_timeouts[attempt]; + let timeout = self.runtime.config().open.timeouts[attempt]; let deadline = Instant::now() .checked_add(timeout) .ok_or(ClientError::DurationOverflow)?; @@ -470,7 +488,7 @@ impl Client { recv_buffer_len: None, negotiated_traffic_class: None, previous_traffic_class: self.applied_traffic_class, - post_open_recv_timeout: self.runtime.config().socket_config.recv_timeout, + post_open_recv_timeout: self.recv_timeout, }); }; let recv_buffer_len = match recv_buffer_size(self.runtime.has_hmac(), Some(negotiated)) { @@ -495,7 +513,7 @@ impl Client { recv_buffer_len: Some(recv_buffer_len), negotiated_traffic_class: Some(negotiated_traffic_class), previous_traffic_class: self.applied_traffic_class, - post_open_recv_timeout: self.runtime.config().socket_config.recv_timeout, + post_open_recv_timeout: self.recv_timeout, }) } diff --git a/crates/irtt-client/src/config.rs b/crates/irtt-client/src/config.rs index 250c46d..f2fb59a 100644 --- a/crates/irtt-client/src/config.rs +++ b/crates/irtt-client/src/config.rs @@ -1,12 +1,12 @@ use std::{net::SocketAddr, time::Duration}; -use crate::Authentication; +use crate::{Authentication, ClientError}; /// Protocol compatibility bound for a requested `server_fill` value, in UTF-8 /// bytes. /// /// This is the maximum encoded server-fill string accepted by this client and -/// protocol decoder. [`ClientConfig::server_fill`] enforces the same bound +/// protocol decoder. [`SessionRequest::server_fill`] enforces the same bound /// before opening a session. pub use irtt_proto::MAX_SERVER_FILL_BYTES; use irtt_proto::{Clock, ReceivedStats, StampAt}; @@ -28,7 +28,7 @@ pub const MAX_TTL: u32 = 255; /// Largest UDP payload length accepted by client configuration, in bytes. /// /// This is the maximum UDP payload size excluding IP and UDP headers. It caps -/// [`ClientConfig::length`] before protocol packets are encoded or sent. +/// [`SessionRequest::length`] before protocol packets are encoded or sent. pub const MAX_UDP_PAYLOAD_LENGTH: u32 = 65_507; pub(crate) const DEFAULT_DURATION: Duration = Duration::from_secs(3); pub(crate) const DEFAULT_INTERVAL: Duration = Duration::from_secs(1); @@ -58,29 +58,82 @@ pub(crate) const DEFAULT_MAX_PENDING: usize = 4096; /// ``` /// use std::time::Duration; /// -/// use irtt_client::{ClientConfig, NegotiationPolicy}; +/// use irtt_client::{ClientConfig, NegotiationPolicy, OpenPolicy, SessionRequest}; /// /// let config = ClientConfig { -/// server_addr: "example.net:2112".into(), -/// duration: Some(Duration::from_secs(10)), -/// interval: Duration::from_millis(250), -/// length: 1200, -/// negotiation_policy: NegotiationPolicy::Loose, +/// request: SessionRequest { +/// duration: Some(Duration::from_secs(10)), +/// interval: Duration::from_millis(250), +/// length: 1200, +/// ..SessionRequest::default() +/// }, +/// open: OpenPolicy { +/// negotiation: NegotiationPolicy::Loose, +/// ..OpenPolicy::default() +/// }, /// ..ClientConfig::default() /// }; /// -/// assert_eq!(config.duration, Some(Duration::from_secs(10))); -/// assert_eq!(config.interval, Duration::from_millis(250)); -/// assert_eq!(config.length, 1200); +/// assert_eq!(config.request.duration, Some(Duration::from_secs(10))); +/// assert_eq!(config.request.interval, Duration::from_millis(250)); +/// assert_eq!(config.request.length, 1200); /// ``` #[derive(Debug, Clone, PartialEq, Eq)] pub struct ClientConfig { - /// Remote server name or address. + /// Remote address families allowed during resolution and socket creation. + pub address_family: AddressFamily, + /// Local UDP socket properties. + pub socket: SocketConfig, + /// Protocol and session values requested from the server. + pub request: SessionRequest, + /// Open retry timeouts and negotiation policy. + pub open: OpenPolicy, + /// Concrete authentication for open, echo, and close packets. /// - /// If no port is present, the default IRTT port 2112 is used. IPv6 - /// literals may be supplied either bracketed or unbracketed when the - /// default port should be used. - pub server_addr: String, + /// HMAC peers must use the same key. An empty HMAC key remains authenticated. + pub auth: Authentication, + /// Time after sending a probe before the client reports it as lost. + /// + /// This timeout is local client behavior; it is not negotiated with the + /// server. It must be greater than zero. + pub probe_timeout: Duration, + /// Maximum number of probes tracked as pending/timed-out/completed. + /// + /// This bounds memory used for reply classification and must be greater + /// than zero. A very small value can reject sends when replies are still + /// outstanding. + pub max_pending_probes: usize, +} + +impl Default for ClientConfig { + fn default() -> Self { + Self { + address_family: AddressFamily::default(), + socket: SocketConfig::default(), + request: SessionRequest::default(), + open: OpenPolicy::default(), + auth: Authentication::Unauthenticated, + probe_timeout: DEFAULT_PROBE_TIMEOUT, + max_pending_probes: DEFAULT_MAX_PENDING, + } + } +} + +/// Allowed remote address families. IPv6-only selection also sets IPV6_V6ONLY. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum AddressFamily { + /// Accept IPv4 and IPv6 addresses. + #[default] + Any, + /// Accept only IPv4 addresses. + Ipv4, + /// Accept only IPv6 addresses and use IPv6-only sockets. + Ipv6, +} + +/// Requested IRTT protocol and session semantics, reusable across endpoints. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SessionRequest { /// Requested run duration. /// /// `Some(duration)` requests a finite test and must be greater than zero. @@ -120,46 +173,20 @@ pub struct ClientConfig { /// This is the six-bit DSCP value, not the full traffic-class byte, and /// must be less than or equal to [`MAX_DSCP_CODEPOINT`]. pub dscp: u8, - /// Concrete authentication for open, echo, and close packets. - /// - /// HMAC peers must use the same key. An empty HMAC key remains authenticated. - pub auth: Authentication, /// Optional server payload fill request. /// /// `None` leaves server fill behavior unspecified. `Some(value)` requests /// a non-empty server fill mode/value and must not exceed /// [`MAX_SERVER_FILL_BYTES`] bytes when UTF-8 encoded. pub server_fill: Option, - /// Per-attempt receive timeouts used while opening the session. - /// - /// The client sends an open request for each entry until a valid open reply - /// is received. The list must not be empty, and each timeout must be at - /// least 200 ms. - pub open_timeouts: Vec, /// Whether opening the session should start a probe test or perform a /// negotiation-only no-test exchange. pub run_mode: RunMode, - /// Policy for server changes to negotiable protocol parameters. - pub negotiation_policy: NegotiationPolicy, - /// Local UDP socket configuration. - pub socket_config: SocketConfig, - /// Time after sending a probe before the client reports it as lost. - /// - /// This timeout is local client behavior; it is not negotiated with the - /// server. It must be greater than zero. - pub probe_timeout: Duration, - /// Maximum number of probes tracked as pending/timed-out/completed. - /// - /// This bounds memory used for reply classification and must be greater - /// than zero. A very small value can reject sends when replies are still - /// outstanding. - pub max_pending_probes: usize, } -impl Default for ClientConfig { +impl Default for SessionRequest { fn default() -> Self { Self { - server_addr: format!("127.0.0.1:{DEFAULT_PORT}"), duration: Some(DEFAULT_DURATION), interval: DEFAULT_INTERVAL, length: 0, @@ -167,23 +194,55 @@ impl Default for ClientConfig { stamp_at: StampAt::Both, clock: Clock::Both, dscp: 0, - auth: Authentication::Unauthenticated, server_fill: None, - open_timeouts: DEFAULT_OPEN_TIMEOUTS.to_vec(), run_mode: RunMode::Normal, - negotiation_policy: NegotiationPolicy::Strict, - socket_config: SocketConfig::default(), - probe_timeout: DEFAULT_PROBE_TIMEOUT, - max_pending_probes: DEFAULT_MAX_PENDING, + } + } +} + +/// Opening behavior shared by blocking, async, and managed clients. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OpenPolicy { + /// Per-attempt receive timeouts used while opening the session. + /// + /// The client sends an open request for each entry until a valid open reply + /// is received. The list must not be empty, and each timeout must be at + /// least 200 ms. + pub timeouts: Vec, + /// Policy for server changes to negotiable protocol parameters. + pub negotiation: NegotiationPolicy, +} + +impl OpenPolicy { + pub(crate) fn validate(&self) -> Result<(), ClientError> { + if self.timeouts.is_empty() { + return Err(ClientError::NoOpenTimeouts); + } + for timeout in &self.timeouts { + if *timeout < MIN_OPEN_TIMEOUT { + return Err(ClientError::OpenTimeoutTooSmall { + timeout: *timeout, + minimum: MIN_OPEN_TIMEOUT, + }); + } + } + Ok(()) + } +} + +impl Default for OpenPolicy { + fn default() -> Self { + Self { + timeouts: DEFAULT_OPEN_TIMEOUTS.to_vec(), + negotiation: NegotiationPolicy::Strict, } } } /// Local UDP socket options used by [`ClientConfig`]. /// -/// These settings affect how the client binds, resolves, and receives from the -/// socket. They do not change the IRTT protocol parameters negotiated with the -/// server. +/// These settings affect how the client binds and sends from the socket. They +/// do not change the IRTT protocol parameters negotiated with the server. /// /// # Example /// @@ -192,20 +251,20 @@ impl Default for ClientConfig { /// ``` /// use std::net::{Ipv4Addr, SocketAddr}; /// -/// use irtt_client::{ClientConfig, SocketConfig}; +/// use irtt_client::{AddressFamily, ClientConfig, SocketConfig}; /// /// let config = ClientConfig { -/// socket_config: SocketConfig { +/// address_family: AddressFamily::Ipv4, +/// socket: SocketConfig { /// bind_addr: Some(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0))), /// ttl: Some(32), -/// ipv4_only: true, /// ..SocketConfig::default() /// }, /// ..ClientConfig::default() /// }; /// -/// assert_eq!(config.socket_config.ttl, Some(32)); -/// assert!(config.socket_config.ipv4_only); +/// assert_eq!(config.socket.ttl, Some(32)); +/// assert_eq!(config.address_family, AddressFamily::Ipv4); /// ``` #[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct SocketConfig { @@ -229,17 +288,6 @@ pub struct SocketConfig { /// platform socket option range; [`MAX_TTL`] is the public configuration /// bound used by this crate. pub ttl: Option, - /// Restrict name resolution to IPv4 addresses. - pub ipv4_only: bool, - /// Restrict name resolution to IPv6 addresses and set IPV6_V6ONLY for IPv6 - /// sockets. - pub ipv6_only: bool, - /// Socket read timeout used after the session is open. - /// - /// `None` leaves reads blocking for APIs that perform a single receive. - /// Managed sessions may replace `None` or long timeouts with a short - /// timeout so cooperative cancellation can be observed promptly. - pub recv_timeout: Option, } /// How strictly to handle server-side negotiation restrictions. diff --git a/crates/irtt-client/src/lib.rs b/crates/irtt-client/src/lib.rs index 64b6062..cdd636b 100644 --- a/crates/irtt-client/src/lib.rs +++ b/crates/irtt-client/src/lib.rs @@ -53,8 +53,9 @@ pub use async_client::AsyncClient; pub use auth::{Authentication, HmacKey}; pub use client::Client; pub use config::{ - ClientConfig, NegotiationPolicy, RecvBudget, RunMode, SocketConfig, MAX_DSCP_CODEPOINT, - MAX_SERVER_FILL_BYTES, MAX_TTL, MAX_UDP_PAYLOAD_LENGTH, + AddressFamily, ClientConfig, NegotiationPolicy, OpenPolicy, RecvBudget, RunMode, + SessionRequest, SocketConfig, MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_TTL, + MAX_UDP_PAYLOAD_LENGTH, }; pub use error::ClientError; pub use event::{ diff --git a/crates/irtt-client/src/managed/model.rs b/crates/irtt-client/src/managed/model.rs index 68f3999..17e824a 100644 --- a/crates/irtt-client/src/managed/model.rs +++ b/crates/irtt-client/src/managed/model.rs @@ -85,6 +85,8 @@ pub enum ManagedCompletionPolicy { /// Shared configuration for one managed task. #[derive(Clone, Debug, Eq, PartialEq)] pub struct ManagedClientConfig { + /// Reusable session/socket configuration shared across target generations. + /// Each target supplies its endpoint and resolves its authentication policy. pub client: ClientConfig, pub pacing: ManagedPacing, pub completion: ManagedCompletionPolicy, diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index f780a4f..17abf70 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -20,7 +20,6 @@ use tokio::{ use crate::{ async_client::{AsyncClient, AsyncOpenState}, session::machine::SessionMachine, - socket::validate_open_timeouts, socket_options::validate_ttl, ClientConfig, ClientError, ClientEvent, OpenOutcome, }; @@ -62,7 +61,10 @@ const MAX_BROADCAST_CHANNEL_CAPACITY: usize = usize::MAX >> 1; /// # async fn run() -> Result<(), Box> { /// let config = ManagedClientConfig { /// client: ClientConfig { -/// duration: Some(std::time::Duration::from_secs(10)), +/// request: irtt_client::SessionRequest { +/// duration: Some(std::time::Duration::from_secs(10)), +/// ..irtt_client::SessionRequest::default() +/// }, /// ..ClientConfig::default() /// }, /// ..ManagedClientConfig::default() @@ -944,7 +946,6 @@ impl ManagedClientTask { return Err(ManagedCommandApplyError::DuplicateTargetId { id: target.id }); } let mut client_config = self.config.client.clone(); - client_config.server_addr.clone_from(&target.server_addr); client_config.auth = target.auth.resolve(&self.config.client.auth); validate_target_config(&client_config).map_err(|source| { ManagedCommandApplyError::InvalidTarget { @@ -1060,7 +1061,8 @@ impl ManagedClientTask { } fn start_connecting(&mut self, index: usize, config: ClientConfig) { - let future = Box::pin(AsyncClient::connect(config)); + let endpoint = Arc::clone(&self.targets[index].server_addr); + let future = Box::pin(async move { AsyncClient::connect(endpoint, config).await }); self.install_target_state(index, TargetState::Connecting { future }); } @@ -2248,7 +2250,6 @@ fn build_task( .checked_add(1) .ok_or(ManagedConfigError::GenerationExhausted)?; let mut client_config = config.client.clone(); - client_config.server_addr.clone_from(&target.server_addr); client_config.auth = target.auth.resolve(&config.client.auth); validate_target_config(&client_config).map_err(|source| { ManagedConfigError::InvalidTarget { @@ -2361,14 +2362,9 @@ fn build_task( } fn validate_target_config(config: &ClientConfig) -> Result<(), ClientError> { - validate_open_timeouts(&config.open_timeouts)?; + config.open.validate()?; SessionMachine::validate_config(config)?; - if config.socket_config.ipv4_only && config.socket_config.ipv6_only { - return Err(ClientError::InvalidConfig { - reason: "ipv4_only and ipv6_only cannot both be true".to_owned(), - }); - } - if let Some(ttl) = config.socket_config.ttl { + if let Some(ttl) = config.socket.ttl { validate_ttl(ttl)?; } Ok(()) @@ -2429,12 +2425,9 @@ mod tests { vec![ManagedTargetConfig::new("target", "127.0.0.1:9")], ) .unwrap(); - let client = AsyncClient::connect(ClientConfig { - server_addr: "127.0.0.1:9".into(), - ..ClientConfig::default() - }) - .await - .unwrap(); + let client = AsyncClient::connect("127.0.0.1:9", ClientConfig::default()) + .await + .unwrap(); task.state = DriverState::Running; task.install_target_state(0, TargetState::Active { client }); task.send_gate = Some(end + Duration::from_secs(1)); diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 260adc9..07a3b7d 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -7,7 +7,8 @@ use irtt_proto::{ use crate::{ config::{ - ClientConfig, RunMode, MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_UDP_PAYLOAD_LENGTH, + ClientConfig, RunMode, SessionRequest, MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, + MAX_UDP_PAYLOAD_LENGTH, }, error::ClientError, event::{ @@ -172,7 +173,7 @@ impl SessionMachine { remote: std::net::SocketAddr, ) -> Result { Self::validate_config(&config)?; - let requested = params_from_config(&config)?; + let requested = params_from_request(&config.request)?; Ok(Self { config, @@ -193,7 +194,7 @@ impl SessionMachine { reason: "probe_timeout must be greater than zero".to_owned(), }); } - params_from_config(config).map(|_| ()) + params_from_request(&config.request).map(|_| ()) } pub(crate) fn config(&self) -> &ClientConfig { @@ -209,7 +210,7 @@ impl SessionMachine { let bytes = encode_request( RequestToEncode::Open { params: &self.requested, - no_test: self.config.run_mode == RunMode::NoTest, + no_test: self.config.request.run_mode == RunMode::NoTest, }, self.config.auth.hmac_key(), )?; @@ -269,7 +270,7 @@ impl SessionMachine { )); } - match self.config.run_mode { + match self.config.request.run_mode { RunMode::Normal if reply_is_close => Err(OpenAcceptanceFailure::new( ClientError::ServerRejected, cleanup_close, @@ -602,14 +603,11 @@ impl SessionMachine { cleanup_close: Option>, ) -> Result { let token = reply.token; - let negotiated = match negotiate_params( - &self.requested, - reply.params, - self.config.negotiation_policy, - ) { - Ok(negotiated) => negotiated, - Err(primary) => return Err(OpenAcceptanceFailure::new(primary, cleanup_close)), - }; + let negotiated = + match negotiate_params(&self.requested, reply.params, self.config.open.negotiation) { + Ok(negotiated) => negotiated, + Err(primary) => return Err(OpenAcceptanceFailure::new(primary, cleanup_close)), + }; let local_close_packet = cleanup_close.expect("normal non-zero-token replies prepare a cleanup close"); let next_state = MachineState::Open(Box::new(ActiveSession { @@ -649,12 +647,9 @@ impl SessionMachine { reply: OpenReply, now: ClientTimestamp, ) -> Result { - let negotiated = negotiate_params( - &self.requested, - reply.params, - self.config.negotiation_policy, - ) - .map_err(OpenAcceptanceFailure::without_cleanup)?; + let negotiated = + negotiate_params(&self.requested, reply.params, self.config.open.negotiation) + .map_err(OpenAcceptanceFailure::without_cleanup)?; let event = ClientEvent::NoTestCompleted { remote: self.remote, negotiated: negotiated.clone(), @@ -888,21 +883,24 @@ pub(crate) fn recv_buffer_size( }) } -pub(crate) fn params_from_config(config: &ClientConfig) -> Result { - validate_protocol_config(config)?; +pub(crate) fn params_from_request(request: &SessionRequest) -> Result { + validate_request(request)?; Ok(Params { protocol_version: PROTOCOL_VERSION, - duration_ns: match config.duration { + duration_ns: match request.duration { Some(duration) => config_duration_to_ns("duration", duration)?, None => 0, }, - interval_ns: config_duration_to_ns("interval", config.interval)?, - length: i64::from(config.length), - received_stats: config.received_stats, - stamp_at: config.stamp_at, - clock: config.clock, - dscp: i64::from(dscp_codepoint_to_traffic_class(config.dscp)?), - server_fill: config.server_fill.clone().map(|value| ServerFill { value }), + interval_ns: config_duration_to_ns("interval", request.interval)?, + length: i64::from(request.length), + received_stats: request.received_stats, + stamp_at: request.stamp_at, + clock: request.clock, + dscp: i64::from(dscp_codepoint_to_traffic_class(request.dscp)?), + server_fill: request + .server_fill + .clone() + .map(|value| ServerFill { value }), }) } @@ -1083,34 +1081,34 @@ fn build_received_stats(reply: &EchoReply) -> Option { }) } -fn validate_protocol_config(config: &ClientConfig) -> Result<(), ClientError> { - if config.duration == Some(Duration::ZERO) { +fn validate_request(request: &SessionRequest) -> Result<(), ClientError> { + if request.duration == Some(Duration::ZERO) { return Err(ClientError::InvalidConfig { reason: "duration must be greater than zero; use None for continuous mode".to_owned(), }); } - if config.interval == Duration::ZERO { + if request.interval == Duration::ZERO { return Err(ClientError::InvalidConfig { reason: "interval must be greater than zero".to_owned(), }); } - if config.clock == Clock::Unspecified { + if request.clock == Clock::Unspecified { return Err(ClientError::InvalidConfig { reason: "clock must be wall, monotonic, or both".to_owned(), }); } - if config.dscp > MAX_DSCP_CODEPOINT { + if request.dscp > MAX_DSCP_CODEPOINT { return Err(ClientError::InvalidConfig { reason: format!("dscp must be <= {MAX_DSCP_CODEPOINT}"), }); } - if config.length > MAX_UDP_PAYLOAD_LENGTH { + if request.length > MAX_UDP_PAYLOAD_LENGTH { return Err(ClientError::InvalidConfig { reason: format!("packet length must be <= {MAX_UDP_PAYLOAD_LENGTH}"), }); } - if let Some(fill) = &config.server_fill { + if let Some(fill) = &request.server_fill { let len = fill.len(); if len == 0 { return Err(ClientError::InvalidConfig { @@ -1191,9 +1189,12 @@ mod tests { fn opened() -> SessionMachine { let mut machine = SessionMachine::new( ClientConfig { + request: crate::SessionRequest { + clock: Clock::Wall, + stamp_at: irtt_proto::StampAt::Receive, + ..Default::default() + }, max_pending_probes: 2, - clock: Clock::Wall, - stamp_at: irtt_proto::StampAt::Receive, ..ClientConfig::default() }, "127.0.0.1:2112".parse().unwrap(), diff --git a/crates/irtt-client/src/session/negotiation.rs b/crates/irtt-client/src/session/negotiation.rs index d1e3fc7..ab6afd3 100644 --- a/crates/irtt-client/src/session/negotiation.rs +++ b/crates/irtt-client/src/session/negotiation.rs @@ -13,7 +13,7 @@ use crate::{config::NegotiationPolicy, error::ClientError}; /// `params.dscp` is the raw IP TOS / Traffic Class byte, not a DSCP /// codepoint: for a configured codepoint of 46 (EF), an unrestricted /// negotiation leaves `params.dscp == 184`. [`NegotiationRestriction::DscpChanged`] -/// reports codepoints instead, for consistency with [`crate::ClientConfig::dscp`]. +/// reports codepoints instead, for consistency with [`crate::SessionRequest::dscp`]. #[derive(Debug, Clone, PartialEq, Eq)] pub struct NegotiatedParams { /// Server-returned protocol parameters used for the session. @@ -61,7 +61,7 @@ pub enum NegotiationRestriction { /// Returned DSCP codepoint differs from the request. /// /// These values are DSCP codepoints (`0..=63`) for human-facing - /// consistency with [`crate::ClientConfig::dscp`], not the raw wire + /// consistency with [`crate::SessionRequest::dscp`], not the raw wire /// `Params::dscp` byte carried by [`NegotiatedParams::params`]. DscpChanged { requested: i64, negotiated: i64 }, /// Returned server payload fill behavior differs from the request. diff --git a/crates/irtt-client/src/socket.rs b/crates/irtt-client/src/socket.rs index 7753e78..bc54b5f 100644 --- a/crates/irtt-client/src/socket.rs +++ b/crates/irtt-client/src/socket.rs @@ -1,12 +1,9 @@ -use std::{ - net::{SocketAddr, ToSocketAddrs, UdpSocket}, - time::Duration, -}; +use std::net::{SocketAddr, ToSocketAddrs, UdpSocket}; use socket2::{Domain, Protocol, Socket, Type}; use crate::{ - config::{ClientConfig, SocketConfig, DEFAULT_PORT, MIN_OPEN_TIMEOUT}, + config::{AddressFamily, SocketConfig, DEFAULT_PORT}, error::ClientError, receive::configure_receive_metadata, socket_options::apply_ttl_to_socket, @@ -20,36 +17,27 @@ use crate::{ ))] use crate::socket_options::apply_routing_options; -pub(crate) fn validate_open_timeouts(timeouts: &[Duration]) -> Result<(), ClientError> { - if timeouts.is_empty() { - return Err(ClientError::NoOpenTimeouts); - } - for timeout in timeouts { - if *timeout < MIN_OPEN_TIMEOUT { - return Err(ClientError::OpenTimeoutTooSmall { - timeout: *timeout, - minimum: MIN_OPEN_TIMEOUT, - }); - } - } - Ok(()) -} - -pub(crate) fn resolve_remote(config: &ClientConfig) -> Result { - let addr = normalize_server_addr(&config.server_addr); +pub(crate) fn resolve_remote( + endpoint: &str, + family: AddressFamily, +) -> Result { + let addr = normalize_server_addr(endpoint); let mut addrs = addr .to_socket_addrs() .map_err(|_| ClientError::Resolve { addr: addr.clone() })?; addrs - .find(|addr| address_family_allowed(config, *addr)) + .find(|addr| address_family_allowed(family, *addr)) .ok_or(ClientError::Resolve { addr }) } #[cfg(feature = "tokio")] -pub(crate) async fn resolve_remote_tokio(config: &ClientConfig) -> Result { - let addr = normalize_server_addr(&config.server_addr); +pub(crate) async fn resolve_remote_tokio( + endpoint: &str, + family: AddressFamily, +) -> Result { + let addr = normalize_server_addr(endpoint); if let Ok(remote) = addr.parse::() { - return address_family_allowed(config, remote) + return address_family_allowed(family, remote) .then_some(remote) .ok_or(ClientError::Resolve { addr }); } @@ -58,13 +46,16 @@ pub(crate) async fn resolve_remote_tokio(config: &ClientConfig) -> Result bool { - (!config.socket_config.ipv4_only || remote.is_ipv4()) - && (!config.socket_config.ipv6_only || remote.is_ipv6()) +fn address_family_allowed(family: AddressFamily, remote: SocketAddr) -> bool { + match family { + AddressFamily::Any => true, + AddressFamily::Ipv4 => remote.is_ipv4(), + AddressFamily::Ipv6 => remote.is_ipv6(), + } } pub(crate) fn normalize_server_addr(addr: &str) -> String { @@ -92,17 +83,9 @@ pub(crate) fn normalize_server_addr(addr: &str) -> String { pub(crate) fn connect_udp_socket( config: &SocketConfig, remote: SocketAddr, + family: AddressFamily, ) -> Result { - let socket = create_connected_udp_socket(config, remote)?; - socket.set_read_timeout(config.recv_timeout)?; - Ok(socket) -} - -fn create_connected_udp_socket( - config: &SocketConfig, - remote: SocketAddr, -) -> Result { - let socket = create_prebind_udp_socket(config, remote)?; + let socket = create_prebind_udp_socket(remote, family)?; #[cfg(any( target_os = "android", target_os = "freebsd", @@ -133,8 +116,8 @@ fn create_connected_udp_socket( } fn create_prebind_udp_socket( - config: &SocketConfig, remote: SocketAddr, + family: AddressFamily, ) -> Result { let domain = if remote.is_ipv4() { Domain::IPV4 @@ -143,7 +126,7 @@ fn create_prebind_udp_socket( }; let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?; - if config.ipv6_only && remote.is_ipv6() { + if family == AddressFamily::Ipv6 && remote.is_ipv6() { socket.set_only_v6(true)?; } Ok(socket) @@ -153,8 +136,9 @@ fn create_prebind_udp_socket( pub(crate) fn connect_tokio_udp_socket( config: &SocketConfig, remote: SocketAddr, + family: AddressFamily, ) -> Result { - let socket = create_connected_udp_socket(config, remote)?; + let socket = connect_udp_socket(config, remote, family)?; socket.set_nonblocking(true)?; Ok(tokio::net::UdpSocket::from_std(socket)?) } diff --git a/crates/irtt-client/tests/client/close.rs b/crates/irtt-client/tests/client/close.rs index 9b70f5c..5c5317b 100644 --- a/crates/irtt-client/tests/client/close.rs +++ b/crates/irtt-client/tests/client/close.rs @@ -9,9 +9,12 @@ fn operations_fail_after_local_close_without_datagrams() { socket.send_to(&reply, peer).unwrap(); let _ = recv_request(&socket, &tx); }); - let mut config = default_test_config(server.addr); - config.socket_config.recv_timeout = Some(Duration::from_millis(50)); - let mut client = Client::connect(config).unwrap(); + let config = default_test_config(); + + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(50))) + .unwrap(); assert_open_started(client.open().unwrap()); client.close().unwrap(); assert!(matches!(client.open(), Err(ClientError::AlreadyClosed))); @@ -52,7 +55,7 @@ fn send_probe_fails_after_close() { } } }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + let mut client = Client::connect(server.addr.to_string(), default_test_config()).unwrap(); assert_open_started(client.open().unwrap()); client.close().unwrap(); assert!(matches!( @@ -89,7 +92,7 @@ fn recv_available_stops_after_peer_close() { socket.send_to(&close, peer).unwrap(); } }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + let mut client = Client::connect(server.addr.to_string(), default_test_config()).unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); @@ -137,14 +140,10 @@ fn close_flagged_duplicate_emits_duplicate_then_closes() { socket.send_to(&close, peer).unwrap(); } }); - let mut client = Client::connect(ClientConfig { - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }) - .unwrap(); + let mut client = Client::connect(server.addr.to_string(), default_test_config()).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); @@ -194,15 +193,17 @@ fn close_flagged_retained_timeout_emits_late_then_closes() { .unwrap(); } }); - let mut client = Client::connect(ClientConfig { - probe_timeout: Duration::from_millis(50), - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() + let mut client = Client::connect( + server.addr.to_string(), + ClientConfig { + probe_timeout: Duration::from_millis(50), + ..default_test_config() }, - ..default_test_config(server.addr) - }) + ) .unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); let sent = client.send_probe().unwrap(); let ClientEvent::EchoSent { sent_at, .. } = &sent[0] else { @@ -289,15 +290,17 @@ fn close_flagged_evicted_sequence_emits_late_then_closes() { .unwrap(); } }); - let mut client = Client::connect(ClientConfig { - max_pending_probes: 1, - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() + let mut client = Client::connect( + server.addr.to_string(), + ClientConfig { + max_pending_probes: 1, + ..default_test_config() }, - ..default_test_config(server.addr) - }) + ) .unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); @@ -354,14 +357,10 @@ fn close_flagged_untracked_sequence_emits_warning_then_closes() { .unwrap(); } }); - let mut client = Client::connect(ClientConfig { - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }) - .unwrap(); + let mut client = Client::connect(server.addr.to_string(), default_test_config()).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); assert!(matches!( @@ -412,14 +411,10 @@ fn wrong_token_close_flag_does_not_close_session() { let _ = recv_request(&socket, &tx); } }); - let mut client = Client::connect(ClientConfig { - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }) - .unwrap(); + let mut client = Client::connect(server.addr.to_string(), default_test_config()).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); @@ -467,14 +462,11 @@ fn close_flagged_echo_reply_emits_reply_then_closes_without_sending_close() { while recv_request_timeout(&socket, &tx).is_some() {} } }); - let config = ClientConfig { - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); + let config = default_test_config(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); diff --git a/crates/irtt-client/tests/client/config.rs b/crates/irtt-client/tests/client/config.rs index 8ea5adc..9b07184 100644 --- a/crates/irtt-client/tests/client/config.rs +++ b/crates/irtt-client/tests/client/config.rs @@ -8,7 +8,10 @@ fn connect_rejects_invalid_configuration() { ( "oversized UDP payload", ClientConfig { - length: MAX_UDP_PAYLOAD_LENGTH + 1, + request: SessionRequest { + length: MAX_UDP_PAYLOAD_LENGTH + 1, + ..Default::default() + }, ..ClientConfig::default() }, "packet length", @@ -16,7 +19,10 @@ fn connect_rejects_invalid_configuration() { ( "zero finite duration", ClientConfig { - duration: Some(Duration::ZERO), + request: SessionRequest { + duration: Some(Duration::ZERO), + ..Default::default() + }, ..ClientConfig::default() }, "duration must be greater than zero; use None for continuous mode", @@ -24,7 +30,10 @@ fn connect_rejects_invalid_configuration() { ( "zero interval", ClientConfig { - interval: Duration::ZERO, + request: SessionRequest { + interval: Duration::ZERO, + ..Default::default() + }, ..ClientConfig::default() }, "interval must be greater than zero", @@ -32,7 +41,10 @@ fn connect_rejects_invalid_configuration() { ( "duration nanosecond overflow", ClientConfig { - duration: Some(too_large), + request: SessionRequest { + duration: Some(too_large), + ..Default::default() + }, ..ClientConfig::default() }, "duration is too large to encode as nanoseconds", @@ -40,7 +52,10 @@ fn connect_rejects_invalid_configuration() { ( "interval nanosecond overflow", ClientConfig { - interval: too_large, + request: SessionRequest { + interval: too_large, + ..Default::default() + }, ..ClientConfig::default() }, "interval is too large to encode as nanoseconds", @@ -48,7 +63,10 @@ fn connect_rejects_invalid_configuration() { ( "invalid DSCP codepoint", ClientConfig { - dscp: 64, + request: SessionRequest { + dscp: 64, + ..Default::default() + }, ..ClientConfig::default() }, "dscp", @@ -58,7 +76,10 @@ fn connect_rejects_invalid_configuration() { // client mode; a client that needs timestamps must name a clock. "unspecified clock", ClientConfig { - clock: Clock::Unspecified, + request: SessionRequest { + clock: Clock::Unspecified, + ..Default::default() + }, ..ClientConfig::default() }, "clock must be wall, monotonic, or both", @@ -66,7 +87,10 @@ fn connect_rejects_invalid_configuration() { ( "empty server fill", ClientConfig { - server_fill: Some("".to_owned()), + request: SessionRequest { + server_fill: Some("".to_owned()), + ..Default::default() + }, ..ClientConfig::default() }, "server_fill", @@ -74,7 +98,10 @@ fn connect_rejects_invalid_configuration() { ( "oversized server fill", ClientConfig { - server_fill: Some("0123456789abcdef0123456789abcdefx".to_owned()), + request: SessionRequest { + server_fill: Some("0123456789abcdef0123456789abcdefx".to_owned()), + ..Default::default() + }, ..ClientConfig::default() }, "server_fill", @@ -84,7 +111,7 @@ fn connect_rejects_invalid_configuration() { for (name, config, expected_reason) in cases { assert!( matches!( - Client::connect(config), + Client::connect("127.0.0.1:2112", config), Err(ClientError::InvalidConfig { .. }) ), "{name} should fail with InvalidConfig ({expected_reason})" @@ -95,11 +122,14 @@ fn connect_rejects_invalid_configuration() { #[test] fn minimum_open_timeout_under_200ms_is_rejected() { let config = ClientConfig { - open_timeouts: vec![Duration::from_millis(199)], + open: OpenPolicy { + timeouts: vec![Duration::from_millis(199)], + ..Default::default() + }, ..ClientConfig::default() }; assert!(matches!( - Client::connect(config), + Client::connect("127.0.0.1:2112", config), Err(ClientError::OpenTimeoutTooSmall { .. }) )); } @@ -107,11 +137,14 @@ fn minimum_open_timeout_under_200ms_is_rejected() { #[test] fn empty_open_timeouts_is_rejected() { let config = ClientConfig { - open_timeouts: vec![], + open: OpenPolicy { + timeouts: vec![], + ..Default::default() + }, ..ClientConfig::default() }; assert!(matches!( - Client::connect(config), + Client::connect("127.0.0.1:2112", config), Err(ClientError::NoOpenTimeouts) )); } diff --git a/crates/irtt-client/tests/client/hmac.rs b/crates/irtt-client/tests/client/hmac.rs index 41cafab..c9f8747 100644 --- a/crates/irtt-client/tests/client/hmac.rs +++ b/crates/irtt-client/tests/client/hmac.rs @@ -4,7 +4,7 @@ use irtt_client::{Authentication, HmacKey}; #[test] fn hmac_open_ignores_missing_hmac_before_valid_reply() { let key = b"secret".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + let mut config = default_test_config(); config.auth = Authentication::Hmac(HmacKey::new(key.clone())); let params = default_params(); let server = start_fake_server(move |socket, tx| { @@ -14,8 +14,7 @@ fn hmac_open_ignores_missing_hmac_before_valid_reply() { socket.send_to(&missing, peer).unwrap(); socket.send_to(&valid, peer).unwrap(); }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); assert_open_started(client.open().unwrap()); assert_eq!(server.rx.iter().take(1).count(), 1); assert!(server.rx.try_recv().is_err()); @@ -26,7 +25,7 @@ fn hmac_open_ignores_missing_hmac_before_valid_reply() { fn hmac_open_ignores_bad_hmac_before_valid_reply() { let key = b"secret".to_vec(); let wrong_key = b"wrong".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + let mut config = default_test_config(); config.auth = Authentication::Hmac(HmacKey::new(key.clone())); let params = default_params(); let server = start_fake_server(move |socket, tx| { @@ -36,8 +35,7 @@ fn hmac_open_ignores_bad_hmac_before_valid_reply() { socket.send_to(&bad, peer).unwrap(); socket.send_to(&valid, peer).unwrap(); }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); assert_open_started(client.open().unwrap()); assert_eq!(server.rx.iter().take(1).count(), 1); assert!(server.rx.try_recv().is_err()); @@ -47,7 +45,7 @@ fn hmac_open_ignores_bad_hmac_before_valid_reply() { #[test] fn post_token_hmac_negotiation_failure_sends_authenticated_cleanup_close() { let key = b"secret".to_vec(); - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + let mut config = default_test_config(); config.auth = Authentication::Hmac(HmacKey::new(key.clone())); let mut returned = default_params(); returned.interval_ns += 1; @@ -58,8 +56,7 @@ fn post_token_hmac_negotiation_failure_sends_authenticated_cleanup_close() { socket.send_to(&reply, peer).unwrap(); let _ = recv_request(&socket, &tx); }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); assert!(matches!( client.open(), diff --git a/crates/irtt-client/tests/client/no_test.rs b/crates/irtt-client/tests/client/no_test.rs index 2ba24cf..6e8e4c2 100644 --- a/crates/irtt-client/tests/client/no_test.rs +++ b/crates/irtt-client/tests/client/no_test.rs @@ -2,8 +2,8 @@ use super::*; #[test] fn no_test_rejects_non_close_open_reply() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; + let mut config = default_test_config(); + config.request.run_mode = RunMode::NoTest; let params = default_params(); let server = start_fake_server(move |socket, tx| { let (_, peer) = recv_request(&socket, &tx); @@ -15,8 +15,7 @@ fn no_test_rejects_non_close_open_reply() { .unwrap(); let _ = recv_request(&socket, &tx); }); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); assert!(matches!( client.open(), Err(ClientError::UnexpectedNoTestReply) @@ -32,12 +31,11 @@ fn no_test_rejects_non_close_open_reply() { #[test] fn no_test_rejects_non_zero_token_with_close_reply() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; + let mut config = default_test_config(); + config.request.run_mode = RunMode::NoTest; let params = default_params(); let server = no_test_server(params, TOKEN); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); assert!(matches!( client.open(), Err(ClientError::NonZeroNoTestToken { token: TOKEN }) @@ -47,13 +45,12 @@ fn no_test_rejects_non_zero_token_with_close_reply() { #[test] fn no_test_strict_negotiation_rejects_changed_params() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; + let mut config = default_test_config(); + config.request.run_mode = RunMode::NoTest; let mut params = default_params(); params.dscp = 1; let server = no_test_server(params, 0); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); assert!(matches!( client.open(), Err(ClientError::NegotiationRejected { .. }) @@ -63,15 +60,14 @@ fn no_test_strict_negotiation_rejects_changed_params() { #[test] fn no_test_loose_negotiation_accepts_restricted_params() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - config.negotiation_policy = NegotiationPolicy::Loose; + let mut config = default_test_config(); + config.request.run_mode = RunMode::NoTest; + config.open.negotiation = NegotiationPolicy::Loose; let requested = default_params(); let mut params = requested.clone(); params.duration_ns /= 2; let server = no_test_server(params.clone(), 0); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); let negotiated = assert_no_test_completed(client.open().unwrap()); assert_eq!(negotiated.params, params); assert_eq!( @@ -86,12 +82,11 @@ fn no_test_loose_negotiation_accepts_restricted_params() { #[test] fn send_probe_fails_after_no_test_completed() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; + let mut config = default_test_config(); + config.request.run_mode = RunMode::NoTest; let params = default_params(); let server = no_test_server(params, 0); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); assert_no_test_completed(client.open().unwrap()); assert!(matches!( client.send_probe(), @@ -102,13 +97,15 @@ fn send_probe_fails_after_no_test_completed() { #[test] fn operations_fail_after_no_test_completed_without_datagrams() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.run_mode = RunMode::NoTest; - config.socket_config.recv_timeout = Some(Duration::from_millis(50)); + let mut config = default_test_config(); + config.request.run_mode = RunMode::NoTest; + let params = default_params(); let server = no_test_server(params, 0); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(50))) + .unwrap(); assert_no_test_completed(client.open().unwrap()); assert!(matches!(client.open(), Err(ClientError::AlreadyCompleted))); let results = [ diff --git a/crates/irtt-client/tests/client/open.rs b/crates/irtt-client/tests/client/open.rs index 456fd82..823d123 100644 --- a/crates/irtt-client/tests/client/open.rs +++ b/crates/irtt-client/tests/client/open.rs @@ -3,10 +3,13 @@ use super::*; #[test] fn receive_before_open_returns_not_open_without_datagrams() { let peer = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); - let mut config = default_test_config(peer.local_addr().unwrap()); + let config = default_test_config(); // Bound the old implementation's blocking receive without sending a packet. - config.socket_config.recv_timeout = Some(Duration::from_millis(50)); - let mut client = Client::connect(config).unwrap(); + + let mut client = Client::connect(peer.local_addr().unwrap().to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(50))) + .unwrap(); let results = [ client.recv_once(), @@ -35,10 +38,13 @@ fn loose_open_rejects_nonpositive_returned_intervals() { returned.interval_ns = interval_ns; let server = open_success_server(returned); let config = ClientConfig { - negotiation_policy: NegotiationPolicy::Loose, - ..default_test_config(server.addr) + open: OpenPolicy { + negotiation: NegotiationPolicy::Loose, + ..default_test_config().open + }, + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); assert!( matches!(client.open(), Err(ClientError::NegotiationRejected { .. })), "returned interval {interval_ns} must be rejected" @@ -51,7 +57,7 @@ fn loose_open_rejects_nonpositive_returned_intervals() { fn open_fails_when_already_open() { let params = default_params(); let server = open_success_server(params); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + let mut client = Client::connect(server.addr.to_string(), default_test_config()).unwrap(); assert_open_started(client.open().unwrap()); assert!(matches!(client.open(), Err(ClientError::AlreadyOpen))); server.join(); @@ -68,10 +74,13 @@ fn open_retries_after_first_timeout() { socket.send_to(&reply, peer).unwrap(); }); let config = ClientConfig { - open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(500)], - ..default_test_config(server.addr) + open: OpenPolicy { + timeouts: vec![Duration::from_millis(200), Duration::from_millis(500)], + ..default_test_config().open + }, + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); let outcome = client.open().unwrap(); assert_open_started(outcome); assert_eq!(server.rx.iter().take(2).count(), 2); @@ -80,12 +89,12 @@ fn open_retries_after_first_timeout() { #[test] fn protocol_version_mismatch_fails() { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.negotiation_policy = NegotiationPolicy::Loose; + let mut config = default_test_config(); + config.open.negotiation = NegotiationPolicy::Loose; let mut params = default_params(); params.protocol_version = 2; let server = open_success_server(params); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + let mut client = Client::connect(server.addr.to_string(), default_test_config()).unwrap(); assert!(matches!( client.open(), Err(ClientError::ProtocolVersionMismatch { received: 2, .. }) @@ -102,7 +111,7 @@ fn server_rejection_fails_in_normal_mode() { let reply = open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, ¶ms, None); socket.send_to(&reply, peer).unwrap(); }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + let mut client = Client::connect(server.addr.to_string(), default_test_config()).unwrap(); assert!(matches!(client.open(), Err(ClientError::ServerRejected))); assert_eq!(server.rx.iter().take(1).count(), 1); server.join(); @@ -119,10 +128,13 @@ fn ignored_datagrams_do_not_restart_the_attempt_deadline() { let _ = recv_request(&socket, &tx); }); let config = ClientConfig { - open_timeouts: vec![Duration::from_millis(250), Duration::from_millis(250)], - ..default_test_config(server.addr) + open: OpenPolicy { + timeouts: vec![Duration::from_millis(250), Duration::from_millis(250)], + ..default_test_config().open + }, + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); let started = Instant::now(); assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); @@ -139,10 +151,13 @@ fn ignored_datagrams_do_not_restart_the_attempt_deadline() { fn opening_deadline_overflow_occurs_before_send() { let server = timeout_server(Duration::from_millis(250)); let config = ClientConfig { - open_timeouts: vec![Duration::MAX], - ..default_test_config(server.addr) + open: OpenPolicy { + timeouts: vec![Duration::MAX], + ..default_test_config().open + }, + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); assert!(matches!(client.open(), Err(ClientError::DurationOverflow))); assert!(server.rx.try_recv().is_err()); @@ -163,7 +178,7 @@ fn post_token_negotiation_failure_sends_cleanup_close() { .unwrap(); let _ = recv_request(&socket, &tx); }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + let mut client = Client::connect(server.addr.to_string(), default_test_config()).unwrap(); assert!(matches!( client.open(), @@ -182,14 +197,16 @@ fn post_token_negotiation_failure_sends_cleanup_close() { fn open_timeout_after_all_timeouts() { let server = timeout_server(Duration::from_millis(700)); let config = ClientConfig { - open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() + open: OpenPolicy { + timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], + ..default_test_config().open }, - ..default_test_config(server.addr) + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(50))) + .unwrap(); assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); assert_eq!(server.rx.iter().take(2).count(), 2); server.join(); @@ -221,7 +238,7 @@ fn multiple_untrusted_datagrams_before_valid_reply_use_one_attempt() { socket.send_to(&packet, peer).unwrap(); } }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + let mut client = Client::connect(server.addr.to_string(), default_test_config()).unwrap(); assert_open_started(client.open().unwrap()); assert_eq!(server.rx.iter().take(1).count(), 1); @@ -241,14 +258,16 @@ fn only_ignored_open_datagrams_eventually_time_out() { thread::sleep(Duration::from_millis(300)); }); let config = ClientConfig { - open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() + open: OpenPolicy { + timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], + ..default_test_config().open }, - ..default_test_config(server.addr) + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(50))) + .unwrap(); assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); assert_eq!(server.rx.iter().take(2).count(), 2); @@ -268,7 +287,7 @@ fn trusted_zero_token_normal_reply_is_terminal() { reply[HMAC_OFFSET..HMAC_OFFSET + 8].copy_from_slice(&0_u64.to_le_bytes()); socket.send_to(&reply, peer).unwrap(); }); - let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + let mut client = Client::connect(server.addr.to_string(), default_test_config()).unwrap(); assert!(matches!( client.open(), diff --git a/crates/irtt-client/tests/client/probes_replies/packet_length.rs b/crates/irtt-client/tests/client/probes_replies/packet_length.rs index 7c34a6e..41d12e0 100644 --- a/crates/irtt-client/tests/client/probes_replies/packet_length.rs +++ b/crates/irtt-client/tests/client/probes_replies/packet_length.rs @@ -23,14 +23,16 @@ fn short_echo_reply_does_not_emit_echo_reply() { } }); let config = ClientConfig { - length: 64, - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() + request: SessionRequest { + length: 64, + ..default_test_config().request }, - ..default_test_config(server.addr) + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); @@ -69,14 +71,16 @@ fn overlong_datagram_detection_uses_extra_receive_byte() { } }); let config = ClientConfig { - length: 4096, - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() + request: SessionRequest { + length: 4096, + ..default_test_config().request }, - ..default_test_config(server.addr) + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); @@ -112,14 +116,16 @@ fn exact_length_echo_reply_still_emits_echo_reply() { } }); let config = ClientConfig { - length: 4096, - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() + request: SessionRequest { + length: 4096, + ..default_test_config().request }, - ..default_test_config(server.addr) + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); diff --git a/crates/irtt-client/tests/client/probes_replies/reply_classification.rs b/crates/irtt-client/tests/client/probes_replies/reply_classification.rs index a85c6cc..1d2ff1b 100644 --- a/crates/irtt-client/tests/client/probes_replies/reply_classification.rs +++ b/crates/irtt-client/tests/client/probes_replies/reply_classification.rs @@ -22,14 +22,11 @@ fn duplicate_reply_emits_duplicate_event() { socket.send_to(&reply_packet, peer).unwrap(); } }); - let config = ClientConfig { - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); + let config = default_test_config(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); thread::sleep(Duration::from_millis(50)); @@ -76,14 +73,11 @@ fn out_of_order_reply_emits_late_event() { let reply0 = echo_reply_packet(TOKEN, seqs[0], ¶ms, &ts, None); socket.send_to(&reply0, peer).unwrap(); }); - let config = ClientConfig { - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); + let config = default_test_config(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); client.send_probe().unwrap(); diff --git a/crates/irtt-client/tests/client/probes_replies/reply_decode.rs b/crates/irtt-client/tests/client/probes_replies/reply_decode.rs index 6a49c63..9118758 100644 --- a/crates/irtt-client/tests/client/probes_replies/reply_decode.rs +++ b/crates/irtt-client/tests/client/probes_replies/reply_decode.rs @@ -4,14 +4,11 @@ use super::*; fn recv_once_returns_empty_on_timeout() { let params = default_params(); let server = open_success_server(params); - let config = ClientConfig { - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); + let config = default_test_config(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(50))) + .unwrap(); assert_open_started(client.open().unwrap()); let events = client.recv_once().unwrap(); assert!(events.is_empty()); @@ -52,14 +49,11 @@ fn recv_once_decodes_only_received_bytes_after_longer_datagram() { ); socket.send_to(&shorter, peer).unwrap(); }); - let config = ClientConfig { - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); + let config = default_test_config(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); diff --git a/crates/irtt-client/tests/client/probes_replies/reply_metadata.rs b/crates/irtt-client/tests/client/probes_replies/reply_metadata.rs index d11cae7..9e9c9c2 100644 --- a/crates/irtt-client/tests/client/probes_replies/reply_metadata.rs +++ b/crates/irtt-client/tests/client/probes_replies/reply_metadata.rs @@ -19,14 +19,11 @@ fn echo_reply_metadata_propagates_observed_dscp_with_ancillary() { echo_reply_packet(TOKEN, seq, ¶ms, &TimestampFields::default(), None); socket.send_to(&reply_packet, peer).unwrap(); }); - let config = ClientConfig { - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); + let config = default_test_config(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); diff --git a/crates/irtt-client/tests/client/probes_replies/timeouts.rs b/crates/irtt-client/tests/client/probes_replies/timeouts.rs index 7e1a88c..edcf79e 100644 --- a/crates/irtt-client/tests/client/probes_replies/timeouts.rs +++ b/crates/irtt-client/tests/client/probes_replies/timeouts.rs @@ -20,9 +20,9 @@ fn probe_timeout_overflow_transmits_no_echo() { }); let config = ClientConfig { probe_timeout: Duration::MAX, - ..default_test_config(server.addr) + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); assert_open_started(client.open().unwrap()); assert!(matches!( client.send_probe(), @@ -46,13 +46,12 @@ fn poll_timeouts_emits_echo_loss() { let server = silent_open_server(params); let config = ClientConfig { probe_timeout: Duration::from_millis(100), - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() - }, - ..default_test_config(server.addr) + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(50))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); @@ -97,13 +96,12 @@ fn late_reply_after_timeout_preserves_measurement_metadata() { }); let config = ClientConfig { probe_timeout: Duration::from_millis(40), - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); @@ -157,15 +155,17 @@ fn pending_full_does_not_send_packet() { }; let server = silent_open_server(params); let config = ClientConfig { - duration: Some(Duration::from_secs(60)), - max_pending_probes: 2, - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(50)), - ..Default::default() + request: SessionRequest { + duration: Some(Duration::from_secs(60)), + ..SessionRequest::default() }, - ..default_test_config(server.addr) + max_pending_probes: 2, + ..default_test_config() }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(50))) + .unwrap(); assert_open_started(client.open().unwrap()); client.send_probe().unwrap(); diff --git a/crates/irtt-client/tests/client/support.rs b/crates/irtt-client/tests/client/support.rs index 8f99479..a6d2028 100644 --- a/crates/irtt-client/tests/client/support.rs +++ b/crates/irtt-client/tests/client/support.rs @@ -25,10 +25,12 @@ impl FakeServer { } } -pub(super) fn default_test_config(addr: SocketAddr) -> ClientConfig { +pub(super) fn default_test_config() -> ClientConfig { ClientConfig { - server_addr: addr.to_string(), - open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], + open: OpenPolicy { + timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], + ..Default::default() + }, ..ClientConfig::default() } } @@ -252,14 +254,11 @@ pub(super) fn assert_no_test_completed(outcome: OpenOutcome) -> NegotiatedParams pub(super) fn open_client_with_echo_server(params: &Params) -> (Client, FakeServer) { let server = echo_server(params.clone()); - let config = ClientConfig { - socket_config: irtt_client::SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..Default::default() - }, - ..default_test_config(server.addr) - }; - let mut client = Client::connect(config).unwrap(); + let config = default_test_config(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert_open_started(client.open().unwrap()); (client, server) } diff --git a/crates/irtt-client/tests/client/ttl.rs b/crates/irtt-client/tests/client/ttl.rs index 5e2aeac..0447e4c 100644 --- a/crates/irtt-client/tests/client/ttl.rs +++ b/crates/irtt-client/tests/client/ttl.rs @@ -3,19 +3,19 @@ use super::*; #[test] fn connect_accepts_valid_ttl_config_values() { for ttl in [None, Some(1), Some(255)] { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 9))); - config.socket_config.ttl = ttl; - Client::connect(config).unwrap(); + let mut config = default_test_config(); + config.socket.ttl = ttl; + Client::connect(SocketAddr::from(([127, 0, 0, 1], 9)).to_string(), config).unwrap(); } } #[test] fn connect_rejects_invalid_ttl_values() { for ttl in [0, 256] { - let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 9))); - config.socket_config.ttl = Some(ttl); + let mut config = default_test_config(); + config.socket.ttl = Some(ttl); assert!(matches!( - Client::connect(config), + Client::connect(SocketAddr::from(([127, 0, 0, 1], 9)).to_string(), config), Err(ClientError::InvalidConfig { .. }) )); } diff --git a/crates/irtt-client/tests/linux_transport.rs b/crates/irtt-client/tests/linux_transport.rs index bf9de55..8414116 100644 --- a/crates/irtt-client/tests/linux_transport.rs +++ b/crates/irtt-client/tests/linux_transport.rs @@ -9,7 +9,9 @@ use std::{ time::{Duration, SystemTime, UNIX_EPOCH}, }; -use irtt_client::{Client, ClientConfig, ClientEvent, NegotiationPolicy, SocketConfig}; +use irtt_client::{ + Client, ClientConfig, ClientEvent, NegotiationPolicy, OpenPolicy, SessionRequest, SocketConfig, +}; use irtt_proto::{decode_request, Clock, DecodedRequestKind, ReceivedStats, StampAt}; use irtt_server::{ServerConfig, ServerCore}; use nix::sys::socket::{recvmsg, ControlMessageOwned, MsgFlags}; @@ -81,21 +83,24 @@ fn peer(dscp_allowed: bool, probes: usize) -> (SocketAddr, thread::JoinHandle ClientConfig { +fn config() -> ClientConfig { ClientConfig { - server_addr: addr.to_string(), - dscp: 46, - negotiation_policy: NegotiationPolicy::Loose, - duration: None, - interval: Duration::from_millis(10), - clock: Clock::Both, - stamp_at: StampAt::Both, - received_stats: ReceivedStats::Both, - length: 128, - open_timeouts: vec![Duration::from_secs(1)], - socket_config: SocketConfig { + open: OpenPolicy { + negotiation: NegotiationPolicy::Loose, + timeouts: vec![Duration::from_secs(1)], + }, + request: SessionRequest { + dscp: 46, + duration: None, + interval: Duration::from_millis(10), + clock: Clock::Both, + stamp_at: StampAt::Both, + received_stats: ReceivedStats::Both, + length: 128, + ..Default::default() + }, + socket: SocketConfig { ttl: Some(TEST_TTL), - recv_timeout: Some(Duration::from_secs(1)), ..SocketConfig::default() }, ..ClientConfig::default() @@ -149,7 +154,10 @@ fn blocking_probes_use_kernel_tx_timing_and_negotiated_packet_marking() { for allowed in [true, false] { const PROBES: usize = 3; let (addr, peer) = peer(allowed, PROBES); - let mut client = Client::connect(config(addr)).unwrap(); + let mut client = Client::connect(addr.to_string(), config()).unwrap(); + client + .set_recv_timeout(Some(Duration::from_secs(1))) + .unwrap(); client.open().unwrap(); assert_eq!( client.negotiated_params().unwrap().params.dscp, @@ -189,7 +197,7 @@ fn async_probes_use_kernel_tx_timing_and_negotiated_packet_marking() { let (addr, peer) = peer(allowed, PROBES); runtime.block_on(async { tokio::time::timeout(Duration::from_secs(5), async { - let mut client = irtt_client::AsyncClient::connect(config(addr)) + let mut client = irtt_client::AsyncClient::connect(addr.to_string(), config()) .await .unwrap(); client.open().await.unwrap(); @@ -245,7 +253,10 @@ fn a_closed_udp_peer_surfaces_a_socket_error() { close_rx.recv_timeout(Duration::from_secs(3)).unwrap(); drop(socket); }); - let mut client = Client::connect(config(addr)).unwrap(); + let mut client = Client::connect(addr.to_string(), config()).unwrap(); + client + .set_recv_timeout(Some(Duration::from_secs(1))) + .unwrap(); client.open().unwrap(); close_tx.send(()).unwrap(); task.join().unwrap(); diff --git a/crates/irtt-integration-tests/tests/async_client.rs b/crates/irtt-integration-tests/tests/async_client.rs index 1b93282..bb8f3cd 100644 --- a/crates/irtt-integration-tests/tests/async_client.rs +++ b/crates/irtt-integration-tests/tests/async_client.rs @@ -1,6 +1,8 @@ use std::time::Duration; -use irtt_client::{AsyncClient, ClientConfig, ClientEvent, OpenOutcome, SocketConfig}; +use irtt_client::{ + AsyncClient, ClientConfig, ClientEvent, OpenOutcome, OpenPolicy, SessionRequest, +}; use irtt_proto::{Clock, ReceivedStats, StampAt}; use irtt_server::{Server, ServerConfig}; use tokio::{sync::oneshot, time::timeout}; @@ -27,22 +29,26 @@ async fn exercise_client_server() { }); let config = ClientConfig { - server_addr: server_addr.to_string(), - duration: Some(Duration::from_secs(1)), - interval: Duration::from_millis(100), - length: 64, - received_stats: ReceivedStats::Both, - stamp_at: StampAt::Both, - clock: Clock::Both, - open_timeouts: vec![Duration::from_millis(200)], - socket_config: SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..SocketConfig::default() + open: OpenPolicy { + timeouts: vec![Duration::from_millis(200)], + ..Default::default() }, + request: SessionRequest { + duration: Some(Duration::from_secs(1)), + interval: Duration::from_millis(100), + length: 64, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Default::default() + }, + probe_timeout: Duration::from_millis(200), ..ClientConfig::default() }; - let mut client = AsyncClient::connect(config).await.unwrap(); + let mut client = AsyncClient::connect(server_addr.to_string(), config) + .await + .unwrap(); let opened = client.open().await.unwrap(); match &opened { diff --git a/crates/irtt-integration-tests/tests/async_client_lifecycle.rs b/crates/irtt-integration-tests/tests/async_client_lifecycle.rs index 210e2ee..194d647 100644 --- a/crates/irtt-integration-tests/tests/async_client_lifecycle.rs +++ b/crates/irtt-integration-tests/tests/async_client_lifecycle.rs @@ -1,4 +1,4 @@ -use irtt_client::{Authentication, HmacKey}; +use irtt_client::{Authentication, HmacKey, OpenPolicy, SessionRequest}; use std::{ future::Future, net::UdpSocket, @@ -17,7 +17,7 @@ use tokio::runtime::{Builder, Runtime}; mod support; use irtt_client::{AsyncClient, ClientConfig, ClientError, ClientEvent, OpenOutcome}; -use irtt_client::{Client, RunMode, SocketConfig}; +use irtt_client::{Client, RunMode}; use irtt_server::ServerConfig; use std::{ io, @@ -49,26 +49,27 @@ fn runtime() -> Runtime { .unwrap() } -fn config(addr: SocketAddr, key: Option>, dscp: u8) -> ClientConfig { +fn config(key: Option>, dscp: u8) -> ClientConfig { ClientConfig { - server_addr: addr.to_string(), - received_stats: ReceivedStats::None, - stamp_at: StampAt::None, - dscp, + open: OpenPolicy { + timeouts: vec![Duration::from_millis(200)], + ..Default::default() + }, + request: SessionRequest { + received_stats: ReceivedStats::None, + stamp_at: StampAt::None, + dscp, + ..Default::default() + }, auth: key.map_or(Authentication::Unauthenticated, |key| { Authentication::Hmac(HmacKey::new(key)) }), - open_timeouts: vec![Duration::from_millis(200)], - socket_config: SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..SocketConfig::default() - }, ..ClientConfig::default() } } async fn opened_client(addr: SocketAddr, dscp: u8) -> AsyncClient { - let mut client = AsyncClient::connect(config(addr, None, dscp)) + let mut client = AsyncClient::connect(addr.to_string(), config(None, dscp)) .await .unwrap(); client.open().await.unwrap(); @@ -251,7 +252,10 @@ fn poll_recv_once(client: &mut AsyncClient) -> Poll, Cli #[test] fn connect_requires_current_runtime_when_polled() { - let mut future = Box::pin(AsyncClient::connect(ClientConfig::default())); + let mut future = Box::pin(AsyncClient::connect( + "127.0.0.1:2112", + ClientConfig::default(), + )); assert!(matches!( poll_once(Pin::as_mut(&mut future)), Poll::Ready(Err(ClientError::NoTokioRuntime)) @@ -262,7 +266,9 @@ fn connect_requires_current_runtime_when_polled() { fn recv_before_open_fails_on_first_poll_without_socket_readiness() { runtime().block_on(async { let remote = SocketAddr::from(([127, 0, 0, 1], 2112)); - let mut client = AsyncClient::connect(config(remote, None, 0)).await.unwrap(); + let mut client = AsyncClient::connect(remote.to_string(), config(None, 0)) + .await + .unwrap(); assert!(matches!( poll_recv_once(&mut client), @@ -312,11 +318,13 @@ fn ignored_open_traffic_cannot_extend_the_absolute_attempt_deadline() { } panic!("second open request did not arrive after the first absolute deadline"); }); - let mut client_config = config(server.addr, None, 0); - client_config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)]; + let mut client_config = config(None, 0); + client_config.open.timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)]; runtime().block_on(async { - let mut client = AsyncClient::connect(client_config).await.unwrap(); + let mut client = AsyncClient::connect(server.addr.to_string(), client_config) + .await + .unwrap(); assert!(matches!( client.open().await.unwrap(), OpenOutcome::Started { .. } @@ -339,11 +347,13 @@ fn authenticated_rejection_is_terminal_without_retry() { 0, ); }); - let mut client_config = config(server.addr, None, 0); - client_config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)]; + let mut client_config = config(None, 0); + client_config.open.timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)]; runtime().block_on(async { - let mut client = AsyncClient::connect(client_config).await.unwrap(); + let mut client = AsyncClient::connect(server.addr.to_string(), client_config) + .await + .unwrap(); assert!(matches!( client.open().await, Err(ClientError::ServerRejected) @@ -371,9 +381,11 @@ fn recv_after_local_close_and_no_test_fails_on_first_poll() { let no_test_server = start_no_test_server(); runtime().block_on(async { - let mut client_config = config(no_test_server.addr, None, 0); - client_config.run_mode = RunMode::NoTest; - let mut client = AsyncClient::connect(client_config).await.unwrap(); + let mut client_config = config(None, 0); + client_config.request.run_mode = RunMode::NoTest; + let mut client = AsyncClient::connect(no_test_server.addr.to_string(), client_config) + .await + .unwrap(); client.open().await.unwrap(); assert!(matches!( poll_recv_once(&mut client), @@ -388,7 +400,7 @@ fn authenticated_async_lifecycle_negotiates_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)) + let mut client = AsyncClient::connect(server.addr.to_string(), config(Some(key), 46)) .await .unwrap(); let opened = client.open().await.unwrap(); @@ -403,7 +415,10 @@ fn authenticated_async_lifecycle_negotiates_dscp() { fn blocking_and_async_peer_close_are_semantically_equivalent() { let blocking_peer_close_server = start_peer_close_server(); let mut blocking_peer_close = - Client::connect(config(blocking_peer_close_server.addr, None, 0)).unwrap(); + Client::connect(blocking_peer_close_server.addr.to_string(), config(None, 0)).unwrap(); + blocking_peer_close + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); blocking_peer_close.open().unwrap(); blocking_peer_close.send_probe().unwrap(); let blocking_peer_close_events = blocking_peer_close.recv_once().unwrap(); @@ -412,9 +427,10 @@ fn blocking_and_async_peer_close_are_semantically_equivalent() { let async_peer_close_server = start_peer_close_server(); let async_peer_close_events = runtime().block_on(async { - let mut client = AsyncClient::connect(config(async_peer_close_server.addr, None, 0)) - .await - .unwrap(); + let mut client = + AsyncClient::connect(async_peer_close_server.addr.to_string(), config(None, 0)) + .await + .unwrap(); client.open().await.unwrap(); client.send_probe().await.unwrap(); let events = client.recv().await.unwrap(); @@ -525,7 +541,10 @@ fn blocking_and_async_roll_back_a_failed_open_and_accept_a_retry() { // One open attempt is ignored, so the first open times out and the second // must be able to open the same connected client. let blocking_server = start_silent_then_open_server(1); - let mut blocking = Client::connect(config(blocking_server.addr, None, 0)).unwrap(); + let mut blocking = Client::connect(blocking_server.addr.to_string(), config(None, 0)).unwrap(); + blocking + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); assert!(!blocking.has_pending_probes()); let blocking_first = blocking.open().unwrap_err(); assert!(!blocking.has_pending_probes()); @@ -539,7 +558,7 @@ fn blocking_and_async_roll_back_a_failed_open_and_accept_a_retry() { let async_server = start_silent_then_open_server(1); let (async_first, async_second, async_sent, async_reply, async_close) = runtime().block_on(async { - let mut client = AsyncClient::connect(config(async_server.addr, None, 0)) + let mut client = AsyncClient::connect(async_server.addr.to_string(), config(None, 0)) .await .unwrap(); assert!(!client.has_pending_probes()); @@ -580,16 +599,16 @@ fn blocking_and_async_roll_back_a_failed_open_and_accept_a_retry() { #[test] fn blocking_and_async_complete_every_caller_paced_probe() { const PROBES: u32 = 3; - fn caller_config(addr: SocketAddr) -> ClientConfig { - ClientConfig { - duration: Some(Duration::from_millis(1)), - interval: Duration::from_secs(1), - ..config(addr, None, 0) - } - } + let mut caller_config = config(None, 0); + caller_config.request.duration = Some(Duration::from_millis(1)); + caller_config.request.interval = Duration::from_secs(1); let blocking_server = InTreeServer::start(ServerConfig::default()); - let mut blocking = Client::connect(caller_config(blocking_server.addr)).unwrap(); + let mut blocking = + Client::connect(blocking_server.addr.to_string(), caller_config.clone()).unwrap(); + blocking + .set_recv_timeout(Some(Duration::from_millis(200))) + .unwrap(); let outcome = blocking.open().unwrap(); assert_eq!( blocking.negotiated_params(), @@ -613,7 +632,7 @@ fn blocking_and_async_complete_every_caller_paced_probe() { let async_server = InTreeServer::start(ServerConfig::default()); let (async_sent, async_replies) = runtime().block_on(async { - let mut client = AsyncClient::connect(caller_config(async_server.addr)) + let mut client = AsyncClient::connect(async_server.addr.to_string(), caller_config) .await .unwrap(); let outcome = client.open().await.unwrap(); @@ -670,11 +689,13 @@ fn ignored_open_traffic_uses_one_request_and_deadline_per_attempt() { } } }); - let mut client_config = config(server.addr, None, 0); - client_config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)]; + let mut client_config = config(None, 0); + client_config.open.timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)]; runtime().block_on(async { - let mut client = AsyncClient::connect(client_config).await.unwrap(); + let mut client = AsyncClient::connect(server.addr.to_string(), client_config) + .await + .unwrap(); assert!(matches!(client.open().await, Err(ClientError::OpenTimeout))); }); assert_eq!(server.finish().len(), 2); diff --git a/crates/irtt-integration-tests/tests/blocking_managed.rs b/crates/irtt-integration-tests/tests/blocking_managed.rs index 3f65168..891191d 100644 --- a/crates/irtt-integration-tests/tests/blocking_managed.rs +++ b/crates/irtt-integration-tests/tests/blocking_managed.rs @@ -1,3 +1,4 @@ +use irtt_client::{OpenPolicy, SessionRequest}; use std::{ net::{SocketAddr, UdpSocket}, sync::{Arc, Condvar, Mutex}, @@ -146,12 +147,18 @@ fn test_runtime() -> tokio::runtime::Runtime { fn config(completion: ManagedCompletionPolicy) -> ManagedClientConfig { ManagedClientConfig { client: ClientConfig { - duration: Some(Duration::from_millis(120)), - interval: Duration::from_millis(20), + open: OpenPolicy { + timeouts: vec![Duration::from_millis(200)], + ..Default::default() + }, + request: SessionRequest { + duration: Some(Duration::from_millis(120)), + interval: Duration::from_millis(20), + received_stats: ReceivedStats::None, + stamp_at: StampAt::None, + ..Default::default() + }, probe_timeout: Duration::from_millis(40), - open_timeouts: vec![Duration::from_millis(200)], - received_stats: ReceivedStats::None, - stamp_at: StampAt::None, ..ClientConfig::default() }, pacing: ManagedPacing::Burst, diff --git a/crates/irtt-integration-tests/tests/fill_negotiation.rs b/crates/irtt-integration-tests/tests/fill_negotiation.rs index c398b49..84227ee 100644 --- a/crates/irtt-integration-tests/tests/fill_negotiation.rs +++ b/crates/irtt-integration-tests/tests/fill_negotiation.rs @@ -1,6 +1,9 @@ +use std::time::Duration; mod support; -use irtt_client::{Client, ClientConfig, ClientError, NegotiationPolicy}; +use irtt_client::{ + Client, ClientConfig, ClientError, NegotiationPolicy, OpenPolicy, SessionRequest, +}; use irtt_proto::{Params, TimestampFields}; use support::{ @@ -52,11 +55,14 @@ fn server_fill_none_short_and_max_map_into_open_params() { fn server_fill_empty_and_oversized_values_are_rejected_at_config_boundary() { for fill in ["", "0123456789abcdef0123456789abcdefx"] { let config = ClientConfig { - server_fill: Some(fill.to_owned()), + request: SessionRequest { + server_fill: Some(fill.to_owned()), + ..Default::default() + }, ..ClientConfig::default() }; assert!(matches!( - Client::connect(config), + Client::connect("127.0.0.1:2112", config), Err(ClientError::InvalidConfig { .. }) )); } @@ -69,7 +75,10 @@ fn server_fill_negotiated_params_reflect_accepted_value() { let mut params = default_params(); params.server_fill = server_fill("rand"); let server = InTreeServer::start(irtt_server::ServerConfig::default()); - let mut client = Client::connect(config_for_params(server.addr, ¶ms)).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config_for_params(¶ms)).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); let outcome = client.open().unwrap(); let negotiated = match outcome { irtt_client::OpenOutcome::Started { negotiated, .. } => negotiated, @@ -99,7 +108,11 @@ fn server_fill_strict_rejects_removed_changed_or_unexpected_fill() { returned.server_fill = returned_fill.and_then(server_fill); let server = start_open_server(returned, None); - let mut client = Client::connect(config_for_params(server.addr, &requested)).unwrap(); + let mut client = + Client::connect(server.addr.to_string(), config_for_params(&requested)).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); assert!(matches!( client.open(), Err(ClientError::NegotiationRejected { .. }) @@ -117,10 +130,13 @@ fn server_fill_loose_allows_server_to_remove_fill() { returned.server_fill = None; let requested_for_config = requested.clone(); - let run = run_one_probe_with_config(returned, TimestampFields::default(), None, |addr| { + let run = run_one_probe_with_config(returned, TimestampFields::default(), None, || { ClientConfig { - negotiation_policy: NegotiationPolicy::Loose, - ..config_for_params(addr, &requested_for_config) + open: OpenPolicy { + negotiation: NegotiationPolicy::Loose, + ..config_for_params(&requested_for_config).open + }, + ..config_for_params(&requested_for_config) } }); diff --git a/crates/irtt-integration-tests/tests/fill_wire_bytes.rs b/crates/irtt-integration-tests/tests/fill_wire_bytes.rs index 7c54ec9..b324e3d 100644 --- a/crates/irtt-integration-tests/tests/fill_wire_bytes.rs +++ b/crates/irtt-integration-tests/tests/fill_wire_bytes.rs @@ -10,7 +10,8 @@ use std::time::Duration; use irtt_client::{ - AsyncClient, ClientConfig, NegotiationPolicy, NegotiationRestriction, OpenOutcome, SocketConfig, + AsyncClient, ClientConfig, NegotiationPolicy, NegotiationRestriction, OpenOutcome, OpenPolicy, + SessionRequest, }; use irtt_proto::{ decode_echo_reply, decode_open_reply, echo_header_len, encode_request, Clock, Params, @@ -132,24 +133,26 @@ async fn exercise_fallback_negotiation() { // `ClientConfig` accepts any non-empty descriptor within the wire bound, so // an unknown one genuinely reaches the server. let config = |policy| ClientConfig { - server_addr: server_addr.to_string(), - duration: Some(Duration::from_secs(1)), - interval: Duration::from_millis(100), - length: 64, - server_fill: Some("bogus".to_owned()), - negotiation_policy: policy, - open_timeouts: vec![Duration::from_millis(200)], - socket_config: SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..SocketConfig::default() + open: OpenPolicy { + negotiation: policy, + timeouts: vec![Duration::from_millis(200)], }, + request: SessionRequest { + duration: Some(Duration::from_secs(1)), + interval: Duration::from_millis(100), + length: 64, + server_fill: Some("bogus".to_owned()), + ..Default::default() + }, + probe_timeout: Duration::from_millis(200), ..ClientConfig::default() }; - let mut strict = AsyncClient::connect(config(NegotiationPolicy::Strict)) - .await - .unwrap(); + let mut strict = + AsyncClient::connect(server_addr.to_string(), config(NegotiationPolicy::Strict)) + .await + .unwrap(); let rejected = strict .open() .await @@ -159,7 +162,7 @@ async fn exercise_fallback_negotiation() { "the restriction names the fill: {rejected}" ); - let mut loose = AsyncClient::connect(config(NegotiationPolicy::Loose)) + let mut loose = AsyncClient::connect(server_addr.to_string(), config(NegotiationPolicy::Loose)) .await .unwrap(); match loose.open().await.unwrap() { diff --git a/crates/irtt-integration-tests/tests/hmac.rs b/crates/irtt-integration-tests/tests/hmac.rs index b40bfa5..4d99bc9 100644 --- a/crates/irtt-integration-tests/tests/hmac.rs +++ b/crates/irtt-integration-tests/tests/hmac.rs @@ -3,9 +3,7 @@ mod support; use std::time::Duration; -use irtt_client::{ - Client, ClientConfig, ClientError, ClientEvent, OpenOutcome, SocketConfig, WarningKind, -}; +use irtt_client::{Client, ClientConfig, ClientError, ClientEvent, OpenOutcome, WarningKind}; use irtt_proto::TimestampFields; use support::{ @@ -27,9 +25,9 @@ fn hmac_open_success_negotiates_without_warnings() { params.clone(), TimestampFields::default(), Some(key), - |addr| ClientConfig { + || ClientConfig { auth: Authentication::Hmac(HmacKey::new(config_key)), - ..config_for_params(addr, &config_params) + ..config_for_params(&config_params) }, ); @@ -51,10 +49,10 @@ fn hmac_echo_success_verifies_request_and_accepts_reply() { let config_params = params.clone(); let config_key = key.clone(); - let run = run_one_probe_with_config(params.clone(), standard_timestamps(), Some(key), |addr| { + let run = run_one_probe_with_config(params.clone(), standard_timestamps(), Some(key), || { ClientConfig { auth: Authentication::Hmac(HmacKey::new(config_key)), - ..config_for_params(addr, &config_params) + ..config_for_params(&config_params) } }); @@ -94,10 +92,13 @@ fn hmac_close_success_sends_authenticated_close_and_closes_session() { let key = b"compat-secret".to_vec(); let params = default_params(); let server = start_hmac_close_server(params.clone(), key.clone()); - let mut config = config_for_params(server.addr, ¶ms); + let mut config = config_for_params(¶ms); config.auth = Authentication::Hmac(HmacKey::new(key)); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); let outcome = client.open().unwrap(); assert_started(outcome, ¶ms); @@ -134,13 +135,16 @@ fn hmac_required_server_rejects_missing_or_wrong_client_key() { ] { let server_key = b"compat-secret".to_vec(); let server = start_hmac_required_open_drop_server(server_key, Duration::from_millis(250)); - let mut config = config_for_params(server.addr, &default_params()); - config.open_timeouts = vec![Duration::from_millis(200)]; + let mut config = config_for_params(&default_params()); + config.open.timeouts = vec![Duration::from_millis(200)]; config.auth = hmac_key.map_or(Authentication::Unauthenticated, |key| { Authentication::Hmac(HmacKey::new(key)) }); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); let observations = server.observations(1); @@ -160,14 +164,13 @@ fn bad_hmac_echo_reply_is_rejected_without_echo_reply_event() { let key = b"compat-secret".to_vec(); let params = default_params(); let server = start_bad_hmac_echo_reply_server(params.clone(), key.clone()); - let mut config = config_for_params(server.addr, ¶ms); + let mut config = config_for_params(¶ms); config.auth = Authentication::Hmac(HmacKey::new(key)); - config.socket_config = SocketConfig { - recv_timeout: Some(Duration::from_millis(500)), - ..Default::default() - }; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); let outcome = client.open().unwrap(); assert_started(outcome, ¶ms); @@ -207,10 +210,13 @@ fn hmac_open_reply_with_bad_hmac_is_ignored_until_timeout() { let wrong_key = b"wrong-secret".to_vec(); let params = default_params(); let server = support::start_bad_hmac_open_reply_server(params.clone(), key.clone(), wrong_key); - let mut config = config_for_params(server.addr, ¶ms); + let mut config = config_for_params(¶ms); config.auth = Authentication::Hmac(HmacKey::new(key)); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); server.join(); } @@ -223,10 +229,13 @@ fn non_hmac_client_open_does_not_set_hmac_flag() { irtt_proto::Clock::Both, ); let server = support::start_open_server(params.clone(), None); - let mut config = config_for_params(server.addr, ¶ms); + let mut config = config_for_params(¶ms); config.auth = Authentication::Unauthenticated; - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); let outcome = client.open().unwrap(); assert_started(outcome, ¶ms); @@ -280,10 +289,13 @@ fn backend_hmac_correct_key_succeeds() { let key = b"compat-secret".to_vec(); let params = default_params(); let peer = BackendPeer::start_open_echo(Some(key.clone())); - let mut config = config_for_params(peer.addr(), ¶ms); + let mut config = config_for_params(¶ms); config.auth = Authentication::Hmac(HmacKey::new(key)); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(peer.addr().to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); let outcome = client.open().unwrap(); assert!(matches!(outcome, OpenOutcome::Started { .. })); @@ -302,13 +314,16 @@ fn backend_hmac_required_rejects_missing_or_wrong_client_key() { for hmac_key in [None, Some(b"wrong-secret".to_vec())] { let server_key = b"compat-secret".to_vec(); let peer = BackendPeer::start_hmac_required(server_key); - let mut config = config_for_params(peer.addr(), &default_params()); - config.open_timeouts = vec![Duration::from_millis(200)]; + let mut config = config_for_params(&default_params()); + config.open.timeouts = vec![Duration::from_millis(200)]; config.auth = hmac_key.map_or(Authentication::Unauthenticated, |key| { Authentication::Hmac(HmacKey::new(key)) }); - let mut client = Client::connect(config).unwrap(); + let mut client = Client::connect(peer.addr().to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); } } diff --git a/crates/irtt-integration-tests/tests/large_packets.rs b/crates/irtt-integration-tests/tests/large_packets.rs index 7aeeda7..b6bce33 100644 --- a/crates/irtt-integration-tests/tests/large_packets.rs +++ b/crates/irtt-integration-tests/tests/large_packets.rs @@ -1,6 +1,6 @@ mod support; -use irtt_client::{Client, ClientConfig, ClientEvent, NegotiationPolicy}; +use irtt_client::{Client, ClientEvent, NegotiationPolicy}; use irtt_proto::{echo_header_len, Params, TimestampFields}; use support::{config_for_params, default_params, run_one_probe, BackendPeer, ServerObservation}; @@ -84,12 +84,13 @@ fn echo_reply_bytes_consistent_with_echo_packet_len_for_various_lengths() { fn backend_large_packet_smoke() { let params = params_with_length(1472); let peer = BackendPeer::start_open_echo(None); - let mut client = Client::connect(ClientConfig { - negotiation_policy: NegotiationPolicy::Loose, - length: 1472, - ..config_for_params(peer.addr(), ¶ms) - }) - .unwrap(); + let mut config = config_for_params(¶ms); + config.open.negotiation = NegotiationPolicy::Loose; + config.request.length = 1472; + let mut client = Client::connect(peer.addr().to_string(), config).unwrap(); + client + .set_recv_timeout(Some(std::time::Duration::from_millis(500))) + .unwrap(); let outcome = client.open().unwrap(); assert!(matches!(outcome, irtt_client::OpenOutcome::Started { .. })); diff --git a/crates/irtt-integration-tests/tests/managed_behavior.rs b/crates/irtt-integration-tests/tests/managed_behavior.rs index 10bd4fa..523548e 100644 --- a/crates/irtt-integration-tests/tests/managed_behavior.rs +++ b/crates/irtt-integration-tests/tests/managed_behavior.rs @@ -2,7 +2,7 @@ mod support; use std::{collections::HashMap, time::Duration}; -use irtt_client::{managed::*, ClientConfig, ClientEvent}; +use irtt_client::{managed::*, ClientConfig, ClientEvent, OpenPolicy, SessionRequest}; use irtt_server::ServerConfig; use support::InTreeServer; use tokio::time::timeout; @@ -18,10 +18,16 @@ async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() { for _ in 0..3 { let config = ManagedClientConfig { client: ClientConfig { - duration: Some(Duration::from_millis(600)), - interval: Duration::from_millis(100), + open: OpenPolicy { + timeouts: vec![Duration::from_millis(200)], + ..Default::default() + }, + request: SessionRequest { + duration: Some(Duration::from_millis(600)), + interval: Duration::from_millis(100), + ..Default::default() + }, probe_timeout: Duration::from_millis(100), - open_timeouts: vec![Duration::from_millis(200)], ..ClientConfig::default() }, pacing, @@ -111,10 +117,16 @@ async fn target_updates_preserve_identical_sessions_and_retire_removed_generatio let server = InTreeServer::start(ServerConfig::default()); let config = ManagedClientConfig { client: ClientConfig { - duration: None, - interval: Duration::from_millis(20), + open: OpenPolicy { + timeouts: vec![Duration::from_millis(200)], + ..Default::default() + }, + request: SessionRequest { + duration: None, + interval: Duration::from_millis(20), + ..Default::default() + }, probe_timeout: Duration::from_millis(100), - open_timeouts: vec![Duration::from_millis(200)], ..ClientConfig::default() }, completion: ManagedCompletionPolicy::ExplicitStop, diff --git a/crates/irtt-integration-tests/tests/managed_lifecycle.rs b/crates/irtt-integration-tests/tests/managed_lifecycle.rs index a9b2114..6fcee0e 100644 --- a/crates/irtt-integration-tests/tests/managed_lifecycle.rs +++ b/crates/irtt-integration-tests/tests/managed_lifecycle.rs @@ -1,4 +1,4 @@ -use irtt_client::{Authentication, HmacKey}; +use irtt_client::{Authentication, HmacKey, OpenPolicy, SessionRequest}; use std::{ future::Future, net::{SocketAddr, UdpSocket}, @@ -176,12 +176,18 @@ fn runtime() -> Runtime { fn config(pacing: ManagedPacing) -> ManagedClientConfig { ManagedClientConfig { client: ClientConfig { - duration: Some(Duration::from_millis(130)), - interval: Duration::from_millis(60), + open: OpenPolicy { + timeouts: vec![Duration::from_millis(200)], + ..Default::default() + }, + request: SessionRequest { + duration: Some(Duration::from_millis(130)), + interval: Duration::from_millis(60), + received_stats: ReceivedStats::None, + stamp_at: StampAt::None, + ..Default::default() + }, probe_timeout: Duration::from_millis(35), - open_timeouts: vec![Duration::from_millis(200)], - received_stats: ReceivedStats::None, - stamp_at: StampAt::None, ..ClientConfig::default() }, pacing, @@ -340,7 +346,7 @@ fn finite_immediate_replies_use_retained_drain_deadline() { fn no_test_target_completion() { let server = start_server(ServerBehavior::NoTest, None); let mut config = config(ManagedPacing::Staggered); - config.client.run_mode = RunMode::NoTest; + config.client.request.run_mode = RunMode::NoTest; let (task, _) = ManagedClient::task(config, vec![target("no-test", server.addr)]).unwrap(); let outcome = runtime().block_on(task); let records = server.finish(); @@ -411,8 +417,8 @@ fn pending_limit_failure_drains_reply_and_closes_session() { Some(key.clone()), ); let mut managed = config(ManagedPacing::Staggered); - managed.client.duration = Some(Duration::from_millis(250)); - managed.client.interval = Duration::from_millis(20); + managed.client.request.duration = Some(Duration::from_millis(250)); + managed.client.request.interval = Duration::from_millis(20); managed.client.probe_timeout = Duration::from_secs(2); managed.client.max_pending_probes = 1; managed.final_drain = Duration::from_millis(50); diff --git a/crates/irtt-integration-tests/tests/negotiation_policy.rs b/crates/irtt-integration-tests/tests/negotiation_policy.rs index 1fde645..dd5cad5 100644 --- a/crates/irtt-integration-tests/tests/negotiation_policy.rs +++ b/crates/irtt-integration-tests/tests/negotiation_policy.rs @@ -10,7 +10,8 @@ use std::time::Duration; use irtt_client::{ - AsyncClient, ClientConfig, NegotiationPolicy, NegotiationRestriction, OpenOutcome, SocketConfig, + AsyncClient, ClientConfig, NegotiationPolicy, NegotiationRestriction, OpenOutcome, OpenPolicy, + SessionRequest, }; use irtt_proto::{Clock, ReceivedStats, StampAt}; use irtt_server::{Server, ServerConfig, TimestampAllowance}; @@ -84,13 +85,19 @@ async fn exercise_dscp_restriction() { // The client's configuration is a DSCP codepoint; 46 (EF) is raw byte 184 on // the wire, and this test does not change either convention. let with_dscp = |policy| ClientConfig { - dscp: 46, - ..test_config(server_addr, policy) + request: SessionRequest { + dscp: 46, + ..test_config(policy).request + }, + ..test_config(policy) }; - let mut strict = AsyncClient::connect(with_dscp(NegotiationPolicy::Strict)) - .await - .unwrap(); + let mut strict = AsyncClient::connect( + server_addr.to_string(), + with_dscp(NegotiationPolicy::Strict), + ) + .await + .unwrap(); let rejected = strict .open() .await @@ -100,9 +107,10 @@ async fn exercise_dscp_restriction() { "the restriction names DSCP: {rejected}" ); - let mut loose = AsyncClient::connect(with_dscp(NegotiationPolicy::Loose)) - .await - .unwrap(); + let mut loose = + AsyncClient::connect(server_addr.to_string(), with_dscp(NegotiationPolicy::Loose)) + .await + .unwrap(); let negotiated = match loose.open().await.unwrap() { OpenOutcome::Started { negotiated, .. } => negotiated, OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"), @@ -138,10 +146,16 @@ async fn a_default_server_restricts_nothing_an_ordinary_client_asks_for() { async fn exercise_default_policy() { let (server_addr, shutdown_tx, server_task) = serve(ServerConfig::default()).await; - let mut client = AsyncClient::connect(ClientConfig { - dscp: 46, - ..test_config(server_addr, NegotiationPolicy::Strict) - }) + let mut client = AsyncClient::connect( + server_addr.to_string(), + ClientConfig { + request: SessionRequest { + dscp: 46, + ..test_config(NegotiationPolicy::Strict).request + }, + ..test_config(NegotiationPolicy::Strict) + }, + ) .await .unwrap(); let negotiated = match client.open().await.unwrap() { @@ -186,28 +200,29 @@ async fn serve( } /// A short deterministic client requesting both timestamps from both clocks. -fn test_config(server_addr: std::net::SocketAddr, policy: NegotiationPolicy) -> ClientConfig { +fn test_config(policy: NegotiationPolicy) -> ClientConfig { ClientConfig { - server_addr: server_addr.to_string(), - duration: Some(Duration::from_secs(1)), - interval: Duration::from_millis(100), - length: 64, - received_stats: ReceivedStats::Both, - stamp_at: StampAt::Both, - clock: Clock::Both, - negotiation_policy: policy, - open_timeouts: vec![Duration::from_millis(200)], - socket_config: SocketConfig { - recv_timeout: Some(Duration::from_millis(200)), - ..SocketConfig::default() + open: OpenPolicy { + negotiation: policy, + timeouts: vec![Duration::from_millis(200)], }, + request: SessionRequest { + duration: Some(Duration::from_secs(1)), + interval: Duration::from_millis(100), + length: 64, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Default::default() + }, + probe_timeout: Duration::from_millis(200), ..ClientConfig::default() } } async fn client(server_addr: std::net::SocketAddr, policy: NegotiationPolicy) -> AsyncClient { - AsyncClient::connect(test_config(server_addr, policy)) + AsyncClient::connect(server_addr.to_string(), test_config(policy)) .await .unwrap() } diff --git a/crates/irtt-integration-tests/tests/protocol_options.rs b/crates/irtt-integration-tests/tests/protocol_options.rs index fe5ac2d..023dbb4 100644 --- a/crates/irtt-integration-tests/tests/protocol_options.rs +++ b/crates/irtt-integration-tests/tests/protocol_options.rs @@ -1,4 +1,4 @@ -use irtt_client::{Authentication, HmacKey}; +use irtt_client::{Authentication, HmacKey, OpenPolicy}; mod support; mod protocol_options { pub mod backend_smoke; @@ -147,10 +147,10 @@ fn hmac_rich_mode_uses_negotiated_echo_layout_and_decodes_reply() { let params = params_for_modes(ReceivedStats::Both, StampAt::Both, Clock::Both); let config_params = params.clone(); let config_key = key.clone(); - let run = run_one_probe_with_config(params.clone(), standard_timestamps(), Some(key), |addr| { + let run = run_one_probe_with_config(params.clone(), standard_timestamps(), Some(key), || { ClientConfig { auth: Authentication::Hmac(HmacKey::new(config_key)), - ..config_for_params(addr, &config_params) + ..config_for_params(&config_params) } }); assert_negotiated_echo_use(&run, ¶ms, true); @@ -171,9 +171,11 @@ fn strict_open_rejects_changed_compatibility_params_from_server() { for returned in changed_compatibility_params(&requested) { let server = start_open_server(returned, None); - let mut config = config_for_params(server.addr, &requested); - config.server_addr = server.addr.to_string(); - let mut client = Client::connect(config).unwrap(); + let config = config_for_params(&requested); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); assert!(matches!( client.open(), Err(ClientError::NegotiationRejected { .. }) @@ -199,10 +201,13 @@ fn loose_open_uses_returned_params_for_echo_layout_and_reply_parsing() { returned.server_fill = None; let requested_for_config = requested.clone(); - let run = run_one_probe_with_config(returned.clone(), standard_timestamps(), None, |addr| { + let run = run_one_probe_with_config(returned.clone(), standard_timestamps(), None, || { ClientConfig { - negotiation_policy: irtt_client::NegotiationPolicy::Loose, - ..config_for_params(addr, &requested_for_config) + open: OpenPolicy { + negotiation: irtt_client::NegotiationPolicy::Loose, + ..config_for_params(&requested_for_config).open + }, + ..config_for_params(&requested_for_config) } }); diff --git a/crates/irtt-integration-tests/tests/protocol_options/backend_smoke.rs b/crates/irtt-integration-tests/tests/protocol_options/backend_smoke.rs index d1c39a2..2d7e0d5 100644 --- a/crates/irtt-integration-tests/tests/protocol_options/backend_smoke.rs +++ b/crates/irtt-integration-tests/tests/protocol_options/backend_smoke.rs @@ -1,5 +1,6 @@ use irtt_client::{Client, ClientEvent}; use irtt_proto::{Clock, ReceivedStats, StampAt}; +use std::time::Duration; use crate::support::{config_for_params, default_params, params_for_modes, BackendPeer}; @@ -7,7 +8,10 @@ use crate::support::{config_for_params, default_params, params_for_modes, Backen fn backend_basic_open_echo_close() { let params = default_params(); let peer = BackendPeer::start_open_echo(None); - let mut client = Client::connect(config_for_params(peer.addr(), ¶ms)).unwrap(); + let mut client = Client::connect(peer.addr().to_string(), config_for_params(¶ms)).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); let outcome = client.open().unwrap(); assert!(matches!(outcome, irtt_client::OpenOutcome::Started { .. })); @@ -32,7 +36,11 @@ fn backend_received_stats_smoke() { ] { let params = params_for_modes(mode, StampAt::None, Clock::Both); let peer = BackendPeer::start_open_echo(None); - let mut client = Client::connect(config_for_params(peer.addr(), ¶ms)).unwrap(); + let mut client = + Client::connect(peer.addr().to_string(), config_for_params(¶ms)).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); client.open().unwrap(); client.send_probe().unwrap(); @@ -55,7 +63,11 @@ fn backend_timestamp_smoke() { ] { let params = params_for_modes(ReceivedStats::None, mode, Clock::Both); let peer = BackendPeer::start_open_echo(None); - let mut client = Client::connect(config_for_params(peer.addr(), ¶ms)).unwrap(); + let mut client = + Client::connect(peer.addr().to_string(), config_for_params(¶ms)).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); client.open().unwrap(); client.send_probe().unwrap(); @@ -102,7 +114,11 @@ fn backend_clock_smoke() { for clock in [Clock::Wall, Clock::Monotonic, Clock::Both] { let params = params_for_modes(ReceivedStats::None, StampAt::Both, clock); let peer = BackendPeer::start_open_echo(None); - let mut client = Client::connect(config_for_params(peer.addr(), ¶ms)).unwrap(); + let mut client = + Client::connect(peer.addr().to_string(), config_for_params(¶ms)).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); client.open().unwrap(); client.send_probe().unwrap(); diff --git a/crates/irtt-integration-tests/tests/support/mod.rs b/crates/irtt-integration-tests/tests/support/mod.rs index 24274e0..1f12793 100644 --- a/crates/irtt-integration-tests/tests/support/mod.rs +++ b/crates/irtt-integration-tests/tests/support/mod.rs @@ -12,7 +12,9 @@ use std::{ time::{Duration, Instant, SystemTime}, }; -use irtt_client::{Client, ClientConfig, ClientEvent, NegotiatedParams, OpenOutcome, SocketConfig}; +use irtt_client::{ + Client, ClientConfig, ClientEvent, NegotiatedParams, OpenOutcome, OpenPolicy, SessionRequest, +}; use irtt_proto::{ compute_hmac_in_place, decode_request, echo_packet_len, encode_echo_reply, encode_open_reply, flags, verify_packet_hmac, Clock, DecodedRequest, DecodedRequestKind, EchoReply, OpenReply, @@ -166,12 +168,12 @@ pub fn params_for_modes(received_stats: ReceivedStats, stamp_at: StampAt, clock: /// Builds a [`ClientConfig`] whose negotiated `Params` will match `params`. /// /// `params.dscp` is the raw IP TOS / Traffic Class byte, while -/// [`ClientConfig::dscp`] is the six-bit codepoint that produces it; shift +/// [`SessionRequest::dscp`] is the six-bit codepoint that produces it; shift /// right by two to recover the codepoint the resulting config must request. /// Panics if `params.dscp` has any ECN bits set, since those cannot be /// recovered from a codepoint and this helper's contract would silently stop /// holding otherwise. -pub fn config_for_params(addr: SocketAddr, params: &Params) -> ClientConfig { +pub fn config_for_params(params: &Params) -> ClientConfig { let traffic_class = u8::try_from(params.dscp).expect("test DSCP must fit the raw traffic-class byte"); assert_eq!( @@ -180,26 +182,27 @@ pub fn config_for_params(addr: SocketAddr, params: &Params) -> ClientConfig { "test Params::dscp must be representable by ClientConfig's DSCP codepoint" ); ClientConfig { - server_addr: addr.to_string(), - duration: if params.duration_ns == 0 { - None - } else { - Some(Duration::from_nanos( - u64::try_from(params.duration_ns).expect("test duration must be non-negative"), - )) + open: OpenPolicy { + timeouts: vec![Duration::from_millis(200)], + ..Default::default() }, - interval: Duration::from_nanos( - u64::try_from(params.interval_ns).expect("test interval must be non-negative"), - ), - length: u32::try_from(params.length).unwrap(), - received_stats: params.received_stats, - stamp_at: params.stamp_at, - clock: params.clock, - dscp: traffic_class >> 2, - server_fill: params.server_fill.as_ref().map(|fill| fill.value.clone()), - open_timeouts: vec![Duration::from_millis(200)], - socket_config: SocketConfig { - recv_timeout: Some(Duration::from_millis(500)), + request: SessionRequest { + duration: if params.duration_ns == 0 { + None + } else { + Some(Duration::from_nanos( + u64::try_from(params.duration_ns).expect("test duration must be non-negative"), + )) + }, + interval: Duration::from_nanos( + u64::try_from(params.interval_ns).expect("test interval must be non-negative"), + ), + length: u32::try_from(params.length).unwrap(), + received_stats: params.received_stats, + stamp_at: params.stamp_at, + clock: params.clock, + dscp: traffic_class >> 2, + server_fill: params.server_fill.as_ref().map(|fill| fill.value.clone()), ..Default::default() }, ..ClientConfig::default() @@ -219,8 +222,8 @@ pub fn standard_timestamps() -> TimestampFields { pub fn run_one_probe(params: Params, timestamps: TimestampFields) -> OneProbeRun { let config_params = params.clone(); - run_one_probe_with_config(params, timestamps, None, |addr| { - config_for_params(addr, &config_params) + run_one_probe_with_config(params, timestamps, None, || { + config_for_params(&config_params) }) } @@ -231,10 +234,13 @@ pub fn run_one_probe_with_config( build_config: F, ) -> OneProbeRun where - F: FnOnce(SocketAddr) -> ClientConfig, + F: FnOnce() -> ClientConfig, { let server = start_one_probe_server(server_params, timestamps, hmac_key); - let mut client = Client::connect(build_config(server.addr)).unwrap(); + let mut client = Client::connect(server.addr.to_string(), build_config()).unwrap(); + client + .set_recv_timeout(Some(Duration::from_millis(500))) + .unwrap(); let negotiated = assert_started(client.open().unwrap()); let sent_events = client.send_probe().unwrap(); diff --git a/crates/irtt-integration-tests/tests/traffic_class.rs b/crates/irtt-integration-tests/tests/traffic_class.rs index 129454d..02c66b4 100644 --- a/crates/irtt-integration-tests/tests/traffic_class.rs +++ b/crates/irtt-integration-tests/tests/traffic_class.rs @@ -14,7 +14,9 @@ use std::{io, net::SocketAddr, time::Duration}; -use irtt_client::{AsyncClient, ClientConfig, ClientEvent, OpenOutcome, PacketMeta, SocketConfig}; +use irtt_client::{ + AsyncClient, ClientConfig, ClientEvent, OpenOutcome, OpenPolicy, PacketMeta, SessionRequest, +}; use irtt_proto::{Clock, ReceivedStats, StampAt}; use irtt_server::{Server, ServerConfig, ServerRuntimeError}; use tokio::{sync::oneshot, task::JoinHandle, time::timeout}; @@ -193,23 +195,27 @@ impl OpenSession { /// A client with an open session requesting `dscp` as a codepoint. async fn opened_session(server_addr: SocketAddr, dscp: u8) -> OpenSession { let config = ClientConfig { - server_addr: server_addr.to_string(), - duration: Some(Duration::from_secs(30)), - interval: Duration::from_millis(10), - length: 64, - dscp, - received_stats: ReceivedStats::Both, - stamp_at: StampAt::Both, - clock: Clock::Both, - open_timeouts: vec![Duration::from_millis(500)], - socket_config: SocketConfig { - recv_timeout: Some(Duration::from_millis(500)), - ..SocketConfig::default() + open: OpenPolicy { + timeouts: vec![Duration::from_millis(500)], + ..Default::default() }, + request: SessionRequest { + duration: Some(Duration::from_secs(30)), + interval: Duration::from_millis(10), + length: 64, + dscp, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Default::default() + }, + probe_timeout: Duration::from_millis(500), ..ClientConfig::default() }; - let mut client = AsyncClient::connect(config).await.unwrap(); + let mut client = AsyncClient::connect(server_addr.to_string(), config) + .await + .unwrap(); match client.open().await.unwrap() { OpenOutcome::Started { negotiated, .. } => assert_eq!( negotiated.params.dscp, diff --git a/crates/irtt-integration-tests/tests/wildcard_source.rs b/crates/irtt-integration-tests/tests/wildcard_source.rs index b772750..24c2a67 100644 --- a/crates/irtt-integration-tests/tests/wildcard_source.rs +++ b/crates/irtt-integration-tests/tests/wildcard_source.rs @@ -18,7 +18,9 @@ use std::{ time::Duration, }; -use irtt_client::{AsyncClient, ClientConfig, ClientEvent, OpenOutcome, PacketMeta, SocketConfig}; +use irtt_client::{ + AsyncClient, ClientConfig, ClientEvent, OpenOutcome, OpenPolicy, PacketMeta, SessionRequest, +}; use irtt_proto::{Clock, ReceivedStats, StampAt}; use irtt_server::{Server, ServerConfig, ServerRuntimeError}; use tokio::{sync::oneshot, task::JoinHandle, time::timeout}; @@ -193,23 +195,27 @@ fn unthrottled() -> ServerConfig { async fn connected_client(server_addr: SocketAddr) -> AsyncClient { let config = ClientConfig { - server_addr: server_addr.to_string(), - duration: Some(Duration::from_secs(30)), - interval: Duration::from_millis(10), - length: 64, - dscp: EF_DSCP, - received_stats: ReceivedStats::Both, - stamp_at: StampAt::Both, - clock: Clock::Both, - open_timeouts: vec![Duration::from_millis(500)], - socket_config: SocketConfig { - recv_timeout: Some(Duration::from_millis(500)), - ..SocketConfig::default() + open: OpenPolicy { + timeouts: vec![Duration::from_millis(500)], + ..Default::default() }, + request: SessionRequest { + duration: Some(Duration::from_secs(30)), + interval: Duration::from_millis(10), + length: 64, + dscp: EF_DSCP, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Default::default() + }, + probe_timeout: Duration::from_millis(500), ..ClientConfig::default() }; - AsyncClient::connect(config).await.unwrap() + AsyncClient::connect(server_addr.to_string(), config) + .await + .unwrap() } async fn assert_started(client: &mut AsyncClient, server_addr: SocketAddr) {