From d467211f417388363fa5d7f0feddb5024dd4d464 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 4 Oct 2026 18:19:28 +0700 Subject: [PATCH] refactor(client): separate accepted negotiation state from peer evidence --- crates/irtt-app/src/cmd/client/output.rs | 25 +- crates/irtt-app/src/cmd/tui/ui.rs | 41 +-- .../irtt-client/examples/blocking_client.rs | 20 +- crates/irtt-client/examples/tokio_client.rs | 20 +- crates/irtt-client/src/async_client.rs | 20 +- crates/irtt-client/src/client.rs | 21 +- crates/irtt-client/src/config.rs | 4 +- crates/irtt-client/src/event.rs | 75 ++--- crates/irtt-client/src/lib.rs | 19 +- crates/irtt-client/src/managed/schedule.rs | 49 ++-- crates/irtt-client/src/managed/task.rs | 48 ++-- crates/irtt-client/src/session/machine.rs | 44 ++- crates/irtt-client/src/session/mod.rs | 2 +- crates/irtt-client/src/session/negotiation.rs | 264 +++++++++++------- crates/irtt-client/tests/client/mod.rs | 2 +- crates/irtt-client/tests/client/no_test.rs | 47 ++-- crates/irtt-client/tests/client/support.rs | 29 +- crates/irtt-client/tests/linux_transport.rs | 8 +- .../tests/async_client.rs | 19 +- .../tests/async_client_lifecycle.rs | 31 +- .../tests/fill_negotiation.rs | 16 +- .../tests/fill_wire_bytes.rs | 22 +- crates/irtt-integration-tests/tests/hmac.rs | 20 +- .../tests/large_packets.rs | 2 +- .../tests/managed_behavior.rs | 2 +- .../tests/negotiation_policy.rs | 51 ++-- .../tests/protocol_options.rs | 21 +- .../tests/protocol_options/backend_smoke.rs | 2 +- .../tests/support/mod.rs | 13 +- .../tests/traffic_class.rs | 9 +- .../tests/wildcard_source.rs | 4 +- crates/irtt-stats/src/normalization.rs | 4 +- 32 files changed, 507 insertions(+), 447 deletions(-) diff --git a/crates/irtt-app/src/cmd/client/output.rs b/crates/irtt-app/src/cmd/client/output.rs index ae3e2e8..d09738b 100644 --- a/crates/irtt-app/src/cmd/client/output.rs +++ b/crates/irtt-app/src/cmd/client/output.rs @@ -5,7 +5,7 @@ use std::{ }; use irtt_client::{ - ClientEvent, NegotiatedParams, OneWayDelaySample, PacketMeta, ReceivedStatsSample, RttSample, + ClientEvent, NegotiationResult, OneWayDelaySample, PacketMeta, ReceivedStatsSample, RttSample, ServerTiming, SignedDuration, WarningKind, }; @@ -234,22 +234,22 @@ enum OutputRow { impl OutputRow { fn from_event(event: &ClientEvent) -> Self { match event { - ClientEvent::SessionStarted { + ClientEvent::SessionStarted(irtt_client::SessionStarted { remote, token, - negotiated, + negotiation: negotiated, at, - } => Self::SessionStarted(LifecycleRow::new( + }) => Self::SessionStarted(LifecycleRow::new( *remote, Some(*token), negotiated, at.wall, )), - ClientEvent::NoTestCompleted { + ClientEvent::NoTestCompleted(irtt_client::NoTestCompleted { remote, - negotiated, + negotiation: negotiated, at, - } => Self::NoTestCompleted(LifecycleRow::new(*remote, None, negotiated, at.wall)), + }) => Self::NoTestCompleted(LifecycleRow::new(*remote, None, negotiated, at.wall)), ClientEvent::SessionClosed { remote, token, at } => Self::SessionClosed { remote: *remote, token: *token, @@ -384,16 +384,19 @@ impl LifecycleRow { fn new( remote: SocketAddr, token: Option, - negotiated: &NegotiatedParams, + negotiated: &NegotiationResult, event_wall: SystemTime, ) -> Self { Self { remote, token, event_wall, - duration_ns: i128::from(negotiated.params.duration_ns), - interval_ns: i128::from(negotiated.params.interval_ns), - payload_length: i128::from(negotiated.params.length), + duration_ns: negotiated + .accepted + .duration + .map_or(0, |duration| duration.as_nanos() as i128), + interval_ns: negotiated.accepted.interval.as_nanos() as i128, + payload_length: i128::from(negotiated.accepted.length), } } } diff --git a/crates/irtt-app/src/cmd/tui/ui.rs b/crates/irtt-app/src/cmd/tui/ui.rs index 2ed05d3..170d105 100644 --- a/crates/irtt-app/src/cmd/tui/ui.rs +++ b/crates/irtt-app/src/cmd/tui/ui.rs @@ -10,7 +10,7 @@ use crossterm::{ terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen}, }; use irtt_client::managed::{ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance}; -use irtt_client::{ClientEvent, NegotiatedParams, SignedDuration}; +use irtt_client::{ClientEvent, NegotiationResult, SignedDuration}; use irtt_stats::{Snapshot, StatsCollector, TimeStats}; use ratatui::{ backend::CrosstermBackend, @@ -266,12 +266,12 @@ impl TuiState { }; target.stats.process(event); match event { - ClientEvent::SessionStarted { + ClientEvent::SessionStarted(irtt_client::SessionStarted { remote, token, - negotiated, + negotiation: negotiated, .. - } => { + }) => { target.remote = Some(remote.to_string()); target.session = Some(format!("{token:#x}")); target.negotiated = Some(negotiated.clone()); @@ -279,9 +279,11 @@ impl TuiState { global_status = Some(TuiStatus::Running); recent = Some(format!("session started token={token:#x}")); } - ClientEvent::NoTestCompleted { - remote, negotiated, .. - } => { + ClientEvent::NoTestCompleted(irtt_client::NoTestCompleted { + remote, + negotiation: negotiated, + .. + }) => { target.remote = Some(remote.to_string()); target.negotiated = Some(negotiated.clone()); target.status = TargetStatus::NoTest; @@ -541,7 +543,7 @@ pub(super) struct TuiTargetState { remote: Option, session: Option, status: TargetStatus, - negotiated: Option, + negotiated: Option, graph_history: VecDeque, last_sample: Option, last_warning: Option, @@ -1897,27 +1899,26 @@ fn push_time_line(lines: &mut Vec>, label: &str, stats: &TimeStats) { ))); } -fn format_negotiated(negotiated: &NegotiatedParams) -> String { - let params = &negotiated.params; - let duration = if params.duration_ns == 0 { - "-".to_owned() - } else { - format_optional_ns_i128(Some(i128::from(params.duration_ns))) - }; - let restrictions = if negotiated.restrictions.is_empty() { +fn format_negotiated(negotiated: &NegotiationResult) -> String { + let params = &negotiated.accepted; + let duration = params.duration.map_or_else( + || "-".to_owned(), + |duration| format_optional_ns_i128(Some(duration.as_nanos() as i128)), + ); + let changes = if negotiated.changes.is_empty() { "none".to_owned() } else { - negotiated.restrictions.len().to_string() + negotiated.changes.len().to_string() }; format!( - "duration={} interval={} length={} clock={:?} timestamps={:?} stats={:?} restrictions={}", + "duration={} interval={} length={} clock={:?} timestamps={:?} stats={:?} changes={}", duration, - format_optional_ns_i128(Some(i128::from(params.interval_ns))), + format_optional_ns_i128(Some(params.interval.as_nanos() as i128)), params.length, params.clock, params.stamp_at, params.received_stats, - restrictions + changes ) } diff --git a/crates/irtt-client/examples/blocking_client.rs b/crates/irtt-client/examples/blocking_client.rs index f248463..b8002ba 100644 --- a/crates/irtt-client/examples/blocking_client.rs +++ b/crates/irtt-client/examples/blocking_client.rs @@ -52,14 +52,12 @@ fn main() { } let negotiated = match client.open() { - Ok(OpenOutcome::Started { - event, negotiated, .. - }) => { - println!("session opened: {event:?}"); - negotiated + Ok(OpenOutcome::Started(started)) => { + println!("session opened: {started:?}"); + started.negotiation } - Ok(OpenOutcome::NoTestCompleted { event, .. }) => { - println!("{event:?}"); + Ok(OpenOutcome::NoTestCompleted(completed)) => { + println!("{completed:?}"); return; } Err(err) => { @@ -69,9 +67,13 @@ fn main() { }; // This loop owns cadence and duration. The transport sends whenever called. - let interval = Duration::from_nanos(negotiated.params.interval_ns as u64); + let interval = negotiated.accepted.interval; let start = Instant::now(); - let end = start + Duration::from_nanos(negotiated.params.duration_ns as u64); + let end = start + + negotiated + .accepted + .duration + .expect("example requests a finite run"); let deadline = end + client.probe_timeout(); let mut next_send = start; while Instant::now() < deadline && !client.is_peer_closed() { diff --git a/crates/irtt-client/examples/tokio_client.rs b/crates/irtt-client/examples/tokio_client.rs index 4f71cc3..cdef84d 100644 --- a/crates/irtt-client/examples/tokio_client.rs +++ b/crates/irtt-client/examples/tokio_client.rs @@ -66,14 +66,12 @@ async fn run() { }; let negotiated = match client.open().await { - Ok(OpenOutcome::Started { - event, negotiated, .. - }) => { - println!("session opened: {event:?}"); - negotiated + Ok(OpenOutcome::Started(started)) => { + println!("session opened: {started:?}"); + started.negotiation } - Ok(OpenOutcome::NoTestCompleted { event, .. }) => { - println!("{event:?}"); + Ok(OpenOutcome::NoTestCompleted(completed)) => { + println!("{completed:?}"); return; } Err(err) => { @@ -83,9 +81,13 @@ async fn run() { }; // This loop owns cadence and duration. The transport sends whenever called. - let interval = Duration::from_nanos(negotiated.params.interval_ns as u64); + let interval = negotiated.accepted.interval; let start = Instant::now(); - let end = start + Duration::from_nanos(negotiated.params.duration_ns as u64); + let end = start + + negotiated + .accepted + .duration + .expect("example requests a finite run"); let deadline = end + client.probe_timeout(); let mut next_send = start; while Instant::now() < deadline && !client.is_peer_closed() { diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index 585a476..c2bf37e 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -654,9 +654,11 @@ impl AsyncClient { self.machine.probe_timeout() } - /// Negotiated parameters of the currently open session. - pub fn negotiated_params(&self) -> Option<&crate::NegotiatedParams> { - self.machine.negotiated_params() + /// Accepted semantics and exact peer parameters of the currently live session. + /// + /// Returns `None` before Open, after Close, and after no-test completion. + pub fn negotiation(&self) -> Option<&crate::NegotiationResult> { + self.machine.negotiation() } /// Whether any sent probes still await a reply or timeout. @@ -707,17 +709,7 @@ impl AsyncClient { return Err(Box::new(PreparedAsyncOpenFailure { primary, machine })); } }; - let negotiated_traffic_class = match u8::try_from(negotiated.params.dscp) { - Ok(traffic_class) => traffic_class, - Err(_) => { - return Err(Box::new(PreparedAsyncOpenFailure { - primary: ClientError::InvalidConfig { - reason: "negotiated dscp must be in range 0..=255".to_owned(), - }, - machine, - })); - } - }; + let negotiated_traffic_class = negotiated.accepted.dscp << 2; let additional = recv_buffer_len.saturating_sub(self.recv_buffer.len()); if let Err(source) = self.recv_buffer.try_reserve(additional) { return Err(Box::new(PreparedAsyncOpenFailure { diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index f9e8f41..2706c97 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -373,9 +373,11 @@ impl Client { self.runtime.poll_timeouts_at(now) } - /// Negotiated parameters of the currently open session. - pub fn negotiated_params(&self) -> Option<&crate::NegotiatedParams> { - self.runtime.negotiated_params() + /// Accepted semantics and exact peer parameters of the currently live session. + /// + /// Returns `None` before Open, after Close, and after no-test completion. + pub fn negotiation(&self) -> Option<&crate::NegotiationResult> { + self.runtime.negotiation() } /// Earliest timeout deadline among probes still awaiting a reply. @@ -496,18 +498,7 @@ impl Client { Err(primary) => return Err(Box::new(PreparedClientOpenFailure { primary, machine })), }; - let negotiated_traffic_class = match u8::try_from(negotiated.params.dscp) { - Ok(traffic_class) => traffic_class, - Err(_) => { - return Err(Box::new(PreparedClientOpenFailure { - primary: ClientError::InvalidConfig { - reason: "negotiated dscp must be in range 0..=255".to_owned(), - }, - machine, - })); - } - }; - + let negotiated_traffic_class = negotiated.accepted.dscp << 2; Ok(PreparedClientOpen { machine, recv_buffer_len: Some(recv_buffer_len), diff --git a/crates/irtt-client/src/config.rs b/crates/irtt-client/src/config.rs index f2fb59a..44cd073 100644 --- a/crates/irtt-client/src/config.rs +++ b/crates/irtt-client/src/config.rs @@ -47,7 +47,7 @@ pub(crate) const DEFAULT_MAX_PENDING: usize = 4096; /// This type describes both the protocol parameters sent in the IRTT open /// request and the local client behavior used to drive the UDP socket. Values /// that are negotiated by the server are available after opening the session -/// through [`NegotiatedParams`](crate::NegotiatedParams). +/// through [`NegotiationResult`](crate::NegotiationResult). /// /// # Example /// @@ -297,7 +297,7 @@ pub enum NegotiationPolicy { /// than requested. Strict, /// Accept documented server restrictions and report them in - /// [`NegotiatedParams`](crate::NegotiatedParams). + /// [`NegotiationResult`](crate::NegotiationResult). Loose, } diff --git a/crates/irtt-client/src/event.rs b/crates/irtt-client/src/event.rs index 0a3f58d..730ad89 100644 --- a/crates/irtt-client/src/event.rs +++ b/crates/irtt-client/src/event.rs @@ -3,35 +3,36 @@ use std::{ time::{Duration, Instant}, }; -use crate::{session::NegotiatedParams, timing::ClientTimestamp}; +use crate::{session::NegotiationResult, timing::ClientTimestamp}; -/// Result of the IRTT open exchange. +/// Successful negotiation with a live session ready for echo probes. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SessionStarted { + /// Resolved remote socket address. + pub remote: SocketAddr, + /// Session token used on Echo and Close packets. + pub token: u64, + pub negotiation: NegotiationResult, + /// Client timestamp when the Open reply was accepted. + pub at: ClientTimestamp, +} + +/// Successful negotiation-only exchange with no live probe session or token. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NoTestCompleted { + pub remote: SocketAddr, + pub negotiation: NegotiationResult, + /// Client timestamp when the Open reply was accepted. + pub at: ClientTimestamp, +} + +/// Result of the IRTT Open exchange, before any echo probes are driven. /// -/// The lower-level [`Client`](crate::Client) API returns this before any echo -/// probes are driven. Managed sessions publish the contained lifecycle event to -/// subscribers immediately after a successful open. +/// Managed sessions publish the same payload as a [`ClientEvent`] after Open. #[derive(Debug, Clone, PartialEq, Eq)] pub enum OpenOutcome { - /// A normal probe session was opened. - Started { - /// Resolved remote socket address used by the UDP socket. - remote: SocketAddr, - /// Session token assigned by the server and used on echo/close packets. - token: u64, - /// Parameters returned by the server after negotiation. - negotiated: NegotiatedParams, - /// Lifecycle event corresponding to this outcome. - event: ClientEvent, - }, - /// A no-test open exchange completed without starting a probe session. - NoTestCompleted { - /// Resolved remote socket address used by the UDP socket. - remote: SocketAddr, - /// Parameters returned by the server after negotiation. - negotiated: NegotiatedParams, - /// Lifecycle event corresponding to this outcome. - event: ClientEvent, - }, + Started(SessionStarted), + NoTestCompleted(NoTestCompleted), } /// Event emitted by an IRTT client session. @@ -44,28 +45,12 @@ pub enum OpenOutcome { #[derive(Debug, Clone, PartialEq, Eq)] pub enum ClientEvent { /// A normal session has opened and is ready to send echo probes. - SessionStarted { - /// Resolved remote socket address. - remote: SocketAddr, - /// Session token assigned by the server. - token: u64, - /// Negotiated protocol parameters for this session. - negotiated: NegotiatedParams, - /// Client timestamp when the open reply was accepted. - at: ClientTimestamp, - }, - /// A negotiation-only no-test exchange completed. + SessionStarted(SessionStarted), + /// A no-test exchange completed without creating a live session. /// /// No `EchoSent`, `EchoReply`, `EchoLoss`, or `SessionClosed` events are - /// expected for this open outcome. - NoTestCompleted { - /// Resolved remote socket address. - remote: SocketAddr, - /// Negotiated protocol parameters returned by the server. - negotiated: NegotiatedParams, - /// Client timestamp when the open reply was accepted. - at: ClientTimestamp, - }, + /// expected for this outcome. + NoTestCompleted(NoTestCompleted), /// The session closed locally or through an authenticated peer close. /// /// For a peer close-flagged echo reply, the reply's measurement or diff --git a/crates/irtt-client/src/lib.rs b/crates/irtt-client/src/lib.rs index cdd636b..1115b3d 100644 --- a/crates/irtt-client/src/lib.rs +++ b/crates/irtt-client/src/lib.rs @@ -19,7 +19,7 @@ //! `tokio` feature, `AsyncClient` provides the corresponding low-level Tokio //! adapter for a caller that owns a runtime and drives socket readiness //! itself. Both adapters send one probe whenever called; callers own cadence -//! and run duration, using `negotiated_params()` after Open. [`managed`] owns +//! and run duration, using `negotiation()` after Open. [`managed`] owns //! negotiated-interval scheduling, finite runs, and staggered/burst pacing: //! `ManagedClientTask` / //! `ManagedClientHandle` (`tokio` feature) run and control one or more @@ -28,6 +28,17 @@ //! current-thread runtime for callers that want managed multi-target //! behavior without touching Tokio themselves. //! +//! A successful [`OpenOutcome`] carries either [`SessionStarted`] (with a live +//! token) or [`NoTestCompleted`] (terminal, with no probe session). Lifecycle +//! events use these same payload types. Their [`NegotiationResult`] separates +//! typed [`AcceptedSessionParameters`] from the exact decoded peer `Params` and +//! accepted [`NegotiationChange`] records. Use `accepted.interval` and +//! `accepted.duration` for cadence and lifetime, `accepted.dscp` for the public +//! DSCP codepoint, and `peer_params` for wire diagnostics. Strict/Loose policy +//! compares the actual request and reply wire values before deriving accepted +//! semantics. Both adapters' `negotiation()` accessors return only a live +//! session's result; no-test completion leaves no current session. +//! //! See `examples/` in the repository for a runnable example of each tier. //! #![forbid(unsafe_code)] @@ -59,8 +70,8 @@ pub use config::{ }; pub use error::ClientError; pub use event::{ - ClientEvent, OneWayDelaySample, OpenOutcome, PacketMeta, ReceivedStatsSample, RttSample, - ServerTiming, SignedDuration, WarningKind, + ClientEvent, NoTestCompleted, OneWayDelaySample, OpenOutcome, PacketMeta, ReceivedStatsSample, + RttSample, ServerTiming, SessionStarted, SignedDuration, WarningKind, }; -pub use session::{NegotiatedParams, NegotiationRestriction}; +pub use session::{AcceptedSessionParameters, NegotiationChange, NegotiationResult}; pub use timing::ClientTimestamp; diff --git a/crates/irtt-client/src/managed/schedule.rs b/crates/irtt-client/src/managed/schedule.rs index ec8d334..388661a 100644 --- a/crates/irtt-client/src/managed/schedule.rs +++ b/crates/irtt-client/src/managed/schedule.rs @@ -1,6 +1,6 @@ use std::time::{Duration, Instant}; -use crate::{error::ClientError, session::NegotiatedParams}; +use crate::{error::ClientError, session::AcceptedSessionParameters}; #[derive(Debug)] pub(crate) struct ProbeSchedule { @@ -18,24 +18,19 @@ pub(crate) struct ScheduleCommit { impl ProbeSchedule { pub(crate) fn new( start_at: Instant, - negotiated: &NegotiatedParams, + accepted: &AcceptedSessionParameters, ) -> Result { - let interval_ns = u64::try_from(negotiated.params.interval_ns) - .expect("validated positive negotiated interval"); - let interval = Duration::from_nanos(interval_ns); - let end_at = if negotiated.params.duration_ns > 0 { - let duration_ns = u64::try_from(negotiated.params.duration_ns) - .expect("validated positive negotiated duration"); - Some( + let interval = accepted.interval; + let end_at = accepted + .duration + .map(|duration| { start_at - .checked_add(Duration::from_nanos(duration_ns)) + .checked_add(duration) .ok_or_else(|| ClientError::NegotiationRejected { reason: "duration is too large to schedule".to_owned(), - })?, - ) - } else { - None - }; + }) + }) + .transpose()?; Ok(Self { end_at, @@ -166,15 +161,17 @@ mod tests { fn lifetime_deadline_exists_only_for_an_unfinished_finite_schedule() { let start = Instant::now(); let end = start + Duration::from_millis(500); - let mut negotiated = NegotiatedParams { - params: irtt_proto::Params { - duration_ns: 500_000_000, - interval_ns: 1_000_000, - ..irtt_proto::Params::default() - }, - restrictions: Vec::new(), + let mut accepted = AcceptedSessionParameters { + duration: Some(Duration::from_millis(500)), + interval: Duration::from_millis(1), + length: 0, + received_stats: irtt_proto::ReceivedStats::None, + stamp_at: irtt_proto::StampAt::None, + clock: irtt_proto::Clock::Both, + dscp: 0, + server_fill: None, }; - let mut finite = ProbeSchedule::new(start, &negotiated).unwrap(); + let mut finite = ProbeSchedule::new(start, &accepted).unwrap(); assert_eq!(finite.end_deadline(), Some(end)); assert!(finite.permit_probe_at(end - Duration::from_nanos(1))); assert_eq!(finite.end_deadline(), Some(end)); @@ -182,7 +179,7 @@ mod tests { assert!(finite.is_finished()); assert_eq!(finite.end_deadline(), None); - let mut finite = ProbeSchedule::new(start, &negotiated).unwrap(); + let mut finite = ProbeSchedule::new(start, &accepted).unwrap(); let commit = finite .preflight_managed_commit(start, end - Duration::from_nanos(1)) .unwrap(); @@ -190,8 +187,8 @@ mod tests { assert!(finite.is_finished()); assert_eq!(finite.end_deadline(), None); - negotiated.params.duration_ns = 0; - let mut continuous = ProbeSchedule::new(start, &negotiated).unwrap(); + accepted.duration = None; + let mut continuous = ProbeSchedule::new(start, &accepted).unwrap(); assert!(continuous.permit_probe_at(end)); assert_eq!(continuous.end_deadline(), None); } diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index 17abf70..b91267e 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -424,8 +424,8 @@ impl TargetCounters { } ClientEvent::EchoSent { .. } | ClientEvent::EchoLoss { .. } - | ClientEvent::SessionStarted { .. } - | ClientEvent::NoTestCompleted { .. } + | ClientEvent::SessionStarted(_) + | ClientEvent::NoTestCompleted(_) | ClientEvent::SessionClosed { .. } => {} } } @@ -1138,13 +1138,8 @@ impl ManagedClientTask { self.targets[index].state = TargetState::Opening { client, open }; false } - Poll::Ready(Ok(OpenOutcome::Started { - event, negotiated, .. - })) => { - let ClientEvent::SessionStarted { at: opened_at, .. } = &event else { - unreachable!("Started carries SessionStarted"); - }; - match ProbeSchedule::new(opened_at.mono, &negotiated) { + Poll::Ready(Ok(OpenOutcome::Started(started))) => { + match ProbeSchedule::new(started.at.mono, &started.negotiation.accepted) { Ok(schedule) => self.targets[index].schedule = Some(schedule), Err(error) => { return self.begin_open_session_failure( @@ -1157,7 +1152,10 @@ impl ManagedClientTask { ) } } - self.publish_client_events(index, vec![event]); + self.publish_client_events( + index, + vec![ClientEvent::SessionStarted(started)], + ); if self.state == DriverState::Stopping || self.effective_retirement(index).is_some() { @@ -1174,8 +1172,11 @@ impl ManagedClientTask { true } } - Poll::Ready(Ok(OpenOutcome::NoTestCompleted { event, .. })) => { - self.publish_client_events(index, vec![event]); + Poll::Ready(Ok(OpenOutcome::NoTestCompleted(completed))) => { + self.publish_client_events( + index, + vec![ClientEvent::NoTestCompleted(completed)], + ); let end_reason = if self.state == DriverState::Stopping || self.effective_retirement(index).is_some() { @@ -2408,13 +2409,15 @@ mod tests { runtime.block_on(async { let start = Instant::now(); let end = start + Duration::from_millis(500); - let negotiated = crate::NegotiatedParams { - params: irtt_proto::Params { - duration_ns: 500_000_000, - interval_ns: 1_000_000, - ..irtt_proto::Params::default() - }, - restrictions: Vec::new(), + let accepted = crate::AcceptedSessionParameters { + duration: Some(Duration::from_millis(500)), + interval: Duration::from_millis(1), + length: 0, + received_stats: irtt_proto::ReceivedStats::None, + stamp_at: irtt_proto::StampAt::None, + clock: irtt_proto::Clock::Both, + dscp: 0, + server_fill: None, }; for pacing in [ManagedPacing::Burst, ManagedPacing::Staggered] { let (mut task, _handle) = ManagedClient::task( @@ -2434,8 +2437,7 @@ mod tests { for send_waiting in [false, true] { task.targets[0].send_waiting = send_waiting; - task.targets[0].schedule = - Some(ProbeSchedule::new(start, &negotiated).unwrap()); + task.targets[0].schedule = Some(ProbeSchedule::new(start, &accepted).unwrap()); let expected = if pacing == ManagedPacing::Burst && !send_waiting { start } else { @@ -2451,8 +2453,8 @@ mod tests { assert_eq!(task.next_deadline(), None); } - let mut continuous = negotiated.clone(); - continuous.params.duration_ns = 0; + let mut continuous = accepted.clone(); + continuous.duration = None; task.targets[0].schedule = Some(ProbeSchedule::new(start, &continuous).unwrap()); assert_eq!(task.next_deadline(), None); } diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 07a3b7d..72247cf 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -12,12 +12,12 @@ use crate::{ }, error::ClientError, event::{ - ClientEvent, OneWayDelaySample, OpenOutcome, ReceivedStatsSample, RttSample, ServerTiming, - SignedDuration, WarningKind, + ClientEvent, NoTestCompleted, OneWayDelaySample, OpenOutcome, ReceivedStatsSample, + RttSample, ServerTiming, SessionStarted, SignedDuration, WarningKind, }, metadata::ReceiveMeta, probe::{CompletedSet, PendingMap, PendingProbe, TimedOutMap}, - session::{negotiate_params, NegotiatedParams}, + session::{negotiate_params, NegotiationResult}, socket_options::dscp_codepoint_to_traffic_class, timing::ClientTimestamp, }; @@ -53,7 +53,7 @@ enum CloseSource { #[derive(Debug)] struct ActiveSession { token: u64, - negotiated: NegotiatedParams, + negotiated: NegotiationResult, local_close_packet: Box<[u8]>, next_wire_seq: u32, highest_received_seq: Option, @@ -82,7 +82,7 @@ pub(crate) struct PreparedOpenAcceptance { } impl PreparedOpenAcceptance { - pub(crate) fn normal_negotiated(&self) -> Option<&NegotiatedParams> { + pub(crate) fn normal_negotiated(&self) -> Option<&NegotiationResult> { match &self.next_state { MachineState::Open(session) => Some(&session.negotiated), _ => None, @@ -311,7 +311,7 @@ impl SessionMachine { RequestToEncode::Echo { token: session.token, sequence: session.next_wire_seq, - params: &session.negotiated.params, + params: &session.negotiated.peer_params, payload: &[], }, self.config.auth.hmac_key(), @@ -540,7 +540,7 @@ impl SessionMachine { } } - pub(crate) fn negotiated_params(&self) -> Option<&NegotiatedParams> { + pub(crate) fn negotiation(&self) -> Option<&NegotiationResult> { match &self.state { MachineState::Open(session) => Some(&session.negotiated), _ => None, @@ -623,19 +623,12 @@ impl SessionMachine { kernel_tx_correlation_valid: true, })); - let event = ClientEvent::SessionStarted { + let outcome = OpenOutcome::Started(SessionStarted { remote: self.remote, token, - negotiated: negotiated.clone(), + negotiation: negotiated, at: now, - }; - - let outcome = OpenOutcome::Started { - remote: self.remote, - token, - negotiated, - event, - }; + }); Ok(PreparedOpenAcceptance { next_state, outcome, @@ -650,16 +643,11 @@ impl SessionMachine { let negotiated = negotiate_params(&self.requested, reply.params, self.config.open.negotiation) .map_err(OpenAcceptanceFailure::without_cleanup)?; - let event = ClientEvent::NoTestCompleted { + let outcome = OpenOutcome::NoTestCompleted(NoTestCompleted { remote: self.remote, - negotiated: negotiated.clone(), + negotiation: negotiated, at: now, - }; - let outcome = OpenOutcome::NoTestCompleted { - remote: self.remote, - negotiated, - event, - }; + }); Ok(PreparedOpenAcceptance { next_state: MachineState::NoTestCompleted, outcome, @@ -673,7 +661,7 @@ impl SessionMachine { decode_echo_reply( packet, - &session.negotiated.params, + &session.negotiated.peer_params, self.config.auth.hmac_key(), ) .ok() @@ -873,10 +861,10 @@ impl SessionMachine { /// rather than asserted away so that no 64-bit assumption is baked in. pub(crate) fn recv_buffer_size( has_hmac: bool, - negotiated: Option<&NegotiatedParams>, + negotiated: Option<&NegotiationResult>, ) -> Result { Ok(match negotiated { - Some(negotiated) => echo_packet_len(has_hmac, &negotiated.params)? + Some(negotiated) => echo_packet_len(has_hmac, &negotiated.peer_params)? .saturating_add(1) .max(MIN_RECV_BUFFER_SIZE), None => MIN_RECV_BUFFER_SIZE, diff --git a/crates/irtt-client/src/session/mod.rs b/crates/irtt-client/src/session/mod.rs index 14be3c4..d0f3c9d 100644 --- a/crates/irtt-client/src/session/mod.rs +++ b/crates/irtt-client/src/session/mod.rs @@ -2,4 +2,4 @@ pub(crate) mod machine; mod negotiation; pub(crate) use negotiation::negotiate_params; -pub use negotiation::{NegotiatedParams, NegotiationRestriction}; +pub use negotiation::{AcceptedSessionParameters, NegotiationChange, NegotiationResult}; diff --git a/crates/irtt-client/src/session/negotiation.rs b/crates/irtt-client/src/session/negotiation.rs index ab6afd3..a414a46 100644 --- a/crates/irtt-client/src/session/negotiation.rs +++ b/crates/irtt-client/src/session/negotiation.rs @@ -1,117 +1,139 @@ -use std::fmt; +use std::{fmt, time::Duration}; use irtt_proto::{Clock, Params, ReceivedStats, StampAt, PROTOCOL_VERSION}; use crate::{config::NegotiationPolicy, error::ClientError}; -/// Protocol parameters accepted for a session. +/// Validated session semantics accepted after a successful Open exchange. /// -/// `params` contains the server-returned values that the client will use for -/// echo packets. `restrictions` records accepted differences from the request -/// when loose negotiation is enabled. -/// -/// `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::SessionRequest::dscp`]. +/// These values drive ordinary client behavior. The exact peer-returned wire +/// representation is available separately in [`NegotiationResult::peer_params`]. #[derive(Debug, Clone, PartialEq, Eq)] -pub struct NegotiatedParams { - /// Server-returned protocol parameters used for the session. - pub params: Params, - /// Accepted server restrictions or parameter changes. - pub restrictions: Vec, +pub struct AcceptedSessionParameters { + /// `None` means continuous mode; finite durations are positive. + pub duration: Option, + /// Positive interval between probes when driven by a scheduler. + pub interval: Duration, + /// Accepted packet length in the usable unsigned domain. + pub length: u32, + pub received_stats: ReceivedStats, + pub stamp_at: StampAt, + pub clock: Clock, + /// DSCP codepoint (`0..=63`), as in [`crate::SessionRequest::dscp`]. + pub dscp: u8, + /// Accepted server payload fill, preserving the peer's string value. + pub server_fill: Option, } -/// A server-side restriction applied during session parameter negotiation. +/// Accepted session semantics, exact protocol evidence, and accepted changes. #[derive(Debug, Clone, PartialEq, Eq)] -pub enum NegotiationRestriction { - /// Run duration was reduced. +pub struct NegotiationResult { + pub accepted: AcceptedSessionParameters, + /// Exact decoded parameters returned by the peer, without normalization. /// - /// A requested duration of `0` means the client requested continuous mode. - /// When `requested_ns == 0` and `negotiated_ns > 0`, the server limited - /// that continuous request to a finite duration. + /// Echo encoding/decoding uses these accepted wire values. `dscp` here is + /// the raw IP TOS / Traffic Class byte, unlike [`AcceptedSessionParameters::dscp`]. + pub peer_params: Params, + /// Differences accepted under [`crate::NegotiationPolicy::Loose`]. + pub changes: Vec, +} + +/// A semantic parameter change accepted during negotiation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum NegotiationChange { + /// A finite duration was reduced, or continuous mode was limited. DurationReduced { - requested_ns: i64, - negotiated_ns: i64, + /// `None` means the request was continuous. + requested: Option, + negotiated: Duration, }, - /// Probe interval was increased. IntervalIncreased { - requested_ns: i64, - negotiated_ns: i64, + requested: Duration, + negotiated: Duration, }, - /// Probe interval was reduced. + /// Loose policy accepts any positive reduced interval. IntervalReduced { - requested_ns: i64, - negotiated_ns: i64, + requested: Duration, + negotiated: Duration, + }, + LengthReduced { + requested: u32, + negotiated: u32, }, - /// Packet length was reduced. - LengthReduced { requested: i64, negotiated: i64 }, - /// Returned received-statistics mode differs from the request. ReceivedStatsChanged { requested: ReceivedStats, negotiated: ReceivedStats, }, - /// Returned timestamp placement differs from the request. StampAtChanged { requested: StampAt, negotiated: StampAt, }, - /// Returned clock source differs from the request. - ClockChanged { requested: Clock, negotiated: Clock }, - /// Returned DSCP codepoint differs from the request. - /// - /// These values are DSCP codepoints (`0..=63`) for human-facing - /// 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. - ServerFillChanged, + ClockChanged { + requested: Clock, + negotiated: Clock, + }, + /// Both values are DSCP codepoints, rather than raw Traffic Class bytes. + DscpChanged { + requested: u8, + negotiated: u8, + }, + ServerFillChanged { + requested: Option, + negotiated: Option, + }, } -impl NegotiationRestriction { - /// Return a human-readable description of this negotiated restriction. +impl NegotiationChange { + /// Return a human-readable description of this accepted change. pub fn message(&self) -> String { self.to_string() } } -impl fmt::Display for NegotiationRestriction { +impl fmt::Display for NegotiationChange { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::DurationReduced { - requested_ns: 0, - negotiated_ns, + requested: None, + negotiated, } => { write!( f, - "server limited continuous duration to {negotiated_ns} ns" + "server limited continuous duration to {} ns", + negotiated.as_nanos() ) } Self::DurationReduced { - requested_ns, - negotiated_ns, + requested: Some(requested), + negotiated, } => { write!( f, - "server reduced duration from {requested_ns} ns to {negotiated_ns} ns" + "server reduced duration from {} ns to {} ns", + requested.as_nanos(), + negotiated.as_nanos() ) } Self::IntervalIncreased { - requested_ns, - negotiated_ns, + requested, + negotiated, } => { write!( f, - "server increased interval from {requested_ns} ns to {negotiated_ns} ns" + "server increased interval from {} ns to {} ns", + requested.as_nanos(), + negotiated.as_nanos() ) } Self::IntervalReduced { - requested_ns, - negotiated_ns, + requested, + negotiated, } => { write!( f, - "server reduced interval from {requested_ns} ns to {negotiated_ns} ns" + "server reduced interval from {} ns to {} ns", + requested.as_nanos(), + negotiated.as_nanos() ) } Self::LengthReduced { @@ -156,23 +178,32 @@ impl fmt::Display for NegotiationRestriction { } => { write!(f, "server changed DSCP from {requested} to {negotiated}") } - Self::ServerFillChanged => write!(f, "server changed payload fill behavior"), + Self::ServerFillChanged { .. } => write!(f, "server changed payload fill behavior"), } } } +// Called only after validating the request and returned numeric domains. +fn positive_duration(nanos: i64) -> Duration { + Duration::from_nanos(u64::try_from(nanos).expect("validated non-negative nanoseconds")) +} + +fn finite_duration(nanos: i64) -> Option { + (nanos != 0).then(|| positive_duration(nanos)) +} + pub(crate) fn negotiate_params( requested: &Params, returned: Params, policy: NegotiationPolicy, -) -> Result { +) -> Result { if returned.protocol_version != PROTOCOL_VERSION { return Err(ClientError::ProtocolVersionMismatch { requested: PROTOCOL_VERSION, received: returned.protocol_version, }); } - let mut restrictions = Vec::new(); + let mut changes = Vec::new(); validate_duration_restriction(requested.duration_ns, returned.duration_ns)?; if returned.length < 0 { @@ -195,70 +226,70 @@ pub(crate) fn negotiate_params( if returned.duration_ns < requested.duration_ns || (requested.duration_ns == 0 && returned.duration_ns > 0) { - record_restriction( + record_change( policy, - &mut restrictions, - NegotiationRestriction::DurationReduced { - requested_ns: requested.duration_ns, - negotiated_ns: returned.duration_ns, + &mut changes, + NegotiationChange::DurationReduced { + requested: finite_duration(requested.duration_ns), + negotiated: positive_duration(returned.duration_ns), }, )?; } if returned.interval_ns > requested.interval_ns { - record_restriction( + record_change( policy, - &mut restrictions, - NegotiationRestriction::IntervalIncreased { - requested_ns: requested.interval_ns, - negotiated_ns: returned.interval_ns, + &mut changes, + NegotiationChange::IntervalIncreased { + requested: positive_duration(requested.interval_ns), + negotiated: positive_duration(returned.interval_ns), }, )?; } if returned.interval_ns < requested.interval_ns { - record_restriction( + record_change( policy, - &mut restrictions, - NegotiationRestriction::IntervalReduced { - requested_ns: requested.interval_ns, - negotiated_ns: returned.interval_ns, + &mut changes, + NegotiationChange::IntervalReduced { + requested: positive_duration(requested.interval_ns), + negotiated: positive_duration(returned.interval_ns), }, )?; } if returned.length < requested.length { - record_restriction( + record_change( policy, - &mut restrictions, - NegotiationRestriction::LengthReduced { - requested: requested.length, - negotiated: returned.length, + &mut changes, + NegotiationChange::LengthReduced { + requested: u32::try_from(requested.length).expect("validated requested length"), + negotiated: u32::try_from(returned.length).expect("validated returned length"), }, )?; } if returned.received_stats != requested.received_stats { - record_restriction( + record_change( policy, - &mut restrictions, - NegotiationRestriction::ReceivedStatsChanged { + &mut changes, + NegotiationChange::ReceivedStatsChanged { requested: requested.received_stats, negotiated: returned.received_stats, }, )?; } if returned.stamp_at != requested.stamp_at { - record_restriction( + record_change( policy, - &mut restrictions, - NegotiationRestriction::StampAtChanged { + &mut changes, + NegotiationChange::StampAtChanged { requested: requested.stamp_at, negotiated: returned.stamp_at, }, )?; } if returned.clock != requested.clock { - record_restriction( + record_change( policy, - &mut restrictions, - NegotiationRestriction::ClockChanged { + &mut changes, + NegotiationChange::ClockChanged { requested: requested.clock, negotiated: returned.clock, }, @@ -270,29 +301,48 @@ pub(crate) fn negotiate_params( }); } if returned.dscp == 0 && requested.dscp != 0 { - record_restriction( + record_change( policy, - &mut restrictions, - NegotiationRestriction::DscpChanged { + &mut changes, + NegotiationChange::DscpChanged { // `requested`/`negotiated` here are raw wire TOS/Traffic Class // bytes; shift back to the codepoint the user actually - // configured for a human-facing restriction. - requested: requested.dscp >> 2, - negotiated: returned.dscp >> 2, + // configured for a semantic change record. + requested: u8::try_from(requested.dscp >> 2).expect("validated requested DSCP"), + negotiated: u8::try_from(returned.dscp >> 2).expect("validated returned DSCP"), }, )?; } if returned.server_fill != requested.server_fill { - record_restriction( + record_change( policy, - &mut restrictions, - NegotiationRestriction::ServerFillChanged, + &mut changes, + NegotiationChange::ServerFillChanged { + requested: requested + .server_fill + .as_ref() + .map(|fill| fill.value.clone()), + negotiated: returned.server_fill.as_ref().map(|fill| fill.value.clone()), + }, )?; } - Ok(NegotiatedParams { - params: returned, - restrictions, + // Compare the exact wire values above before deriving accepted semantics. + // In particular, do not normalize fill or traffic class before Strict policy. + let accepted = AcceptedSessionParameters { + duration: finite_duration(returned.duration_ns), + interval: positive_duration(returned.interval_ns), + length: u32::try_from(returned.length).expect("validated returned length"), + received_stats: returned.received_stats, + stamp_at: returned.stamp_at, + clock: returned.clock, + dscp: u8::try_from(returned.dscp >> 2).expect("validated returned DSCP"), + server_fill: returned.server_fill.as_ref().map(|fill| fill.value.clone()), + }; + Ok(NegotiationResult { + accepted, + peer_params: returned, + changes, }) } @@ -332,17 +382,17 @@ fn validate_dscp_restriction(returned: i64) -> Result<(), ClientError> { Ok(()) } -fn record_restriction( +fn record_change( policy: NegotiationPolicy, - restrictions: &mut Vec, - restriction: NegotiationRestriction, + changes: &mut Vec, + change: NegotiationChange, ) -> Result<(), ClientError> { if policy == NegotiationPolicy::Strict { return Err(ClientError::NegotiationRejected { - reason: restriction.message(), + reason: change.message(), }); } - restrictions.push(restriction); + changes.push(change); Ok(()) } diff --git a/crates/irtt-client/tests/client/mod.rs b/crates/irtt-client/tests/client/mod.rs index a084b7f..5754828 100644 --- a/crates/irtt-client/tests/client/mod.rs +++ b/crates/irtt-client/tests/client/mod.rs @@ -1,6 +1,6 @@ use super::*; use irtt_client::NegotiationPolicy; -use irtt_client::{NegotiatedParams, RunMode, WarningKind, MAX_UDP_PAYLOAD_LENGTH}; +use irtt_client::{NegotiationResult, RunMode, WarningKind, MAX_UDP_PAYLOAD_LENGTH}; use irtt_proto::{flags, Params, TimestampFields}; use irtt_proto::{ flags::FLAG_HMAC, flags::FLAG_OPEN, flags::FLAG_REPLY, verify_packet_hmac, Clock, HMAC_SIZE, diff --git a/crates/irtt-client/tests/client/no_test.rs b/crates/irtt-client/tests/client/no_test.rs index 6e8e4c2..98b88d6 100644 --- a/crates/irtt-client/tests/client/no_test.rs +++ b/crates/irtt-client/tests/client/no_test.rs @@ -60,24 +60,35 @@ fn no_test_strict_negotiation_rejects_changed_params() { #[test] fn no_test_loose_negotiation_accepts_restricted_params() { - 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); - 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!( - negotiated.restrictions, - vec![irtt_client::NegotiationRestriction::DurationReduced { - requested_ns: requested.duration_ns, - negotiated_ns: params.duration_ns, - }] - ); - server.join(); + for requested_duration in [Some(Duration::from_secs(3)), None] { + let mut config = default_test_config(); + config.request.run_mode = RunMode::NoTest; + config.request.duration = requested_duration; + config.open.negotiation = NegotiationPolicy::Loose; + let mut params = default_params(); + params.duration_ns = 1_500_000_000; + let server = no_test_server(params.clone(), 0); + let mut client = Client::connect(server.addr.to_string(), config).unwrap(); + let negotiated = assert_no_test_completed(client.open().unwrap()); + assert_eq!(negotiated.peer_params, params); + assert_eq!( + negotiated.accepted.duration, + Some(Duration::from_millis(1500)) + ); + assert_eq!( + negotiated.changes, + vec![irtt_client::NegotiationChange::DurationReduced { + requested: requested_duration, + negotiated: Duration::from_millis(1500), + }] + ); + assert_eq!( + client.negotiation(), + None, + "no-test creates no live session" + ); + server.join(); + } } #[test] diff --git a/crates/irtt-client/tests/client/support.rs b/crates/irtt-client/tests/client/support.rs index a6d2028..0f05fea 100644 --- a/crates/irtt-client/tests/client/support.rs +++ b/crates/irtt-client/tests/client/support.rs @@ -222,33 +222,24 @@ pub(super) fn echo_server(params: Params) -> FakeServer { }) } -pub(super) fn assert_open_started(outcome: OpenOutcome) -> NegotiatedParams { +pub(super) fn assert_open_started(outcome: OpenOutcome) -> NegotiationResult { match outcome { - OpenOutcome::Started { - token, negotiated, .. - } => { + OpenOutcome::Started(irtt_client::SessionStarted { + token, + negotiation: negotiated, + .. + }) => { assert_eq!(token, TOKEN); negotiated } - OpenOutcome::NoTestCompleted { .. } => panic!("unexpected no-test outcome"), + OpenOutcome::NoTestCompleted(_) => panic!("unexpected no-test outcome"), } } -pub(super) fn assert_no_test_completed(outcome: OpenOutcome) -> NegotiatedParams { +pub(super) fn assert_no_test_completed(outcome: OpenOutcome) -> NegotiationResult { match outcome { - OpenOutcome::NoTestCompleted { - negotiated, event, .. - } => { - assert!(matches!( - event, - ClientEvent::NoTestCompleted { - negotiated: ref event_params, - .. - } if *event_params == negotiated - )); - negotiated - } - OpenOutcome::Started { .. } => panic!("unexpected started outcome"), + OpenOutcome::NoTestCompleted(completed) => completed.negotiation, + OpenOutcome::Started(_) => panic!("unexpected started outcome"), } } diff --git a/crates/irtt-client/tests/linux_transport.rs b/crates/irtt-client/tests/linux_transport.rs index 8414116..c6abee7 100644 --- a/crates/irtt-client/tests/linux_transport.rs +++ b/crates/irtt-client/tests/linux_transport.rs @@ -160,8 +160,8 @@ fn blocking_probes_use_kernel_tx_timing_and_negotiated_packet_marking() { .unwrap(); client.open().unwrap(); assert_eq!( - client.negotiated_params().unwrap().params.dscp, - if allowed { 184 } else { 0 } + client.negotiation().unwrap().accepted.dscp, + if allowed { 46 } else { 0 } ); for _ in 0..PROBES { let before = SystemTime::now(); @@ -202,8 +202,8 @@ fn async_probes_use_kernel_tx_timing_and_negotiated_packet_marking() { .unwrap(); client.open().await.unwrap(); assert_eq!( - client.negotiated_params().unwrap().params.dscp, - if allowed { 184 } else { 0 } + client.negotiation().unwrap().accepted.dscp, + if allowed { 46 } else { 0 } ); for _ in 0..PROBES { let before = SystemTime::now(); diff --git a/crates/irtt-integration-tests/tests/async_client.rs b/crates/irtt-integration-tests/tests/async_client.rs index bb8f3cd..949a626 100644 --- a/crates/irtt-integration-tests/tests/async_client.rs +++ b/crates/irtt-integration-tests/tests/async_client.rs @@ -52,25 +52,24 @@ async fn exercise_client_server() { let opened = client.open().await.unwrap(); match &opened { - OpenOutcome::Started { + OpenOutcome::Started(irtt_client::SessionStarted { remote, token, - negotiated, - event, - } => { + negotiation: negotiated, + .. + }) => { assert_eq!(*remote, server_addr); assert_ne!(*token, 0); - assert_eq!(negotiated.params.received_stats, ReceivedStats::Both); - assert_eq!(negotiated.params.stamp_at, StampAt::Both); - assert_eq!(negotiated.params.clock, Clock::Both); + assert_eq!(negotiated.accepted.received_stats, ReceivedStats::Both); + assert_eq!(negotiated.accepted.stamp_at, StampAt::Both); + assert_eq!(negotiated.accepted.clock, Clock::Both); // A default client requests no fill, and this open succeeded under // the default strict negotiation policy — which it could not have // if the server had answered an absent request with its own default // descriptor. The server still fills the payload with it. - assert_eq!(negotiated.params.server_fill, None); - assert!(matches!(event, ClientEvent::SessionStarted { .. })); + assert_eq!(negotiated.accepted.server_fill, None); } - OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"), + OpenOutcome::NoTestCompleted(_) => panic!("normal client unexpectedly ran no-test"), } assert!(matches!( diff --git a/crates/irtt-integration-tests/tests/async_client_lifecycle.rs b/crates/irtt-integration-tests/tests/async_client_lifecycle.rs index 194d647..dbd874a 100644 --- a/crates/irtt-integration-tests/tests/async_client_lifecycle.rs +++ b/crates/irtt-integration-tests/tests/async_client_lifecycle.rs @@ -327,7 +327,7 @@ fn ignored_open_traffic_cannot_extend_the_absolute_attempt_deadline() { .unwrap(); assert!(matches!( client.open().await.unwrap(), - OpenOutcome::Started { .. } + OpenOutcome::Started(_) )); }); assert_eq!(server.finish().len(), 2); @@ -404,7 +404,7 @@ fn authenticated_async_lifecycle_negotiates_dscp() { .await .unwrap(); let opened = client.open().await.unwrap(); - assert_eq!(open_negotiated(&opened).params.dscp, 184); + assert_eq!(open_negotiated(&opened).accepted.dscp, 46); client.send_probe().await.unwrap(); assert!(matches!(client.recv().await.unwrap().as_slice(), [ClientEvent::EchoReply { seq: 0, .. }])); assert!(matches!(client.close().await.unwrap().as_slice(), [ClientEvent::SessionClosed { token, .. }] if *token == open_token(&opened))); @@ -450,17 +450,23 @@ fn blocking_and_async_peer_close_are_semantically_equivalent() { } } -fn open_negotiated(outcome: &OpenOutcome) -> &irtt_client::NegotiatedParams { +fn open_negotiated(outcome: &OpenOutcome) -> &irtt_client::NegotiationResult { match outcome { - OpenOutcome::Started { negotiated, .. } - | OpenOutcome::NoTestCompleted { negotiated, .. } => negotiated, + OpenOutcome::Started(irtt_client::SessionStarted { + negotiation: negotiated, + .. + }) + | OpenOutcome::NoTestCompleted(irtt_client::NoTestCompleted { + negotiation: negotiated, + .. + }) => negotiated, } } fn open_token(outcome: &OpenOutcome) -> u64 { match outcome { - OpenOutcome::Started { token, .. } => *token, - OpenOutcome::NoTestCompleted { .. } => panic!("no-test open has no session token"), + OpenOutcome::Started(irtt_client::SessionStarted { token, .. }) => *token, + OpenOutcome::NoTestCompleted(_) => panic!("no-test open has no session token"), } } @@ -577,8 +583,8 @@ fn blocking_and_async_roll_back_a_failed_open_and_accept_a_retry() { // state that refuses the retry. assert_eq!(error_name(&blocking_first), "open timeout"); assert_eq!(error_name(&blocking_first), error_name(&async_first)); - assert!(matches!(blocking_second, OpenOutcome::Started { .. })); - assert!(matches!(async_second, OpenOutcome::Started { .. })); + assert!(matches!(blocking_second, OpenOutcome::Started(_))); + assert!(matches!(async_second, OpenOutcome::Started(_))); assert_eq!( open_negotiated(&blocking_second), open_negotiated(&async_second) @@ -610,10 +616,7 @@ fn blocking_and_async_complete_every_caller_paced_probe() { .set_recv_timeout(Some(Duration::from_millis(200))) .unwrap(); let outcome = blocking.open().unwrap(); - assert_eq!( - blocking.negotiated_params(), - Some(open_negotiated(&outcome)) - ); + assert_eq!(blocking.negotiation(), Some(open_negotiated(&outcome))); assert!(!blocking.has_pending_probes()); // Oversleeping cannot invalidate this: low-level sends remain caller-owned // after the negotiated duration. The compliant server has no duration cap. @@ -636,7 +639,7 @@ fn blocking_and_async_complete_every_caller_paced_probe() { .await .unwrap(); let outcome = client.open().await.unwrap(); - assert_eq!(client.negotiated_params(), Some(open_negotiated(&outcome))); + assert_eq!(client.negotiation(), Some(open_negotiated(&outcome))); tokio::time::sleep(Duration::from_millis(2)).await; let mut sent = Vec::new(); let mut replies = Vec::new(); diff --git a/crates/irtt-integration-tests/tests/fill_negotiation.rs b/crates/irtt-integration-tests/tests/fill_negotiation.rs index 84227ee..99b7807 100644 --- a/crates/irtt-integration-tests/tests/fill_negotiation.rs +++ b/crates/irtt-integration-tests/tests/fill_negotiation.rs @@ -81,10 +81,16 @@ fn server_fill_negotiated_params_reflect_accepted_value() { .unwrap(); let outcome = client.open().unwrap(); let negotiated = match outcome { - irtt_client::OpenOutcome::Started { negotiated, .. } => negotiated, + irtt_client::OpenOutcome::Started(irtt_client::SessionStarted { + negotiation: negotiated, + .. + }) => negotiated, other => panic!("expected started outcome, got {other:?}"), }; - assert_eq!(negotiated.params.server_fill, params.server_fill); + assert_eq!( + negotiated.accepted.server_fill, + params.server_fill.map(|fill| fill.value) + ); client.send_probe().unwrap(); assert!(matches!( @@ -144,7 +150,7 @@ fn server_fill_loose_allows_server_to_remove_fill() { open_observation_params(&run).server_fill, requested.server_fill ); - assert_eq!(run.negotiated.params.server_fill, None); + assert_eq!(run.negotiated.accepted.server_fill, None); } // ─── Echo behavior ─── @@ -169,11 +175,11 @@ fn server_fill_does_not_affect_client_side_packet_layout() { let (echo_len_without, _, _, _) = echo_observation(&run_without); assert_eq!(echo_len_with, echo_len_without); assert_eq!( - test_echo_packet_len(false, &run_with.negotiated.params), + test_echo_packet_len(false, &run_with.negotiated.peer_params), echo_len_with ); assert_eq!( - test_echo_packet_len(false, &run_without.negotiated.params), + test_echo_packet_len(false, &run_without.negotiated.peer_params), echo_len_without ); } diff --git a/crates/irtt-integration-tests/tests/fill_wire_bytes.rs b/crates/irtt-integration-tests/tests/fill_wire_bytes.rs index b324e3d..ef6e286 100644 --- a/crates/irtt-integration-tests/tests/fill_wire_bytes.rs +++ b/crates/irtt-integration-tests/tests/fill_wire_bytes.rs @@ -10,7 +10,7 @@ use std::time::Duration; use irtt_client::{ - AsyncClient, ClientConfig, NegotiationPolicy, NegotiationRestriction, OpenOutcome, OpenPolicy, + AsyncClient, ClientConfig, NegotiationChange, NegotiationPolicy, OpenOutcome, OpenPolicy, SessionRequest, }; use irtt_proto::{ @@ -166,21 +166,23 @@ async fn exercise_fallback_negotiation() { .await .unwrap(); match loose.open().await.unwrap() { - OpenOutcome::Started { negotiated, .. } => { + OpenOutcome::Started(irtt_client::SessionStarted { + negotiation: negotiated, + .. + }) => { assert_eq!( - negotiated - .params - .server_fill - .map(|fill| fill.value) - .as_deref(), + negotiated.accepted.server_fill.as_deref(), Some("pattern:69727474"), "the server reports the default descriptor it fell back to" ); assert!(negotiated - .restrictions - .contains(&NegotiationRestriction::ServerFillChanged)); + .changes + .contains(&NegotiationChange::ServerFillChanged { + requested: Some("bogus".to_owned()), + negotiated: Some("pattern:69727474".to_owned()) + })); } - OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"), + OpenOutcome::NoTestCompleted(_) => panic!("normal client unexpectedly ran no-test"), } loose.close().await.unwrap(); diff --git a/crates/irtt-integration-tests/tests/hmac.rs b/crates/irtt-integration-tests/tests/hmac.rs index 4d99bc9..a4e512e 100644 --- a/crates/irtt-integration-tests/tests/hmac.rs +++ b/crates/irtt-integration-tests/tests/hmac.rs @@ -31,7 +31,7 @@ fn hmac_open_success_negotiates_without_warnings() { }, ); - assert_eq!(run.negotiated.params, params); + assert_eq!(run.negotiated.peer_params, params); match &run.observations[0] { ServerObservation::Open { params: got, hmac } => { assert_eq!(got, ¶ms); @@ -249,21 +249,13 @@ fn non_hmac_client_open_does_not_set_hmac_flag() { fn assert_started(outcome: OpenOutcome, expected: &irtt_proto::Params) { match outcome { - OpenOutcome::Started { + OpenOutcome::Started(irtt_client::SessionStarted { token, - negotiated, - event: - ClientEvent::SessionStarted { - token: event_token, - negotiated: event_negotiated, - .. - }, + negotiation: negotiated, .. - } => { + }) => { assert_eq!(token, TOKEN); - assert_eq!(event_token, TOKEN); - assert_eq!(negotiated.params, *expected); - assert_eq!(event_negotiated.params, *expected); + assert_eq!(negotiated.peer_params, *expected); } other => panic!("expected started open outcome, got {other:?}"), } @@ -297,7 +289,7 @@ fn backend_hmac_correct_key_succeeds() { .set_recv_timeout(Some(Duration::from_millis(500))) .unwrap(); let outcome = client.open().unwrap(); - assert!(matches!(outcome, OpenOutcome::Started { .. })); + assert!(matches!(outcome, OpenOutcome::Started(_))); let sent = client.send_probe().unwrap(); assert_eq!(sent.len(), 1); diff --git a/crates/irtt-integration-tests/tests/large_packets.rs b/crates/irtt-integration-tests/tests/large_packets.rs index b6bce33..3a35b5e 100644 --- a/crates/irtt-integration-tests/tests/large_packets.rs +++ b/crates/irtt-integration-tests/tests/large_packets.rs @@ -93,7 +93,7 @@ fn backend_large_packet_smoke() { .unwrap(); let outcome = client.open().unwrap(); - assert!(matches!(outcome, irtt_client::OpenOutcome::Started { .. })); + assert!(matches!(outcome, irtt_client::OpenOutcome::Started(_))); client.send_probe().unwrap(); let events = client.recv_once().unwrap(); diff --git a/crates/irtt-integration-tests/tests/managed_behavior.rs b/crates/irtt-integration-tests/tests/managed_behavior.rs index 523548e..9785da5 100644 --- a/crates/irtt-integration-tests/tests/managed_behavior.rs +++ b/crates/irtt-integration-tests/tests/managed_behavior.rs @@ -162,7 +162,7 @@ async fn target_updates_preserve_identical_sessions_and_retire_removed_generatio match events.recv().await.unwrap() { ManagedEvent::Client { target, - event: ClientEvent::SessionStarted { .. }, + event: ClientEvent::SessionStarted(_), } if target == first => panic!("identical update reopened the session"), ManagedEvent::Client { target, diff --git a/crates/irtt-integration-tests/tests/negotiation_policy.rs b/crates/irtt-integration-tests/tests/negotiation_policy.rs index dd5cad5..5e6d76a 100644 --- a/crates/irtt-integration-tests/tests/negotiation_policy.rs +++ b/crates/irtt-integration-tests/tests/negotiation_policy.rs @@ -10,7 +10,7 @@ use std::time::Duration; use irtt_client::{ - AsyncClient, ClientConfig, NegotiationPolicy, NegotiationRestriction, OpenOutcome, OpenPolicy, + AsyncClient, ClientConfig, NegotiationChange, NegotiationPolicy, OpenOutcome, OpenPolicy, SessionRequest, }; use irtt_proto::{Clock, ReceivedStats, StampAt}; @@ -40,22 +40,25 @@ async fn exercise_timestamp_restriction() { let mut loose = client(server_addr, NegotiationPolicy::Loose).await; let negotiated = match loose.open().await.unwrap() { - OpenOutcome::Started { negotiated, .. } => negotiated, - OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"), + OpenOutcome::Started(irtt_client::SessionStarted { + negotiation: negotiated, + .. + }) => negotiated, + OpenOutcome::NoTestCompleted(_) => panic!("normal client unexpectedly ran no-test"), }; assert_eq!( - negotiated.params.stamp_at, + negotiated.accepted.stamp_at, StampAt::Midpoint, "a request for both instants is answered with the midpoint" ); assert_eq!( - negotiated.params.clock, + negotiated.accepted.clock, Clock::Both, "the clock is not restricted with it" ); assert!(negotiated - .restrictions - .contains(&NegotiationRestriction::StampAtChanged { + .changes + .contains(&NegotiationChange::StampAtChanged { requested: StampAt::Both, negotiated: StampAt::Midpoint, })); @@ -112,16 +115,19 @@ async fn exercise_dscp_restriction() { .await .unwrap(); let negotiated = match loose.open().await.unwrap() { - OpenOutcome::Started { negotiated, .. } => negotiated, - OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"), + OpenOutcome::Started(irtt_client::SessionStarted { + negotiation: negotiated, + .. + }) => negotiated, + OpenOutcome::NoTestCompleted(_) => panic!("normal client unexpectedly ran no-test"), }; assert_eq!( - negotiated.params.dscp, 0, - "the raw wire parameter is negotiated to zero" + negotiated.accepted.dscp, 0, + "the accepted DSCP codepoint is zero" ); assert!(negotiated - .restrictions - .contains(&NegotiationRestriction::DscpChanged { + .changes + .contains(&NegotiationChange::DscpChanged { requested: 46, negotiated: 0, })); @@ -159,14 +165,21 @@ async fn exercise_default_policy() { .await .unwrap(); let negotiated = match client.open().await.unwrap() { - OpenOutcome::Started { negotiated, .. } => negotiated, - OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"), + OpenOutcome::Started(irtt_client::SessionStarted { + negotiation: negotiated, + .. + }) => negotiated, + OpenOutcome::NoTestCompleted(_) => panic!("normal client unexpectedly ran no-test"), }; - assert_eq!(negotiated.restrictions, Vec::new()); - assert_eq!(negotiated.params.stamp_at, StampAt::Both); - assert_eq!(negotiated.params.clock, Clock::Both); - assert_eq!(negotiated.params.dscp, 184, "the raw byte for codepoint 46"); + assert_eq!(negotiated.changes, Vec::new()); + assert_eq!(negotiated.accepted.stamp_at, StampAt::Both); + assert_eq!(negotiated.accepted.clock, Clock::Both); + assert_eq!(negotiated.accepted.dscp, 46); + assert_eq!( + negotiated.peer_params.dscp, 184, + "the raw byte for codepoint 46" + ); probe(&mut client).await; client.close().await.unwrap(); diff --git a/crates/irtt-integration-tests/tests/protocol_options.rs b/crates/irtt-integration-tests/tests/protocol_options.rs index 023dbb4..e89b7a1 100644 --- a/crates/irtt-integration-tests/tests/protocol_options.rs +++ b/crates/irtt-integration-tests/tests/protocol_options.rs @@ -7,8 +7,8 @@ mod protocol_options { use std::time::Duration; use irtt_client::{ - Client, ClientConfig, ClientError, ClientEvent, OneWayDelaySample, ReceivedStatsSample, - RttSample, ServerTiming, SignedDuration, + AcceptedSessionParameters, Client, ClientConfig, ClientError, ClientEvent, OneWayDelaySample, + ReceivedStatsSample, RttSample, ServerTiming, SignedDuration, }; use irtt_proto::{Clock, Params, ReceivedStats, StampAt, TimestampFields}; @@ -212,7 +212,20 @@ fn loose_open_uses_returned_params_for_echo_layout_and_reply_parsing() { }); assert_eq!(open_params(&run), &requested); - assert_eq!(run.negotiated.params, returned); + assert_eq!(run.negotiated.peer_params, returned); + assert_eq!( + run.negotiated.accepted, + AcceptedSessionParameters { + duration: Some(Duration::from_secs(3)), + interval: Duration::from_secs(1), + length: 28, + received_stats: ReceivedStats::Count, + stamp_at: StampAt::Receive, + clock: Clock::Wall, + dscp: 0, + server_fill: None, + } + ); assert_echo_uses_params(&run, &returned, false); let reply = expect_echo_reply(&run.reply); @@ -253,7 +266,7 @@ fn changed_compatibility_params(requested: &Params) -> Vec { } fn assert_negotiated_echo_use(run: &OneProbeRun, params: &Params, hmac: bool) { - assert_eq!(run.negotiated.params, *params); + assert_eq!(run.negotiated.peer_params, *params); assert_eq!(open_params(run), params); assert_echo_uses_params(run, params, hmac); } 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 2d7e0d5..3b5c195 100644 --- a/crates/irtt-integration-tests/tests/protocol_options/backend_smoke.rs +++ b/crates/irtt-integration-tests/tests/protocol_options/backend_smoke.rs @@ -14,7 +14,7 @@ fn backend_basic_open_echo_close() { .unwrap(); let outcome = client.open().unwrap(); - assert!(matches!(outcome, irtt_client::OpenOutcome::Started { .. })); + assert!(matches!(outcome, irtt_client::OpenOutcome::Started(_))); let sent = client.send_probe().unwrap(); assert_eq!(sent.len(), 1); diff --git a/crates/irtt-integration-tests/tests/support/mod.rs b/crates/irtt-integration-tests/tests/support/mod.rs index 1f12793..f5b38b0 100644 --- a/crates/irtt-integration-tests/tests/support/mod.rs +++ b/crates/irtt-integration-tests/tests/support/mod.rs @@ -13,7 +13,7 @@ use std::{ }; use irtt_client::{ - Client, ClientConfig, ClientEvent, NegotiatedParams, OpenOutcome, OpenPolicy, SessionRequest, + Client, ClientConfig, ClientEvent, NegotiationResult, OpenOutcome, OpenPolicy, SessionRequest, }; use irtt_proto::{ compute_hmac_in_place, decode_request, echo_packet_len, encode_echo_reply, encode_open_reply, @@ -138,7 +138,7 @@ pub enum ServerObservation { } pub struct OneProbeRun { - pub negotiated: NegotiatedParams, + pub negotiated: NegotiationResult, pub sent: ClientEvent, pub reply: ClientEvent, pub observations: Vec, @@ -514,10 +514,13 @@ fn recv_request_timeout(socket: &UdpSocket) -> Option> { .map(|(size, _)| buf[..size].to_vec()) } -fn assert_started(outcome: OpenOutcome) -> NegotiatedParams { +fn assert_started(outcome: OpenOutcome) -> NegotiationResult { match outcome { - OpenOutcome::Started { negotiated, .. } => negotiated, - OpenOutcome::NoTestCompleted { .. } => panic!("expected started outcome"), + OpenOutcome::Started(irtt_client::SessionStarted { + negotiation: negotiated, + .. + }) => negotiated, + OpenOutcome::NoTestCompleted(_) => panic!("expected started outcome"), } } diff --git a/crates/irtt-integration-tests/tests/traffic_class.rs b/crates/irtt-integration-tests/tests/traffic_class.rs index 02c66b4..161aac3 100644 --- a/crates/irtt-integration-tests/tests/traffic_class.rs +++ b/crates/irtt-integration-tests/tests/traffic_class.rs @@ -217,12 +217,15 @@ async fn opened_session(server_addr: SocketAddr, dscp: u8) -> OpenSession { .await .unwrap(); match client.open().await.unwrap() { - OpenOutcome::Started { negotiated, .. } => assert_eq!( - negotiated.params.dscp, + OpenOutcome::Started(irtt_client::SessionStarted { + negotiation: negotiated, + .. + }) => assert_eq!( + negotiated.peer_params.dscp, i64::from(dscp) << 2, "the server negotiates the raw byte the codepoint asks for" ), - OpenOutcome::NoTestCompleted { .. } => panic!("a normal client unexpectedly ran no-test"), + OpenOutcome::NoTestCompleted(_) => panic!("a normal client unexpectedly ran no-test"), } OpenSession { client, diff --git a/crates/irtt-integration-tests/tests/wildcard_source.rs b/crates/irtt-integration-tests/tests/wildcard_source.rs index 24c2a67..f48a285 100644 --- a/crates/irtt-integration-tests/tests/wildcard_source.rs +++ b/crates/irtt-integration-tests/tests/wildcard_source.rs @@ -220,11 +220,11 @@ async fn connected_client(server_addr: SocketAddr) -> AsyncClient { async fn assert_started(client: &mut AsyncClient, server_addr: SocketAddr) { match client.open().await.unwrap() { - OpenOutcome::Started { remote, token, .. } => { + OpenOutcome::Started(irtt_client::SessionStarted { remote, token, .. }) => { assert_eq!(remote, server_addr); assert_ne!(token, 0); } - OpenOutcome::NoTestCompleted { .. } => panic!("a normal client unexpectedly ran no-test"), + OpenOutcome::NoTestCompleted(_) => panic!("a normal client unexpectedly ran no-test"), } } diff --git a/crates/irtt-stats/src/normalization.rs b/crates/irtt-stats/src/normalization.rs index ea6cf65..bc016d1 100644 --- a/crates/irtt-stats/src/normalization.rs +++ b/crates/irtt-stats/src/normalization.rs @@ -124,8 +124,8 @@ pub(crate) fn normalize_event(event: &ClientEvent) -> Option { at: (*timeout_at).max(sent_at.mono), }), ClientEvent::Warning { at, .. } => Some(StatsEvent::Warning { at: at.mono }), - ClientEvent::SessionStarted { .. } - | ClientEvent::NoTestCompleted { .. } + ClientEvent::SessionStarted(_) + | ClientEvent::NoTestCompleted(_) | ClientEvent::SessionClosed { .. } => None, } }