From dad4d2a29812a696c7e6992894d79da65b8df818 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 01:23:03 +0700 Subject: [PATCH 1/5] refactor(session): prepare opening transactions --- crates/irtt-client/src/session/machine.rs | 513 ++++++++++++++++++++-- 1 file changed, 465 insertions(+), 48 deletions(-) diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 2c53753..d14755c 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -61,6 +61,71 @@ struct ActiveSession { completed: CompletedSet, } +#[derive(Debug)] +pub(crate) struct PreparedOpenRequest { + pub(crate) bytes: Box<[u8]>, +} + +#[derive(Debug)] +pub(crate) enum OpenDatagramDisposition { + Ignore, + Trusted(OpenReply), +} + +#[derive(Debug)] +pub(crate) struct PreparedOpenAcceptance { + next_state: MachineState, + outcome: OpenOutcome, + cleanup_close: Option>, +} + +impl PreparedOpenAcceptance { + pub(crate) fn negotiated(&self) -> &NegotiatedParams { + match &self.outcome { + OpenOutcome::Started { negotiated, .. } + | OpenOutcome::NoTestCompleted { negotiated, .. } => negotiated, + } + } + + pub(crate) fn normal_negotiated(&self) -> Option<&NegotiatedParams> { + match &self.next_state { + MachineState::Open(session) => Some(&session.negotiated), + MachineState::NoTestCompleted => None, + MachineState::Connected | MachineState::Closed { .. } => { + unreachable!("open acceptance only prepares open or no-test state") + } + } + } + + pub(crate) fn cleanup_close_packet(&self) -> Option<&[u8]> { + self.cleanup_close.as_deref() + } + + #[cfg(test)] + pub(crate) fn has_prepared_active_session(&self) -> bool { + matches!(self.next_state, MachineState::Open(_)) + } +} + +#[derive(Debug)] +pub(crate) struct OpenAcceptanceFailure { + pub(crate) primary: ClientError, + pub(crate) cleanup_close: Option>, +} + +impl OpenAcceptanceFailure { + fn new(primary: ClientError, cleanup_close: Option>) -> Self { + Self { + primary, + cleanup_close, + } + } + + fn without_cleanup(primary: ClientError) -> Self { + Self::new(primary, None) + } +} + #[derive(Debug)] pub(crate) struct PreparedProbe { pub(crate) bytes: Box<[u8]>, @@ -121,22 +186,106 @@ impl SessionMachine { self.config.hmac_key.is_some() } - pub(crate) fn open_packet(&self) -> Result, ClientError> { - match self.state { - MachineState::Connected => {} - MachineState::Open(_) => return Err(ClientError::AlreadyOpen), - MachineState::Closed { .. } => return Err(ClientError::AlreadyClosed), - MachineState::NoTestCompleted => return Err(ClientError::AlreadyCompleted), - } - + pub(crate) fn prepare_open_request(&self) -> Result { + self.ensure_connected()?; let request = OpenRequest { params: self.requested.clone(), close: self.config.run_mode == RunMode::NoTest, }; - Ok(encode_open_request( - &request, - self.config.hmac_key.as_deref(), - )?) + let bytes = encode_open_request(&request, self.config.hmac_key.as_deref())?; + Ok(PreparedOpenRequest { + bytes: bytes.into_boxed_slice(), + }) + } + + pub(crate) fn inspect_open_datagram( + &self, + packet: &[u8], + ) -> Result { + self.ensure_connected()?; + match irtt_proto::decode_open_reply(packet, self.config.hmac_key.as_deref()) { + Ok(reply) => Ok(OpenDatagramDisposition::Trusted(reply)), + Err(irtt_proto::ProtoError::ZeroToken) => Err(ClientError::ZeroToken), + Err( + error @ (irtt_proto::ProtoError::TruncatedVarint + | irtt_proto::ProtoError::VarintOverflow + | irtt_proto::ProtoError::InvalidUtf8 + | irtt_proto::ProtoError::InvalidEnum { .. } + | irtt_proto::ProtoError::NegativePacketLength { .. } + | irtt_proto::ProtoError::ParameterLengthTooLarge { .. } + | irtt_proto::ProtoError::MalformedParams), + ) => Err(ClientError::Protocol(error)), + Err(_) => Ok(OpenDatagramDisposition::Ignore), + } + } + + pub(crate) fn prepare_open_acceptance( + &self, + reply: OpenReply, + now: ClientTimestamp, + ) -> Result { + self.ensure_connected() + .map_err(OpenAcceptanceFailure::without_cleanup)?; + + let reply_is_close = flags::has(reply.flags, flags::FLAG_CLOSE); + let cleanup_close = + if self.config.run_mode == RunMode::Normal && !reply_is_close && reply.token != 0 { + let bytes = encode_close_request( + &CloseRequest { token: reply.token }, + self.config.hmac_key.as_deref(), + ) + .map_err(ClientError::from) + .map_err(OpenAcceptanceFailure::without_cleanup)?; + Some(bytes.into_boxed_slice()) + } else { + None + }; + + if reply.params.protocol_version != PROTOCOL_VERSION { + return Err(OpenAcceptanceFailure::new( + ClientError::ProtocolVersionMismatch { + requested: PROTOCOL_VERSION, + received: reply.params.protocol_version, + }, + cleanup_close, + )); + } + + match self.config.run_mode { + RunMode::Normal if reply_is_close => Err(OpenAcceptanceFailure::new( + ClientError::ServerRejected, + cleanup_close, + )), + RunMode::Normal if reply.token == 0 => Err(OpenAcceptanceFailure::new( + ClientError::ZeroToken, + cleanup_close, + )), + RunMode::Normal => self.prepare_normal_open(reply, now, cleanup_close), + RunMode::NoTest if !reply_is_close => Err(OpenAcceptanceFailure::new( + ClientError::UnexpectedNoTestReply, + cleanup_close, + )), + RunMode::NoTest if reply.token != 0 => Err(OpenAcceptanceFailure::new( + ClientError::NonZeroNoTestToken { token: reply.token }, + cleanup_close, + )), + RunMode::NoTest => self.prepare_no_test_open(reply, now), + } + } + + pub(crate) fn commit_open(&mut self, prepared: PreparedOpenAcceptance) -> OpenOutcome { + debug_assert!( + matches!(self.state, MachineState::Connected), + "open acceptance commits only from connected state" + ); + self.state = prepared.next_state; + prepared.outcome + } + + // Temporary private bridges keep this first commit independently compiling. + // The blocking and managed migrations remove them before the PR is complete. + pub(crate) fn open_packet(&self) -> Result, ClientError> { + Ok(self.prepare_open_request()?.bytes.into_vec()) } pub(crate) fn decode_open_reply(&self, packet: &[u8]) -> Result { @@ -155,24 +304,13 @@ impl SessionMachine { where F: FnOnce(&NegotiatedParams) -> Result<(), ClientError>, { - if reply.params.protocol_version != PROTOCOL_VERSION { - return Err(ClientError::ProtocolVersionMismatch { - requested: PROTOCOL_VERSION, - received: reply.params.protocol_version, - }); - } - - let reply_is_close = flags::has(reply.flags, flags::FLAG_CLOSE); - match self.config.run_mode { - RunMode::Normal if reply_is_close => Err(ClientError::ServerRejected), - RunMode::Normal if reply.token == 0 => Err(ClientError::ZeroToken), - RunMode::Normal => self.accept_normal_open(reply, now, before_normal_open), - RunMode::NoTest if !reply_is_close => Err(ClientError::UnexpectedNoTestReply), - RunMode::NoTest if reply.token != 0 => { - Err(ClientError::NonZeroNoTestToken { token: reply.token }) - } - RunMode::NoTest => self.accept_no_test_open(reply, now), + let prepared = self + .prepare_open_acceptance(reply, now) + .map_err(|failure| failure.primary)?; + if let Some(negotiated) = prepared.normal_negotiated() { + before_normal_open(negotiated)?; } + Ok(self.commit_open(prepared)) } pub(crate) fn probe_timeout(&self) -> Duration { @@ -374,24 +512,22 @@ impl SessionMachine { self.open_session().map(|_| ()) } - fn accept_normal_open( - &mut self, + fn prepare_normal_open( + &self, reply: OpenReply, now: ClientTimestamp, - before_normal_open: F, - ) -> Result - where - F: FnOnce(&NegotiatedParams) -> Result<(), ClientError>, - { + cleanup_close: Option>, + ) -> Result { let token = reply.token; - let negotiated = negotiate_params( + let negotiated = match negotiate_params( &self.requested, reply.params, self.config.negotiation_policy, - )?; - before_normal_open(&negotiated)?; - - self.state = MachineState::Open(Box::new(ActiveSession { + ) { + Ok(negotiated) => negotiated, + Err(primary) => return Err(OpenAcceptanceFailure::new(primary, cleanup_close)), + }; + let next_state = MachineState::Open(Box::new(ActiveSession { token, negotiated: negotiated.clone(), next_wire_seq: 0, @@ -409,34 +545,44 @@ impl SessionMachine { at: now, }; - Ok(OpenOutcome::Started { + let outcome = OpenOutcome::Started { remote: self.remote, token, negotiated, event, + }; + Ok(PreparedOpenAcceptance { + next_state, + outcome, + cleanup_close, }) } - fn accept_no_test_open( - &mut self, + fn prepare_no_test_open( + &self, reply: OpenReply, now: ClientTimestamp, - ) -> Result { + ) -> Result { let negotiated = negotiate_params( &self.requested, reply.params, self.config.negotiation_policy, - )?; - self.state = MachineState::NoTestCompleted; + ) + .map_err(OpenAcceptanceFailure::without_cleanup)?; let event = ClientEvent::NoTestCompleted { remote: self.remote, negotiated: negotiated.clone(), at: now, }; - Ok(OpenOutcome::NoTestCompleted { + let outcome = OpenOutcome::NoTestCompleted { remote: self.remote, negotiated, event, + }; + Ok(PreparedOpenAcceptance { + next_state: MachineState::NoTestCompleted, + outcome, + cleanup_close: None, }) } @@ -620,6 +766,15 @@ impl SessionMachine { } } + fn ensure_connected(&self) -> Result<(), ClientError> { + match self.state { + MachineState::Connected => Ok(()), + MachineState::Open(_) => Err(ClientError::AlreadyOpen), + MachineState::Closed { .. } => Err(ClientError::AlreadyClosed), + MachineState::NoTestCompleted => Err(ClientError::AlreadyCompleted), + } + } + fn open_session_mut(&mut self) -> Result<&mut ActiveSession, ClientError> { match &mut self.state { MachineState::Open(session) => Ok(session), @@ -896,6 +1051,268 @@ mod tests { } } + fn connected_machine(config: ClientConfig) -> SessionMachine { + SessionMachine::new(config, "127.0.0.1:2112".parse().unwrap()).unwrap() + } + + fn normal_open_reply(machine: &SessionMachine, token: u64) -> OpenReply { + OpenReply { + flags: flags::FLAG_OPEN | flags::FLAG_REPLY, + token, + params: machine.requested.clone(), + } + } + + fn encoded_open_reply( + machine: &SessionMachine, + reply: &OpenReply, + ) -> Result, irtt_proto::ProtoError> { + irtt_proto::encode_open_reply(reply, machine.config.hmac_key.as_deref()) + } + + #[test] + fn prepared_open_request_is_exact_and_inert_when_dropped() { + let machine = connected_machine(ClientConfig::default()); + let first = machine.prepare_open_request().unwrap(); + let second = machine.prepare_open_request().unwrap(); + + assert_eq!(first.bytes, second.bytes); + assert!(!first.bytes.is_empty()); + drop(first); + drop(second); + assert!(matches!(machine.state, MachineState::Connected)); + } + + #[test] + fn malformed_wrong_direction_and_invalid_flags_are_ignored() { + let machine = connected_machine(ClientConfig::default()); + assert!(matches!( + machine.inspect_open_datagram(&[0_u8]), + Ok(OpenDatagramDisposition::Ignore) + )); + + let request = machine.prepare_open_request().unwrap(); + assert!(matches!( + machine.inspect_open_datagram(&request.bytes), + Ok(OpenDatagramDisposition::Ignore) + )); + + let reply = normal_open_reply(&machine, 0x1020_3040_5060_7080); + let mut reserved = encoded_open_reply(&machine, &reply).unwrap(); + reserved[3] |= 0x10; + assert!(matches!( + machine.inspect_open_datagram(&reserved), + Ok(OpenDatagramDisposition::Ignore) + )); + assert!(matches!(machine.state, MachineState::Connected)); + } + + #[test] + fn missing_unexpected_and_bad_hmac_are_ignored() { + let key = b"open-key".to_vec(); + let authenticated = connected_machine(ClientConfig { + hmac_key: Some(key.clone()), + ..ClientConfig::default() + }); + let reply = normal_open_reply(&authenticated, 0x1020_3040_5060_7080); + let plain = irtt_proto::encode_open_reply(&reply, None).unwrap(); + assert!(matches!( + authenticated.inspect_open_datagram(&plain), + Ok(OpenDatagramDisposition::Ignore) + )); + + let mut bad = encoded_open_reply(&authenticated, &reply).unwrap(); + *bad.last_mut().unwrap() ^= 0x80; + assert!(matches!( + authenticated.inspect_open_datagram(&bad), + Ok(OpenDatagramDisposition::Ignore) + )); + + let plain_machine = connected_machine(ClientConfig::default()); + let unexpected = irtt_proto::encode_open_reply(&reply, Some(&key)).unwrap(); + assert!(matches!( + plain_machine.inspect_open_datagram(&unexpected), + Ok(OpenDatagramDisposition::Ignore) + )); + } + + #[test] + fn authenticated_zero_token_is_trusted_and_terminal() { + let key = b"open-key".to_vec(); + let machine = connected_machine(ClientConfig { + hmac_key: Some(key.clone()), + ..ClientConfig::default() + }); + let reply = normal_open_reply(&machine, 0x1020_3040_5060_7080); + let mut packet = encoded_open_reply(&machine, &reply).unwrap(); + let token_offset = 4 + irtt_proto::HMAC_SIZE; + packet[token_offset..token_offset + 8].fill(0); + irtt_proto::compute_hmac_in_place(&key, &mut packet, 4).unwrap(); + + assert!(matches!( + machine.inspect_open_datagram(&packet), + Err(ClientError::ZeroToken) + )); + assert!(matches!(machine.state, MachineState::Connected)); + } + + #[test] + fn trusted_invalid_parameter_encoding_is_terminal() { + let machine = connected_machine(ClientConfig::default()); + let mut packet = irtt_proto::MAGIC.to_vec(); + packet.push(flags::FLAG_OPEN | flags::FLAG_REPLY); + packet.extend_from_slice(&0x1020_3040_5060_7080_u64.to_le_bytes()); + packet.push(0x80); + + assert!(matches!( + machine.inspect_open_datagram(&packet), + Err(ClientError::Protocol( + irtt_proto::ProtoError::TruncatedVarint + )) + )); + assert!(matches!(machine.state, MachineState::Connected)); + } + + #[test] + fn trusted_rejection_version_and_run_mode_failures_do_not_open() { + let machine = connected_machine(ClientConfig::default()); + let rejection = OpenReply { + flags: flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE, + token: 0, + params: machine.requested.clone(), + }; + assert!(matches!( + machine + .prepare_open_acceptance(rejection, timestamp(Instant::now())) + .unwrap_err() + .primary, + ClientError::ServerRejected + )); + + let mut version = normal_open_reply(&machine, 0x1020_3040_5060_7080); + version.params.protocol_version += 1; + assert!(matches!( + machine + .prepare_open_acceptance(version, timestamp(Instant::now())) + .unwrap_err() + .primary, + ClientError::ProtocolVersionMismatch { .. } + )); + + let no_test = connected_machine(ClientConfig { + run_mode: RunMode::NoTest, + ..ClientConfig::default() + }); + assert!(matches!( + no_test + .prepare_open_acceptance( + normal_open_reply(&no_test, 0x1020_3040_5060_7080), + timestamp(Instant::now()), + ) + .unwrap_err() + .primary, + ClientError::UnexpectedNoTestReply + )); + assert!(matches!(machine.state, MachineState::Connected)); + assert!(matches!(no_test.state, MachineState::Connected)); + } + + #[test] + fn dropping_prepared_acceptance_leaves_connected_state() { + let machine = connected_machine(ClientConfig::default()); + let prepared = machine + .prepare_open_acceptance( + normal_open_reply(&machine, 0x1020_3040_5060_7080), + timestamp(Instant::now()), + ) + .unwrap(); + + assert!(prepared.has_prepared_active_session()); + assert!(prepared.cleanup_close_packet().is_some()); + drop(prepared); + assert!(matches!(machine.state, MachineState::Connected)); + } + + #[test] + fn commit_open_assigns_prebuilt_state_once() { + let mut machine = connected_machine(ClientConfig::default()); + let token = 0x1020_3040_5060_7080; + let prepared = machine + .prepare_open_acceptance( + normal_open_reply(&machine, token), + timestamp(Instant::now()), + ) + .unwrap(); + + let outcome = machine.commit_open(prepared); + + assert!(matches!( + outcome, + OpenOutcome::Started { + token: outcome_token, + .. + } if outcome_token == token + )); + assert!(machine.is_open()); + assert!(matches!( + machine.prepare_open_request(), + Err(ClientError::AlreadyOpen) + )); + } + + #[test] + fn no_test_preparation_completes_without_active_session() { + let config = ClientConfig { + run_mode: RunMode::NoTest, + ..ClientConfig::default() + }; + let mut machine = connected_machine(config); + let reply = OpenReply { + flags: flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE, + token: 0, + params: machine.requested.clone(), + }; + let prepared = machine + .prepare_open_acceptance(reply, timestamp(Instant::now())) + .unwrap(); + + assert!(prepared.normal_negotiated().is_none()); + assert!(prepared.cleanup_close_packet().is_none()); + assert!(matches!(machine.state, MachineState::Connected)); + let outcome = machine.commit_open(prepared); + assert!(matches!(outcome, OpenOutcome::NoTestCompleted { .. })); + assert!(matches!(machine.state, MachineState::NoTestCompleted)); + } + + #[test] + fn cleanup_close_is_preencoded_with_token_and_hmac() { + let key = b"cleanup-key".to_vec(); + let machine = connected_machine(ClientConfig { + hmac_key: Some(key.clone()), + ..ClientConfig::default() + }); + let token = 0x1020_3040_5060_7080; + let prepared = machine + .prepare_open_acceptance( + normal_open_reply(&machine, token), + timestamp(Instant::now()), + ) + .unwrap(); + let cleanup = prepared.cleanup_close_packet().unwrap(); + + assert_eq!(cleanup[3], flags::FLAG_CLOSE | flags::FLAG_HMAC); + irtt_proto::verify_hmac(&key, cleanup, 4).unwrap(); + assert_eq!( + u64::from_le_bytes( + cleanup[4 + irtt_proto::HMAC_SIZE..12 + irtt_proto::HMAC_SIZE] + .try_into() + .unwrap() + ), + token + ); + assert!(matches!(machine.state, MachineState::Connected)); + } + #[test] fn dropping_prepared_probe_changes_nothing() { let machine = open_machine(4, Duration::from_secs(1)); From ff27ec24c6ee3befeda44112de083ef8c0cdb28c Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 01:29:07 +0700 Subject: [PATCH 2/5] refactor(client): transact blocking session opening --- crates/irtt-client/src/client.rs | 341 +++++++++++++--- crates/irtt-client/src/client/tests/hmac.rs | 62 ++- .../irtt-client/src/client/tests/no_test.rs | 1 + crates/irtt-client/src/client/tests/open.rs | 374 ++++++++++++++++++ crates/irtt-client/src/error.rs | 4 +- crates/irtt-client/tests/hmac.rs | 9 +- 6 files changed, 709 insertions(+), 82 deletions(-) diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index a674ad8..592e3ce 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -4,6 +4,9 @@ use std::{ time::{Duration, Instant}, }; +#[cfg(test)] +use std::cell::Cell; + #[cfg(test)] use irtt_proto::{flags, Params, TimestampFields, PROTOCOL_VERSION}; @@ -12,7 +15,10 @@ use crate::{ error::ClientError, event::{ClientEvent, OpenOutcome}, receive::recv_datagram, - session::machine::{recv_buffer_size, ProbeSent, SessionMachine, MAX_OPEN_PACKET_SIZE}, + session::machine::{ + recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, ProbeSent, + SessionMachine, MAX_OPEN_PACKET_SIZE, + }, socket::{connect_udp_socket, resolve_remote, validate_open_timeouts}, socket_options::{apply_dscp_to_socket, clear_dscp_on_socket}, timing::ClientTimestamp, @@ -28,6 +34,34 @@ enum ProbeScheduleMode { Managed { scheduled_at: Instant }, } +#[derive(Debug)] +struct PreparedClientOpen { + machine: PreparedOpenAcceptance, + schedule: Option, + recv_buffer_len: Option, + negotiated_dscp: Option, + previous_dscp: Option, + post_open_recv_timeout: Option, +} + +#[derive(Debug)] +struct PreparedClientOpenFailure { + primary: ClientError, + machine: PreparedOpenAcceptance, +} + +#[cfg(test)] +#[derive(Debug, Default)] +struct ClientTestHooks { + fail_open_dscp: Cell, + fail_open_timeout_restore: Cell, + fail_cleanup_send: Cell, + recv_buffer_len_override: Cell>, + fail_dscp_restore: Cell, + last_restored_read_timeout: Cell>, + prepared_active_session_before_dscp: Cell, +} + #[cfg(test)] #[derive(Debug, Clone, Copy)] pub(crate) struct ProbeSendTimestamps { @@ -60,6 +94,9 @@ pub struct Client { socket: UdpSocket, remote: SocketAddr, recv_buffer: Vec, + applied_dscp: Option, + #[cfg(test)] + test_hooks: ClientTestHooks, #[cfg(test)] probe_reported_len: Option, #[cfg(test)] @@ -86,6 +123,9 @@ impl Client { socket, remote, recv_buffer: vec![0_u8; recv_buffer_size(false, None)?], + applied_dscp: None, + #[cfg(test)] + test_hooks: ClientTestHooks::default(), #[cfg(test)] probe_reported_len: None, #[cfg(test)] @@ -99,66 +139,24 @@ impl Client { /// /// On success, returns the negotiated open outcome and transitions the /// client into either an open probe session or completed no-test state. - /// Open attempts use [`ClientConfig::open_timeouts`]. + /// Open attempts use [`ClientConfig::open_timeouts`]. Malformed, unrelated, + /// or unauthenticated datagrams are ignored until the current attempt's + /// absolute deadline, so one attempt may consume several datagrams without + /// retransmitting. Silence or ignored traffic eventually produces + /// [`ClientError::OpenTimeout`], while authenticated incompatibility remains + /// terminal. + /// + /// When a trusted reply allocates a token but later negotiation or socket + /// preparation fails, the client sends a best-effort cleanup close and + /// preserves the original failure. Failed opening never leaves the session + /// machine open. pub fn open(&mut self) -> Result { - let outcome = (|| -> Result { - let packet = self.runtime.open_packet()?; - let mut buf = [0_u8; MAX_OPEN_PACKET_SIZE]; - - for timeout in &self.runtime.config().open_timeouts { - self.socket.set_read_timeout(Some(*timeout))?; - self.socket.send(&packet)?; - - match self.socket.recv(&mut buf) { - Ok(size) => { - let reply = self.runtime.decode_open_reply(&buf[..size])?; - let opened_at = ClientTimestamp::now(); - let remote = self.remote; - let has_hmac = self.runtime.has_hmac(); - let socket = &self.socket; - let recv_buffer = &mut self.recv_buffer; - let mut schedule = None; - - let outcome = - self.runtime - .accept_open_reply(reply, opened_at, |negotiated| { - let size = recv_buffer_size(has_hmac, Some(negotiated))?; - let negotiated_dscp = u8::try_from(negotiated.params.dscp) - .map_err(|_| ClientError::InvalidConfig { - reason: "negotiated dscp must be in range 0..=63" - .to_owned(), - })?; - apply_dscp_to_socket(socket, remote, negotiated_dscp)?; - recv_buffer.resize(size, 0); - schedule = - Some(ProbeSchedule::new(opened_at.mono, negotiated)?); - Ok(()) - })?; - self.schedule = schedule; - return Ok(outcome); - } - Err(err) - if matches!( - err.kind(), - io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut - ) => {} - Err(err) => return Err(ClientError::Socket(err)), - } - } - - Err(ClientError::OpenTimeout) - })(); - - let restore = self - .socket - .set_read_timeout(self.runtime.config().socket_config.recv_timeout); - - match (outcome, restore) { - (Ok(outcome), Ok(())) => Ok(outcome), - (Ok(_), Err(source)) => Err(ClientError::ReadTimeoutRestore { source }), - (Err(err), Ok(())) => Err(err), - (Err(err), Err(_)) => Err(err), + let result = self.open_transaction(); + if result.is_err() { + let _ = + self.restore_open_read_timeout(self.runtime.config().socket_config.recv_timeout); } + result } /// Send a close request and emit a [`ClientEvent::SessionClosed`] event. @@ -417,6 +415,231 @@ impl Client { )); Ok(events) } + + fn open_transaction(&mut self) -> Result { + let request = self.runtime.prepare_open_request()?; + let mut buf = [0_u8; MAX_OPEN_PACKET_SIZE]; + let attempt_count = self.runtime.config().open_timeouts.len(); + + for attempt in 0..attempt_count { + let timeout = self.runtime.config().open_timeouts[attempt]; + let deadline = Instant::now() + .checked_add(timeout) + .ok_or(ClientError::DurationOverflow)?; + self.socket.set_read_timeout(Some(timeout))?; + self.socket.send(&request.bytes)?; + + loop { + let now = Instant::now(); + let Some(remaining) = deadline.checked_duration_since(now) else { + break; + }; + if remaining.is_zero() { + break; + } + self.socket.set_read_timeout(Some(remaining))?; + + let datagram = match recv_datagram(&self.socket, &mut buf) { + Ok(datagram) => datagram, + Err(err) + if matches!( + err.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => + { + if Instant::now() >= deadline { + break; + } + continue; + } + Err(err) => return Err(ClientError::Socket(err)), + }; + if datagram.received_at.mono > deadline { + break; + } + + let reply = match self.runtime.inspect_open_datagram(&buf[..datagram.len])? { + OpenDatagramDisposition::Ignore => continue, + OpenDatagramDisposition::Trusted(reply) => reply, + }; + let machine = match self + .runtime + .prepare_open_acceptance(reply, datagram.received_at) + { + Ok(machine) => machine, + Err(failure) => { + self.send_cleanup_close_best_effort(failure.cleanup_close.as_deref()); + return Err(failure.primary); + } + }; + let prepared = match self.prepare_client_open(machine, datagram.received_at) { + Ok(prepared) => prepared, + Err(failure) => { + self.send_cleanup_close_best_effort(failure.machine.cleanup_close_packet()); + return Err(failure.primary); + } + }; + return self.apply_prepared_open(prepared); + } + } + + Err(ClientError::OpenTimeout) + } + + fn prepare_client_open( + &self, + machine: PreparedOpenAcceptance, + opened_at: ClientTimestamp, + ) -> Result> { + let Some(negotiated) = machine.normal_negotiated() else { + return Ok(PreparedClientOpen { + machine, + schedule: None, + recv_buffer_len: None, + negotiated_dscp: None, + previous_dscp: self.applied_dscp, + post_open_recv_timeout: self.runtime.config().socket_config.recv_timeout, + }); + }; + let schedule = match ProbeSchedule::new(opened_at.mono, negotiated) { + Ok(schedule) => schedule, + Err(primary) => return Err(Box::new(PreparedClientOpenFailure { primary, machine })), + }; + let recv_buffer_len = match recv_buffer_size(self.runtime.has_hmac(), Some(negotiated)) { + Ok(size) => size, + Err(primary) => return Err(Box::new(PreparedClientOpenFailure { primary, machine })), + }; + #[cfg(test)] + let recv_buffer_len = self + .test_hooks + .recv_buffer_len_override + .get() + .unwrap_or(recv_buffer_len); + let negotiated_dscp = match u8::try_from(negotiated.params.dscp) { + Ok(dscp) => dscp, + Err(_) => { + return Err(Box::new(PreparedClientOpenFailure { + primary: ClientError::InvalidConfig { + reason: "negotiated dscp must be in range 0..=63".to_owned(), + }, + machine, + })); + } + }; + + Ok(PreparedClientOpen { + machine, + schedule: Some(schedule), + recv_buffer_len: Some(recv_buffer_len), + negotiated_dscp: Some(negotiated_dscp), + previous_dscp: self.applied_dscp, + post_open_recv_timeout: self.runtime.config().socket_config.recv_timeout, + }) + } + + fn apply_prepared_open( + &mut self, + prepared: PreparedClientOpen, + ) -> Result { + let PreparedClientOpen { + machine, + schedule, + recv_buffer_len, + negotiated_dscp, + previous_dscp, + post_open_recv_timeout, + } = prepared; + + if let (Some(recv_buffer_len), Some(negotiated_dscp)) = (recv_buffer_len, negotiated_dscp) { + let previous_len = self.recv_buffer.len(); + let additional = recv_buffer_len.saturating_sub(previous_len); + if let Err(source) = self.recv_buffer.try_reserve(additional) { + self.send_cleanup_close_best_effort(machine.cleanup_close_packet()); + return Err(ClientError::AllocationFailed { + operation: "negotiated receive buffer", + source, + }); + } + self.recv_buffer.resize(recv_buffer_len, 0); + + #[cfg(test)] + self.test_hooks + .prepared_active_session_before_dscp + .set(machine.has_prepared_active_session()); + if let Err(primary) = self.apply_open_dscp(negotiated_dscp) { + self.recv_buffer.truncate(previous_len); + self.restore_dscp_best_effort(previous_dscp); + self.send_cleanup_close_best_effort(machine.cleanup_close_packet()); + return Err(primary); + } + if let Err(source) = self.restore_open_read_timeout(post_open_recv_timeout) { + let primary = ClientError::ReadTimeoutRestore { source }; + self.recv_buffer.truncate(previous_len); + self.restore_dscp_best_effort(previous_dscp); + self.send_cleanup_close_best_effort(machine.cleanup_close_packet()); + return Err(primary); + } + + let outcome = self.runtime.commit_open(machine); + self.schedule = schedule; + self.applied_dscp = Some(negotiated_dscp); + Ok(outcome) + } else { + debug_assert!(schedule.is_none()); + if let Err(source) = self.restore_open_read_timeout(post_open_recv_timeout) { + return Err(ClientError::ReadTimeoutRestore { source }); + } + let outcome = self.runtime.commit_open(machine); + self.schedule = None; + self.applied_dscp = None; + Ok(outcome) + } + } + + fn restore_dscp_best_effort(&self, previous_dscp: Option) { + #[cfg(test)] + if self.test_hooks.fail_dscp_restore.replace(false) { + return; + } + let _ = match previous_dscp { + Some(dscp) => apply_dscp_to_socket(&self.socket, self.remote, dscp), + None => clear_dscp_on_socket(&self.socket, self.remote), + }; + } + + fn send_cleanup_close_best_effort(&self, packet: Option<&[u8]>) { + if let Some(packet) = packet { + #[cfg(test)] + if self.test_hooks.fail_cleanup_send.replace(false) { + return; + } + let _ = self.socket.send(packet); + } + } + + fn apply_open_dscp(&self, dscp: u8) -> Result<(), ClientError> { + #[cfg(test)] + if self.test_hooks.fail_open_dscp.replace(false) { + return Err(ClientError::SocketOption { + operation: "set negotiated DSCP", + remote: self.remote, + source: io::Error::other("injected negotiated DSCP failure"), + }); + } + apply_dscp_to_socket(&self.socket, self.remote, dscp) + } + + fn restore_open_read_timeout(&self, timeout: Option) -> io::Result<()> { + #[cfg(test)] + self.test_hooks.last_restored_read_timeout.set(timeout); + #[cfg(test)] + if self.test_hooks.fail_open_timeout_restore.replace(false) { + return Err(io::Error::other( + "injected configured read timeout restoration failure", + )); + } + self.socket.set_read_timeout(timeout) + } } pub(crate) fn echo_sent_event( diff --git a/crates/irtt-client/src/client/tests/hmac.rs b/crates/irtt-client/src/client/tests/hmac.rs index a49d106..dc36a99 100644 --- a/crates/irtt-client/src/client/tests/hmac.rs +++ b/crates/irtt-client/src/client/tests/hmac.rs @@ -19,45 +19,77 @@ fn hmac_open_success() { } #[test] -fn hmac_open_rejects_missing_hmac() { +fn hmac_open_ignores_missing_hmac_before_valid_reply() { let key = b"secret".to_vec(); let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key); + config.hmac_key = Some(key.clone()); let params = params_from_config(&config).unwrap(); let server = start_fake_server(move |socket, tx| { let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); - socket.send_to(&reply, peer).unwrap(); + let missing = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + let valid = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&key)); + socket.send_to(&missing, peer).unwrap(); + socket.send_to(&valid, peer).unwrap(); }); config.server_addr = server.addr.to_string(); let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(), - Err(ClientError::Protocol( - irtt_proto::ProtoError::HmacPresenceMismatch - )) - )); + assert_open_started(client.open().unwrap()); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(server.rx.try_recv().is_err()); server.join(); } #[test] -fn hmac_open_rejects_bad_hmac() { +fn hmac_open_ignores_bad_hmac_before_valid_reply() { let key = b"secret".to_vec(); let wrong_key = b"wrong".to_vec(); let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); - config.hmac_key = Some(key); + config.hmac_key = Some(key.clone()); let params = params_from_config(&config).unwrap(); let server = start_fake_server(move |socket, tx| { let (_, peer) = recv_request(&socket, &tx); - let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&wrong_key)); - socket.send_to(&reply, peer).unwrap(); + let bad = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&wrong_key)); + let valid = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(&key)); + socket.send_to(&bad, peer).unwrap(); + socket.send_to(&valid, peer).unwrap(); }); config.server_addr = server.addr.to_string(); let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open().unwrap()); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(server.rx.try_recv().is_err()); + server.join(); +} + +#[test] +fn post_token_hmac_negotiation_failure_sends_authenticated_cleanup_close() { + let key = b"secret".to_vec(); + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.hmac_key = Some(key.clone()); + let mut returned = params_from_config(&config).unwrap(); + returned.interval_ns += 1; + let server_key = key.clone(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, Some(&server_key)); + socket.send_to(&reply, peer).unwrap(); + let _ = recv_request(&socket, &tx); + }); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + assert!(matches!( client.open(), - Err(ClientError::Protocol(irtt_proto::ProtoError::BadHmac)) + Err(ClientError::NegotiationRejected { .. }) )); + let packets: Vec<_> = server.rx.iter().take(2).collect(); + let cleanup = &packets[1]; + assert_eq!(cleanup[3], flags::FLAG_CLOSE | FLAG_HMAC); + verify_hmac(&key, cleanup, HMAC_OFFSET).unwrap(); + assert_eq!( + u64::from_le_bytes(cleanup[4 + HMAC_SIZE..12 + HMAC_SIZE].try_into().unwrap()), + TOKEN + ); server.join(); } diff --git a/crates/irtt-client/src/client/tests/no_test.rs b/crates/irtt-client/src/client/tests/no_test.rs index d1fc893..2af264f 100644 --- a/crates/irtt-client/src/client/tests/no_test.rs +++ b/crates/irtt-client/src/client/tests/no_test.rs @@ -13,6 +13,7 @@ fn no_test_success_validates_params() { assert_eq!(negotiated.params, params); assert!(client.next_send_deadline().is_none()); assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); server.join(); } diff --git a/crates/irtt-client/src/client/tests/open.rs b/crates/irtt-client/src/client/tests/open.rs index ae19c5f..12b8859 100644 --- a/crates/irtt-client/src/client/tests/open.rs +++ b/crates/irtt-client/src/client/tests/open.rs @@ -9,6 +9,30 @@ fn successful_open_handshake() { let negotiated = assert_open_started(client.open().unwrap()); assert_eq!(negotiated.params, params); + assert!(client.schedule.is_some()); + assert_eq!(client.applied_dscp, Some(0)); + server.join(); +} + +#[test] +fn successful_open_restores_configured_receive_timeout() { + let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + let params = params_from_config(&config).unwrap(); + let server = open_success_server(params); + let mut config = default_test_config(server.addr); + config.socket_config.recv_timeout = Some(Duration::from_millis(75)); + let mut client = Client::connect(config).unwrap(); + + assert_open_started(client.open().unwrap()); + + assert_eq!( + client.test_hooks.last_restored_read_timeout.get(), + Some(Duration::from_millis(75)) + ); + assert_eq!( + client.socket.read_timeout().unwrap(), + Some(Duration::from_millis(75)) + ); server.join(); } @@ -49,11 +73,19 @@ fn open_timeout_after_all_timeouts() { let server = timeout_server(Duration::from_millis(700)); let config = ClientConfig { open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, ..default_test_config(server.addr) }; let mut client = Client::connect(config).unwrap(); assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); assert_eq!(server.rx.iter().take(2).count(), 2); + assert_eq!( + client.test_hooks.last_restored_read_timeout.get(), + Some(Duration::from_millis(50)) + ); server.join(); } @@ -69,6 +101,8 @@ fn protocol_version_mismatch_fails() { client.open(), Err(ClientError::ProtocolVersionMismatch { received: 2, .. }) )); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(client.runtime.prepare_open_request().is_ok()); server.join(); } @@ -83,5 +117,345 @@ fn server_rejection_fails_in_normal_mode() { }); let mut client = Client::connect(default_test_config(server.addr)).unwrap(); assert!(matches!(client.open(), Err(ClientError::ServerRejected))); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(client.runtime.prepare_open_request().is_ok()); + server.join(); +} + +#[test] +fn multiple_untrusted_datagrams_before_valid_reply_use_one_attempt() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + let valid = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None); + let mut wrong_type = valid.clone(); + wrong_type[3] = FLAG_REPLY; + let mut reserved_flags = valid.clone(); + reserved_flags[3] |= 0x10; + let mut bad_magic = valid.clone(); + bad_magic[0] ^= 0xff; + let unexpected_hmac = + open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, Some(b"unexpected")); + + for packet in [ + vec![0_u8], + bad_magic, + wrong_type, + reserved_flags, + unexpected_hmac, + valid, + ] { + socket.send_to(&packet, peer).unwrap(); + } + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + + assert_open_started(client.open().unwrap()); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(server.rx.try_recv().is_err()); + assert!(client.schedule.is_some()); + server.join(); +} + +#[test] +fn ignored_datagrams_do_not_restart_the_attempt_deadline() { + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + thread::sleep(Duration::from_millis(150)); + socket.send_to(&[0_u8], peer).unwrap(); + thread::sleep(Duration::from_millis(150)); + socket.send_to(&[0_u8], peer).unwrap(); + let _ = recv_request(&socket, &tx); + }); + let config = ClientConfig { + open_timeouts: vec![Duration::from_millis(250), Duration::from_millis(250)], + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + let started = Instant::now(); + + assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); + let requests: Vec<_> = server.rx.iter().take(2).collect(); + assert_eq!(requests.len(), 2); + assert!( + started.elapsed() < Duration::from_millis(700), + "ignored datagrams restarted an attempt deadline" + ); + server.join(); +} + +#[test] +fn only_ignored_open_datagrams_eventually_time_out() { + let server = start_fake_server(move |socket, tx| { + for _ in 0..2 { + let (_, peer) = recv_request(&socket, &tx); + socket.send_to(&[0_u8], peer).unwrap(); + socket.send_to(&MAGIC, peer).unwrap(); + } + thread::sleep(Duration::from_millis(300)); + }); + let config = ClientConfig { + open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)], + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(50)), + ..Default::default() + }, + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + + assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); + assert_eq!(server.rx.iter().take(2).count(), 2); + assert!(server.rx.try_recv().is_err()); + assert_eq!( + client.test_hooks.last_restored_read_timeout.get(), + Some(Duration::from_millis(50)) + ); + server.join(); +} + +#[test] +fn trusted_zero_token_normal_reply_is_terminal() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, 0, ¶ms, None), peer) + .unwrap(); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + + assert!(matches!(client.open(), Err(ClientError::ZeroToken))); + assert_eq!(server.rx.iter().take(1).count(), 1); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + server.join(); +} + +#[test] +fn opening_deadline_overflow_occurs_before_send() { + let server = timeout_server(Duration::from_millis(250)); + let config = ClientConfig { + open_timeouts: vec![Duration::MAX], + ..default_test_config(server.addr) + }; + let mut client = Client::connect(config).unwrap(); + + assert!(matches!(client.open(), Err(ClientError::DurationOverflow))); + assert!(server.rx.try_recv().is_err()); + server.join(); +} + +#[test] +fn schedule_duration_overflow_cannot_partially_open() { + let server = start_fake_server(|_socket, _tx| {}); + let client = Client::connect(default_test_config(server.addr)).unwrap(); + let reply = irtt_proto::OpenReply { + flags: FLAG_OPEN | FLAG_REPLY, + token: TOKEN, + params: params_from_config(client.runtime.config()).unwrap(), + }; + let mut latest = Instant::now(); + let mut step = Duration::MAX; + while !step.is_zero() { + if let Some(candidate) = latest.checked_add(step) { + latest = candidate; + } + step /= 2; + } + let opened_at = ClientTimestamp { + mono: latest, + wall: SystemTime::now(), + }; + let machine = client + .runtime + .prepare_open_acceptance(reply, opened_at) + .unwrap(); + let previous_recv_len = client.recv_buffer.len(); + + let failure = client.prepare_client_open(machine, opened_at).unwrap_err(); + + assert!(matches!( + failure.primary, + ClientError::NegotiationRejected { .. } + )); + assert!(failure.machine.cleanup_close_packet().is_some()); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); + assert_eq!(client.recv_buffer.len(), previous_recv_len); + server.join(); +} + +#[test] +fn post_token_negotiation_failure_sends_cleanup_close() { + let mut returned = default_params(); + returned.interval_ns += 1; + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, None), + peer, + ) + .unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + + assert!(matches!( + client.open(), + Err(ClientError::NegotiationRejected { .. }) + )); + let packets: Vec<_> = server.rx.iter().take(2).collect(); + assert_eq!(packets[1][3], flags::FLAG_CLOSE); + assert_eq!( + u64::from_le_bytes(packets[1][4..12].try_into().unwrap()), + TOKEN + ); + assert!(client.runtime.prepare_open_request().is_ok()); + server.join(); +} + +#[test] +fn cleanup_send_failure_does_not_replace_primary_open_error() { + let mut returned = default_params(); + returned.interval_ns += 1; + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, None), + peer, + ) + .unwrap(); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + client.test_hooks.fail_cleanup_send.set(true); + + assert!(matches!( + client.open(), + Err(ClientError::NegotiationRejected { .. }) + )); + assert!(client.runtime.prepare_open_request().is_ok()); + server.join(); +} + +#[test] +fn receive_buffer_reservation_failure_precedes_dscp_application() { + let mut returned = default_params(); + returned.dscp = 46; + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, None), + peer, + ) + .unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut config = default_test_config(server.addr); + config.dscp = 46; + let mut client = Client::connect(config).unwrap(); + client + .test_hooks + .recv_buffer_len_override + .set(Some(usize::MAX)); + client.test_hooks.fail_open_dscp.set(true); + + assert!(matches!( + client.open(), + Err(ClientError::AllocationFailed { + operation: "negotiated receive buffer", + .. + }) + )); + assert!(client.test_hooks.fail_open_dscp.get()); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); + server.join(); +} + +#[test] +fn dscp_application_failure_leaves_machine_connected() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None), + peer, + ) + .unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + client.test_hooks.fail_open_dscp.set(true); + + assert!(matches!( + client.open(), + Err(ClientError::SocketOption { + operation: "set negotiated DSCP", + .. + }) + )); + assert!(client.test_hooks.prepared_active_session_before_dscp.get()); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); + server.join(); +} + +#[test] +fn read_timeout_restoration_failure_leaves_machine_connected() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None), + peer, + ) + .unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + client.test_hooks.fail_open_timeout_restore.set(true); + + assert!(matches!( + client.open(), + Err(ClientError::ReadTimeoutRestore { .. }) + )); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); + server.join(); +} + +#[test] +fn dscp_rollback_failure_keeps_timeout_restoration_primary() { + let params = default_params(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None), + peer, + ) + .unwrap(); + let _ = recv_request(&socket, &tx); + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + client.test_hooks.fail_open_timeout_restore.set(true); + client.test_hooks.fail_dscp_restore.set(true); + + let error = client.open().unwrap_err(); + + assert!(matches!(error, ClientError::ReadTimeoutRestore { .. })); + assert!(client.runtime.prepare_open_request().is_ok()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); server.join(); } diff --git a/crates/irtt-client/src/error.rs b/crates/irtt-client/src/error.rs index 325946c..ee03b2a 100644 --- a/crates/irtt-client/src/error.rs +++ b/crates/irtt-client/src/error.rs @@ -38,8 +38,8 @@ pub enum ClientError { }, /// Restoring the configured read timeout after open negotiation failed. /// - /// The session may have opened successfully, but the socket could not be - /// returned to the caller's requested receive-timeout behavior. + /// The prepared session was not committed, and the client remains + /// connected rather than partially open. #[error("failed to restore configured socket read timeout after open negotiation: {source}")] ReadTimeoutRestore { source: io::Error }, /// A packet failed IRTT protocol parsing or authentication. diff --git a/crates/irtt-client/tests/hmac.rs b/crates/irtt-client/tests/hmac.rs index fbd099e..1ccd0a9 100644 --- a/crates/irtt-client/tests/hmac.rs +++ b/crates/irtt-client/tests/hmac.rs @@ -5,7 +5,7 @@ use std::time::Duration; use irtt_client::{ Client, ClientConfig, ClientError, ClientEvent, OpenOutcome, SocketConfig, WarningKind, }; -use irtt_proto::{ProtoError, TimestampFields}; +use irtt_proto::TimestampFields; use support::{ config_for_params, default_params, params_for_modes, run_one_probe_with_config, @@ -192,7 +192,7 @@ fn bad_hmac_echo_reply_is_rejected_without_echo_reply_event() { } #[test] -fn hmac_open_reply_with_bad_hmac_fails_with_protocol_error() { +fn hmac_open_reply_with_bad_hmac_is_ignored_until_timeout() { let key = b"compat-secret".to_vec(); let wrong_key = b"wrong-secret".to_vec(); let params = default_params(); @@ -201,10 +201,7 @@ fn hmac_open_reply_with_bad_hmac_fails_with_protocol_error() { config.hmac_key = Some(key); let mut client = Client::connect(config).unwrap(); - assert!(matches!( - client.open(), - Err(ClientError::Protocol(ProtoError::BadHmac)) - )); + assert!(matches!(client.open(), Err(ClientError::OpenTimeout))); server.join(); } From e270183f5d5d1a91540ade4862580e37a0227a0b Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 01:35:00 +0700 Subject: [PATCH 3/5] refactor(managed): transact group session opening --- crates/irtt-client/src/client/tests/open.rs | 5 +- crates/irtt-client/src/managed/group.rs | 674 ++++++++++++++++++-- crates/irtt-client/src/session/machine.rs | 35 +- 3 files changed, 637 insertions(+), 77 deletions(-) diff --git a/crates/irtt-client/src/client/tests/open.rs b/crates/irtt-client/src/client/tests/open.rs index 12b8859..2b75380 100644 --- a/crates/irtt-client/src/client/tests/open.rs +++ b/crates/irtt-client/src/client/tests/open.rs @@ -225,7 +225,10 @@ fn trusted_zero_token_normal_reply_is_terminal() { }); let mut client = Client::connect(default_test_config(server.addr)).unwrap(); - assert!(matches!(client.open(), Err(ClientError::ZeroToken))); + assert!(matches!( + client.open(), + Err(ClientError::Protocol(irtt_proto::ProtoError::ZeroToken)) + )); assert_eq!(server.rx.iter().take(1).count(), 1); assert!(client.runtime.prepare_open_request().is_ok()); assert!(client.schedule.is_none()); diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index 012a249..06d10ad 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -4,7 +4,7 @@ use std::{ hash::{Hash, Hasher}, net::{SocketAddr, UdpSocket}, sync::{ - atomic::{AtomicU64, Ordering}, + atomic::{AtomicBool, AtomicU64, Ordering}, mpsc, Arc, Mutex, }, thread::{self, JoinHandle}, @@ -22,7 +22,10 @@ use crate::{ event::{ClientEvent, OpenOutcome}, metadata::ReceiveMeta, receive::recv_datagram_from, - session::machine::{params_from_config, SessionMachine}, + session::machine::{ + params_from_config, OpenDatagramDisposition, PreparedOpenAcceptance, PreparedOpenRequest, + SessionMachine, + }, socket::{bind_unconnected_udp_socket, validate_open_timeouts}, socket_options::apply_dscp_to_socket, timing::ClientTimestamp, @@ -190,6 +193,12 @@ impl ManagedGroupEvent { pub type TargetEventSubscription = EventSubscription; /// Entry point for running a shared-socket multi-target managed client group. +/// +/// Each opening attempt sends one reusable request and may inspect several +/// datagrams until its absolute deadline. Malformed, unrelated, and +/// unauthenticated traffic is ignored; authenticated incompatibility is +/// terminal. A trusted token followed by group-policy or schedule rejection +/// triggers a best-effort cleanup close without replacing the primary failure. #[derive(Debug)] pub struct ManagedClientGroup; @@ -473,6 +482,8 @@ impl ManagedClientGroup { cancellation: cancellation.clone(), control_tx: control_tx.clone(), peer_closed_target_count: peer_closed_target_count.clone(), + receiver_processed_through: Mutex::new(now), + receiver_progress_waited_on: AtomicBool::new(false), requested_interval_ns: requested.interval_ns, requested_dscp: requested.dscp, family_remote, @@ -604,11 +615,63 @@ struct GroupShared { cancellation: CancellationToken, control_tx: mpsc::Sender, peer_closed_target_count: Arc, + receiver_processed_through: Mutex, + receiver_progress_waited_on: AtomicBool, requested_interval_ns: i64, requested_dscp: i64, family_remote: SocketAddr, } +struct ReceiverProgressGuard<'a> { + shared: &'a GroupShared, + processed_through: Instant, +} + +impl<'a> ReceiverProgressGuard<'a> { + fn new(shared: &'a GroupShared, processed_through: Instant) -> Self { + Self { + shared, + processed_through, + } + } +} + +impl Drop for ReceiverProgressGuard<'_> { + fn drop(&mut self) { + record_receiver_progress(self.shared, self.processed_through); + } +} + +fn record_receiver_progress(shared: &GroupShared, processed_through: Instant) { + let advanced = { + let mut current = shared + .receiver_processed_through + .lock() + .expect("receiver progress mutex poisoned"); + if processed_through > *current { + *current = processed_through; + true + } else { + false + } + }; + if advanced + && shared + .receiver_progress_waited_on + .swap(false, Ordering::AcqRel) + { + let _ = shared.control_tx.send(ControlMessage::Wake); + } +} + +fn receiver_processed_deadline(shared: &GroupShared, deadline: Instant) -> bool { + *shared + .receiver_processed_through + .lock() + .expect("receiver progress mutex poisoned") + >= deadline +} + struct GroupSchedulerCleanup { shared: Arc, } @@ -636,7 +699,7 @@ struct TargetState { runtime: SessionMachine, schedule: Option, status: TargetStatus, - open_packet: Vec, + open_request: PreparedOpenRequest, counters: TargetCounters, order: u64, final_reason: Option, @@ -655,6 +718,8 @@ enum TargetStatus { Opening { attempt: usize, next_send_at: Instant, + active_window: Option, + awaiting_receiver: bool, }, Active, Draining { @@ -663,6 +728,30 @@ enum TargetStatus { Finished, } +#[derive(Debug, Clone, Copy)] +struct OpenAttemptWindow { + sent_at: Instant, + deadline: Instant, +} + +impl OpenAttemptWindow { + fn contains(self, received_at: Instant) -> bool { + received_at >= self.sent_at && received_at <= self.deadline + } +} + +#[derive(Debug)] +struct PreparedGroupOpen { + machine: PreparedOpenAcceptance, + schedule: Option, +} + +#[derive(Debug)] +struct PreparedGroupOpenFailure { + primary: ClientError, + machine: PreparedOpenAcceptance, +} + #[derive(Debug, Default, Clone, PartialEq, Eq)] struct TargetCounters { replies_received: u64, @@ -725,7 +814,7 @@ impl TargetState { config.hmac_key = auth.hmac_key.clone(); } let runtime = SessionMachine::new(config, target.remote)?; - let open_packet = runtime.open_packet()?; + let open_request = runtime.prepare_open_request()?; Ok(Self { id: target.id, remote: target.remote, @@ -735,8 +824,10 @@ impl TargetState { status: TargetStatus::Opening { attempt: 0, next_send_at: now, + active_window: None, + awaiting_receiver: false, }, - open_packet, + open_request, counters: TargetCounters::default(), order, final_reason: None, @@ -867,6 +958,7 @@ fn run_group_receiver(socket: UdpSocket, shared: Arc) -> Result<(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut ) => { + record_receiver_progress(&shared, Instant::now()); continue; } Err(err) => { @@ -881,6 +973,7 @@ fn run_group_receiver(socket: UdpSocket, shared: Arc) -> Result<(), return Err(ClientError::Socket(err)); } }; + let _progress = ReceiverProgressGuard::new(&shared, datagram.received_at.mono); let target = { let registry = shared @@ -901,29 +994,56 @@ fn run_group_receiver(socket: UdpSocket, shared: Arc) -> Result<(), { let mut target = target.lock().expect("target mutex poisoned"); match target.status { - TargetStatus::Opening { .. } => { + TargetStatus::Opening { active_window, .. } => { + if !active_window + .is_some_and(|window| window.contains(datagram.received_at.mono)) + { + continue; + } let opened_at = datagram.received_at; - let mut schedule = None; - let result = target.runtime.decode_open_reply(packet).and_then(|reply| { - target - .runtime - .accept_open_reply(reply, opened_at, |negotiated| { - validate_group_negotiation(&shared, negotiated)?; - schedule = Some(ProbeSchedule::new(opened_at.mono, negotiated)?); - Ok(()) - }) - }); - - match result { - Ok(outcome) => { - target.schedule = schedule; - publish_open_outcome(&shared.hub, &mut target, outcome); - wake_scheduler = true; - } + let reply = match target.runtime.inspect_open_datagram(packet) { + Ok(OpenDatagramDisposition::Ignore) => continue, + Ok(OpenDatagramDisposition::Trusted(reply)) => reply, Err(err) => { target.mark_finished(ManagedTargetEndReason::OpenFailed( ManagedTargetFailure::opening(&err), )); + let _ = shared.control_tx.send(ControlMessage::Wake); + continue; + } + }; + let machine = match target.runtime.prepare_open_acceptance(reply, opened_at) { + Ok(machine) => machine, + Err(failure) => { + send_group_cleanup_close_best_effort( + &socket, + target.remote, + failure.cleanup_close.as_deref(), + ); + target.mark_finished(ManagedTargetEndReason::OpenFailed( + ManagedTargetFailure::opening(&failure.primary), + )); + let _ = shared.control_tx.send(ControlMessage::Wake); + continue; + } + }; + + match prepare_group_open(&shared, machine, opened_at) { + Ok(prepared) => { + let outcome = target.runtime.commit_open(prepared.machine); + target.schedule = prepared.schedule; + publish_open_outcome(&shared.hub, &mut target, outcome); + wake_scheduler = true; + } + Err(failure) => { + send_group_cleanup_close_best_effort( + &socket, + target.remote, + failure.machine.cleanup_close_packet(), + ); + target.mark_finished(ManagedTargetEndReason::OpenFailed( + ManagedTargetFailure::opening(&failure.primary), + )); wake_scheduler = true; } } @@ -967,6 +1087,35 @@ fn validate_group_negotiation( Ok(()) } +fn prepare_group_open( + shared: &GroupShared, + machine: PreparedOpenAcceptance, + opened_at: ClientTimestamp, +) -> Result> { + if let Err(primary) = validate_group_negotiation(shared, machine.negotiated()) { + return Err(Box::new(PreparedGroupOpenFailure { primary, machine })); + } + let schedule = if let Some(negotiated) = machine.normal_negotiated() { + match ProbeSchedule::new(opened_at.mono, negotiated) { + Ok(schedule) => Some(schedule), + Err(primary) => return Err(Box::new(PreparedGroupOpenFailure { primary, machine })), + } + } else { + None + }; + Ok(PreparedGroupOpen { machine, schedule }) +} + +fn send_group_cleanup_close_best_effort( + socket: &UdpSocket, + remote: SocketAddr, + packet: Option<&[u8]>, +) { + if let Some(packet) = packet { + let _ = socket.send_to(packet, remote); + } +} + fn publish_open_outcome( hub: &EventHub, target: &mut TargetState, @@ -1229,6 +1378,8 @@ fn drive_open_attempts( let TargetStatus::Opening { attempt, next_send_at, + active_window, + .. } = target.status else { continue; @@ -1238,6 +1389,21 @@ fn drive_open_attempts( continue; } + if attempt > 0 && !receiver_processed_deadline(shared, next_send_at) { + shared + .receiver_progress_waited_on + .store(true, Ordering::Release); + if !receiver_processed_deadline(shared, next_send_at) { + target.status = TargetStatus::Opening { + attempt, + next_send_at, + active_window, + awaiting_receiver: true, + }; + continue; + } + } + if attempt >= config.open_timeouts.len() { target.mark_finished(ManagedTargetEndReason::OpenFailed( ManagedTargetFailure::opening(&ClientError::OpenTimeout), @@ -1245,11 +1411,24 @@ fn drive_open_attempts( continue; } - match socket.send_to(&target.open_packet, target.remote) { + let sent_at = Instant::now(); + let deadline = match sent_at.checked_add(config.open_timeouts[attempt]) { + Some(deadline) => deadline, + None => { + target.mark_finished(ManagedTargetEndReason::OpenFailed( + ManagedTargetFailure::opening(&ClientError::DurationOverflow), + )); + continue; + } + }; + + match socket.send_to(&target.open_request.bytes, target.remote) { Ok(_) => { target.status = TargetStatus::Opening { attempt: attempt + 1, - next_send_at: now + config.open_timeouts[attempt], + next_send_at: deadline, + active_window: Some(OpenAttemptWindow { sent_at, deadline }), + awaiting_receiver: false, }; } Err(err) => { @@ -1811,7 +1990,11 @@ fn next_open_deadline(shared: &GroupShared) -> Option { .filter_map(|target| { let target = target.lock().expect("target mutex poisoned"); match target.status { - TargetStatus::Opening { next_send_at, .. } => Some(next_send_at), + TargetStatus::Opening { + next_send_at, + awaiting_receiver: false, + .. + } => Some(next_send_at), _ => None, } }) @@ -1846,10 +2029,10 @@ mod tests { config::NegotiationPolicy, managed::SubscriberOverflow, EventSubscriptionError, WarningKind, }; use irtt_proto::{ - echo_packet_len, + compute_hmac_in_place, echo_packet_len, flags::{self, FLAG_OPEN, FLAG_REPLY}, layout::PacketLayout, - Clock, Params, ReceivedStats, StampAt, TimestampFields, MAGIC, PROTOCOL_VERSION, + Clock, Params, ReceivedStats, StampAt, TimestampFields, HMAC_SIZE, MAGIC, PROTOCOL_VERSION, }; use std::sync::mpsc; @@ -1951,16 +2134,19 @@ mod tests { TargetState::new(&client_config, target("peer", remote), 0, Instant::now()).unwrap(); let open_packet = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms); - let open_reply = target.runtime.decode_open_reply(&open_packet).unwrap(); + let open_reply = match target.runtime.inspect_open_datagram(&open_packet).unwrap() { + OpenDatagramDisposition::Trusted(reply) => reply, + OpenDatagramDisposition::Ignore => panic!("fixture open reply must be trusted"), + }; let opened_at = ClientTimestamp::now(); - let mut opened_schedule = None; - let open_outcome = target + let prepared = target .runtime - .accept_open_reply(open_reply, opened_at, |negotiated| { - opened_schedule = Some(ProbeSchedule::new(opened_at.mono, negotiated)?); - Ok(()) - }) + .prepare_open_acceptance(open_reply, opened_at) .unwrap(); + let opened_schedule = Some( + ProbeSchedule::new(opened_at.mono, prepared.normal_negotiated().unwrap()).unwrap(), + ); + let open_outcome = target.runtime.commit_open(prepared); target.schedule = opened_schedule; publish_open_outcome(&hub, &mut target, open_outcome); @@ -2033,6 +2219,8 @@ mod tests { cancellation: CancellationToken::new(), control_tx, peer_closed_target_count: Arc::new(AtomicU64::new(0)), + receiver_processed_through: Mutex::new(Instant::now()), + receiver_progress_waited_on: AtomicBool::new(false), requested_interval_ns: requested.interval_ns, requested_dscp: requested.dscp, family_remote: remote, @@ -2066,16 +2254,18 @@ mod tests { let mut target = TargetState::new(config, target("peer", remote), 0, opened_at.mono).unwrap(); let packet = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, params); - let reply = target.runtime.decode_open_reply(&packet).unwrap(); - let mut schedule = None; - target + let reply = match target.runtime.inspect_open_datagram(&packet).unwrap() { + OpenDatagramDisposition::Trusted(reply) => reply, + OpenDatagramDisposition::Ignore => panic!("test open reply must be trusted"), + }; + let prepared = target .runtime - .accept_open_reply(reply, opened_at, |negotiated| { - schedule = Some(ProbeSchedule::new(opened_at.mono, negotiated)?); - Ok(()) - }) + .prepare_open_acceptance(reply, opened_at) .unwrap(); - target.schedule = schedule; + target.schedule = Some( + ProbeSchedule::new(opened_at.mono, prepared.normal_negotiated().unwrap()).unwrap(), + ); + target.runtime.commit_open(prepared); target.status = TargetStatus::Active; target } @@ -2348,6 +2538,17 @@ mod tests { packet } + fn hmac_open_reply(flags: u8, token: u64, params: &Params, key: &[u8]) -> Vec { + let mut packet = Vec::new(); + packet.extend_from_slice(&MAGIC); + packet.push(flags | flags::FLAG_HMAC); + packet.extend_from_slice(&[0_u8; HMAC_SIZE]); + packet.extend_from_slice(&token.to_le_bytes()); + packet.extend_from_slice(¶ms.encode()); + compute_hmac_in_place(key, &mut packet, 4).unwrap(); + packet + } + fn echo_reply_packet( token: u64, seq: u32, @@ -3029,6 +3230,393 @@ mod tests { .all(|target| target.end_reason.failure().is_some())); } + #[test] + fn bad_open_packet_followed_by_valid_reply_succeeds_without_retry() { + let interval = Duration::from_millis(10); + let params = test_params(None, interval); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + let addr = socket.local_addr().unwrap(); + let (tx, rx) = mpsc::channel(); + let done = thread::spawn(move || { + let (open, peer) = recv_request_timeout(&socket).expect("missing open request"); + assert_ne!(open[3] & FLAG_OPEN, 0); + tx.send(ServerObservation::Open { at: Instant::now() }) + .unwrap(); + socket.send_to(&[0_u8], peer).unwrap(); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, ¶ms), + peer, + ) + .unwrap(); + }); + let server = FakeServer { + addr, + _observations: rx, + done, + }; + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered); + config.client.run_mode = crate::RunMode::NoTest; + + let outcome = ManagedClientGroup::start(config, vec![target("peer", server.addr)]) + .unwrap() + .join() + .unwrap(); + + assert_eq!(outcome.successful_target_outcomes, 1); + assert_eq!(server._observations.try_iter().collect::>().len(), 1); + server.join(); + } + + #[test] + fn bad_hmac_open_packet_followed_by_valid_reply_succeeds_without_retry() { + let interval = Duration::from_millis(10); + let params = test_params(None, interval); + let key = b"group-secret".to_vec(); + let wrong_key = b"wrong".to_vec(); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + let addr = socket.local_addr().unwrap(); + let (tx, rx) = mpsc::channel(); + let server_key = key.clone(); + let done = thread::spawn(move || { + let (open, peer) = recv_request_timeout(&socket).expect("missing open request"); + assert_ne!(open[3] & FLAG_OPEN, 0); + tx.send(ServerObservation::Open { at: Instant::now() }) + .unwrap(); + socket + .send_to( + &hmac_open_reply( + FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, + 0, + ¶ms, + &wrong_key, + ), + peer, + ) + .unwrap(); + socket + .send_to( + &hmac_open_reply( + FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, + 0, + ¶ms, + &server_key, + ), + peer, + ) + .unwrap(); + }); + let server = FakeServer { + addr, + _observations: rx, + done, + }; + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered); + config.client.run_mode = crate::RunMode::NoTest; + config.client.hmac_key = Some(key); + + let outcome = ManagedClientGroup::start(config, vec![target("peer", server.addr)]) + .unwrap() + .join() + .unwrap(); + + assert_eq!(outcome.successful_target_outcomes, 1); + assert_eq!(server._observations.try_iter().collect::>().len(), 1); + server.join(); + } + + #[test] + fn trusted_group_incompatibility_sends_cleanup_and_fails_without_retry() { + let interval = Duration::from_millis(10); + let mut returned = test_params(None, interval); + returned.interval_ns = duration_ns_i64(Duration::from_millis(20)); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let addr = socket.local_addr().unwrap(); + let (tx, rx) = mpsc::channel(); + let done = thread::spawn(move || { + let (open, peer) = recv_request_timeout(&socket).expect("missing open request"); + assert_ne!(open[3] & FLAG_OPEN, 0); + tx.send(ServerObservation::Open { at: Instant::now() }) + .unwrap(); + socket + .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned), peer) + .unwrap(); + let (cleanup, _) = + recv_request_timeout(&socket).expect("missing post-token cleanup close"); + assert_eq!(cleanup[3], flags::FLAG_CLOSE); + assert_eq!( + u64::from_le_bytes(cleanup[4..12].try_into().unwrap()), + TOKEN + ); + tx.send(ServerObservation::Close { at: Instant::now() }) + .unwrap(); + }); + let server = FakeServer { + addr, + _observations: rx, + done, + }; + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered); + config.client.negotiation_policy = NegotiationPolicy::Loose; + + let outcome = ManagedClientGroup::start(config, vec![target("peer", server.addr)]) + .unwrap() + .join() + .unwrap(); + + assert_eq!(outcome.failed_target_outcomes, 1); + assert!(matches!( + outcome.targets[0].end_reason, + ManagedTargetEndReason::OpenFailed(ManagedTargetFailure { + kind: ManagedTargetFailureKind::OpeningProtocol, + .. + }) + )); + assert_eq!(server._observations.try_iter().collect::>().len(), 2); + server.join(); + } + + #[test] + fn group_open_deadline_overflow_fails_before_send() { + let silent_peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = silent_peer.local_addr().unwrap(); + let mut config = group_config( + None, + Duration::from_millis(10), + ManagedGroupPacing::Staggered, + ); + config.client.open_timeouts = vec![Duration::MAX]; + + let outcome = ManagedClientGroup::start(config, vec![target("peer", remote)]) + .unwrap() + .join() + .unwrap(); + + assert_eq!(outcome.failed_target_outcomes, 1); + let failure = outcome.targets[0].end_reason.failure().unwrap(); + assert!(failure.message.contains("duration")); + silent_peer.set_nonblocking(true).unwrap(); + let mut packet = [0_u8; 512]; + assert!(matches!( + silent_peer.recv_from(&mut packet), + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock + )); + } + + #[test] + fn receiver_progress_is_monotonic_and_wakes_only_waiters() { + let interval = Duration::from_millis(10); + let config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let now = Instant::now(); + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, now).unwrap(), + )); + let shared = shared_with_target(&config, EventHub::new(), target); + let first = now + Duration::from_millis(10); + let later = now + Duration::from_millis(20); + + record_receiver_progress(&shared, later); + record_receiver_progress(&shared, first); + + assert!(receiver_processed_deadline(&shared, later)); + assert!(!shared.receiver_progress_waited_on.load(Ordering::Acquire)); + } + + #[test] + fn expired_attempt_waits_for_receiver_before_retrying() { + let interval = Duration::from_millis(10); + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)]; + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + peer.set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + let remote = peer.local_addr().unwrap(); + let now = Instant::now(); + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, now).unwrap(), + )); + let shared = shared_with_target(&config, EventHub::new(), target.clone()); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + drive_open_attempts(&config, &socket, &shared, now); + let deadline = { + let target = target.lock().expect("target mutex poisoned"); + match target.status { + TargetStatus::Opening { + attempt: 1, + next_send_at, + .. + } => next_send_at, + ref status => panic!("unexpected target status after open send: {status:?}"), + } + }; + assert!(recv_request_timeout(&peer).is_some()); + + drive_open_attempts(&config, &socket, &shared, deadline); + assert!(matches!( + target.lock().expect("target mutex poisoned").status, + TargetStatus::Opening { + attempt: 1, + awaiting_receiver: true, + .. + } + )); + assert!(recv_request_timeout(&peer).is_none()); + + record_receiver_progress(&shared, deadline); + drive_open_attempts(&config, &socket, &shared, deadline); + assert!(matches!( + target.lock().expect("target mutex poisoned").status, + TargetStatus::Opening { + attempt: 2, + awaiting_receiver: false, + .. + } + )); + assert!(recv_request_timeout(&peer).is_some()); + } + + #[test] + fn in_window_reply_remains_eligible_after_scheduler_deadline() { + let interval = Duration::from_millis(10); + let config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + let params = test_params(None, interval); + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let now = Instant::now(); + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, now).unwrap(), + )); + let shared = shared_with_target(&config, EventHub::new(), target.clone()); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + drive_open_attempts(&config, &socket, &shared, now); + let (window, deadline) = { + let target = target.lock().expect("target mutex poisoned"); + match target.status { + TargetStatus::Opening { + active_window: Some(window), + next_send_at, + .. + } => (window, next_send_at), + ref status => panic!("unexpected opening status: {status:?}"), + } + }; + let received_at = ClientTimestamp { + wall: std::time::SystemTime::now(), + mono: window.deadline, + }; + let progress = ReceiverProgressGuard::new(&shared, received_at.mono); + + drive_open_attempts(&config, &socket, &shared, deadline); + assert!(matches!( + target.lock().expect("target mutex poisoned").status, + TargetStatus::Opening { + awaiting_receiver: true, + .. + } + )); + + { + let mut target = target.lock().expect("target mutex poisoned"); + assert!(window.contains(received_at.mono)); + let reply = match target + .runtime + .inspect_open_datagram(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms)) + .unwrap() + { + OpenDatagramDisposition::Trusted(reply) => reply, + OpenDatagramDisposition::Ignore => panic!("in-window reply must be trusted"), + }; + let machine = target + .runtime + .prepare_open_acceptance(reply, received_at) + .unwrap(); + let prepared = prepare_group_open(&shared, machine, received_at).unwrap(); + let outcome = target.runtime.commit_open(prepared.machine); + target.schedule = prepared.schedule; + publish_open_outcome(&shared.hub, &mut target, outcome); + } + drop(progress); + + assert!(matches!( + target.lock().expect("target mutex poisoned").status, + TargetStatus::Active + )); + assert!(receiver_processed_deadline(&shared, deadline)); + } + + #[test] + fn post_deadline_reply_is_not_eligible_for_expired_attempt() { + let now = Instant::now(); + let window = OpenAttemptWindow { + sent_at: now, + deadline: now + Duration::from_millis(200), + }; + + assert!(window.contains(window.sent_at)); + assert!(window.contains(window.deadline)); + assert!(!window.contains(window.deadline + Duration::from_nanos(1))); + } + + #[test] + fn final_open_timeout_waits_until_receiver_processes_through_deadline() { + let interval = Duration::from_millis(10); + let config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let now = Instant::now(); + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, now).unwrap(), + )); + let shared = shared_with_target(&config, EventHub::new(), target.clone()); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + drive_open_attempts(&config, &socket, &shared, now); + let deadline = { + let target = target.lock().expect("target mutex poisoned"); + match target.status { + TargetStatus::Opening { + attempt: 1, + next_send_at, + .. + } => next_send_at, + ref status => panic!("unexpected target status after open send: {status:?}"), + } + }; + + drive_open_attempts(&config, &socket, &shared, deadline); + { + let target = target.lock().expect("target mutex poisoned"); + assert!(matches!( + target.status, + TargetStatus::Opening { + attempt: 1, + awaiting_receiver: true, + .. + } + )); + assert!(target.final_reason.is_none()); + } + + record_receiver_progress(&shared, deadline); + drive_open_attempts(&config, &socket, &shared, deadline); + let target = target.lock().expect("target mutex poisoned"); + assert!(matches!(target.status, TargetStatus::Finished)); + assert!(matches!( + target.final_reason, + Some(ManagedTargetEndReason::OpenFailed(ManagedTargetFailure { + kind: ManagedTargetFailureKind::OpeningTimeout, + .. + })) + )); + } + #[test] fn group_probe_separates_permission_committed_send_and_send_call_timing() { let duration = Duration::from_secs(1); diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index d14755c..e8c1c96 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -205,7 +205,7 @@ impl SessionMachine { self.ensure_connected()?; match irtt_proto::decode_open_reply(packet, self.config.hmac_key.as_deref()) { Ok(reply) => Ok(OpenDatagramDisposition::Trusted(reply)), - Err(irtt_proto::ProtoError::ZeroToken) => Err(ClientError::ZeroToken), + Err(error @ irtt_proto::ProtoError::ZeroToken) => Err(ClientError::Protocol(error)), Err( error @ (irtt_proto::ProtoError::TruncatedVarint | irtt_proto::ProtoError::VarintOverflow @@ -282,37 +282,6 @@ impl SessionMachine { prepared.outcome } - // Temporary private bridges keep this first commit independently compiling. - // The blocking and managed migrations remove them before the PR is complete. - pub(crate) fn open_packet(&self) -> Result, ClientError> { - Ok(self.prepare_open_request()?.bytes.into_vec()) - } - - pub(crate) fn decode_open_reply(&self, packet: &[u8]) -> Result { - Ok(irtt_proto::decode_open_reply( - packet, - self.config.hmac_key.as_deref(), - )?) - } - - pub(crate) fn accept_open_reply( - &mut self, - reply: OpenReply, - now: ClientTimestamp, - before_normal_open: F, - ) -> Result - where - F: FnOnce(&NegotiatedParams) -> Result<(), ClientError>, - { - let prepared = self - .prepare_open_acceptance(reply, now) - .map_err(|failure| failure.primary)?; - if let Some(negotiated) = prepared.normal_negotiated() { - before_normal_open(negotiated)?; - } - Ok(self.commit_open(prepared)) - } - pub(crate) fn probe_timeout(&self) -> Duration { self.config.probe_timeout } @@ -1151,7 +1120,7 @@ mod tests { assert!(matches!( machine.inspect_open_datagram(&packet), - Err(ClientError::ZeroToken) + Err(ClientError::Protocol(irtt_proto::ProtoError::ZeroToken)) )); assert!(matches!(machine.state, MachineState::Connected)); } From dd74f3800e64007413cb7794d0e02d0d6bca8b47 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 12:18:31 +0700 Subject: [PATCH 4/5] fix(client): preserve no-test opening semantics --- .../irtt-client/src/client/tests/no_test.rs | 44 ++++- crates/irtt-client/src/client/tests/open.rs | 4 - crates/irtt-client/src/managed/group.rs | 162 +++++++++++++++++- crates/irtt-client/src/session/machine.rs | 93 ++++++++-- 4 files changed, 276 insertions(+), 27 deletions(-) diff --git a/crates/irtt-client/src/client/tests/no_test.rs b/crates/irtt-client/src/client/tests/no_test.rs index 2af264f..6b64075 100644 --- a/crates/irtt-client/src/client/tests/no_test.rs +++ b/crates/irtt-client/src/client/tests/no_test.rs @@ -22,13 +22,55 @@ fn no_test_rejects_non_close_open_reply() { let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); config.run_mode = RunMode::NoTest; let params = params_from_config(&config).unwrap(); - let server = open_success_server(params); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None), + peer, + ) + .unwrap(); + let _ = recv_request(&socket, &tx); + }); config.server_addr = server.addr.to_string(); let mut client = Client::connect(config).unwrap(); assert!(matches!( client.open(), Err(ClientError::UnexpectedNoTestReply) )); + let packets: Vec<_> = server.rx.iter().take(2).collect(); + assert_eq!(packets[1][3], flags::FLAG_CLOSE); + assert_eq!( + u64::from_le_bytes(packets[1][4..12].try_into().unwrap()), + TOKEN + ); + assert!(client.runtime.prepare_open_request().is_ok()); + server.join(); +} + +#[test] +fn no_test_cleanup_send_failure_preserves_unexpected_reply() { + let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1))); + config.run_mode = RunMode::NoTest; + let params = params_from_config(&config).unwrap(); + let server = start_fake_server(move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None), + peer, + ) + .unwrap(); + }); + config.server_addr = server.addr.to_string(); + let mut client = Client::connect(config).unwrap(); + client.test_hooks.fail_cleanup_send.set(true); + + assert!(matches!( + client.open(), + Err(ClientError::UnexpectedNoTestReply) + )); + assert!(client.runtime.prepare_open_request().is_ok()); server.join(); } diff --git a/crates/irtt-client/src/client/tests/open.rs b/crates/irtt-client/src/client/tests/open.rs index 2b75380..f7f4c21 100644 --- a/crates/irtt-client/src/client/tests/open.rs +++ b/crates/irtt-client/src/client/tests/open.rs @@ -29,10 +29,6 @@ fn successful_open_restores_configured_receive_timeout() { client.test_hooks.last_restored_read_timeout.get(), Some(Duration::from_millis(75)) ); - assert_eq!( - client.socket.read_timeout().unwrap(), - Some(Duration::from_millis(75)) - ); server.join(); } diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index 06d10ad..bb11dc3 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -1092,10 +1092,10 @@ fn prepare_group_open( machine: PreparedOpenAcceptance, opened_at: ClientTimestamp, ) -> Result> { - if let Err(primary) = validate_group_negotiation(shared, machine.negotiated()) { - return Err(Box::new(PreparedGroupOpenFailure { primary, machine })); - } let schedule = if let Some(negotiated) = machine.normal_negotiated() { + if let Err(primary) = validate_group_negotiation(shared, negotiated) { + return Err(Box::new(PreparedGroupOpenFailure { primary, machine })); + } match ProbeSchedule::new(opened_at.mono, negotiated) { Ok(schedule) => Some(schedule), Err(primary) => return Err(Box::new(PreparedGroupOpenFailure { primary, machine })), @@ -3379,6 +3379,162 @@ mod tests { server.join(); } + #[test] + fn no_test_non_close_reply_sends_cleanup_and_fails_without_retry() { + let interval = Duration::from_millis(10); + let params = test_params(None, interval); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let addr = socket.local_addr().unwrap(); + let (tx, rx) = mpsc::channel(); + let done = thread::spawn(move || { + let (open, peer) = recv_request_timeout(&socket).expect("missing open request"); + assert_ne!(open[3] & flags::FLAG_CLOSE, 0); + tx.send(ServerObservation::Open { at: Instant::now() }) + .unwrap(); + socket + .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), peer) + .unwrap(); + let (cleanup, _) = + recv_request_timeout(&socket).expect("missing no-test cleanup close"); + assert_eq!(cleanup[3], flags::FLAG_CLOSE); + assert_eq!( + u64::from_le_bytes(cleanup[4..12].try_into().unwrap()), + TOKEN + ); + tx.send(ServerObservation::Close { at: Instant::now() }) + .unwrap(); + }); + let server = FakeServer { + addr, + _observations: rx, + done, + }; + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered); + config.client.run_mode = crate::RunMode::NoTest; + + let outcome = ManagedClientGroup::start(config, vec![target("peer", server.addr)]) + .unwrap() + .join() + .unwrap(); + + assert_eq!(outcome.failed_target_outcomes, 1); + assert!(matches!( + outcome.targets[0].end_reason, + ManagedTargetEndReason::OpenFailed(ManagedTargetFailure { + kind: ManagedTargetFailureKind::OpeningProtocol, + .. + }) + )); + assert_eq!(server._observations.try_iter().collect::>().len(), 2); + server.join(); + } + + #[test] + fn loose_no_test_group_skips_active_interval_and_dscp_policy() { + let interval = Duration::from_millis(10); + let mut changed_interval = test_params(None, interval); + changed_interval.interval_ns += 1; + let changed_dscp = test_params(None, interval); + + for (requested_dscp, params) in [(0, changed_interval), (1, changed_dscp)] { + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + config.run_mode = crate::RunMode::NoTest; + config.negotiation_policy = NegotiationPolicy::Loose; + config.dscp = requested_dscp; + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, Instant::now()).unwrap(), + )); + let shared = shared_with_target(&config, EventHub::new(), target.clone()); + let opened_at = ClientTimestamp::now(); + + let mut target = target.lock().expect("target mutex poisoned"); + let packet = open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, ¶ms); + let reply = match target.runtime.inspect_open_datagram(&packet).unwrap() { + OpenDatagramDisposition::Trusted(reply) => reply, + OpenDatagramDisposition::Ignore => panic!("no-test reply must be trusted"), + }; + let machine = target + .runtime + .prepare_open_acceptance(reply, opened_at) + .unwrap(); + let prepared = prepare_group_open(&shared, machine, opened_at).unwrap(); + assert!(prepared.schedule.is_none()); + let outcome = target.runtime.commit_open(prepared.machine); + target.schedule = prepared.schedule; + publish_open_outcome(&shared.hub, &mut target, outcome); + + assert!(matches!(target.status, TargetStatus::Finished)); + assert_eq!( + target.final_reason, + Some(ManagedTargetEndReason::NoTestComplete) + ); + } + } + + #[test] + fn normal_and_strict_sessions_preserve_group_negotiation_checks() { + let interval = Duration::from_millis(10); + let mut changed_interval = test_params(None, interval); + changed_interval.interval_ns += 1; + let changed_dscp = test_params(None, interval); + + for (requested_dscp, params) in [(0, changed_interval.clone()), (1, changed_dscp)] { + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + config.negotiation_policy = NegotiationPolicy::Loose; + config.dscp = requested_dscp; + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, Instant::now()).unwrap(), + )); + let shared = shared_with_target(&config, EventHub::new(), target.clone()); + let opened_at = ClientTimestamp::now(); + let target = target.lock().expect("target mutex poisoned"); + let packet = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms); + let reply = match target.runtime.inspect_open_datagram(&packet).unwrap() { + OpenDatagramDisposition::Trusted(reply) => reply, + OpenDatagramDisposition::Ignore => panic!("normal reply must be trusted"), + }; + let machine = target + .runtime + .prepare_open_acceptance(reply, opened_at) + .unwrap(); + let failure = prepare_group_open(&shared, machine, opened_at).unwrap_err(); + assert!(matches!( + failure.primary, + ClientError::NegotiationRejected { .. } + )); + } + + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + config.run_mode = crate::RunMode::NoTest; + let target = TargetState::new(&config, target("peer", remote), 0, Instant::now()).unwrap(); + let packet = open_reply( + FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, + 0, + &changed_interval, + ); + let reply = match target.runtime.inspect_open_datagram(&packet).unwrap() { + OpenDatagramDisposition::Trusted(reply) => reply, + OpenDatagramDisposition::Ignore => panic!("strict no-test reply must be trusted"), + }; + let failure = target + .runtime + .prepare_open_acceptance(reply, ClientTimestamp::now()) + .unwrap_err(); + assert!(matches!( + failure.primary, + ClientError::NegotiationRejected { .. } + )); + } + #[test] fn group_open_deadline_overflow_fails_before_send() { let silent_peer = UdpSocket::bind("127.0.0.1:0").unwrap(); diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index e8c1c96..74cced0 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -80,13 +80,6 @@ pub(crate) struct PreparedOpenAcceptance { } impl PreparedOpenAcceptance { - pub(crate) fn negotiated(&self) -> &NegotiatedParams { - match &self.outcome { - OpenOutcome::Started { negotiated, .. } - | OpenOutcome::NoTestCompleted { negotiated, .. } => negotiated, - } - } - pub(crate) fn normal_negotiated(&self) -> Option<&NegotiatedParams> { match &self.next_state { MachineState::Open(session) => Some(&session.negotiated), @@ -228,18 +221,17 @@ impl SessionMachine { .map_err(OpenAcceptanceFailure::without_cleanup)?; let reply_is_close = flags::has(reply.flags, flags::FLAG_CLOSE); - let cleanup_close = - if self.config.run_mode == RunMode::Normal && !reply_is_close && reply.token != 0 { - let bytes = encode_close_request( - &CloseRequest { token: reply.token }, - self.config.hmac_key.as_deref(), - ) - .map_err(ClientError::from) - .map_err(OpenAcceptanceFailure::without_cleanup)?; - Some(bytes.into_boxed_slice()) - } else { - None - }; + let cleanup_close = if !reply_is_close && reply.token != 0 { + let bytes = encode_close_request( + &CloseRequest { token: reply.token }, + self.config.hmac_key.as_deref(), + ) + .map_err(ClientError::from) + .map_err(OpenAcceptanceFailure::without_cleanup)?; + Some(bytes.into_boxed_slice()) + } else { + None + }; if reply.params.protocol_version != PROTOCOL_VERSION { return Err(OpenAcceptanceFailure::new( @@ -1186,6 +1178,69 @@ mod tests { assert!(matches!(no_test.state, MachineState::Connected)); } + #[test] + fn no_test_non_close_reply_prepares_authenticated_cleanup_without_state_change() { + let key = b"cleanup-key".to_vec(); + let machine = connected_machine(ClientConfig { + run_mode: RunMode::NoTest, + hmac_key: Some(key.clone()), + ..ClientConfig::default() + }); + let token = 0x1020_3040_5060_7080; + + let failure = machine + .prepare_open_acceptance( + normal_open_reply(&machine, token), + timestamp(Instant::now()), + ) + .unwrap_err(); + + assert!(matches!( + failure.primary, + ClientError::UnexpectedNoTestReply + )); + let cleanup = failure.cleanup_close.as_deref().unwrap(); + assert_eq!(cleanup[3], flags::FLAG_CLOSE | flags::FLAG_HMAC); + irtt_proto::verify_hmac(&key, cleanup, 4).unwrap(); + assert_eq!( + u64::from_le_bytes( + cleanup[4 + irtt_proto::HMAC_SIZE..12 + irtt_proto::HMAC_SIZE] + .try_into() + .unwrap() + ), + token + ); + drop(failure); + assert!(matches!(machine.state, MachineState::Connected)); + } + + #[test] + fn no_test_close_replies_do_not_prepare_cleanup() { + let machine = connected_machine(ClientConfig { + run_mode: RunMode::NoTest, + ..ClientConfig::default() + }); + let reply = |token| OpenReply { + flags: flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE, + token, + params: machine.requested.clone(), + }; + + let prepared = machine + .prepare_open_acceptance(reply(0), timestamp(Instant::now())) + .unwrap(); + assert!(prepared.cleanup_close_packet().is_none()); + + let failure = machine + .prepare_open_acceptance(reply(0x1020_3040_5060_7080), timestamp(Instant::now())) + .unwrap_err(); + assert!(matches!( + failure.primary, + ClientError::NonZeroNoTestToken { .. } + )); + assert!(failure.cleanup_close.is_none()); + } + #[test] fn dropping_prepared_acceptance_leaves_connected_state() { let machine = connected_machine(ClientConfig::default()); From 185e0bf3e446cb1a0f95d910ced021cd855dc830 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 12:34:13 +0700 Subject: [PATCH 5/5] fix(managed): drain queued opening replies before retry --- crates/irtt-client/src/managed/group.rs | 927 +++++++++++++++++------- 1 file changed, 659 insertions(+), 268 deletions(-) diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index bb11dc3..e2fda5e 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -4,7 +4,7 @@ use std::{ hash::{Hash, Hasher}, net::{SocketAddr, UdpSocket}, sync::{ - atomic::{AtomicBool, AtomicU64, Ordering}, + atomic::{AtomicU64, Ordering}, mpsc, Arc, Mutex, }, thread::{self, JoinHandle}, @@ -21,7 +21,7 @@ use crate::{ error::ClientError, event::{ClientEvent, OpenOutcome}, metadata::ReceiveMeta, - receive::recv_datagram_from, + receive::{recv_datagram_from, ReceivedDatagramFrom}, session::machine::{ params_from_config, OpenDatagramDisposition, PreparedOpenAcceptance, PreparedOpenRequest, SessionMachine, @@ -482,8 +482,7 @@ impl ManagedClientGroup { cancellation: cancellation.clone(), control_tx: control_tx.clone(), peer_closed_target_count: peer_closed_target_count.clone(), - receiver_processed_through: Mutex::new(now), - receiver_progress_waited_on: AtomicBool::new(false), + receiver_drain: Mutex::new(ReceiverDrainState::default()), requested_interval_ns: requested.interval_ns, requested_dscp: requested.dscp, family_remote, @@ -615,61 +614,78 @@ struct GroupShared { cancellation: CancellationToken, control_tx: mpsc::Sender, peer_closed_target_count: Arc, - receiver_processed_through: Mutex, - receiver_progress_waited_on: AtomicBool, + receiver_drain: Mutex, requested_interval_ns: i64, requested_dscp: i64, family_remote: SocketAddr, } -struct ReceiverProgressGuard<'a> { - shared: &'a GroupShared, - processed_through: Instant, +#[derive(Debug, Default)] +struct ReceiverDrainState { + requested: u64, + completed: u64, } -impl<'a> ReceiverProgressGuard<'a> { - fn new(shared: &'a GroupShared, processed_through: Instant) -> Self { - Self { - shared, - processed_through, +impl ReceiverDrainState { + fn request_or_join(&mut self) -> Result { + if self.requested == self.completed { + self.requested = self + .requested + .checked_add(1) + .ok_or(ClientError::CounterOverflow { + counter: "receiver_drain_generation", + })?; } + Ok(self.requested) } -} -impl Drop for ReceiverProgressGuard<'_> { - fn drop(&mut self) { - record_receiver_progress(self.shared, self.processed_through); - } -} - -fn record_receiver_progress(shared: &GroupShared, processed_through: Instant) { - let advanced = { - let mut current = shared - .receiver_processed_through - .lock() - .expect("receiver progress mutex poisoned"); - if processed_through > *current { - *current = processed_through; + fn complete_observed(&mut self, observed_generation: u64) -> bool { + let observed_generation = observed_generation.min(self.requested); + if observed_generation > self.completed { + self.completed = observed_generation; true } else { false } - }; - if advanced - && shared - .receiver_progress_waited_on - .swap(false, Ordering::AcqRel) - { - let _ = shared.control_tx.send(ControlMessage::Wake); } } -fn receiver_processed_deadline(shared: &GroupShared, deadline: Instant) -> bool { - *shared - .receiver_processed_through +#[cfg(test)] +fn request_receiver_drain(shared: &GroupShared) -> Result { + shared + .receiver_drain .lock() - .expect("receiver progress mutex poisoned") - >= deadline + .expect("receiver drain mutex poisoned") + .request_or_join() +} + +fn requested_receiver_drain(shared: &GroupShared) -> u64 { + shared + .receiver_drain + .lock() + .expect("receiver drain mutex poisoned") + .requested +} + +fn completed_receiver_drain(shared: &GroupShared) -> u64 { + shared + .receiver_drain + .lock() + .expect("receiver drain mutex poisoned") + .completed +} + +fn complete_receiver_drain(shared: &GroupShared, observed_generation: u64) { + let advanced = { + let mut state = shared + .receiver_drain + .lock() + .expect("receiver drain mutex poisoned"); + state.complete_observed(observed_generation) + }; + if advanced { + let _ = shared.control_tx.send(ControlMessage::Wake); + } } struct GroupSchedulerCleanup { @@ -718,8 +734,7 @@ enum TargetStatus { Opening { attempt: usize, next_send_at: Instant, - active_window: Option, - awaiting_receiver: bool, + awaiting_receiver_generation: Option, }, Active, Draining { @@ -728,18 +743,6 @@ enum TargetStatus { Finished, } -#[derive(Debug, Clone, Copy)] -struct OpenAttemptWindow { - sent_at: Instant, - deadline: Instant, -} - -impl OpenAttemptWindow { - fn contains(self, received_at: Instant) -> bool { - received_at >= self.sent_at && received_at <= self.deadline - } -} - #[derive(Debug)] struct PreparedGroupOpen { machine: PreparedOpenAcceptance, @@ -824,8 +827,7 @@ impl TargetState { status: TargetStatus::Opening { attempt: 0, next_send_at: now, - active_window: None, - awaiting_receiver: false, + awaiting_receiver_generation: None, }, open_request, counters: TargetCounters::default(), @@ -950,6 +952,8 @@ fn run_group_scheduler( fn run_group_receiver(socket: UdpSocket, shared: Arc) -> Result<(), ClientError> { let mut buf = vec![0_u8; RECV_BUFFER_SIZE]; while !shared.cancellation.is_cancelled() { + // A timeout only proves requests visible before this receive began. + let observed_drain_generation = requested_receiver_drain(&shared); let datagram = match recv_datagram_from(&socket, &mut buf) { Ok(datagram) => datagram, Err(err) @@ -958,7 +962,7 @@ fn run_group_receiver(socket: UdpSocket, shared: Arc) -> Result<(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut ) => { - record_receiver_progress(&shared, Instant::now()); + complete_receiver_drain(&shared, observed_drain_generation); continue; } Err(err) => { @@ -973,101 +977,116 @@ fn run_group_receiver(socket: UdpSocket, shared: Arc) -> Result<(), return Err(ClientError::Socket(err)); } }; - let _progress = ReceiverProgressGuard::new(&shared, datagram.received_at.mono); - let target = { - let registry = shared - .registry - .lock() - .expect("target registry mutex poisoned"); - let Some(id) = registry.remotes.get(&datagram.source) else { - continue; - }; - registry.targets.get(id).cloned() - }; - let Some(target) = target else { - continue; - }; - - let packet = &buf[..datagram.len]; - let mut wake_scheduler = false; - { - let mut target = target.lock().expect("target mutex poisoned"); - match target.status { - TargetStatus::Opening { active_window, .. } => { - if !active_window - .is_some_and(|window| window.contains(datagram.received_at.mono)) - { - continue; - } - let opened_at = datagram.received_at; - let reply = match target.runtime.inspect_open_datagram(packet) { - Ok(OpenDatagramDisposition::Ignore) => continue, - Ok(OpenDatagramDisposition::Trusted(reply)) => reply, - Err(err) => { - target.mark_finished(ManagedTargetEndReason::OpenFailed( - ManagedTargetFailure::opening(&err), - )); - let _ = shared.control_tx.send(ControlMessage::Wake); - continue; - } - }; - let machine = match target.runtime.prepare_open_acceptance(reply, opened_at) { - Ok(machine) => machine, - Err(failure) => { - send_group_cleanup_close_best_effort( - &socket, - target.remote, - failure.cleanup_close.as_deref(), - ); - target.mark_finished(ManagedTargetEndReason::OpenFailed( - ManagedTargetFailure::opening(&failure.primary), - )); - let _ = shared.control_tx.send(ControlMessage::Wake); - continue; - } - }; - - match prepare_group_open(&shared, machine, opened_at) { - Ok(prepared) => { - let outcome = target.runtime.commit_open(prepared.machine); - target.schedule = prepared.schedule; - publish_open_outcome(&shared.hub, &mut target, outcome); - wake_scheduler = true; - } - Err(failure) => { - send_group_cleanup_close_best_effort( - &socket, - target.remote, - failure.machine.cleanup_close_packet(), - ); - target.mark_finished(ManagedTargetEndReason::OpenFailed( - ManagedTargetFailure::opening(&failure.primary), - )); - wake_scheduler = true; - } - } - } - TargetStatus::Active | TargetStatus::Draining { .. } => { - wake_scheduler = process_active_target_packet( - &shared.hub, - &mut target, - packet, - datagram.received_at, - datagram.meta, - ); - } - TargetStatus::Finished => {} - } - } - - if wake_scheduler { - let _ = shared.control_tx.send(ControlMessage::Wake); - } + process_group_datagram(&socket, &shared, datagram, &buf[..datagram.len]); } Ok(()) } +fn process_group_datagram( + socket: &UdpSocket, + shared: &GroupShared, + datagram: ReceivedDatagramFrom, + packet: &[u8], +) { + // The single receiver thread cannot advance completion while processing + // this datagram, so one snapshot consistently classifies its target. + let completed_drain_generation = completed_receiver_drain(shared); + let target = { + let registry = shared + .registry + .lock() + .expect("target registry mutex poisoned"); + let Some(id) = registry.remotes.get(&datagram.source) else { + return; + }; + registry.targets.get(id).cloned() + }; + let Some(target) = target else { + return; + }; + + let mut wake_scheduler = false; + { + let mut target = target.lock().expect("target mutex poisoned"); + match target.status { + TargetStatus::Opening { + attempt, + awaiting_receiver_generation, + .. + } => { + if attempt == 0 + || awaiting_receiver_generation + .is_some_and(|generation| completed_drain_generation >= generation) + { + return; + } + let opened_at = datagram.received_at; + let reply = match target.runtime.inspect_open_datagram(packet) { + Ok(OpenDatagramDisposition::Ignore) => return, + Ok(OpenDatagramDisposition::Trusted(reply)) => reply, + Err(err) => { + target.mark_finished(ManagedTargetEndReason::OpenFailed( + ManagedTargetFailure::opening(&err), + )); + let _ = shared.control_tx.send(ControlMessage::Wake); + return; + } + }; + let machine = match target.runtime.prepare_open_acceptance(reply, opened_at) { + Ok(machine) => machine, + Err(failure) => { + send_group_cleanup_close_best_effort( + socket, + target.remote, + failure.cleanup_close.as_deref(), + ); + target.mark_finished(ManagedTargetEndReason::OpenFailed( + ManagedTargetFailure::opening(&failure.primary), + )); + let _ = shared.control_tx.send(ControlMessage::Wake); + return; + } + }; + + match prepare_group_open(shared, machine, opened_at) { + Ok(prepared) => { + let outcome = target.runtime.commit_open(prepared.machine); + target.schedule = prepared.schedule; + publish_open_outcome(&shared.hub, &mut target, outcome); + wake_scheduler = true; + } + Err(failure) => { + send_group_cleanup_close_best_effort( + socket, + target.remote, + failure.machine.cleanup_close_packet(), + ); + target.mark_finished(ManagedTargetEndReason::OpenFailed( + ManagedTargetFailure::opening(&failure.primary), + )); + wake_scheduler = true; + } + } + } + TargetStatus::Active | TargetStatus::Draining { .. } => { + wake_scheduler = process_active_target_packet( + &shared.hub, + &mut target, + packet, + datagram.received_at, + datagram.meta, + ); + } + TargetStatus::Finished => {} + } + } + + if wake_scheduler { + let _ = shared.control_tx.send(ControlMessage::Wake); + } +} + fn validate_group_negotiation( shared: &GroupShared, negotiated: &crate::NegotiatedParams, @@ -1373,13 +1392,16 @@ fn drive_open_attempts( shared: &GroupShared, now: Instant, ) { - for target in all_targets(shared) { + let targets = all_targets(shared); + register_due_open_drains(shared, &targets, now); + let completed_drain_generation = completed_receiver_drain(shared); + + for target in targets { let mut target = target.lock().expect("target mutex poisoned"); let TargetStatus::Opening { attempt, next_send_at, - active_window, - .. + awaiting_receiver_generation, } = target.status else { continue; @@ -1389,18 +1411,11 @@ fn drive_open_attempts( continue; } - if attempt > 0 && !receiver_processed_deadline(shared, next_send_at) { - shared - .receiver_progress_waited_on - .store(true, Ordering::Release); - if !receiver_processed_deadline(shared, next_send_at) { - target.status = TargetStatus::Opening { - attempt, - next_send_at, - active_window, - awaiting_receiver: true, - }; - continue; + if attempt > 0 { + match awaiting_receiver_generation { + None => continue, + Some(generation) if completed_drain_generation < generation => continue, + Some(_) => {} } } @@ -1427,8 +1442,7 @@ fn drive_open_attempts( target.status = TargetStatus::Opening { attempt: attempt + 1, next_send_at: deadline, - active_window: Some(OpenAttemptWindow { sent_at, deadline }), - awaiting_receiver: false, + awaiting_receiver_generation: None, }; } Err(err) => { @@ -1440,6 +1454,45 @@ fn drive_open_attempts( } } +fn register_due_open_drains( + shared: &GroupShared, + targets: &[Arc>], + now: Instant, +) { + // Register every currently due target while completion is excluded so one + // empty-queue boundary can release the whole batch. + let mut drain = shared + .receiver_drain + .lock() + .expect("receiver drain mutex poisoned"); + for target in targets { + let mut target = target.lock().expect("target mutex poisoned"); + let TargetStatus::Opening { + attempt, + next_send_at, + awaiting_receiver_generation: None, + } = target.status + else { + continue; + }; + if attempt == 0 || next_send_at > now { + continue; + } + match drain.request_or_join() { + Ok(generation) => { + target.status = TargetStatus::Opening { + attempt, + next_send_at, + awaiting_receiver_generation: Some(generation), + }; + } + Err(err) => target.mark_finished(ManagedTargetEndReason::OpenFailed( + ManagedTargetFailure::opening(&err), + )), + } + } +} + fn poll_active_timeouts(shared: &GroupShared, now: Instant) { for target in all_targets(shared) { let mut target = target.lock().expect("target mutex poisoned"); @@ -1992,7 +2045,7 @@ fn next_open_deadline(shared: &GroupShared) -> Option { match target.status { TargetStatus::Opening { next_send_at, - awaiting_receiver: false, + awaiting_receiver_generation: None, .. } => Some(next_send_at), _ => None, @@ -2202,29 +2255,43 @@ mod tests { hub: EventHub, target: Arc>, ) -> GroupShared { - let requested = params_from_config(client_config).unwrap(); - let (id, remote) = { - let target = target.lock().expect("target mutex poisoned"); - (target.id.clone(), target.remote) - }; - let mut registry = TargetRegistry::default(); - registry.desired.insert(id.clone()); - registry.remotes.insert(remote, id.clone()); - registry.targets.insert(id, target); - let (control_tx, _control_rx) = mpsc::channel(); + shared_with_targets(client_config, hub, vec![target]).0 + } - GroupShared { - registry: Mutex::new(registry), - hub, - cancellation: CancellationToken::new(), - control_tx, - peer_closed_target_count: Arc::new(AtomicU64::new(0)), - receiver_processed_through: Mutex::new(Instant::now()), - receiver_progress_waited_on: AtomicBool::new(false), - requested_interval_ns: requested.interval_ns, - requested_dscp: requested.dscp, - family_remote: remote, + fn shared_with_targets( + client_config: &ClientConfig, + hub: EventHub, + targets: Vec>>, + ) -> (GroupShared, mpsc::Receiver) { + let requested = params_from_config(client_config).unwrap(); + let mut registry = TargetRegistry::default(); + let mut family_remote = None; + for target in targets { + let (id, remote) = { + let target = target.lock().expect("target mutex poisoned"); + (target.id.clone(), target.remote) + }; + family_remote.get_or_insert(remote); + registry.desired.insert(id.clone()); + registry.remotes.insert(remote, id.clone()); + registry.targets.insert(id, target); } + let (control_tx, control_rx) = mpsc::channel(); + + ( + GroupShared { + registry: Mutex::new(registry), + hub, + cancellation: CancellationToken::new(), + control_tx, + peer_closed_target_count: Arc::new(AtomicU64::new(0)), + receiver_drain: Mutex::new(ReceiverDrainState::default()), + requested_interval_ns: requested.interval_ns, + requested_dscp: requested.dscp, + family_remote: family_remote.expect("test shared state requires a target"), + }, + control_rx, + ) } fn drain_available_group_events( @@ -2529,6 +2596,13 @@ mod tests { .map(|(size, peer)| (buf[..size].to_vec(), peer)) } + fn recv_group_datagram(socket: &UdpSocket) -> (ReceivedDatagramFrom, Vec) { + let mut packet = vec![0_u8; RECV_BUFFER_SIZE]; + let datagram = recv_datagram_from(socket, &mut packet).unwrap(); + packet.truncate(datagram.len); + (datagram, packet) + } + fn open_reply(flags: u8, token: u64, params: &Params) -> Vec { let mut packet = Vec::new(); packet.extend_from_slice(&MAGIC); @@ -3563,7 +3637,7 @@ mod tests { } #[test] - fn receiver_progress_is_monotonic_and_wakes_only_waiters() { + fn receiver_drain_completion_is_monotonic_and_wakes_waiting_scheduler() { let interval = Duration::from_millis(10); let config = group_config(None, interval, ManagedGroupPacing::Staggered).client; let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); @@ -3572,19 +3646,25 @@ mod tests { let target = Arc::new(Mutex::new( TargetState::new(&config, target("peer", remote), 0, now).unwrap(), )); - let shared = shared_with_target(&config, EventHub::new(), target); - let first = now + Duration::from_millis(10); - let later = now + Duration::from_millis(20); + let (shared, control_rx) = shared_with_targets(&config, EventHub::new(), vec![target]); + let shared = Arc::new(shared); + let generation = request_receiver_drain(&shared).unwrap(); + let waiter_shared = shared.clone(); + let waiter = thread::spawn(move || { + wait_for_scheduler_control(&control_rx, &waiter_shared.cancellation, None) + }); - record_receiver_progress(&shared, later); - record_receiver_progress(&shared, first); + complete_receiver_drain(&shared, generation); - assert!(receiver_processed_deadline(&shared, later)); - assert!(!shared.receiver_progress_waited_on.load(Ordering::Acquire)); + assert!(matches!(waiter.join().unwrap(), Some(ControlMessage::Wake))); + assert_eq!(completed_receiver_drain(&shared), generation); + complete_receiver_drain(&shared, 0); + complete_receiver_drain(&shared, generation); + assert_eq!(completed_receiver_drain(&shared), generation); } #[test] - fn expired_attempt_waits_for_receiver_before_retrying() { + fn expired_attempt_waits_for_receiver_drain_before_retrying() { let interval = Duration::from_millis(10); let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client; config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)]; @@ -3614,23 +3694,23 @@ mod tests { assert!(recv_request_timeout(&peer).is_some()); drive_open_attempts(&config, &socket, &shared, deadline); - assert!(matches!( - target.lock().expect("target mutex poisoned").status, + let generation = match target.lock().expect("target mutex poisoned").status { TargetStatus::Opening { attempt: 1, - awaiting_receiver: true, + awaiting_receiver_generation: Some(generation), .. - } - )); + } => generation, + ref status => panic!("unexpected waiting status: {status:?}"), + }; assert!(recv_request_timeout(&peer).is_none()); - record_receiver_progress(&shared, deadline); + complete_receiver_drain(&shared, generation); drive_open_attempts(&config, &socket, &shared, deadline); assert!(matches!( target.lock().expect("target mutex poisoned").status, TargetStatus::Opening { attempt: 2, - awaiting_receiver: false, + awaiting_receiver_generation: None, .. } )); @@ -3638,11 +3718,14 @@ mod tests { } #[test] - fn in_window_reply_remains_eligible_after_scheduler_deadline() { + fn queued_reply_dequeued_after_deadline_opens_without_retry() { let interval = Duration::from_millis(10); - let config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + config.open_timeouts = vec![Duration::from_millis(200)]; let params = test_params(None, interval); let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + peer.set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); let remote = peer.local_addr().unwrap(); let now = Instant::now(); let target = Arc::new(Mutex::new( @@ -3652,76 +3735,374 @@ mod tests { let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); drive_open_attempts(&config, &socket, &shared, now); - let (window, deadline) = { + let deadline = { let target = target.lock().expect("target mutex poisoned"); match target.status { - TargetStatus::Opening { - active_window: Some(window), - next_send_at, - .. - } => (window, next_send_at), + TargetStatus::Opening { next_send_at, .. } => next_send_at, ref status => panic!("unexpected opening status: {status:?}"), } }; - let received_at = ClientTimestamp { - wall: std::time::SystemTime::now(), - mono: window.deadline, - }; - let progress = ReceiverProgressGuard::new(&shared, received_at.mono); + let (_, client_addr) = recv_request_timeout(&peer).expect("missing first open request"); + peer.send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), + client_addr, + ) + .unwrap(); + assert!( + Instant::now() < deadline, + "reply must queue before deadline" + ); + while Instant::now() < deadline { + thread::yield_now(); + } drive_open_attempts(&config, &socket, &shared, deadline); assert!(matches!( target.lock().expect("target mutex poisoned").status, TargetStatus::Opening { - awaiting_receiver: true, + awaiting_receiver_generation: Some(_), .. } )); - { - let mut target = target.lock().expect("target mutex poisoned"); - assert!(window.contains(received_at.mono)); - let reply = match target - .runtime - .inspect_open_datagram(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms)) - .unwrap() - { - OpenDatagramDisposition::Trusted(reply) => reply, - OpenDatagramDisposition::Ignore => panic!("in-window reply must be trusted"), - }; - let machine = target - .runtime - .prepare_open_acceptance(reply, received_at) - .unwrap(); - let prepared = prepare_group_open(&shared, machine, received_at).unwrap(); - let outcome = target.runtime.commit_open(prepared.machine); - target.schedule = prepared.schedule; - publish_open_outcome(&shared.hub, &mut target, outcome); - } - drop(progress); + let (datagram, packet) = recv_group_datagram(&socket); + assert!(datagram.received_at.mono >= deadline); + process_group_datagram(&socket, &shared, datagram, &packet); - assert!(matches!( - target.lock().expect("target mutex poisoned").status, - TargetStatus::Active - )); - assert!(receiver_processed_deadline(&shared, deadline)); + let target = target.lock().expect("target mutex poisoned"); + assert!(matches!(target.status, TargetStatus::Active)); + assert!(target.final_reason.is_none()); + drop(target); + assert!(recv_request_timeout(&peer).is_none()); } #[test] - fn post_deadline_reply_is_not_eligible_for_expired_attempt() { + fn generation_requested_after_receive_snapshot_needs_later_empty_boundary() { + let interval = Duration::from_millis(10); + let config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, Instant::now()).unwrap(), + )); + let shared = shared_with_target(&config, EventHub::new(), target); + + let observed_before_request = requested_receiver_drain(&shared); + let generation = request_receiver_drain(&shared).unwrap(); + complete_receiver_drain(&shared, observed_before_request); + assert!(completed_receiver_drain(&shared) < generation); + + let observed_after_request = requested_receiver_drain(&shared); + complete_receiver_drain(&shared, observed_after_request); + assert_eq!(completed_receiver_drain(&shared), generation); + } + + #[test] + fn malformed_packet_requires_empty_receive_to_complete_drain() { + let interval = Duration::from_millis(10); + let config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + peer.set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + let remote = peer.local_addr().unwrap(); let now = Instant::now(); - let window = OpenAttemptWindow { - sent_at: now, - deadline: now + Duration::from_millis(200), + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, now).unwrap(), + )); + let (shared, control_rx) = + shared_with_targets(&config, EventHub::new(), vec![target.clone()]); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + drive_open_attempts(&config, &socket, &shared, now); + let deadline = match target.lock().expect("target mutex poisoned").status { + TargetStatus::Opening { next_send_at, .. } => next_send_at, + ref status => panic!("unexpected opening status: {status:?}"), + }; + let (_, client_addr) = recv_request_timeout(&peer).expect("missing open request"); + drive_open_attempts(&config, &socket, &shared, deadline); + let generation = match target.lock().expect("target mutex poisoned").status { + TargetStatus::Opening { + awaiting_receiver_generation: Some(generation), + .. + } => generation, + ref status => panic!("unexpected waiting status: {status:?}"), }; - assert!(window.contains(window.sent_at)); - assert!(window.contains(window.deadline)); - assert!(!window.contains(window.deadline + Duration::from_nanos(1))); + peer.send_to(&[0_u8], client_addr).unwrap(); + let (datagram, packet) = recv_group_datagram(&socket); + process_group_datagram(&socket, &shared, datagram, &packet); + assert!(completed_receiver_drain(&shared) < generation); + assert!(matches!( + control_rx.try_recv(), + Err(mpsc::TryRecvError::Empty) + )); + + let observed_generation = requested_receiver_drain(&shared); + socket.set_nonblocking(true).unwrap(); + let mut buf = [0_u8; 1]; + assert!(matches!( + recv_datagram_from(&socket, &mut buf), + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock + )); + complete_receiver_drain(&shared, observed_generation); + assert_eq!(completed_receiver_drain(&shared), generation); + assert!(matches!(control_rx.recv().unwrap(), ControlMessage::Wake)); } #[test] - fn final_open_timeout_waits_until_receiver_processes_through_deadline() { + fn opening_reply_before_first_request_is_ignored() { + let interval = Duration::from_millis(10); + let config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + let params = test_params(None, interval); + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = peer.local_addr().unwrap(); + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, Instant::now()).unwrap(), + )); + let shared = shared_with_target(&config, EventHub::new(), target.clone()); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + peer.send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), + socket.local_addr().unwrap(), + ) + .unwrap(); + let (datagram, packet) = recv_group_datagram(&socket); + process_group_datagram(&socket, &shared, datagram, &packet); + + let target = target.lock().expect("target mutex poisoned"); + assert!(matches!( + target.status, + TargetStatus::Opening { + attempt: 0, + awaiting_receiver_generation: None, + .. + } + )); + assert!(target.runtime.prepare_open_request().is_ok()); + } + + #[test] + fn post_drain_reply_cannot_open_expired_attempt() { + let interval = Duration::from_millis(10); + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)]; + let params = test_params(None, interval); + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + peer.set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + let remote = peer.local_addr().unwrap(); + let now = Instant::now(); + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", remote), 0, now).unwrap(), + )); + let shared = shared_with_target(&config, EventHub::new(), target.clone()); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + drive_open_attempts(&config, &socket, &shared, now); + let deadline = match target.lock().expect("target mutex poisoned").status { + TargetStatus::Opening { next_send_at, .. } => next_send_at, + ref status => panic!("unexpected opening status: {status:?}"), + }; + let (_, client_addr) = recv_request_timeout(&peer).expect("missing open request"); + drive_open_attempts(&config, &socket, &shared, deadline); + let generation = match target.lock().expect("target mutex poisoned").status { + TargetStatus::Opening { + awaiting_receiver_generation: Some(generation), + .. + } => generation, + ref status => panic!("unexpected waiting status: {status:?}"), + }; + complete_receiver_drain(&shared, generation); + + peer.send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), + client_addr, + ) + .unwrap(); + let (datagram, packet) = recv_group_datagram(&socket); + process_group_datagram(&socket, &shared, datagram, &packet); + { + let target = target.lock().expect("target mutex poisoned"); + assert!(matches!( + target.status, + TargetStatus::Opening { + attempt: 1, + awaiting_receiver_generation: Some(waiting), + .. + } if waiting == generation + )); + assert!(target.runtime.prepare_open_request().is_ok()); + } + + drive_open_attempts(&config, &socket, &shared, deadline); + assert!(matches!( + target.lock().expect("target mutex poisoned").status, + TargetStatus::Opening { + attempt: 2, + awaiting_receiver_generation: None, + .. + } + )); + assert!(recv_request_timeout(&peer).is_some()); + } + + #[test] + fn multiple_expired_targets_share_one_receiver_drain_generation() { + let interval = Duration::from_millis(10); + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)]; + let first_peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let second_peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + for peer in [&first_peer, &second_peer] { + peer.set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + } + let now = Instant::now(); + let first = Arc::new(Mutex::new( + TargetState::new( + &config, + target("first", first_peer.local_addr().unwrap()), + 0, + now, + ) + .unwrap(), + )); + let second = Arc::new(Mutex::new( + TargetState::new( + &config, + target("second", second_peer.local_addr().unwrap()), + 1, + now, + ) + .unwrap(), + )); + let (shared, _control_rx) = shared_with_targets( + &config, + EventHub::new(), + vec![first.clone(), second.clone()], + ); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + drive_open_attempts(&config, &socket, &shared, now); + assert!(recv_request_timeout(&first_peer).is_some()); + assert!(recv_request_timeout(&second_peer).is_some()); + let deadline = [&first, &second] + .into_iter() + .filter_map(|target| match target.lock().unwrap().status { + TargetStatus::Opening { next_send_at, .. } => Some(next_send_at), + _ => None, + }) + .max() + .unwrap(); + + drive_open_attempts(&config, &socket, &shared, deadline); + let generations = [&first, &second].map(|target| match target.lock().unwrap().status { + TargetStatus::Opening { + awaiting_receiver_generation: Some(generation), + .. + } => generation, + ref status => panic!("unexpected waiting status: {status:?}"), + }); + assert_eq!(generations[0], generations[1]); + + complete_receiver_drain(&shared, generations[0]); + drive_open_attempts(&config, &socket, &shared, deadline); + for target in [&first, &second] { + assert!(matches!( + target.lock().unwrap().status, + TargetStatus::Opening { + attempt: 2, + awaiting_receiver_generation: None, + .. + } + )); + } + assert!(recv_request_timeout(&first_peer).is_some()); + assert!(recv_request_timeout(&second_peer).is_some()); + + drive_open_attempts(&config, &socket, &shared, deadline); + assert!(recv_request_timeout(&first_peer).is_none()); + assert!(recv_request_timeout(&second_peer).is_none()); + } + + #[test] + fn removal_and_cancellation_release_targets_waiting_for_receiver_drain() { + let interval = Duration::from_millis(10); + let config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + let first_peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let second_peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + let now = Instant::now(); + let first_config = target("first", first_peer.local_addr().unwrap()); + let second_config = target("second", second_peer.local_addr().unwrap()); + let first = Arc::new(Mutex::new( + TargetState::new(&config, first_config.clone(), 0, now).unwrap(), + )); + let second = Arc::new(Mutex::new( + TargetState::new(&config, second_config, 1, now).unwrap(), + )); + let (shared, _control_rx) = shared_with_targets( + &config, + EventHub::new(), + vec![first.clone(), second.clone()], + ); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + drive_open_attempts(&config, &socket, &shared, now); + let deadline = [&first, &second] + .into_iter() + .filter_map(|target| match target.lock().unwrap().status { + TargetStatus::Opening { next_send_at, .. } => Some(next_send_at), + _ => None, + }) + .max() + .unwrap(); + drive_open_attempts(&config, &socket, &shared, deadline); + for target in [&first, &second] { + assert!(matches!( + target.lock().unwrap().status, + TargetStatus::Opening { + awaiting_receiver_generation: Some(_), + .. + } + )); + } + + let mut next_order = 2; + let mut records = TargetOutcomeHistory::default(); + apply_target_update( + &config, + &socket, + &shared, + vec![first_config], + &mut next_order, + &mut records, + ) + .unwrap(); + assert!(matches!( + second.lock().unwrap().final_reason, + Some(ManagedTargetEndReason::Removed) + )); + + shared.cancellation.cancel(); + cancel_remaining_targets(&socket, &shared, &mut records); + assert!(matches!( + first.lock().unwrap().final_reason, + Some(ManagedTargetEndReason::Cancelled) + )); + assert_eq!(records.total, 2); + assert!(matches!( + records.recent.front().map(|outcome| &outcome.end_reason), + Some(ManagedTargetEndReason::Removed) + )); + assert!(matches!( + records.recent.back().map(|outcome| &outcome.end_reason), + Some(ManagedTargetEndReason::Cancelled) + )); + } + + #[test] + fn final_open_timeout_waits_until_receiver_drain_completes() { let interval = Duration::from_millis(10); let config = group_config(None, interval, ManagedGroupPacing::Staggered).client; let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); @@ -3747,30 +4128,40 @@ mod tests { }; drive_open_attempts(&config, &socket, &shared, deadline); - { + let generation = { let target = target.lock().expect("target mutex poisoned"); - assert!(matches!( - target.status, + let generation = match target.status { TargetStatus::Opening { attempt: 1, - awaiting_receiver: true, + awaiting_receiver_generation: Some(generation), .. - } - )); + } => generation, + ref status => panic!("unexpected waiting status: {status:?}"), + }; assert!(target.final_reason.is_none()); - } + generation + }; - record_receiver_progress(&shared, deadline); + complete_receiver_drain(&shared, generation); drive_open_attempts(&config, &socket, &shared, deadline); - let target = target.lock().expect("target mutex poisoned"); - assert!(matches!(target.status, TargetStatus::Finished)); - assert!(matches!( - target.final_reason, - Some(ManagedTargetEndReason::OpenFailed(ManagedTargetFailure { - kind: ManagedTargetFailureKind::OpeningTimeout, - .. - })) - )); + let reason = { + let target = target.lock().expect("target mutex poisoned"); + assert!(matches!(target.status, TargetStatus::Finished)); + assert!(matches!( + target.final_reason, + Some(ManagedTargetEndReason::OpenFailed(ManagedTargetFailure { + kind: ManagedTargetFailureKind::OpeningTimeout, + .. + })) + )); + target.final_reason.clone() + }; + + drive_open_attempts(&config, &socket, &shared, deadline); + assert_eq!( + target.lock().expect("target mutex poisoned").final_reason, + reason + ); } #[test]