refactor(client): separate accepted negotiation state from peer evidence
This commit is contained in:
32 files changed
+507
-447
No files matched your search
@@ -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 {
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
@@ -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<Self, ClientError> {
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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<u32>,
|
||||
@@ -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<usize, ClientError> {
|
||||
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,
|
||||
|
||||
@@ -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};
|
||||
@@ -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<NegotiationRestriction>,
|
||||
pub struct AcceptedSessionParameters {
|
||||
/// `None` means continuous mode; finite durations are positive.
|
||||
pub duration: Option<Duration>,
|
||||
/// 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<String>,
|
||||
}
|
||||
|
||||
/// 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<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 {
|
||||
requested_ns: i64,
|
||||
negotiated_ns: i64,
|
||||
/// `None` means the request was continuous.
|
||||
requested: Option<Duration>,
|
||||
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<String>,
|
||||
negotiated: Option<String>,
|
||||
},
|
||||
}
|
||||
|
||||
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<Duration> {
|
||||
(nanos != 0).then(|| positive_duration(nanos))
|
||||
}
|
||||
|
||||
pub(crate) fn negotiate_params(
|
||||
requested: &Params,
|
||||
returned: Params,
|
||||
policy: NegotiationPolicy,
|
||||
) -> Result<NegotiatedParams, ClientError> {
|
||||
) -> Result<NegotiationResult, ClientError> {
|
||||
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<NegotiationRestriction>,
|
||||
restriction: NegotiationRestriction,
|
||||
changes: &mut Vec<NegotiationChange>,
|
||||
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(())
|
||||
}
|
||||
Reference in new issue
Block a user