Merge branch 'refactor/typed-client-negotiation'
Rust CI / cargo fmt (push) Successful in 17s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 29s
Rust CI / cargo clippy (push) Successful in 33s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Failing after 9s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Failing after 8s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m13s
Rust CI / cargo doc (push) Successful in 24s
Fuzz Smoke / cargo fuzz smoke (push) Failing after 27s
Supply chain / cargo-deny advisories (push) Successful in 7s
Supply chain / cargo-deny bans licenses sources (push) Successful in 7s
Supply chain / actionlint (push) Successful in 2s
Netem Interop / netem interop (push) Successful in 1m12s

This commit is contained in:
Nils Andreas Svee committed 2026-10-04 18:34:31 +07:00
commit eebe83986a
32 files changed
+507 -447

No files matched your search

+14 -11
View File
@@ -5,7 +5,7 @@ use std::{
}; };
use irtt_client::{ use irtt_client::{
ClientEvent, NegotiatedParams, OneWayDelaySample, PacketMeta, ReceivedStatsSample, RttSample, ClientEvent, NegotiationResult, OneWayDelaySample, PacketMeta, ReceivedStatsSample, RttSample,
ServerTiming, SignedDuration, WarningKind, ServerTiming, SignedDuration, WarningKind,
}; };
@@ -234,22 +234,22 @@ enum OutputRow {
impl OutputRow { impl OutputRow {
fn from_event(event: &ClientEvent) -> Self { fn from_event(event: &ClientEvent) -> Self {
match event { match event {
ClientEvent::SessionStarted { ClientEvent::SessionStarted(irtt_client::SessionStarted {
remote, remote,
token, token,
negotiated, negotiation: negotiated,
at, at,
} => Self::SessionStarted(LifecycleRow::new( }) => Self::SessionStarted(LifecycleRow::new(
*remote, *remote,
Some(*token), Some(*token),
negotiated, negotiated,
at.wall, at.wall,
)), )),
ClientEvent::NoTestCompleted { ClientEvent::NoTestCompleted(irtt_client::NoTestCompleted {
remote, remote,
negotiated, negotiation: negotiated,
at, 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 { ClientEvent::SessionClosed { remote, token, at } => Self::SessionClosed {
remote: *remote, remote: *remote,
token: *token, token: *token,
@@ -384,16 +384,19 @@ impl LifecycleRow {
fn new( fn new(
remote: SocketAddr, remote: SocketAddr,
token: Option<u64>, token: Option<u64>,
negotiated: &NegotiatedParams, negotiated: &NegotiationResult,
event_wall: SystemTime, event_wall: SystemTime,
) -> Self { ) -> Self {
Self { Self {
remote, remote,
token, token,
event_wall, event_wall,
duration_ns: i128::from(negotiated.params.duration_ns), duration_ns: negotiated
interval_ns: i128::from(negotiated.params.interval_ns), .accepted
payload_length: i128::from(negotiated.params.length), .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),
} }
} }
} }
+21 -20
View File
@@ -10,7 +10,7 @@ use crossterm::{
terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen}, terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen},
}; };
use irtt_client::managed::{ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance}; 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 irtt_stats::{Snapshot, StatsCollector, TimeStats};
use ratatui::{ use ratatui::{
backend::CrosstermBackend, backend::CrosstermBackend,
@@ -266,12 +266,12 @@ impl TuiState {
}; };
target.stats.process(event); target.stats.process(event);
match event { match event {
ClientEvent::SessionStarted { ClientEvent::SessionStarted(irtt_client::SessionStarted {
remote, remote,
token, token,
negotiated, negotiation: negotiated,
.. ..
} => { }) => {
target.remote = Some(remote.to_string()); target.remote = Some(remote.to_string());
target.session = Some(format!("{token:#x}")); target.session = Some(format!("{token:#x}"));
target.negotiated = Some(negotiated.clone()); target.negotiated = Some(negotiated.clone());
@@ -279,9 +279,11 @@ impl TuiState {
global_status = Some(TuiStatus::Running); global_status = Some(TuiStatus::Running);
recent = Some(format!("session started token={token:#x}")); recent = Some(format!("session started token={token:#x}"));
} }
ClientEvent::NoTestCompleted { ClientEvent::NoTestCompleted(irtt_client::NoTestCompleted {
remote, negotiated, .. remote,
} => { negotiation: negotiated,
..
}) => {
target.remote = Some(remote.to_string()); target.remote = Some(remote.to_string());
target.negotiated = Some(negotiated.clone()); target.negotiated = Some(negotiated.clone());
target.status = TargetStatus::NoTest; target.status = TargetStatus::NoTest;
@@ -541,7 +543,7 @@ pub(super) struct TuiTargetState {
remote: Option<String>, remote: Option<String>,
session: Option<String>, session: Option<String>,
status: TargetStatus, status: TargetStatus,
negotiated: Option<NegotiatedParams>, negotiated: Option<NegotiationResult>,
graph_history: VecDeque<GraphSample>, graph_history: VecDeque<GraphSample>,
last_sample: Option<LastSample>, last_sample: Option<LastSample>,
last_warning: Option<String>, last_warning: Option<String>,
@@ -1897,27 +1899,26 @@ fn push_time_line(lines: &mut Vec<Line<'_>>, label: &str, stats: &TimeStats) {
))); )));
} }
fn format_negotiated(negotiated: &NegotiatedParams) -> String { fn format_negotiated(negotiated: &NegotiationResult) -> String {
let params = &negotiated.params; let params = &negotiated.accepted;
let duration = if params.duration_ns == 0 { let duration = params.duration.map_or_else(
"-".to_owned() || "-".to_owned(),
} else { |duration| format_optional_ns_i128(Some(duration.as_nanos() as i128)),
format_optional_ns_i128(Some(i128::from(params.duration_ns))) );
}; let changes = if negotiated.changes.is_empty() {
let restrictions = if negotiated.restrictions.is_empty() {
"none".to_owned() "none".to_owned()
} else { } else {
negotiated.restrictions.len().to_string() negotiated.changes.len().to_string()
}; };
format!( format!(
"duration={} interval={} length={} clock={:?} timestamps={:?} stats={:?} restrictions={}", "duration={} interval={} length={} clock={:?} timestamps={:?} stats={:?} changes={}",
duration, 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.length,
params.clock, params.clock,
params.stamp_at, params.stamp_at,
params.received_stats, params.received_stats,
restrictions changes
) )
} }
+11 -9
View File
@@ -52,14 +52,12 @@ fn main() {
} }
let negotiated = match client.open() { let negotiated = match client.open() {
Ok(OpenOutcome::Started { Ok(OpenOutcome::Started(started)) => {
event, negotiated, .. println!("session opened: {started:?}");
}) => { started.negotiation
println!("session opened: {event:?}");
negotiated
} }
Ok(OpenOutcome::NoTestCompleted { event, .. }) => { Ok(OpenOutcome::NoTestCompleted(completed)) => {
println!("{event:?}"); println!("{completed:?}");
return; return;
} }
Err(err) => { Err(err) => {
@@ -69,9 +67,13 @@ fn main() {
}; };
// This loop owns cadence and duration. The transport sends whenever called. // 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 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 deadline = end + client.probe_timeout();
let mut next_send = start; let mut next_send = start;
while Instant::now() < deadline && !client.is_peer_closed() { while Instant::now() < deadline && !client.is_peer_closed() {
+11 -9
View File
@@ -66,14 +66,12 @@ async fn run() {
}; };
let negotiated = match client.open().await { let negotiated = match client.open().await {
Ok(OpenOutcome::Started { Ok(OpenOutcome::Started(started)) => {
event, negotiated, .. println!("session opened: {started:?}");
}) => { started.negotiation
println!("session opened: {event:?}");
negotiated
} }
Ok(OpenOutcome::NoTestCompleted { event, .. }) => { Ok(OpenOutcome::NoTestCompleted(completed)) => {
println!("{event:?}"); println!("{completed:?}");
return; return;
} }
Err(err) => { Err(err) => {
@@ -83,9 +81,13 @@ async fn run() {
}; };
// This loop owns cadence and duration. The transport sends whenever called. // 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 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 deadline = end + client.probe_timeout();
let mut next_send = start; let mut next_send = start;
while Instant::now() < deadline && !client.is_peer_closed() { while Instant::now() < deadline && !client.is_peer_closed() {
+6 -14
View File
@@ -654,9 +654,11 @@ impl AsyncClient {
self.machine.probe_timeout() self.machine.probe_timeout()
} }
/// Negotiated parameters of the currently open session. /// Accepted semantics and exact peer parameters of the currently live session.
pub fn negotiated_params(&self) -> Option<&crate::NegotiatedParams> { ///
self.machine.negotiated_params() /// 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. /// Whether any sent probes still await a reply or timeout.
@@ -707,17 +709,7 @@ impl AsyncClient {
return Err(Box::new(PreparedAsyncOpenFailure { primary, machine })); return Err(Box::new(PreparedAsyncOpenFailure { primary, machine }));
} }
}; };
let negotiated_traffic_class = match u8::try_from(negotiated.params.dscp) { let negotiated_traffic_class = negotiated.accepted.dscp << 2;
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 additional = recv_buffer_len.saturating_sub(self.recv_buffer.len()); let additional = recv_buffer_len.saturating_sub(self.recv_buffer.len());
if let Err(source) = self.recv_buffer.try_reserve(additional) { if let Err(source) = self.recv_buffer.try_reserve(additional) {
return Err(Box::new(PreparedAsyncOpenFailure { return Err(Box::new(PreparedAsyncOpenFailure {
+6 -15
View File
@@ -373,9 +373,11 @@ impl Client {
self.runtime.poll_timeouts_at(now) self.runtime.poll_timeouts_at(now)
} }
/// Negotiated parameters of the currently open session. /// Accepted semantics and exact peer parameters of the currently live session.
pub fn negotiated_params(&self) -> Option<&crate::NegotiatedParams> { ///
self.runtime.negotiated_params() /// 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. /// Earliest timeout deadline among probes still awaiting a reply.
@@ -496,18 +498,7 @@ impl Client {
Err(primary) => return Err(Box::new(PreparedClientOpenFailure { primary, machine })), Err(primary) => return Err(Box::new(PreparedClientOpenFailure { primary, machine })),
}; };
let negotiated_traffic_class = match u8::try_from(negotiated.params.dscp) { let negotiated_traffic_class = negotiated.accepted.dscp << 2;
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,
}));
}
};
Ok(PreparedClientOpen { Ok(PreparedClientOpen {
machine, machine,
recv_buffer_len: Some(recv_buffer_len), recv_buffer_len: Some(recv_buffer_len),
+2 -2
View File
@@ -47,7 +47,7 @@ pub(crate) const DEFAULT_MAX_PENDING: usize = 4096;
/// This type describes both the protocol parameters sent in the IRTT open /// 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 /// 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 /// that are negotiated by the server are available after opening the session
/// through [`NegotiatedParams`](crate::NegotiatedParams). /// through [`NegotiationResult`](crate::NegotiationResult).
/// ///
/// # Example /// # Example
/// ///
@@ -297,7 +297,7 @@ pub enum NegotiationPolicy {
/// than requested. /// than requested.
Strict, Strict,
/// Accept documented server restrictions and report them in /// Accept documented server restrictions and report them in
/// [`NegotiatedParams`](crate::NegotiatedParams). /// [`NegotiationResult`](crate::NegotiationResult).
Loose, Loose,
} }
+30 -45
View File
@@ -3,35 +3,36 @@ use std::{
time::{Duration, Instant}, 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 /// Managed sessions publish the same payload as a [`ClientEvent`] after Open.
/// probes are driven. Managed sessions publish the contained lifecycle event to
/// subscribers immediately after a successful open.
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub enum OpenOutcome { pub enum OpenOutcome {
/// A normal probe session was opened. Started(SessionStarted),
Started { NoTestCompleted(NoTestCompleted),
/// 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,
},
} }
/// Event emitted by an IRTT client session. /// Event emitted by an IRTT client session.
@@ -44,28 +45,12 @@ pub enum OpenOutcome {
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub enum ClientEvent { pub enum ClientEvent {
/// A normal session has opened and is ready to send echo probes. /// A normal session has opened and is ready to send echo probes.
SessionStarted { SessionStarted(SessionStarted),
/// Resolved remote socket address. /// A no-test exchange completed without creating a live session.
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.
/// ///
/// No `EchoSent`, `EchoReply`, `EchoLoss`, or `SessionClosed` events are /// No `EchoSent`, `EchoReply`, `EchoLoss`, or `SessionClosed` events are
/// expected for this open outcome. /// expected for this outcome.
NoTestCompleted { NoTestCompleted(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,
},
/// The session closed locally or through an authenticated peer close. /// The session closed locally or through an authenticated peer close.
/// ///
/// For a peer close-flagged echo reply, the reply's measurement or /// For a peer close-flagged echo reply, the reply's measurement or
+15 -4
View File
@@ -19,7 +19,7 @@
//! `tokio` feature, `AsyncClient` provides the corresponding low-level Tokio //! `tokio` feature, `AsyncClient` provides the corresponding low-level Tokio
//! adapter for a caller that owns a runtime and drives socket readiness //! adapter for a caller that owns a runtime and drives socket readiness
//! itself. Both adapters send one probe whenever called; callers own cadence //! 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: //! negotiated-interval scheduling, finite runs, and staggered/burst pacing:
//! `ManagedClientTask` / //! `ManagedClientTask` /
//! `ManagedClientHandle` (`tokio` feature) run and control one or more //! `ManagedClientHandle` (`tokio` feature) run and control one or more
@@ -28,6 +28,17 @@
//! current-thread runtime for callers that want managed multi-target //! current-thread runtime for callers that want managed multi-target
//! behavior without touching Tokio themselves. //! 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. //! See `examples/` in the repository for a runnable example of each tier.
//! //!
#![forbid(unsafe_code)] #![forbid(unsafe_code)]
@@ -59,8 +70,8 @@ pub use config::{
}; };
pub use error::ClientError; pub use error::ClientError;
pub use event::{ pub use event::{
ClientEvent, OneWayDelaySample, OpenOutcome, PacketMeta, ReceivedStatsSample, RttSample, ClientEvent, NoTestCompleted, OneWayDelaySample, OpenOutcome, PacketMeta, ReceivedStatsSample,
ServerTiming, SignedDuration, WarningKind, RttSample, ServerTiming, SessionStarted, SignedDuration, WarningKind,
}; };
pub use session::{NegotiatedParams, NegotiationRestriction}; pub use session::{AcceptedSessionParameters, NegotiationChange, NegotiationResult};
pub use timing::ClientTimestamp; pub use timing::ClientTimestamp;
+23 -26
View File
@@ -1,6 +1,6 @@
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use crate::{error::ClientError, session::NegotiatedParams}; use crate::{error::ClientError, session::AcceptedSessionParameters};
#[derive(Debug)] #[derive(Debug)]
pub(crate) struct ProbeSchedule { pub(crate) struct ProbeSchedule {
@@ -18,24 +18,19 @@ pub(crate) struct ScheduleCommit {
impl ProbeSchedule { impl ProbeSchedule {
pub(crate) fn new( pub(crate) fn new(
start_at: Instant, start_at: Instant,
negotiated: &NegotiatedParams, accepted: &AcceptedSessionParameters,
) -> Result<Self, ClientError> { ) -> Result<Self, ClientError> {
let interval_ns = u64::try_from(negotiated.params.interval_ns) let interval = accepted.interval;
.expect("validated positive negotiated interval"); let end_at = accepted
let interval = Duration::from_nanos(interval_ns); .duration
let end_at = if negotiated.params.duration_ns > 0 { .map(|duration| {
let duration_ns = u64::try_from(negotiated.params.duration_ns)
.expect("validated positive negotiated duration");
Some(
start_at start_at
.checked_add(Duration::from_nanos(duration_ns)) .checked_add(duration)
.ok_or_else(|| ClientError::NegotiationRejected { .ok_or_else(|| ClientError::NegotiationRejected {
reason: "duration is too large to schedule".to_owned(), reason: "duration is too large to schedule".to_owned(),
})?, })
) })
} else { .transpose()?;
None
};
Ok(Self { Ok(Self {
end_at, end_at,
@@ -166,15 +161,17 @@ mod tests {
fn lifetime_deadline_exists_only_for_an_unfinished_finite_schedule() { fn lifetime_deadline_exists_only_for_an_unfinished_finite_schedule() {
let start = Instant::now(); let start = Instant::now();
let end = start + Duration::from_millis(500); let end = start + Duration::from_millis(500);
let mut negotiated = NegotiatedParams { let mut accepted = AcceptedSessionParameters {
params: irtt_proto::Params { duration: Some(Duration::from_millis(500)),
duration_ns: 500_000_000, interval: Duration::from_millis(1),
interval_ns: 1_000_000, length: 0,
..irtt_proto::Params::default() received_stats: irtt_proto::ReceivedStats::None,
}, stamp_at: irtt_proto::StampAt::None,
restrictions: Vec::new(), 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_eq!(finite.end_deadline(), Some(end));
assert!(finite.permit_probe_at(end - Duration::from_nanos(1))); assert!(finite.permit_probe_at(end - Duration::from_nanos(1)));
assert_eq!(finite.end_deadline(), Some(end)); assert_eq!(finite.end_deadline(), Some(end));
@@ -182,7 +179,7 @@ mod tests {
assert!(finite.is_finished()); assert!(finite.is_finished());
assert_eq!(finite.end_deadline(), None); 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 let commit = finite
.preflight_managed_commit(start, end - Duration::from_nanos(1)) .preflight_managed_commit(start, end - Duration::from_nanos(1))
.unwrap(); .unwrap();
@@ -190,8 +187,8 @@ mod tests {
assert!(finite.is_finished()); assert!(finite.is_finished());
assert_eq!(finite.end_deadline(), None); assert_eq!(finite.end_deadline(), None);
negotiated.params.duration_ns = 0; accepted.duration = None;
let mut continuous = ProbeSchedule::new(start, &negotiated).unwrap(); let mut continuous = ProbeSchedule::new(start, &accepted).unwrap();
assert!(continuous.permit_probe_at(end)); assert!(continuous.permit_probe_at(end));
assert_eq!(continuous.end_deadline(), None); assert_eq!(continuous.end_deadline(), None);
} }
+25 -23
View File
@@ -424,8 +424,8 @@ impl TargetCounters {
} }
ClientEvent::EchoSent { .. } ClientEvent::EchoSent { .. }
| ClientEvent::EchoLoss { .. } | ClientEvent::EchoLoss { .. }
| ClientEvent::SessionStarted { .. } | ClientEvent::SessionStarted(_)
| ClientEvent::NoTestCompleted { .. } | ClientEvent::NoTestCompleted(_)
| ClientEvent::SessionClosed { .. } => {} | ClientEvent::SessionClosed { .. } => {}
} }
} }
@@ -1138,13 +1138,8 @@ impl ManagedClientTask {
self.targets[index].state = TargetState::Opening { client, open }; self.targets[index].state = TargetState::Opening { client, open };
false false
} }
Poll::Ready(Ok(OpenOutcome::Started { Poll::Ready(Ok(OpenOutcome::Started(started))) => {
event, negotiated, .. match ProbeSchedule::new(started.at.mono, &started.negotiation.accepted) {
})) => {
let ClientEvent::SessionStarted { at: opened_at, .. } = &event else {
unreachable!("Started carries SessionStarted");
};
match ProbeSchedule::new(opened_at.mono, &negotiated) {
Ok(schedule) => self.targets[index].schedule = Some(schedule), Ok(schedule) => self.targets[index].schedule = Some(schedule),
Err(error) => { Err(error) => {
return self.begin_open_session_failure( 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 if self.state == DriverState::Stopping
|| self.effective_retirement(index).is_some() || self.effective_retirement(index).is_some()
{ {
@@ -1174,8 +1172,11 @@ impl ManagedClientTask {
true true
} }
} }
Poll::Ready(Ok(OpenOutcome::NoTestCompleted { event, .. })) => { Poll::Ready(Ok(OpenOutcome::NoTestCompleted(completed))) => {
self.publish_client_events(index, vec![event]); self.publish_client_events(
index,
vec![ClientEvent::NoTestCompleted(completed)],
);
let end_reason = if self.state == DriverState::Stopping let end_reason = if self.state == DriverState::Stopping
|| self.effective_retirement(index).is_some() || self.effective_retirement(index).is_some()
{ {
@@ -2408,13 +2409,15 @@ mod tests {
runtime.block_on(async { runtime.block_on(async {
let start = Instant::now(); let start = Instant::now();
let end = start + Duration::from_millis(500); let end = start + Duration::from_millis(500);
let negotiated = crate::NegotiatedParams { let accepted = crate::AcceptedSessionParameters {
params: irtt_proto::Params { duration: Some(Duration::from_millis(500)),
duration_ns: 500_000_000, interval: Duration::from_millis(1),
interval_ns: 1_000_000, length: 0,
..irtt_proto::Params::default() received_stats: irtt_proto::ReceivedStats::None,
}, stamp_at: irtt_proto::StampAt::None,
restrictions: Vec::new(), clock: irtt_proto::Clock::Both,
dscp: 0,
server_fill: None,
}; };
for pacing in [ManagedPacing::Burst, ManagedPacing::Staggered] { for pacing in [ManagedPacing::Burst, ManagedPacing::Staggered] {
let (mut task, _handle) = ManagedClient::task( let (mut task, _handle) = ManagedClient::task(
@@ -2434,8 +2437,7 @@ mod tests {
for send_waiting in [false, true] { for send_waiting in [false, true] {
task.targets[0].send_waiting = send_waiting; task.targets[0].send_waiting = send_waiting;
task.targets[0].schedule = task.targets[0].schedule = Some(ProbeSchedule::new(start, &accepted).unwrap());
Some(ProbeSchedule::new(start, &negotiated).unwrap());
let expected = if pacing == ManagedPacing::Burst && !send_waiting { let expected = if pacing == ManagedPacing::Burst && !send_waiting {
start start
} else { } else {
@@ -2451,8 +2453,8 @@ mod tests {
assert_eq!(task.next_deadline(), None); assert_eq!(task.next_deadline(), None);
} }
let mut continuous = negotiated.clone(); let mut continuous = accepted.clone();
continuous.params.duration_ns = 0; continuous.duration = None;
task.targets[0].schedule = Some(ProbeSchedule::new(start, &continuous).unwrap()); task.targets[0].schedule = Some(ProbeSchedule::new(start, &continuous).unwrap());
assert_eq!(task.next_deadline(), None); assert_eq!(task.next_deadline(), None);
} }
+16 -28
View File
@@ -12,12 +12,12 @@ use crate::{
}, },
error::ClientError, error::ClientError,
event::{ event::{
ClientEvent, OneWayDelaySample, OpenOutcome, ReceivedStatsSample, RttSample, ServerTiming, ClientEvent, NoTestCompleted, OneWayDelaySample, OpenOutcome, ReceivedStatsSample,
SignedDuration, WarningKind, RttSample, ServerTiming, SessionStarted, SignedDuration, WarningKind,
}, },
metadata::ReceiveMeta, metadata::ReceiveMeta,
probe::{CompletedSet, PendingMap, PendingProbe, TimedOutMap}, probe::{CompletedSet, PendingMap, PendingProbe, TimedOutMap},
session::{negotiate_params, NegotiatedParams}, session::{negotiate_params, NegotiationResult},
socket_options::dscp_codepoint_to_traffic_class, socket_options::dscp_codepoint_to_traffic_class,
timing::ClientTimestamp, timing::ClientTimestamp,
}; };
@@ -53,7 +53,7 @@ enum CloseSource {
#[derive(Debug)] #[derive(Debug)]
struct ActiveSession { struct ActiveSession {
token: u64, token: u64,
negotiated: NegotiatedParams, negotiated: NegotiationResult,
local_close_packet: Box<[u8]>, local_close_packet: Box<[u8]>,
next_wire_seq: u32, next_wire_seq: u32,
highest_received_seq: Option<u32>, highest_received_seq: Option<u32>,
@@ -82,7 +82,7 @@ pub(crate) struct PreparedOpenAcceptance {
} }
impl PreparedOpenAcceptance { impl PreparedOpenAcceptance {
pub(crate) fn normal_negotiated(&self) -> Option<&NegotiatedParams> { pub(crate) fn normal_negotiated(&self) -> Option<&NegotiationResult> {
match &self.next_state { match &self.next_state {
MachineState::Open(session) => Some(&session.negotiated), MachineState::Open(session) => Some(&session.negotiated),
_ => None, _ => None,
@@ -311,7 +311,7 @@ impl SessionMachine {
RequestToEncode::Echo { RequestToEncode::Echo {
token: session.token, token: session.token,
sequence: session.next_wire_seq, sequence: session.next_wire_seq,
params: &session.negotiated.params, params: &session.negotiated.peer_params,
payload: &[], payload: &[],
}, },
self.config.auth.hmac_key(), 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 { match &self.state {
MachineState::Open(session) => Some(&session.negotiated), MachineState::Open(session) => Some(&session.negotiated),
_ => None, _ => None,
@@ -623,19 +623,12 @@ impl SessionMachine {
kernel_tx_correlation_valid: true, kernel_tx_correlation_valid: true,
})); }));
let event = ClientEvent::SessionStarted { let outcome = OpenOutcome::Started(SessionStarted {
remote: self.remote, remote: self.remote,
token, token,
negotiated: negotiated.clone(), negotiation: negotiated,
at: now, at: now,
}; });
let outcome = OpenOutcome::Started {
remote: self.remote,
token,
negotiated,
event,
};
Ok(PreparedOpenAcceptance { Ok(PreparedOpenAcceptance {
next_state, next_state,
outcome, outcome,
@@ -650,16 +643,11 @@ impl SessionMachine {
let negotiated = let negotiated =
negotiate_params(&self.requested, reply.params, self.config.open.negotiation) negotiate_params(&self.requested, reply.params, self.config.open.negotiation)
.map_err(OpenAcceptanceFailure::without_cleanup)?; .map_err(OpenAcceptanceFailure::without_cleanup)?;
let event = ClientEvent::NoTestCompleted { let outcome = OpenOutcome::NoTestCompleted(NoTestCompleted {
remote: self.remote, remote: self.remote,
negotiated: negotiated.clone(), negotiation: negotiated,
at: now, at: now,
}; });
let outcome = OpenOutcome::NoTestCompleted {
remote: self.remote,
negotiated,
event,
};
Ok(PreparedOpenAcceptance { Ok(PreparedOpenAcceptance {
next_state: MachineState::NoTestCompleted, next_state: MachineState::NoTestCompleted,
outcome, outcome,
@@ -673,7 +661,7 @@ impl SessionMachine {
decode_echo_reply( decode_echo_reply(
packet, packet,
&session.negotiated.params, &session.negotiated.peer_params,
self.config.auth.hmac_key(), self.config.auth.hmac_key(),
) )
.ok() .ok()
@@ -873,10 +861,10 @@ impl SessionMachine {
/// rather than asserted away so that no 64-bit assumption is baked in. /// rather than asserted away so that no 64-bit assumption is baked in.
pub(crate) fn recv_buffer_size( pub(crate) fn recv_buffer_size(
has_hmac: bool, has_hmac: bool,
negotiated: Option<&NegotiatedParams>, negotiated: Option<&NegotiationResult>,
) -> Result<usize, ClientError> { ) -> Result<usize, ClientError> {
Ok(match negotiated { 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) .saturating_add(1)
.max(MIN_RECV_BUFFER_SIZE), .max(MIN_RECV_BUFFER_SIZE),
None => MIN_RECV_BUFFER_SIZE, None => MIN_RECV_BUFFER_SIZE,
+1 -1
View File
@@ -2,4 +2,4 @@ pub(crate) mod machine;
mod negotiation; mod negotiation;
pub(crate) use negotiation::negotiate_params; pub(crate) use negotiation::negotiate_params;
pub use negotiation::{NegotiatedParams, NegotiationRestriction}; pub use negotiation::{AcceptedSessionParameters, NegotiationChange, NegotiationResult};
+157 -107
View File
@@ -1,117 +1,139 @@
use std::fmt; use std::{fmt, time::Duration};
use irtt_proto::{Clock, Params, ReceivedStats, StampAt, PROTOCOL_VERSION}; use irtt_proto::{Clock, Params, ReceivedStats, StampAt, PROTOCOL_VERSION};
use crate::{config::NegotiationPolicy, error::ClientError}; 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 /// These values drive ordinary client behavior. The exact peer-returned wire
/// echo packets. `restrictions` records accepted differences from the request /// representation is available separately in [`NegotiationResult::peer_params`].
/// 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`].
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct NegotiatedParams { pub struct AcceptedSessionParameters {
/// Server-returned protocol parameters used for the session. /// `None` means continuous mode; finite durations are positive.
pub params: Params, pub duration: Option<Duration>,
/// Accepted server restrictions or parameter changes. /// Positive interval between probes when driven by a scheduler.
pub restrictions: Vec<NegotiationRestriction>, 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<String>,
} }
/// A server-side restriction applied during session parameter negotiation. /// Accepted session semantics, exact protocol evidence, and accepted changes.
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub enum NegotiationRestriction { pub struct NegotiationResult {
/// Run duration was reduced. pub accepted: AcceptedSessionParameters,
/// Exact decoded parameters returned by the peer, without normalization.
/// ///
/// A requested duration of `0` means the client requested continuous mode. /// Echo encoding/decoding uses these accepted wire values. `dscp` here is
/// When `requested_ns == 0` and `negotiated_ns > 0`, the server limited /// the raw IP TOS / Traffic Class byte, unlike [`AcceptedSessionParameters::dscp`].
/// that continuous request to a finite duration. pub peer_params: Params,
/// Differences accepted under [`crate::NegotiationPolicy::Loose`].
pub changes: Vec<NegotiationChange>,
}
/// 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 { DurationReduced {
requested_ns: i64, /// `None` means the request was continuous.
negotiated_ns: i64, requested: Option<Duration>,
negotiated: Duration,
}, },
/// Probe interval was increased.
IntervalIncreased { IntervalIncreased {
requested_ns: i64, requested: Duration,
negotiated_ns: i64, negotiated: Duration,
}, },
/// Probe interval was reduced. /// Loose policy accepts any positive reduced interval.
IntervalReduced { IntervalReduced {
requested_ns: i64, requested: Duration,
negotiated_ns: i64, 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 { ReceivedStatsChanged {
requested: ReceivedStats, requested: ReceivedStats,
negotiated: ReceivedStats, negotiated: ReceivedStats,
}, },
/// Returned timestamp placement differs from the request.
StampAtChanged { StampAtChanged {
requested: StampAt, requested: StampAt,
negotiated: StampAt, negotiated: StampAt,
}, },
/// Returned clock source differs from the request. ClockChanged {
ClockChanged { requested: Clock, negotiated: Clock }, requested: Clock,
/// Returned DSCP codepoint differs from the request. negotiated: Clock,
/// },
/// These values are DSCP codepoints (`0..=63`) for human-facing /// Both values are DSCP codepoints, rather than raw Traffic Class bytes.
/// consistency with [`crate::SessionRequest::dscp`], not the raw wire DscpChanged {
/// `Params::dscp` byte carried by [`NegotiatedParams::params`]. requested: u8,
DscpChanged { requested: i64, negotiated: i64 }, negotiated: u8,
/// Returned server payload fill behavior differs from the request. },
ServerFillChanged, ServerFillChanged {
requested: Option<String>,
negotiated: Option<String>,
},
} }
impl NegotiationRestriction { impl NegotiationChange {
/// Return a human-readable description of this negotiated restriction. /// Return a human-readable description of this accepted change.
pub fn message(&self) -> String { pub fn message(&self) -> String {
self.to_string() self.to_string()
} }
} }
impl fmt::Display for NegotiationRestriction { impl fmt::Display for NegotiationChange {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self { match self {
Self::DurationReduced { Self::DurationReduced {
requested_ns: 0, requested: None,
negotiated_ns, negotiated,
} => { } => {
write!( write!(
f, f,
"server limited continuous duration to {negotiated_ns} ns" "server limited continuous duration to {} ns",
negotiated.as_nanos()
) )
} }
Self::DurationReduced { Self::DurationReduced {
requested_ns, requested: Some(requested),
negotiated_ns, negotiated,
} => { } => {
write!( write!(
f, 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 { Self::IntervalIncreased {
requested_ns, requested,
negotiated_ns, negotiated,
} => { } => {
write!( write!(
f, 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 { Self::IntervalReduced {
requested_ns, requested,
negotiated_ns, negotiated,
} => { } => {
write!( write!(
f, 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 { Self::LengthReduced {
@@ -156,23 +178,32 @@ impl fmt::Display for NegotiationRestriction {
} => { } => {
write!(f, "server changed DSCP from {requested} to {negotiated}") 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<Duration> {
(nanos != 0).then(|| positive_duration(nanos))
}
pub(crate) fn negotiate_params( pub(crate) fn negotiate_params(
requested: &Params, requested: &Params,
returned: Params, returned: Params,
policy: NegotiationPolicy, policy: NegotiationPolicy,
) -> Result<NegotiatedParams, ClientError> { ) -> Result<NegotiationResult, ClientError> {
if returned.protocol_version != PROTOCOL_VERSION { if returned.protocol_version != PROTOCOL_VERSION {
return Err(ClientError::ProtocolVersionMismatch { return Err(ClientError::ProtocolVersionMismatch {
requested: PROTOCOL_VERSION, requested: PROTOCOL_VERSION,
received: returned.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)?; validate_duration_restriction(requested.duration_ns, returned.duration_ns)?;
if returned.length < 0 { if returned.length < 0 {
@@ -195,70 +226,70 @@ pub(crate) fn negotiate_params(
if returned.duration_ns < requested.duration_ns if returned.duration_ns < requested.duration_ns
|| (requested.duration_ns == 0 && returned.duration_ns > 0) || (requested.duration_ns == 0 && returned.duration_ns > 0)
{ {
record_restriction( record_change(
policy, policy,
&mut restrictions, &mut changes,
NegotiationRestriction::DurationReduced { NegotiationChange::DurationReduced {
requested_ns: requested.duration_ns, requested: finite_duration(requested.duration_ns),
negotiated_ns: returned.duration_ns, negotiated: positive_duration(returned.duration_ns),
}, },
)?; )?;
} }
if returned.interval_ns > requested.interval_ns { if returned.interval_ns > requested.interval_ns {
record_restriction( record_change(
policy, policy,
&mut restrictions, &mut changes,
NegotiationRestriction::IntervalIncreased { NegotiationChange::IntervalIncreased {
requested_ns: requested.interval_ns, requested: positive_duration(requested.interval_ns),
negotiated_ns: returned.interval_ns, negotiated: positive_duration(returned.interval_ns),
}, },
)?; )?;
} }
if returned.interval_ns < requested.interval_ns { if returned.interval_ns < requested.interval_ns {
record_restriction( record_change(
policy, policy,
&mut restrictions, &mut changes,
NegotiationRestriction::IntervalReduced { NegotiationChange::IntervalReduced {
requested_ns: requested.interval_ns, requested: positive_duration(requested.interval_ns),
negotiated_ns: returned.interval_ns, negotiated: positive_duration(returned.interval_ns),
}, },
)?; )?;
} }
if returned.length < requested.length { if returned.length < requested.length {
record_restriction( record_change(
policy, policy,
&mut restrictions, &mut changes,
NegotiationRestriction::LengthReduced { NegotiationChange::LengthReduced {
requested: requested.length, requested: u32::try_from(requested.length).expect("validated requested length"),
negotiated: returned.length, negotiated: u32::try_from(returned.length).expect("validated returned length"),
}, },
)?; )?;
} }
if returned.received_stats != requested.received_stats { if returned.received_stats != requested.received_stats {
record_restriction( record_change(
policy, policy,
&mut restrictions, &mut changes,
NegotiationRestriction::ReceivedStatsChanged { NegotiationChange::ReceivedStatsChanged {
requested: requested.received_stats, requested: requested.received_stats,
negotiated: returned.received_stats, negotiated: returned.received_stats,
}, },
)?; )?;
} }
if returned.stamp_at != requested.stamp_at { if returned.stamp_at != requested.stamp_at {
record_restriction( record_change(
policy, policy,
&mut restrictions, &mut changes,
NegotiationRestriction::StampAtChanged { NegotiationChange::StampAtChanged {
requested: requested.stamp_at, requested: requested.stamp_at,
negotiated: returned.stamp_at, negotiated: returned.stamp_at,
}, },
)?; )?;
} }
if returned.clock != requested.clock { if returned.clock != requested.clock {
record_restriction( record_change(
policy, policy,
&mut restrictions, &mut changes,
NegotiationRestriction::ClockChanged { NegotiationChange::ClockChanged {
requested: requested.clock, requested: requested.clock,
negotiated: returned.clock, negotiated: returned.clock,
}, },
@@ -270,29 +301,48 @@ pub(crate) fn negotiate_params(
}); });
} }
if returned.dscp == 0 && requested.dscp != 0 { if returned.dscp == 0 && requested.dscp != 0 {
record_restriction( record_change(
policy, policy,
&mut restrictions, &mut changes,
NegotiationRestriction::DscpChanged { NegotiationChange::DscpChanged {
// `requested`/`negotiated` here are raw wire TOS/Traffic Class // `requested`/`negotiated` here are raw wire TOS/Traffic Class
// bytes; shift back to the codepoint the user actually // bytes; shift back to the codepoint the user actually
// configured for a human-facing restriction. // configured for a semantic change record.
requested: requested.dscp >> 2, requested: u8::try_from(requested.dscp >> 2).expect("validated requested DSCP"),
negotiated: returned.dscp >> 2, negotiated: u8::try_from(returned.dscp >> 2).expect("validated returned DSCP"),
}, },
)?; )?;
} }
if returned.server_fill != requested.server_fill { if returned.server_fill != requested.server_fill {
record_restriction( record_change(
policy, policy,
&mut restrictions, &mut changes,
NegotiationRestriction::ServerFillChanged, 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 { // Compare the exact wire values above before deriving accepted semantics.
params: returned, // In particular, do not normalize fill or traffic class before Strict policy.
restrictions, 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(()) Ok(())
} }
fn record_restriction( fn record_change(
policy: NegotiationPolicy, policy: NegotiationPolicy,
restrictions: &mut Vec<NegotiationRestriction>, changes: &mut Vec<NegotiationChange>,
restriction: NegotiationRestriction, change: NegotiationChange,
) -> Result<(), ClientError> { ) -> Result<(), ClientError> {
if policy == NegotiationPolicy::Strict { if policy == NegotiationPolicy::Strict {
return Err(ClientError::NegotiationRejected { return Err(ClientError::NegotiationRejected {
reason: restriction.message(), reason: change.message(),
}); });
} }
restrictions.push(restriction); changes.push(change);
Ok(()) Ok(())
} }
+1 -1
View File
@@ -1,6 +1,6 @@
use super::*; use super::*;
use irtt_client::NegotiationPolicy; 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, Params, TimestampFields};
use irtt_proto::{ use irtt_proto::{
flags::FLAG_HMAC, flags::FLAG_OPEN, flags::FLAG_REPLY, verify_packet_hmac, Clock, HMAC_SIZE, flags::FLAG_HMAC, flags::FLAG_OPEN, flags::FLAG_REPLY, verify_packet_hmac, Clock, HMAC_SIZE,
+29 -18
View File
@@ -60,24 +60,35 @@ fn no_test_strict_negotiation_rejects_changed_params() {
#[test] #[test]
fn no_test_loose_negotiation_accepts_restricted_params() { fn no_test_loose_negotiation_accepts_restricted_params() {
let mut config = default_test_config(); for requested_duration in [Some(Duration::from_secs(3)), None] {
config.request.run_mode = RunMode::NoTest; let mut config = default_test_config();
config.open.negotiation = NegotiationPolicy::Loose; config.request.run_mode = RunMode::NoTest;
let requested = default_params(); config.request.duration = requested_duration;
let mut params = requested.clone(); config.open.negotiation = NegotiationPolicy::Loose;
params.duration_ns /= 2; let mut params = default_params();
let server = no_test_server(params.clone(), 0); params.duration_ns = 1_500_000_000;
let mut client = Client::connect(server.addr.to_string(), config).unwrap(); let server = no_test_server(params.clone(), 0);
let negotiated = assert_no_test_completed(client.open().unwrap()); let mut client = Client::connect(server.addr.to_string(), config).unwrap();
assert_eq!(negotiated.params, params); let negotiated = assert_no_test_completed(client.open().unwrap());
assert_eq!( assert_eq!(negotiated.peer_params, params);
negotiated.restrictions, assert_eq!(
vec![irtt_client::NegotiationRestriction::DurationReduced { negotiated.accepted.duration,
requested_ns: requested.duration_ns, Some(Duration::from_millis(1500))
negotiated_ns: params.duration_ns, );
}] assert_eq!(
); negotiated.changes,
server.join(); 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] #[test]
+10 -19
View File
@@ -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 { match outcome {
OpenOutcome::Started { OpenOutcome::Started(irtt_client::SessionStarted {
token, negotiated, .. token,
} => { negotiation: negotiated,
..
}) => {
assert_eq!(token, TOKEN); assert_eq!(token, TOKEN);
negotiated 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 { match outcome {
OpenOutcome::NoTestCompleted { OpenOutcome::NoTestCompleted(completed) => completed.negotiation,
negotiated, event, .. OpenOutcome::Started(_) => panic!("unexpected started outcome"),
} => {
assert!(matches!(
event,
ClientEvent::NoTestCompleted {
negotiated: ref event_params,
..
} if *event_params == negotiated
));
negotiated
}
OpenOutcome::Started { .. } => panic!("unexpected started outcome"),
} }
} }
+4 -4
View File
@@ -160,8 +160,8 @@ fn blocking_probes_use_kernel_tx_timing_and_negotiated_packet_marking() {
.unwrap(); .unwrap();
client.open().unwrap(); client.open().unwrap();
assert_eq!( assert_eq!(
client.negotiated_params().unwrap().params.dscp, client.negotiation().unwrap().accepted.dscp,
if allowed { 184 } else { 0 } if allowed { 46 } else { 0 }
); );
for _ in 0..PROBES { for _ in 0..PROBES {
let before = SystemTime::now(); let before = SystemTime::now();
@@ -202,8 +202,8 @@ fn async_probes_use_kernel_tx_timing_and_negotiated_packet_marking() {
.unwrap(); .unwrap();
client.open().await.unwrap(); client.open().await.unwrap();
assert_eq!( assert_eq!(
client.negotiated_params().unwrap().params.dscp, client.negotiation().unwrap().accepted.dscp,
if allowed { 184 } else { 0 } if allowed { 46 } else { 0 }
); );
for _ in 0..PROBES { for _ in 0..PROBES {
let before = SystemTime::now(); let before = SystemTime::now();
@@ -52,25 +52,24 @@ async fn exercise_client_server() {
let opened = client.open().await.unwrap(); let opened = client.open().await.unwrap();
match &opened { match &opened {
OpenOutcome::Started { OpenOutcome::Started(irtt_client::SessionStarted {
remote, remote,
token, token,
negotiated, negotiation: negotiated,
event, ..
} => { }) => {
assert_eq!(*remote, server_addr); assert_eq!(*remote, server_addr);
assert_ne!(*token, 0); assert_ne!(*token, 0);
assert_eq!(negotiated.params.received_stats, ReceivedStats::Both); assert_eq!(negotiated.accepted.received_stats, ReceivedStats::Both);
assert_eq!(negotiated.params.stamp_at, StampAt::Both); assert_eq!(negotiated.accepted.stamp_at, StampAt::Both);
assert_eq!(negotiated.params.clock, Clock::Both); assert_eq!(negotiated.accepted.clock, Clock::Both);
// A default client requests no fill, and this open succeeded under // A default client requests no fill, and this open succeeded under
// the default strict negotiation policy — which it could not have // the default strict negotiation policy — which it could not have
// if the server had answered an absent request with its own default // if the server had answered an absent request with its own default
// descriptor. The server still fills the payload with it. // descriptor. The server still fills the payload with it.
assert_eq!(negotiated.params.server_fill, None); assert_eq!(negotiated.accepted.server_fill, None);
assert!(matches!(event, ClientEvent::SessionStarted { .. }));
} }
OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"), OpenOutcome::NoTestCompleted(_) => panic!("normal client unexpectedly ran no-test"),
} }
assert!(matches!( assert!(matches!(
@@ -327,7 +327,7 @@ fn ignored_open_traffic_cannot_extend_the_absolute_attempt_deadline() {
.unwrap(); .unwrap();
assert!(matches!( assert!(matches!(
client.open().await.unwrap(), client.open().await.unwrap(),
OpenOutcome::Started { .. } OpenOutcome::Started(_)
)); ));
}); });
assert_eq!(server.finish().len(), 2); assert_eq!(server.finish().len(), 2);
@@ -404,7 +404,7 @@ fn authenticated_async_lifecycle_negotiates_dscp() {
.await .await
.unwrap(); .unwrap();
let opened = client.open().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(); client.send_probe().await.unwrap();
assert!(matches!(client.recv().await.unwrap().as_slice(), [ClientEvent::EchoReply { seq: 0, .. }])); 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))); 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 { match outcome {
OpenOutcome::Started { negotiated, .. } OpenOutcome::Started(irtt_client::SessionStarted {
| OpenOutcome::NoTestCompleted { negotiated, .. } => negotiated, negotiation: negotiated,
..
})
| OpenOutcome::NoTestCompleted(irtt_client::NoTestCompleted {
negotiation: negotiated,
..
}) => negotiated,
} }
} }
fn open_token(outcome: &OpenOutcome) -> u64 { fn open_token(outcome: &OpenOutcome) -> u64 {
match outcome { match outcome {
OpenOutcome::Started { token, .. } => *token, OpenOutcome::Started(irtt_client::SessionStarted { token, .. }) => *token,
OpenOutcome::NoTestCompleted { .. } => panic!("no-test open has no session 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. // state that refuses the retry.
assert_eq!(error_name(&blocking_first), "open timeout"); assert_eq!(error_name(&blocking_first), "open timeout");
assert_eq!(error_name(&blocking_first), error_name(&async_first)); assert_eq!(error_name(&blocking_first), error_name(&async_first));
assert!(matches!(blocking_second, OpenOutcome::Started { .. })); assert!(matches!(blocking_second, OpenOutcome::Started(_)));
assert!(matches!(async_second, OpenOutcome::Started { .. })); assert!(matches!(async_second, OpenOutcome::Started(_)));
assert_eq!( assert_eq!(
open_negotiated(&blocking_second), open_negotiated(&blocking_second),
open_negotiated(&async_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))) .set_recv_timeout(Some(Duration::from_millis(200)))
.unwrap(); .unwrap();
let outcome = blocking.open().unwrap(); let outcome = blocking.open().unwrap();
assert_eq!( assert_eq!(blocking.negotiation(), Some(open_negotiated(&outcome)));
blocking.negotiated_params(),
Some(open_negotiated(&outcome))
);
assert!(!blocking.has_pending_probes()); assert!(!blocking.has_pending_probes());
// Oversleeping cannot invalidate this: low-level sends remain caller-owned // Oversleeping cannot invalidate this: low-level sends remain caller-owned
// after the negotiated duration. The compliant server has no duration cap. // 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 .await
.unwrap(); .unwrap();
let outcome = client.open().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; tokio::time::sleep(Duration::from_millis(2)).await;
let mut sent = Vec::new(); let mut sent = Vec::new();
let mut replies = Vec::new(); let mut replies = Vec::new();
@@ -81,10 +81,16 @@ fn server_fill_negotiated_params_reflect_accepted_value() {
.unwrap(); .unwrap();
let outcome = client.open().unwrap(); let outcome = client.open().unwrap();
let negotiated = match outcome { 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:?}"), 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(); client.send_probe().unwrap();
assert!(matches!( assert!(matches!(
@@ -144,7 +150,7 @@ fn server_fill_loose_allows_server_to_remove_fill() {
open_observation_params(&run).server_fill, open_observation_params(&run).server_fill,
requested.server_fill requested.server_fill
); );
assert_eq!(run.negotiated.params.server_fill, None); assert_eq!(run.negotiated.accepted.server_fill, None);
} }
// ─── Echo behavior ─── // ─── Echo behavior ───
@@ -169,11 +175,11 @@ fn server_fill_does_not_affect_client_side_packet_layout() {
let (echo_len_without, _, _, _) = echo_observation(&run_without); let (echo_len_without, _, _, _) = echo_observation(&run_without);
assert_eq!(echo_len_with, echo_len_without); assert_eq!(echo_len_with, echo_len_without);
assert_eq!( assert_eq!(
test_echo_packet_len(false, &run_with.negotiated.params), test_echo_packet_len(false, &run_with.negotiated.peer_params),
echo_len_with echo_len_with
); );
assert_eq!( assert_eq!(
test_echo_packet_len(false, &run_without.negotiated.params), test_echo_packet_len(false, &run_without.negotiated.peer_params),
echo_len_without echo_len_without
); );
} }
@@ -10,7 +10,7 @@
use std::time::Duration; use std::time::Duration;
use irtt_client::{ use irtt_client::{
AsyncClient, ClientConfig, NegotiationPolicy, NegotiationRestriction, OpenOutcome, OpenPolicy, AsyncClient, ClientConfig, NegotiationChange, NegotiationPolicy, OpenOutcome, OpenPolicy,
SessionRequest, SessionRequest,
}; };
use irtt_proto::{ use irtt_proto::{
@@ -166,21 +166,23 @@ async fn exercise_fallback_negotiation() {
.await .await
.unwrap(); .unwrap();
match loose.open().await.unwrap() { match loose.open().await.unwrap() {
OpenOutcome::Started { negotiated, .. } => { OpenOutcome::Started(irtt_client::SessionStarted {
negotiation: negotiated,
..
}) => {
assert_eq!( assert_eq!(
negotiated negotiated.accepted.server_fill.as_deref(),
.params
.server_fill
.map(|fill| fill.value)
.as_deref(),
Some("pattern:69727474"), Some("pattern:69727474"),
"the server reports the default descriptor it fell back to" "the server reports the default descriptor it fell back to"
); );
assert!(negotiated assert!(negotiated
.restrictions .changes
.contains(&NegotiationRestriction::ServerFillChanged)); .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(); loose.close().await.unwrap();
+6 -14
View File
@@ -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] { match &run.observations[0] {
ServerObservation::Open { params: got, hmac } => { ServerObservation::Open { params: got, hmac } => {
assert_eq!(got, &params); assert_eq!(got, &params);
@@ -249,21 +249,13 @@ fn non_hmac_client_open_does_not_set_hmac_flag() {
fn assert_started(outcome: OpenOutcome, expected: &irtt_proto::Params) { fn assert_started(outcome: OpenOutcome, expected: &irtt_proto::Params) {
match outcome { match outcome {
OpenOutcome::Started { OpenOutcome::Started(irtt_client::SessionStarted {
token, token,
negotiated, negotiation: negotiated,
event:
ClientEvent::SessionStarted {
token: event_token,
negotiated: event_negotiated,
..
},
.. ..
} => { }) => {
assert_eq!(token, TOKEN); assert_eq!(token, TOKEN);
assert_eq!(event_token, TOKEN); assert_eq!(negotiated.peer_params, *expected);
assert_eq!(negotiated.params, *expected);
assert_eq!(event_negotiated.params, *expected);
} }
other => panic!("expected started open outcome, got {other:?}"), 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))) .set_recv_timeout(Some(Duration::from_millis(500)))
.unwrap(); .unwrap();
let outcome = client.open().unwrap(); let outcome = client.open().unwrap();
assert!(matches!(outcome, OpenOutcome::Started { .. })); assert!(matches!(outcome, OpenOutcome::Started(_)));
let sent = client.send_probe().unwrap(); let sent = client.send_probe().unwrap();
assert_eq!(sent.len(), 1); assert_eq!(sent.len(), 1);
@@ -93,7 +93,7 @@ fn backend_large_packet_smoke() {
.unwrap(); .unwrap();
let outcome = client.open().unwrap(); let outcome = client.open().unwrap();
assert!(matches!(outcome, irtt_client::OpenOutcome::Started { .. })); assert!(matches!(outcome, irtt_client::OpenOutcome::Started(_)));
client.send_probe().unwrap(); client.send_probe().unwrap();
let events = client.recv_once().unwrap(); let events = client.recv_once().unwrap();
@@ -162,7 +162,7 @@ async fn target_updates_preserve_identical_sessions_and_retire_removed_generatio
match events.recv().await.unwrap() { match events.recv().await.unwrap() {
ManagedEvent::Client { ManagedEvent::Client {
target, target,
event: ClientEvent::SessionStarted { .. }, event: ClientEvent::SessionStarted(_),
} if target == first => panic!("identical update reopened the session"), } if target == first => panic!("identical update reopened the session"),
ManagedEvent::Client { ManagedEvent::Client {
target, target,
@@ -10,7 +10,7 @@
use std::time::Duration; use std::time::Duration;
use irtt_client::{ use irtt_client::{
AsyncClient, ClientConfig, NegotiationPolicy, NegotiationRestriction, OpenOutcome, OpenPolicy, AsyncClient, ClientConfig, NegotiationChange, NegotiationPolicy, OpenOutcome, OpenPolicy,
SessionRequest, SessionRequest,
}; };
use irtt_proto::{Clock, ReceivedStats, StampAt}; 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 mut loose = client(server_addr, NegotiationPolicy::Loose).await;
let negotiated = match loose.open().await.unwrap() { let negotiated = match loose.open().await.unwrap() {
OpenOutcome::Started { negotiated, .. } => negotiated, OpenOutcome::Started(irtt_client::SessionStarted {
OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"), negotiation: negotiated,
..
}) => negotiated,
OpenOutcome::NoTestCompleted(_) => panic!("normal client unexpectedly ran no-test"),
}; };
assert_eq!( assert_eq!(
negotiated.params.stamp_at, negotiated.accepted.stamp_at,
StampAt::Midpoint, StampAt::Midpoint,
"a request for both instants is answered with the midpoint" "a request for both instants is answered with the midpoint"
); );
assert_eq!( assert_eq!(
negotiated.params.clock, negotiated.accepted.clock,
Clock::Both, Clock::Both,
"the clock is not restricted with it" "the clock is not restricted with it"
); );
assert!(negotiated assert!(negotiated
.restrictions .changes
.contains(&NegotiationRestriction::StampAtChanged { .contains(&NegotiationChange::StampAtChanged {
requested: StampAt::Both, requested: StampAt::Both,
negotiated: StampAt::Midpoint, negotiated: StampAt::Midpoint,
})); }));
@@ -112,16 +115,19 @@ async fn exercise_dscp_restriction() {
.await .await
.unwrap(); .unwrap();
let negotiated = match loose.open().await.unwrap() { let negotiated = match loose.open().await.unwrap() {
OpenOutcome::Started { negotiated, .. } => negotiated, OpenOutcome::Started(irtt_client::SessionStarted {
OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"), negotiation: negotiated,
..
}) => negotiated,
OpenOutcome::NoTestCompleted(_) => panic!("normal client unexpectedly ran no-test"),
}; };
assert_eq!( assert_eq!(
negotiated.params.dscp, 0, negotiated.accepted.dscp, 0,
"the raw wire parameter is negotiated to zero" "the accepted DSCP codepoint is zero"
); );
assert!(negotiated assert!(negotiated
.restrictions .changes
.contains(&NegotiationRestriction::DscpChanged { .contains(&NegotiationChange::DscpChanged {
requested: 46, requested: 46,
negotiated: 0, negotiated: 0,
})); }));
@@ -159,14 +165,21 @@ async fn exercise_default_policy() {
.await .await
.unwrap(); .unwrap();
let negotiated = match client.open().await.unwrap() { let negotiated = match client.open().await.unwrap() {
OpenOutcome::Started { negotiated, .. } => negotiated, OpenOutcome::Started(irtt_client::SessionStarted {
OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"), negotiation: negotiated,
..
}) => negotiated,
OpenOutcome::NoTestCompleted(_) => panic!("normal client unexpectedly ran no-test"),
}; };
assert_eq!(negotiated.restrictions, Vec::new()); assert_eq!(negotiated.changes, Vec::new());
assert_eq!(negotiated.params.stamp_at, StampAt::Both); assert_eq!(negotiated.accepted.stamp_at, StampAt::Both);
assert_eq!(negotiated.params.clock, Clock::Both); assert_eq!(negotiated.accepted.clock, Clock::Both);
assert_eq!(negotiated.params.dscp, 184, "the raw byte for codepoint 46"); assert_eq!(negotiated.accepted.dscp, 46);
assert_eq!(
negotiated.peer_params.dscp, 184,
"the raw byte for codepoint 46"
);
probe(&mut client).await; probe(&mut client).await;
client.close().await.unwrap(); client.close().await.unwrap();
@@ -7,8 +7,8 @@ mod protocol_options {
use std::time::Duration; use std::time::Duration;
use irtt_client::{ use irtt_client::{
Client, ClientConfig, ClientError, ClientEvent, OneWayDelaySample, ReceivedStatsSample, AcceptedSessionParameters, Client, ClientConfig, ClientError, ClientEvent, OneWayDelaySample,
RttSample, ServerTiming, SignedDuration, ReceivedStatsSample, RttSample, ServerTiming, SignedDuration,
}; };
use irtt_proto::{Clock, Params, ReceivedStats, StampAt, TimestampFields}; 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!(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); assert_echo_uses_params(&run, &returned, false);
let reply = expect_echo_reply(&run.reply); let reply = expect_echo_reply(&run.reply);
@@ -253,7 +266,7 @@ fn changed_compatibility_params(requested: &Params) -> Vec<Params> {
} }
fn assert_negotiated_echo_use(run: &OneProbeRun, params: &Params, hmac: bool) { 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_eq!(open_params(run), params);
assert_echo_uses_params(run, params, hmac); assert_echo_uses_params(run, params, hmac);
} }
@@ -14,7 +14,7 @@ fn backend_basic_open_echo_close() {
.unwrap(); .unwrap();
let outcome = client.open().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(); let sent = client.send_probe().unwrap();
assert_eq!(sent.len(), 1); assert_eq!(sent.len(), 1);
@@ -13,7 +13,7 @@ use std::{
}; };
use irtt_client::{ use irtt_client::{
Client, ClientConfig, ClientEvent, NegotiatedParams, OpenOutcome, OpenPolicy, SessionRequest, Client, ClientConfig, ClientEvent, NegotiationResult, OpenOutcome, OpenPolicy, SessionRequest,
}; };
use irtt_proto::{ use irtt_proto::{
compute_hmac_in_place, decode_request, echo_packet_len, encode_echo_reply, encode_open_reply, 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 struct OneProbeRun {
pub negotiated: NegotiatedParams, pub negotiated: NegotiationResult,
pub sent: ClientEvent, pub sent: ClientEvent,
pub reply: ClientEvent, pub reply: ClientEvent,
pub observations: Vec<ServerObservation>, pub observations: Vec<ServerObservation>,
@@ -514,10 +514,13 @@ fn recv_request_timeout(socket: &UdpSocket) -> Option<Vec<u8>> {
.map(|(size, _)| buf[..size].to_vec()) .map(|(size, _)| buf[..size].to_vec())
} }
fn assert_started(outcome: OpenOutcome) -> NegotiatedParams { fn assert_started(outcome: OpenOutcome) -> NegotiationResult {
match outcome { match outcome {
OpenOutcome::Started { negotiated, .. } => negotiated, OpenOutcome::Started(irtt_client::SessionStarted {
OpenOutcome::NoTestCompleted { .. } => panic!("expected started outcome"), negotiation: negotiated,
..
}) => negotiated,
OpenOutcome::NoTestCompleted(_) => panic!("expected started outcome"),
} }
} }
@@ -217,12 +217,15 @@ async fn opened_session(server_addr: SocketAddr, dscp: u8) -> OpenSession {
.await .await
.unwrap(); .unwrap();
match client.open().await.unwrap() { match client.open().await.unwrap() {
OpenOutcome::Started { negotiated, .. } => assert_eq!( OpenOutcome::Started(irtt_client::SessionStarted {
negotiated.params.dscp, negotiation: negotiated,
..
}) => assert_eq!(
negotiated.peer_params.dscp,
i64::from(dscp) << 2, i64::from(dscp) << 2,
"the server negotiates the raw byte the codepoint asks for" "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 { OpenSession {
client, client,
@@ -220,11 +220,11 @@ async fn connected_client(server_addr: SocketAddr) -> AsyncClient {
async fn assert_started(client: &mut AsyncClient, server_addr: SocketAddr) { async fn assert_started(client: &mut AsyncClient, server_addr: SocketAddr) {
match client.open().await.unwrap() { match client.open().await.unwrap() {
OpenOutcome::Started { remote, token, .. } => { OpenOutcome::Started(irtt_client::SessionStarted { remote, token, .. }) => {
assert_eq!(remote, server_addr); assert_eq!(remote, server_addr);
assert_ne!(token, 0); assert_ne!(token, 0);
} }
OpenOutcome::NoTestCompleted { .. } => panic!("a normal client unexpectedly ran no-test"), OpenOutcome::NoTestCompleted(_) => panic!("a normal client unexpectedly ran no-test"),
} }
} }
+2 -2
View File
@@ -124,8 +124,8 @@ pub(crate) fn normalize_event(event: &ClientEvent) -> Option<StatsEvent> {
at: (*timeout_at).max(sent_at.mono), at: (*timeout_at).max(sent_at.mono),
}), }),
ClientEvent::Warning { at, .. } => Some(StatsEvent::Warning { at: at.mono }), ClientEvent::Warning { at, .. } => Some(StatsEvent::Warning { at: at.mono }),
ClientEvent::SessionStarted { .. } ClientEvent::SessionStarted(_)
| ClientEvent::NoTestCompleted { .. } | ClientEvent::NoTestCompleted(_)
| ClientEvent::SessionClosed { .. } => None, | ClientEvent::SessionClosed { .. } => None,
} }
} }