feat(client): track and report specific negotiation restrictions

Replace basic validation with a detailed negotiation process that records specific parameter changes. This allows the client to provide better feedback when a server restricts requested session parameters like duration or DSCP.
This commit is contained in:
Nils Andreas Svee committed 2026-05-12 15:18:02 +07:00
1 parent a7e10358fd
commit cbe91f2fe0
7 files changed
+409 -87

No files matched your search

+1
View File
@@ -1456,6 +1456,7 @@ mod tests {
interval_ns: 1_000_000_000,
..Params::default()
},
restrictions: vec![],
},
at: test_timestamp(Duration::from_secs(1)),
};
+9 -14
View File
@@ -23,7 +23,7 @@ use crate::{
metadata::ReceiveMeta,
probe::{CompletedSet, PendingMap, PendingProbe, TimedOutMap},
receive::recv_datagram,
session::{validate_negotiated_params, ActiveSession, ClientPhase, NegotiatedParams},
session::{negotiate_params, ActiveSession, ClientPhase, NegotiatedParams},
socket::{connect_udp_socket, resolve_remote, validate_open_timeouts},
socket_options::{apply_dscp_to_socket, clear_dscp_on_socket},
timing::ClientTimestamp,
@@ -605,14 +605,12 @@ impl Client {
reply: OpenReply,
now: ClientTimestamp,
) -> Result<OpenOutcome, ClientError> {
validate_negotiated_params(
let token = reply.token;
let negotiated = negotiate_params(
&self.requested,
&reply.params,
reply.params,
self.config.negotiation_policy,
)?;
let negotiated = NegotiatedParams {
params: reply.params.clone(),
};
let recv_buffer_size = recv_buffer_size(self.config.hmac_key.is_some(), Some(&negotiated))?;
let negotiated_dscp =
u8::try_from(negotiated.params.dscp).map_err(|_| ClientError::InvalidConfig {
@@ -621,7 +619,7 @@ impl Client {
apply_dscp_to_socket(&self.socket, self.remote, negotiated_dscp)?;
self.negotiated = Some(negotiated.clone());
self.recv_buffer.resize(recv_buffer_size, 0);
self.phase = ClientPhase::Open { token: reply.token };
self.phase = ClientPhase::Open { token };
let end_mono = if negotiated.params.duration_ns > 0 {
Some(negotiated_end_mono(
@@ -647,14 +645,14 @@ impl Client {
let event = ClientEvent::SessionStarted {
remote: self.remote,
token: reply.token,
token,
negotiated: negotiated.clone(),
at: now,
};
Ok(OpenOutcome::Started {
remote: self.remote,
token: reply.token,
token,
negotiated,
event,
})
@@ -665,14 +663,11 @@ impl Client {
reply: OpenReply,
now: ClientTimestamp,
) -> Result<OpenOutcome, ClientError> {
validate_negotiated_params(
let negotiated = negotiate_params(
&self.requested,
&reply.params,
reply.params,
self.config.negotiation_policy,
)?;
let negotiated = NegotiatedParams {
params: reply.params.clone(),
};
self.negotiated = Some(negotiated.clone());
self.phase = ClientPhase::NoTestCompleted;
let event = ClientEvent::NoTestCompleted {
@@ -60,6 +60,13 @@ fn normal_open_uses_negotiated_dscp_not_requested_dscp() {
let negotiated = assert_open_started(client.open().unwrap());
assert_eq!(negotiated.params.dscp, 0);
assert_eq!(
negotiated.restrictions,
vec![crate::NegotiationRestriction::DscpChanged {
requested: 46,
negotiated: 0,
}]
);
assert_eq!(
socket_traffic_class(&client.socket, client.remote).unwrap(),
0
@@ -1,10 +1,29 @@
use super::*;
use crate::NegotiationRestriction;
fn assert_negotiates(
requested: &Params,
returned: &Params,
policy: NegotiationPolicy,
) -> NegotiatedParams {
negotiate_params(requested, returned.clone(), policy)
.unwrap_or_else(|err| panic!("expected negotiation success, got {err:?}"))
}
fn rejection_reason(requested: &Params, returned: &Params, policy: NegotiationPolicy) -> String {
match negotiate_params(requested, returned.clone(), policy) {
Err(ClientError::NegotiationRejected { reason }) => reason,
other => panic!("expected negotiation rejection, got {other:?}"),
}
}
#[test]
fn strict_negotiation_accepts_identical_params() {
let config = ClientConfig::default();
let params = params_from_config(&config).unwrap();
assert!(validate_negotiated_params(&params, &params, NegotiationPolicy::Strict).is_ok());
let negotiated = assert_negotiates(&params, &params, NegotiationPolicy::Strict);
assert_eq!(negotiated.params, params);
assert!(negotiated.restrictions.is_empty());
}
#[test]
@@ -13,20 +32,38 @@ fn strict_negotiation_rejects_changed_params() {
let requested = params_from_config(&config).unwrap();
let mut returned = requested.clone();
returned.dscp = 1;
assert!(matches!(
validate_negotiated_params(&requested, &returned, NegotiationPolicy::Strict),
Err(ClientError::NegotiationRejected { .. })
));
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
NegotiationRestriction::DscpChanged {
requested: requested.dscp,
negotiated: returned.dscp,
}
.message()
);
}
#[test]
fn loose_negotiation_accepts_server_restricted_params() {
let config = ClientConfig::default();
let requested = params_from_config(&config).unwrap();
let mut requested = params_from_config(&config).unwrap();
requested.length = 128;
let mut returned = requested.clone();
returned.duration_ns /= 2;
returned.length = 0;
assert!(validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose).is_ok());
let negotiated = assert_negotiates(&requested, &returned, NegotiationPolicy::Loose);
assert_eq!(
negotiated.restrictions,
vec![
NegotiationRestriction::DurationReduced {
requested_ns: requested.duration_ns,
negotiated_ns: returned.duration_ns,
},
NegotiationRestriction::LengthReduced {
requested: requested.length,
negotiated: returned.length,
},
]
);
}
#[test]
@@ -37,11 +74,10 @@ fn loose_negotiation_rejects_non_positive_returned_interval() {
for interval_ns in [0, -1] {
let mut returned = requested.clone();
returned.interval_ns = interval_ns;
assert!(matches!(
validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose),
Err(ClientError::NegotiationRejected { reason })
if reason == "interval must be positive"
));
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Loose),
"interval must be positive"
);
}
}
@@ -51,21 +87,28 @@ fn loose_negotiation_accepts_documented_duration_combinations() {
let requested_finite = params_from_config(&finite_config).unwrap();
let returned_same_finite = requested_finite.clone();
assert!(validate_negotiated_params(
assert!(assert_negotiates(
&requested_finite,
&returned_same_finite,
NegotiationPolicy::Loose
)
.is_ok());
.restrictions
.is_empty());
let mut returned_shorter_finite = requested_finite.clone();
returned_shorter_finite.duration_ns /= 2;
assert!(validate_negotiated_params(
&requested_finite,
&returned_shorter_finite,
NegotiationPolicy::Loose
)
.is_ok());
assert_eq!(
assert_negotiates(
&requested_finite,
&returned_shorter_finite,
NegotiationPolicy::Loose
)
.restrictions,
vec![NegotiationRestriction::DurationReduced {
requested_ns: requested_finite.duration_ns,
negotiated_ns: returned_shorter_finite.duration_ns,
}]
);
let continuous_config = ClientConfig {
duration: None,
@@ -74,21 +117,28 @@ fn loose_negotiation_accepts_documented_duration_combinations() {
let requested_continuous = params_from_config(&continuous_config).unwrap();
let returned_continuous = requested_continuous.clone();
assert!(validate_negotiated_params(
assert!(assert_negotiates(
&requested_continuous,
&returned_continuous,
NegotiationPolicy::Loose
)
.is_ok());
.restrictions
.is_empty());
let mut returned_finite = requested_continuous.clone();
returned_finite.duration_ns = 1_000_000_000;
assert!(validate_negotiated_params(
&requested_continuous,
&returned_finite,
NegotiationPolicy::Loose
)
.is_ok());
assert_eq!(
assert_negotiates(
&requested_continuous,
&returned_finite,
NegotiationPolicy::Loose
)
.restrictions,
vec![NegotiationRestriction::DurationReduced {
requested_ns: requested_continuous.duration_ns,
negotiated_ns: returned_finite.duration_ns,
}]
);
}
#[test]
@@ -98,11 +148,10 @@ fn loose_negotiation_rejects_finite_request_returned_continuous() {
let mut returned = requested.clone();
returned.duration_ns = 0;
assert!(matches!(
validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose),
Err(ClientError::NegotiationRejected { reason })
if reason == "server returned continuous duration for finite request"
));
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Loose),
"server returned continuous duration for finite request"
);
}
#[test]
@@ -111,10 +160,10 @@ fn loose_negotiation_rejects_negative_returned_length() {
let requested = params_from_config(&config).unwrap();
let mut returned = requested.clone();
returned.length = -1;
assert!(matches!(
validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose),
Err(ClientError::NegotiationRejected { reason }) if reason == "length must be non-negative"
));
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Loose),
"length must be non-negative"
);
}
#[test]
@@ -123,10 +172,10 @@ fn strict_negotiation_rejects_negative_returned_length() {
let requested = params_from_config(&config).unwrap();
let mut returned = requested.clone();
returned.length = -1;
assert!(matches!(
validate_negotiated_params(&requested, &returned, NegotiationPolicy::Strict),
Err(ClientError::NegotiationRejected { reason }) if reason == "length must be non-negative"
));
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
"length must be non-negative"
);
}
#[test]
@@ -137,10 +186,9 @@ fn loose_negotiation_rejects_runtime_invalid_returned_dscp() {
for dscp in [-1, 64] {
let mut returned = requested.clone();
returned.dscp = dscp;
assert!(matches!(
validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose),
Err(ClientError::NegotiationRejected { reason })
if reason == "dscp must be in range 0..=63"
));
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Loose),
"dscp must be in range 0..=63"
);
}
}
@@ -79,13 +79,21 @@ fn no_test_loose_negotiation_accepts_restricted_params() {
let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
config.run_mode = RunMode::NoTest;
config.negotiation_policy = NegotiationPolicy::Loose;
let mut params = params_from_config(&config).unwrap();
let requested = params_from_config(&config).unwrap();
let mut params = requested.clone();
params.duration_ns /= 2;
let server = no_test_server(params.clone(), 0);
config.server_addr = server.addr.to_string();
let mut client = Client::connect(config).unwrap();
let negotiated = assert_no_test_completed(client.open().unwrap());
assert_eq!(negotiated.params, params);
assert_eq!(
negotiated.restrictions,
vec![crate::NegotiationRestriction::DurationReduced {
requested_ns: requested.duration_ns,
negotiated_ns: params.duration_ns,
}]
);
server.join();
}
+1 -1
View File
@@ -31,5 +31,5 @@ pub use managed::{
CancellationToken, EventHub, EventSubscription, ManagedClient, ManagedClientSession,
SessionEndReason, SessionOutcome, SubscriberConfig, SubscriberOverflow,
};
pub use session::NegotiatedParams;
pub use session::{NegotiatedParams, NegotiationRestriction};
pub use timing::ClientTimestamp;
+288 -25
View File
@@ -1,6 +1,6 @@
use std::time::Instant;
use irtt_proto::{Params, PROTOCOL_VERSION};
use irtt_proto::{Clock, Params, ReceivedStats, StampAt, PROTOCOL_VERSION};
use crate::{
config::{NegotiationPolicy, MAX_DSCP_CODEPOINT},
@@ -11,6 +11,106 @@ use crate::{
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NegotiatedParams {
pub params: Params,
pub restrictions: Vec<NegotiationRestriction>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum NegotiationRestriction {
DurationReduced {
requested_ns: i64,
negotiated_ns: i64,
},
IntervalIncreased {
requested_ns: i64,
negotiated_ns: i64,
},
IntervalReduced {
requested_ns: i64,
negotiated_ns: i64,
},
LengthReduced {
requested: i64,
negotiated: i64,
},
ReceivedStatsChanged {
requested: ReceivedStats,
negotiated: ReceivedStats,
},
StampAtChanged {
requested: StampAt,
negotiated: StampAt,
},
ClockChanged {
requested: Clock,
negotiated: Clock,
},
DscpChanged {
requested: i64,
negotiated: i64,
},
ServerFillChanged,
}
impl NegotiationRestriction {
pub fn message(&self) -> String {
match self {
Self::DurationReduced {
requested_ns: 0,
negotiated_ns,
} => {
format!("server limited continuous duration to {negotiated_ns} ns")
}
Self::DurationReduced {
requested_ns,
negotiated_ns,
} => {
format!("server reduced duration from {requested_ns} ns to {negotiated_ns} ns")
}
Self::IntervalIncreased {
requested_ns,
negotiated_ns,
} => {
format!("server increased interval from {requested_ns} ns to {negotiated_ns} ns")
}
Self::IntervalReduced {
requested_ns,
negotiated_ns,
} => {
format!("server reduced interval from {requested_ns} ns to {negotiated_ns} ns")
}
Self::LengthReduced {
requested,
negotiated,
} => {
format!("server reduced packet length from {requested} bytes to {negotiated} bytes")
}
Self::ReceivedStatsChanged {
requested,
negotiated,
} => {
format!("server changed received-stats from {requested:?} to {negotiated:?}")
}
Self::StampAtChanged {
requested,
negotiated,
} => {
format!("server changed stamp-at from {requested:?} to {negotiated:?}")
}
Self::ClockChanged {
requested,
negotiated,
} => {
format!("server changed clock from {requested:?} to {negotiated:?}")
}
Self::DscpChanged {
requested,
negotiated,
} => {
format!("server changed dscp from {requested} to {negotiated}")
}
Self::ServerFillChanged => "server changed payload fill behavior".to_owned(),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -35,17 +135,19 @@ pub(crate) struct ActiveSession {
pub sending_done: bool,
}
pub(crate) fn validate_negotiated_params(
pub(crate) fn negotiate_params(
requested: &Params,
returned: &Params,
returned: Params,
policy: NegotiationPolicy,
) -> Result<(), ClientError> {
) -> Result<NegotiatedParams, ClientError> {
if returned.protocol_version != PROTOCOL_VERSION {
return Err(ClientError::ProtocolVersionMismatch {
requested: PROTOCOL_VERSION,
received: returned.protocol_version,
});
}
let mut restrictions = Vec::new();
validate_duration_restriction(requested.duration_ns, returned.duration_ns)?;
if returned.length < 0 {
return Err(ClientError::NegotiationRejected {
@@ -64,12 +166,100 @@ pub(crate) fn validate_negotiated_params(
}
validate_dscp_restriction(returned.dscp)?;
if policy == NegotiationPolicy::Strict && returned != requested {
return Err(ClientError::NegotiationRejected {
reason: "returned params differ from requested params".to_owned(),
});
if returned.duration_ns < requested.duration_ns
|| (requested.duration_ns == 0 && returned.duration_ns > 0)
{
record_restriction(
policy,
&mut restrictions,
NegotiationRestriction::DurationReduced {
requested_ns: requested.duration_ns,
negotiated_ns: returned.duration_ns,
},
)?;
}
Ok(())
if returned.interval_ns > requested.interval_ns {
record_restriction(
policy,
&mut restrictions,
NegotiationRestriction::IntervalIncreased {
requested_ns: requested.interval_ns,
negotiated_ns: returned.interval_ns,
},
)?;
}
if returned.interval_ns < requested.interval_ns {
record_restriction(
policy,
&mut restrictions,
NegotiationRestriction::IntervalReduced {
requested_ns: requested.interval_ns,
negotiated_ns: returned.interval_ns,
},
)?;
}
if returned.length < requested.length {
record_restriction(
policy,
&mut restrictions,
NegotiationRestriction::LengthReduced {
requested: requested.length,
negotiated: returned.length,
},
)?;
}
if returned.received_stats != requested.received_stats {
record_restriction(
policy,
&mut restrictions,
NegotiationRestriction::ReceivedStatsChanged {
requested: requested.received_stats,
negotiated: returned.received_stats,
},
)?;
}
if returned.stamp_at != requested.stamp_at {
record_restriction(
policy,
&mut restrictions,
NegotiationRestriction::StampAtChanged {
requested: requested.stamp_at,
negotiated: returned.stamp_at,
},
)?;
}
if returned.clock != requested.clock {
record_restriction(
policy,
&mut restrictions,
NegotiationRestriction::ClockChanged {
requested: requested.clock,
negotiated: returned.clock,
},
)?;
}
if returned.dscp != requested.dscp {
record_restriction(
policy,
&mut restrictions,
NegotiationRestriction::DscpChanged {
requested: requested.dscp,
negotiated: returned.dscp,
},
)?;
}
if returned.server_fill != requested.server_fill {
record_restriction(
policy,
&mut restrictions,
NegotiationRestriction::ServerFillChanged,
)?;
}
Ok(NegotiatedParams {
params: returned,
restrictions,
})
}
fn validate_duration_restriction(requested: i64, returned: i64) -> Result<(), ClientError> {
@@ -104,6 +294,21 @@ fn validate_dscp_restriction(returned: i64) -> Result<(), ClientError> {
Ok(())
}
fn record_restriction(
policy: NegotiationPolicy,
restrictions: &mut Vec<NegotiationRestriction>,
restriction: NegotiationRestriction,
) -> Result<(), ClientError> {
if policy == NegotiationPolicy::Strict {
return Err(ClientError::NegotiationRejected {
reason: restriction.message(),
});
}
restrictions.push(restriction);
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
@@ -127,17 +332,26 @@ mod tests {
fn assert_rejected(requested: &Params, returned: &Params, policy: NegotiationPolicy) {
assert!(matches!(
validate_negotiated_params(requested, returned, policy),
negotiate_params(requested, returned.clone(), policy),
Err(ClientError::NegotiationRejected { .. })
));
}
fn assert_negotiates(
requested: &Params,
returned: &Params,
policy: NegotiationPolicy,
) -> NegotiatedParams {
negotiate_params(requested, returned.clone(), policy)
.unwrap_or_else(|err| panic!("expected negotiation success, got {err:?}"))
}
fn rejection_reason(
requested: &Params,
returned: &Params,
policy: NegotiationPolicy,
) -> String {
match validate_negotiated_params(requested, returned, policy) {
match negotiate_params(requested, returned.clone(), policy) {
Err(ClientError::NegotiationRejected { reason }) => reason,
other => panic!("expected negotiation rejection, got {other:?}"),
}
@@ -149,27 +363,65 @@ mod tests {
let mut returned = requested.clone();
returned.length = 128;
assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
NegotiationRestriction::LengthReduced {
requested: requested.length,
negotiated: returned.length,
}
.message()
);
let mut returned = requested.clone();
returned.dscp = 8;
assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
NegotiationRestriction::DscpChanged {
requested: requested.dscp,
negotiated: returned.dscp,
}
.message()
);
let mut returned = requested.clone();
returned.received_stats = ReceivedStats::Count;
assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
NegotiationRestriction::ReceivedStatsChanged {
requested: requested.received_stats,
negotiated: returned.received_stats,
}
.message()
);
let mut returned = requested.clone();
returned.stamp_at = StampAt::Midpoint;
assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
NegotiationRestriction::StampAtChanged {
requested: requested.stamp_at,
negotiated: returned.stamp_at,
}
.message()
);
let mut returned = requested.clone();
returned.clock = Clock::Wall;
assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
NegotiationRestriction::ClockChanged {
requested: requested.clock,
negotiated: returned.clock,
}
.message()
);
let mut returned = requested.clone();
returned.server_fill = None;
assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
assert_eq!(
rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
NegotiationRestriction::ServerFillChanged.message()
);
}
#[test]
@@ -178,8 +430,13 @@ mod tests {
let mut returned = requested.clone();
returned.duration_ns = requested.duration_ns / 2;
assert!(
validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose).is_ok()
let negotiated = assert_negotiates(&requested, &returned, NegotiationPolicy::Loose);
assert_eq!(
negotiated.restrictions,
vec![NegotiationRestriction::DurationReduced {
requested_ns: requested.duration_ns,
negotiated_ns: returned.duration_ns,
}]
);
let mut returned = requested.clone();
@@ -200,12 +457,18 @@ mod tests {
continuous_requested.duration_ns = 0;
let mut finite_returned = continuous_requested.clone();
finite_returned.duration_ns = 1_000_000_000;
assert!(validate_negotiated_params(
let negotiated = assert_negotiates(
&continuous_requested,
&finite_returned,
NegotiationPolicy::Loose
)
.is_ok());
NegotiationPolicy::Loose,
);
assert_eq!(
negotiated.restrictions,
vec![NegotiationRestriction::DurationReduced {
requested_ns: 0,
negotiated_ns: finite_returned.duration_ns,
}]
);
assert_rejected(
&continuous_requested,
@@ -213,9 +476,9 @@ mod tests {
NegotiationPolicy::Strict,
);
assert!(validate_negotiated_params(
&continuous_requested,
assert!(negotiate_params(
&continuous_requested,
continuous_requested.clone(),
NegotiationPolicy::Strict
)
.is_ok());