From da731edd0d9ae6b1aee7008a4473dd4e415240d4 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 12:59:27 +0700 Subject: [PATCH 1/4] refactor(session): prepare close transactions --- crates/irtt-client/src/session/machine.rs | 208 +++++++++++++++++++--- 1 file changed, 186 insertions(+), 22 deletions(-) diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 74cced0..451368b 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -53,6 +53,7 @@ enum CloseSource { struct ActiveSession { token: u64, negotiated: NegotiatedParams, + local_close_packet: Box<[u8]>, next_wire_seq: u32, highest_received_seq: Option, packets_sent: u64, @@ -76,7 +77,6 @@ pub(crate) enum OpenDatagramDisposition { pub(crate) struct PreparedOpenAcceptance { next_state: MachineState, outcome: OpenOutcome, - cleanup_close: Option>, } impl PreparedOpenAcceptance { @@ -91,7 +91,13 @@ impl PreparedOpenAcceptance { } pub(crate) fn cleanup_close_packet(&self) -> Option<&[u8]> { - self.cleanup_close.as_deref() + match &self.next_state { + MachineState::Open(session) => Some(&session.local_close_packet), + MachineState::NoTestCompleted => None, + MachineState::Connected | MachineState::Closed { .. } => { + unreachable!("open acceptance only prepares open or no-test state") + } + } } #[cfg(test)] @@ -100,6 +106,18 @@ impl PreparedOpenAcceptance { } } +#[derive(Debug)] +pub(crate) struct PreparedClose<'a> { + pub(crate) bytes: &'a [u8], + pub(crate) commit: CloseCommit, +} + +#[derive(Debug)] +pub(crate) struct CloseCommit { + packets_sent: u64, + event: ClientEvent, +} + #[derive(Debug)] pub(crate) struct OpenAcceptanceFailure { pub(crate) primary: ClientError, @@ -401,28 +419,40 @@ impl SessionMachine { Ok(events) } + pub(crate) fn prepare_close(&self) -> Result, ClientError> { + let session = self.open_session()?; + Ok(PreparedClose { + bytes: &session.local_close_packet, + commit: CloseCommit { + packets_sent: session.packets_sent, + event: ClientEvent::SessionClosed { + remote: self.remote, + token: session.token, + at: ClientTimestamp::now(), + }, + }, + }) + } + + pub(crate) fn commit_local_close(&mut self, commit: CloseCommit) -> ClientEvent { + debug_assert!( + matches!(self.state, MachineState::Open(_)), + "local close commits only from open state" + ); + self.state = MachineState::Closed { + source: CloseSource::Local, + packets_sent: commit.packets_sent, + }; + commit.event + } + pub(crate) fn close_with(&mut self, send: F) -> Result, ClientError> where F: FnOnce(&[u8]) -> Result<(), ClientError>, { - let token = match &self.state { - MachineState::Open(session) => session.token, - MachineState::Closed { .. } => return Err(ClientError::AlreadyClosed), - MachineState::Connected | MachineState::NoTestCompleted => { - return Err(ClientError::NotOpen) - } - }; - - let packet = - encode_close_request(&CloseRequest { token }, self.config.hmac_key.as_deref())?; - send(&packet)?; - self.transition_to_closed(CloseSource::Local); - - Ok(vec![ClientEvent::SessionClosed { - remote: self.remote, - token, - at: ClientTimestamp::now(), - }]) + let prepared = self.prepare_close()?; + send(prepared.bytes)?; + Ok(vec![self.commit_local_close(prepared.commit)]) } pub(crate) fn pending_is_empty(&self) -> bool { @@ -488,9 +518,12 @@ impl SessionMachine { Ok(negotiated) => negotiated, Err(primary) => return Err(OpenAcceptanceFailure::new(primary, cleanup_close)), }; + let local_close_packet = + cleanup_close.expect("normal non-zero-token replies prepare a cleanup close"); let next_state = MachineState::Open(Box::new(ActiveSession { token, negotiated: negotiated.clone(), + local_close_packet, next_wire_seq: 0, highest_received_seq: None, packets_sent: 0, @@ -515,7 +548,6 @@ impl SessionMachine { Ok(PreparedOpenAcceptance { next_state, outcome, - cleanup_close, }) } @@ -543,7 +575,6 @@ impl SessionMachine { Ok(PreparedOpenAcceptance { next_state: MachineState::NoTestCompleted, outcome, - cleanup_close: None, }) } @@ -981,6 +1012,14 @@ mod tests { machine.state = MachineState::Open(Box::new(ActiveSession { token: 0x0102_0304_0506_0708, negotiated, + local_close_packet: encode_close_request( + &CloseRequest { + token: 0x0102_0304_0506_0708, + }, + None, + ) + .unwrap() + .into_boxed_slice(), next_wire_seq: 0, highest_received_seq: None, packets_sent: 0, @@ -1268,6 +1307,7 @@ mod tests { ) .unwrap(); + let expected_close = prepared.cleanup_close_packet().unwrap().to_vec(); let outcome = machine.commit_open(prepared); assert!(matches!( @@ -1278,6 +1318,7 @@ mod tests { } if outcome_token == token )); assert!(machine.is_open()); + assert_eq!(active(&machine).local_close_packet.as_ref(), expected_close); assert!(matches!( machine.prepare_open_request(), Err(ClientError::AlreadyOpen) @@ -1337,6 +1378,129 @@ mod tests { assert!(matches!(machine.state, MachineState::Connected)); } + #[test] + fn prepared_close_is_stable_inert_and_commits_preserved_state() { + let mut machine = open_machine(4, Duration::from_secs(1)); + let remote = machine.remote; + active_mut(&mut machine).packets_sent = 7; + + let first_bytes = { + let prepared = machine.prepare_close().unwrap(); + assert_eq!( + prepared.bytes, + encode_close_request( + &CloseRequest { + token: 0x0102_0304_0506_0708, + }, + None, + ) + .unwrap() + ); + prepared.bytes.to_vec() + }; + assert!(machine.is_open()); + assert_eq!(active(&machine).packets_sent, 7); + + let prepared = machine.prepare_close().unwrap(); + assert_eq!(prepared.bytes, first_bytes); + let event = machine.commit_local_close(prepared.commit); + + assert!(matches!( + event, + ClientEvent::SessionClosed { + remote: event_remote, + token: 0x0102_0304_0506_0708, + .. + } if event_remote == remote + )); + assert!(matches!( + machine.state, + MachineState::Closed { + source: CloseSource::Local, + packets_sent: 7, + } + )); + assert!(matches!( + machine.prepare_close(), + Err(ClientError::AlreadyClosed) + )); + } + + #[test] + fn prepared_close_reuses_hmac_packet_built_during_open_acceptance() { + let key = b"close-key".to_vec(); + let mut machine = connected_machine(ClientConfig { + hmac_key: Some(key.clone()), + ..ClientConfig::default() + }); + let token = 0x1020_3040_5060_7080; + let prepared_open = machine + .prepare_open_acceptance( + normal_open_reply(&machine, token), + timestamp(Instant::now()), + ) + .unwrap(); + let opening_close = prepared_open.cleanup_close_packet().unwrap().to_vec(); + machine.commit_open(prepared_open); + + let prepared_close = machine.prepare_close().unwrap(); + assert_eq!(prepared_close.bytes, opening_close); + assert_eq!( + prepared_close.bytes[3], + flags::FLAG_CLOSE | flags::FLAG_HMAC + ); + irtt_proto::verify_hmac(&key, prepared_close.bytes, 4).unwrap(); + assert_eq!( + u64::from_le_bytes( + prepared_close.bytes[4 + irtt_proto::HMAC_SIZE..12 + irtt_proto::HMAC_SIZE] + .try_into() + .unwrap() + ), + token + ); + } + + #[test] + fn close_preparation_preserves_non_open_errors_and_peer_source() { + let connected = connected_machine(ClientConfig::default()); + assert!(matches!( + connected.prepare_close(), + Err(ClientError::NotOpen) + )); + + let mut no_test = connected_machine(ClientConfig { + run_mode: RunMode::NoTest, + ..ClientConfig::default() + }); + let reply = OpenReply { + flags: flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE, + token: 0, + params: no_test.requested.clone(), + }; + let prepared = no_test + .prepare_open_acceptance(reply, timestamp(Instant::now())) + .unwrap(); + no_test.commit_open(prepared); + assert!(matches!( + no_test.prepare_close(), + Err(ClientError::AlreadyCompleted) + )); + + let mut peer_closed = open_machine(4, Duration::from_secs(1)); + peer_closed.transition_to_closed(CloseSource::Peer); + assert!(matches!( + peer_closed.state, + MachineState::Closed { + source: CloseSource::Peer, + .. + } + )); + assert!(matches!( + peer_closed.prepare_close(), + Err(ClientError::AlreadyClosed) + )); + } + #[test] fn dropping_prepared_probe_changes_nothing() { let machine = open_machine(4, Duration::from_secs(1)); -- 2.54.0 From 357e1356ea0e1ff17e7f8a28b472d2278925caf2 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 13:03:51 +0700 Subject: [PATCH 2/4] fix(client): transact blocking session close --- crates/irtt-client/src/client.rs | 84 ++++++- crates/irtt-client/src/client/tests/close.rs | 251 ++++++++++++++++++- crates/irtt-client/src/client/tests/dscp.rs | 129 ++++++++++ 3 files changed, 454 insertions(+), 10 deletions(-) diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 592e3ce..1d1c85f 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -55,6 +55,11 @@ struct PreparedClientOpenFailure { struct ClientTestHooks { fail_open_dscp: Cell, fail_open_timeout_restore: Cell, + fail_close_event_reserve: Cell, + fail_close_dscp_clear: Cell, + fail_close_send: Cell, + close_reported_len: Cell>, + close_send_attempts: Cell, fail_cleanup_send: Cell, recv_buffer_len_override: Cell>, fail_dscp_restore: Cell, @@ -162,16 +167,44 @@ impl Client { /// Send a close request and emit a [`ClientEvent::SessionClosed`] event. /// /// The close event means the client has sent its close packet and stopped - /// tracking the session locally; it is not a server acknowledgement. + /// tracking the session locally; it is not a server acknowledgement. If the + /// send fails, the negotiated DSCP is restored best-effort and the session + /// and probe schedule remain open. pub fn close(&mut self) -> Result, ClientError> { - let socket = &self.socket; - let remote = self.remote; - let events = self.runtime.close_with(|packet| { - clear_dscp_on_socket(socket, remote)?; - socket.send(packet)?; - Ok(()) - })?; + let prepared = self.runtime.prepare_close()?; + let mut events = Vec::new(); + #[cfg(test)] + if self.test_hooks.fail_close_event_reserve.replace(false) { + events + .try_reserve(usize::MAX) + .map_err(|source| ClientError::AllocationFailed { + operation: "close event result", + source, + })?; + } + events + .try_reserve(1) + .map_err(|source| ClientError::AllocationFailed { + operation: "close event result", + source, + })?; + let previous_dscp = self.applied_dscp; + let expected_bytes = prepared.bytes.len(); + + self.clear_close_dscp()?; + let bytes = match self.send_close_datagram(prepared.bytes) { + Ok(bytes) => bytes, + Err(err) => { + self.restore_dscp_best_effort(previous_dscp); + return Err(ClientError::Socket(err)); + } + }; + let event = self.runtime.commit_local_close(prepared.commit); self.schedule = None; + self.applied_dscp = None; + + validate_datagram_length(expected_bytes, bytes)?; + events.push(event); Ok(events) } @@ -244,8 +277,11 @@ impl Client { datagram.received_at, datagram.meta, )?; - if self.runtime.is_terminal() { + if self.runtime.is_peer_closed() { self.schedule = None; + if self.clear_close_dscp().is_ok() { + self.applied_dscp = None; + } } Ok(events) } @@ -607,6 +643,18 @@ impl Client { }; } + fn clear_close_dscp(&self) -> Result<(), ClientError> { + #[cfg(test)] + if self.test_hooks.fail_close_dscp_clear.replace(false) { + return Err(ClientError::SocketOption { + operation: "clear negotiated DSCP", + remote: self.remote, + source: io::Error::other("injected negotiated DSCP clear failure"), + }); + } + 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)] @@ -640,6 +688,24 @@ impl Client { } self.socket.set_read_timeout(timeout) } + + fn send_close_datagram(&self, packet: &[u8]) -> io::Result { + #[cfg(test)] + { + self.test_hooks + .close_send_attempts + .set(self.test_hooks.close_send_attempts.get() + 1); + if self.test_hooks.fail_close_send.replace(false) { + return Err(io::Error::other("injected close send failure")); + } + } + let bytes = self.socket.send(packet)?; + #[cfg(test)] + if let Some(reported) = self.test_hooks.close_reported_len.take() { + return Ok(reported); + } + Ok(bytes) + } } pub(crate) fn echo_sent_event( diff --git a/crates/irtt-client/src/client/tests/close.rs b/crates/irtt-client/src/client/tests/close.rs index f5c7f23..656692f 100644 --- a/crates/irtt-client/src/client/tests/close.rs +++ b/crates/irtt-client/src/client/tests/close.rs @@ -30,7 +30,13 @@ fn close_sends_one_close_packet_with_negotiated_token() { let mut client = Client::connect(default_test_config(server.addr)).unwrap(); assert_open_started(client.open().unwrap()); let events = client.close().unwrap(); - assert_eq!(events.len(), 1); + assert!(matches!( + events.as_slice(), + [ClientEvent::SessionClosed { token: TOKEN, .. }] + )); + assert!(!client.runtime.is_open()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); let packets: Vec<_> = server.rx.iter().take(2).collect(); let close = &packets[1]; assert_eq!(close[3], flags::FLAG_CLOSE); @@ -38,6 +44,138 @@ fn close_sends_one_close_packet_with_negotiated_token() { server.join(); } +#[test] +fn close_event_reservation_failure_precedes_dscp_clear_and_send() { + 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(); + assert_open_started(client.open().unwrap()); + client.test_hooks.fail_close_event_reserve.set(true); + client.test_hooks.fail_close_dscp_clear.set(true); + + assert!(matches!( + client.close(), + Err(ClientError::AllocationFailed { + operation: "close event result", + .. + }) + )); + assert_eq!(client.test_hooks.close_send_attempts.get(), 0); + assert!(client.test_hooks.fail_close_dscp_clear.get()); + assert!(client.runtime.is_open()); + assert!(client.schedule.is_some()); + assert_eq!(client.applied_dscp, Some(0)); + + client.test_hooks.fail_close_dscp_clear.set(false); + client.close().unwrap(); + assert_eq!(client.test_hooks.close_send_attempts.get(), 1); + server.join(); +} + +#[test] +fn close_dscp_clear_failure_precedes_send_and_preserves_open_state() { + 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(); + assert_open_started(client.open().unwrap()); + client.test_hooks.fail_close_dscp_clear.set(true); + + assert!(matches!( + client.close(), + Err(ClientError::SocketOption { .. }) + )); + assert_eq!(client.test_hooks.close_send_attempts.get(), 0); + assert!(client.runtime.is_open()); + assert!(client.schedule.is_some()); + assert_eq!(client.applied_dscp, Some(0)); + + client.close().unwrap(); + assert_eq!(client.test_hooks.close_send_attempts.get(), 1); + server.join(); +} + +#[test] +fn close_send_failure_leaves_machine_and_schedule_open_for_retry() { + 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(); + assert_open_started(client.open().unwrap()); + client.test_hooks.fail_close_send.set(true); + + assert!(matches!(client.close(), Err(ClientError::Socket(_)))); + assert!(client.runtime.is_open()); + assert!(client.schedule.is_some()); + assert_eq!(client.applied_dscp, Some(0)); + assert_eq!(client.test_hooks.close_send_attempts.get(), 1); + + assert!(matches!( + client.close().unwrap().as_slice(), + [ClientEvent::SessionClosed { token: TOKEN, .. }] + )); + assert_eq!(client.test_hooks.close_send_attempts.get(), 2); + assert!(matches!(client.close(), Err(ClientError::AlreadyClosed))); + server.join(); +} + +#[test] +fn short_successful_close_commits_before_length_mismatch() { + 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(); + assert_open_started(client.open().unwrap()); + let expected = client.runtime.prepare_close().unwrap().bytes.len(); + client.test_hooks.close_reported_len.set(Some(expected - 1)); + + assert!(matches!( + client.close(), + Err(ClientError::DatagramLengthMismatch { + expected: error_expected, + actual, + }) if error_expected == expected && actual + 1 == expected + )); + assert!(!client.runtime.is_open()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); + assert!(matches!(client.close(), Err(ClientError::AlreadyClosed))); + server.join(); +} + #[test] fn send_probe_fails_after_close() { let params = default_params(); @@ -132,6 +270,117 @@ fn close_flagged_echo_reply_emits_reply_then_closes_without_sending_close() { server.join(); } +#[test] +fn peer_close_dscp_cleanup_failure_preserves_events_and_closed_state() { + let params = default_params(); + let server = start_fake_server({ + let params = params.clone(); + move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None), + peer, + ) + .unwrap(); + let (request, _) = recv_request(&socket, &tx); + let seq = u32::from_le_bytes(request[12..16].try_into().unwrap()); + socket + .send_to( + &echo_reply_packet_with_flags( + TOKEN, + seq, + ¶ms, + &TimestampFields::default(), + None, + FLAG_REPLY | flags::FLAG_CLOSE, + ), + peer, + ) + .unwrap(); + socket + .set_read_timeout(Some(Duration::from_millis(250))) + .unwrap(); + while recv_request_timeout(&socket, &tx).is_some() {} + } + }); + let mut client = Client::connect(ClientConfig { + socket_config: crate::SocketConfig { + recv_timeout: Some(Duration::from_millis(200)), + ..Default::default() + }, + ..default_test_config(server.addr) + }) + .unwrap(); + assert_open_started(client.open().unwrap()); + client.send_probe().unwrap(); + client.test_hooks.fail_close_dscp_clear.set(true); + + assert!(matches!( + client.recv_once().unwrap().as_slice(), + [ + ClientEvent::EchoReply { .. }, + ClientEvent::SessionClosed { token: TOKEN, .. } + ] + )); + assert!(client.is_peer_closed()); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, Some(0)); + assert_eq!(client.test_hooks.close_send_attempts.get(), 0); + let open = server.rx.recv_timeout(Duration::from_millis(100)).unwrap(); + let echo = server.rx.recv_timeout(Duration::from_millis(100)).unwrap(); + assert_ne!(open[3] & FLAG_OPEN, 0); + assert_eq!(echo[3] & flags::FLAG_CLOSE, 0); + assert!(server.rx.recv_timeout(Duration::from_millis(400)).is_err()); + server.join(); +} + +#[test] +fn recv_available_stops_after_peer_close() { + let params = default_params(); + let server = start_fake_server({ + let params = params.clone(); + move |socket, tx| { + let (_, peer) = recv_request(&socket, &tx); + socket + .send_to( + &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms, None), + peer, + ) + .unwrap(); + let (request, _) = recv_request(&socket, &tx); + let seq = u32::from_le_bytes(request[12..16].try_into().unwrap()); + let close = echo_reply_packet_with_flags( + TOKEN, + seq, + ¶ms, + &TimestampFields::default(), + None, + FLAG_REPLY | flags::FLAG_CLOSE, + ); + socket.send_to(&close, peer).unwrap(); + socket.send_to(&close, peer).unwrap(); + } + }); + let mut client = Client::connect(default_test_config(server.addr)).unwrap(); + assert_open_started(client.open().unwrap()); + client.send_probe().unwrap(); + + let events = client + .recv_available(RecvBudget { max_packets: 8 }) + .unwrap(); + + assert!(matches!( + events.as_slice(), + [ + ClientEvent::EchoReply { .. }, + ClientEvent::SessionClosed { token: TOKEN, .. } + ] + )); + assert!(client.is_peer_closed()); + server.join(); +} + #[test] fn normal_echo_reply_does_not_close_session() { let params = default_params(); diff --git a/crates/irtt-client/src/client/tests/dscp.rs b/crates/irtt-client/src/client/tests/dscp.rs index 21956c8..c98e876 100644 --- a/crates/irtt-client/src/client/tests/dscp.rs +++ b/crates/irtt-client/src/client/tests/dscp.rs @@ -73,3 +73,132 @@ fn normal_open_uses_negotiated_dscp_not_requested_dscp() { ); server.join(); } + +#[test] +#[cfg(not(any( + target_os = "fuchsia", + target_os = "redox", + target_os = "solaris", + target_os = "illumos", + target_os = "haiku", +)))] +fn failed_close_send_restores_negotiated_dscp_and_keeps_session_open() { + let mut params = default_params(); + params.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, ¶ms, 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(); + assert_open_started(client.open().unwrap()); + client.test_hooks.fail_close_send.set(true); + + let error = client.close().unwrap_err(); + + assert!(matches!(error, ClientError::Socket(_))); + assert!(client.runtime.is_open()); + assert!(client.schedule.is_some()); + assert_eq!(client.applied_dscp, Some(46)); + assert_eq!( + socket_traffic_class(&client.socket, client.remote).unwrap() & 0xfc, + 184 + ); + client.close().unwrap(); + server.join(); +} + +#[test] +fn failed_close_keeps_send_error_primary_when_dscp_restoration_also_fails() { + 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(); + assert_open_started(client.open().unwrap()); + client.test_hooks.fail_close_send.set(true); + client.test_hooks.fail_dscp_restore.set(true); + + let error = client.close().unwrap_err(); + + assert!(matches!(error, ClientError::Socket(_))); + assert!(error.to_string().contains("injected close send failure")); + assert!(client.runtime.is_open()); + assert!(client.schedule.is_some()); + assert_eq!(client.applied_dscp, Some(0)); + client.close().unwrap(); + server.join(); +} + +#[test] +#[cfg(not(any( + target_os = "fuchsia", + target_os = "redox", + target_os = "solaris", + target_os = "illumos", + target_os = "haiku", +)))] +fn authenticated_peer_close_clears_schedule_and_negotiated_dscp() { + let mut params = default_params(); + params.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, ¶ms, None), + peer, + ) + .unwrap(); + let (request, _) = recv_request(&socket, &tx); + let seq = u32::from_le_bytes(request[12..16].try_into().unwrap()); + socket + .send_to( + &echo_reply_packet_with_flags( + TOKEN, + seq, + ¶ms, + &TimestampFields::default(), + None, + FLAG_REPLY | flags::FLAG_CLOSE, + ), + peer, + ) + .unwrap(); + }); + let mut config = default_test_config(server.addr); + config.dscp = 46; + let mut client = Client::connect(config).unwrap(); + assert_open_started(client.open().unwrap()); + client.send_probe().unwrap(); + + let events = client.recv_once().unwrap(); + + assert!(matches!( + events.as_slice(), + [ + ClientEvent::EchoReply { .. }, + ClientEvent::SessionClosed { .. } + ] + )); + assert!(client.schedule.is_none()); + assert_eq!(client.applied_dscp, None); + assert_eq!( + socket_traffic_class(&client.socket, client.remote).unwrap(), + 0 + ); + server.join(); +} -- 2.54.0 From ef19c45c1f1cfc9186cbf4253e8f345e261a64fd Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 13:08:38 +0700 Subject: [PATCH 3/4] fix(managed): transact group session close --- crates/irtt-client/src/managed/group.rs | 365 +++++++++++++++++++++- crates/irtt-client/src/session/machine.rs | 9 - 2 files changed, 355 insertions(+), 19 deletions(-) diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index e2fda5e..9d1145e 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -725,6 +725,14 @@ struct TargetState { probe_send_error: bool, #[cfg(test)] probe_send_timestamps: Option, + #[cfg(test)] + close_reported_len: Option, + #[cfg(test)] + close_send_error: bool, + #[cfg(test)] + close_send_attempts: usize, + #[cfg(test)] + close_event_reserve_error: bool, } type ScheduledTarget = (Arc>, Instant); @@ -839,6 +847,14 @@ impl TargetState { probe_send_error: false, #[cfg(test)] probe_send_timestamps: None, + #[cfg(test)] + close_reported_len: None, + #[cfg(test)] + close_send_error: false, + #[cfg(test)] + close_send_attempts: 0, + #[cfg(test)] + close_event_reserve_error: false, }) } @@ -1795,21 +1811,79 @@ fn close_locked_target( ) { if target.runtime.is_open() && !target.runtime.is_peer_closed() { let remote = target.remote; - match target.runtime.close_with(|packet| { - socket.send_to(packet, remote)?; - Ok(()) - }) { - Ok(events) => { - publish_events(hub, target, events); - target.schedule = None; - } + let prepared = match target.runtime.prepare_close() { + Ok(prepared) => prepared, Err(err) => { target.mark_finished(ManagedTargetEndReason::Failed( ManagedTargetFailure::runtime(&err), )); return; } + }; + let mut events = Vec::new(); + #[cfg(test)] + if std::mem::take(&mut target.close_event_reserve_error) { + if let Err(source) = events.try_reserve(usize::MAX) { + let err = ClientError::AllocationFailed { + operation: "managed close event result", + source, + }; + target.mark_finished(ManagedTargetEndReason::Failed( + ManagedTargetFailure::runtime(&err), + )); + return; + } } + if let Err(source) = events.try_reserve(1) { + let err = ClientError::AllocationFailed { + operation: "managed close event result", + source, + }; + target.mark_finished(ManagedTargetEndReason::Failed( + ManagedTargetFailure::runtime(&err), + )); + return; + } + let expected_bytes = prepared.bytes.len(); + #[cfg(test)] + let reported_bytes = target.close_reported_len.take(); + #[cfg(test)] + let fail_send = std::mem::take(&mut target.close_send_error); + #[cfg(test)] + { + target.close_send_attempts += 1; + if fail_send { + let err = ClientError::Socket(std::io::Error::other( + "injected managed close send failure", + )); + target.mark_finished(ManagedTargetEndReason::Failed( + ManagedTargetFailure::runtime(&err), + )); + return; + } + } + let bytes = match socket.send_to(prepared.bytes, remote) { + Ok(bytes) => bytes, + Err(err) => { + let err = ClientError::Socket(err); + target.mark_finished(ManagedTargetEndReason::Failed( + ManagedTargetFailure::runtime(&err), + )); + return; + } + }; + let event = target.runtime.commit_local_close(prepared.commit); + target.schedule = None; + #[cfg(test)] + let bytes = reported_bytes.unwrap_or(bytes); + if let Err(err) = validate_datagram_length(expected_bytes, bytes) { + target.mark_finished(ManagedTargetEndReason::Failed( + ManagedTargetFailure::runtime(&err), + )); + return; + } + events.push(event); + publish_events(hub, target, events); } target.mark_finished(reason); } @@ -2085,7 +2159,8 @@ mod tests { compute_hmac_in_place, echo_packet_len, flags::{self, FLAG_OPEN, FLAG_REPLY}, layout::PacketLayout, - Clock, Params, ReceivedStats, StampAt, TimestampFields, HMAC_SIZE, MAGIC, PROTOCOL_VERSION, + Clock, OpenReply, Params, ReceivedStats, StampAt, TimestampFields, HMAC_SIZE, MAGIC, + PROTOCOL_VERSION, }; use std::sync::mpsc; @@ -2172,10 +2247,13 @@ mod tests { } fn active_target_fixture() -> ActiveTargetFixture { + active_target_fixture_at("127.0.0.1:2112".parse().unwrap()) + } + + fn active_target_fixture_at(remote: SocketAddr) -> ActiveTargetFixture { let interval = Duration::from_millis(10); let params = test_params(None, interval); let client_config = group_config(None, interval, ManagedGroupPacing::Staggered).client; - let remote: SocketAddr = "127.0.0.1:2112".parse().unwrap(); let hub = EventHub::new(); let subscription = hub .subscribe(SubscriberConfig { @@ -2894,6 +2972,7 @@ mod tests { ReceiveMeta::default(), )); assert!(matches!(target.status, TargetStatus::Finished)); + assert!(target.schedule.is_none()); assert_eq!( target.final_reason, Some(ManagedTargetEndReason::PeerClosed) @@ -2959,6 +3038,272 @@ mod tests { assert_eq!(outcome.targets[0].replies_received, 1); } + #[test] + fn managed_close_success_commits_clears_schedule_and_publishes_once() { + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + peer.set_read_timeout(Some(Duration::from_millis(200))) + .unwrap(); + let fixture = active_target_fixture_at(peer.local_addr().unwrap()); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + { + let mut target = fixture.target.lock().expect("target mutex poisoned"); + close_locked_target( + &socket, + &fixture.hub, + &mut target, + ManagedTargetEndReason::TestComplete, + ); + assert!(!target.runtime.is_open()); + assert!(target.schedule.is_none()); + assert_eq!(target.close_send_attempts, 1); + assert_eq!( + target.final_reason, + Some(ManagedTargetEndReason::TestComplete) + ); + } + + let mut packet = [0_u8; 512]; + let (len, _) = peer.recv_from(&mut packet).unwrap(); + assert_eq!(packet[3], flags::FLAG_CLOSE); + assert_eq!(u64::from_le_bytes(packet[4..12].try_into().unwrap()), TOKEN); + assert_eq!(len, 12); + let events = drain_available_group_events(&fixture.subscription); + assert_eq!( + events + .iter() + .filter(|event| matches!( + event, + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::SessionClosed { token: TOKEN, .. }, + .. + }) + )) + .count(), + 1 + ); + } + + #[test] + fn managed_removal_and_cancellation_each_send_one_close() { + for reason in [ + ManagedTargetEndReason::Removed, + ManagedTargetEndReason::Cancelled, + ] { + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + peer.set_read_timeout(Some(Duration::from_millis(200))) + .unwrap(); + let fixture = active_target_fixture_at(peer.local_addr().unwrap()); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + { + let mut target = fixture.target.lock().expect("target mutex poisoned"); + close_locked_target(&socket, &fixture.hub, &mut target, reason.clone()); + assert_eq!(target.final_reason, Some(reason)); + assert_eq!(target.close_send_attempts, 1); + assert!(!target.runtime.is_open()); + assert!(target.schedule.is_none()); + } + + let mut packet = [0_u8; 512]; + let (len, _) = peer.recv_from(&mut packet).unwrap(); + assert_eq!(packet[3], flags::FLAG_CLOSE); + assert_eq!(len, 12); + assert!(matches!( + peer.recv_from(&mut packet), + Err(err) + if matches!( + err.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + ) + )); + } + } + + #[test] + fn managed_close_send_failure_preserves_machine_and_requested_reason_is_not_recorded() { + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + peer.set_nonblocking(true).unwrap(); + let fixture = active_target_fixture_at(peer.local_addr().unwrap()); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + { + let mut target = fixture.target.lock().expect("target mutex poisoned"); + target.close_send_error = true; + close_locked_target( + &socket, + &fixture.hub, + &mut target, + ManagedTargetEndReason::Removed, + ); + assert!(target.runtime.is_open()); + assert!(target.schedule.is_some()); + assert_eq!(target.close_send_attempts, 1); + assert!(matches!( + target.final_reason, + Some(ManagedTargetEndReason::Failed(_)) + )); + } + + let mut packet = [0_u8; 512]; + assert!(matches!( + peer.recv_from(&mut packet), + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock + )); + assert!(!drain_available_group_events(&fixture.subscription) + .iter() + .any(|event| matches!( + event, + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::SessionClosed { .. }, + .. + }) + ))); + } + + #[test] + fn managed_short_close_commits_and_clears_schedule_before_failure() { + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + peer.set_read_timeout(Some(Duration::from_millis(200))) + .unwrap(); + let fixture = active_target_fixture_at(peer.local_addr().unwrap()); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + { + let mut target = fixture.target.lock().expect("target mutex poisoned"); + let expected = target.runtime.prepare_close().unwrap().bytes.len(); + target.close_reported_len = Some(expected - 1); + close_locked_target( + &socket, + &fixture.hub, + &mut target, + ManagedTargetEndReason::Cancelled, + ); + assert!(!target.runtime.is_open()); + assert!(target.schedule.is_none()); + assert_eq!(target.close_send_attempts, 1); + let failure = target + .final_reason + .as_ref() + .and_then(ManagedTargetEndReason::failure) + .expect("short close must use managed failure policy"); + assert_eq!(failure.kind, ManagedTargetFailureKind::RuntimeProtocol); + assert!(failure.message.contains("UDP accepted")); + } + + let mut packet = [0_u8; 512]; + peer.recv_from(&mut packet).unwrap(); + assert!(!drain_available_group_events(&fixture.subscription) + .iter() + .any(|event| matches!( + event, + ManagedGroupEvent::Client(TargetEvent { + event: ClientEvent::SessionClosed { .. }, + .. + }) + ))); + } + + #[test] + fn managed_close_reservation_failure_precedes_send() { + let peer = UdpSocket::bind("127.0.0.1:0").unwrap(); + peer.set_nonblocking(true).unwrap(); + let fixture = active_target_fixture_at(peer.local_addr().unwrap()); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + + { + let mut target = fixture.target.lock().expect("target mutex poisoned"); + target.close_event_reserve_error = true; + close_locked_target( + &socket, + &fixture.hub, + &mut target, + ManagedTargetEndReason::TestComplete, + ); + assert!(target.runtime.is_open()); + assert!(target.schedule.is_some()); + assert_eq!(target.close_send_attempts, 0); + assert!(matches!( + target.final_reason, + Some(ManagedTargetEndReason::Failed(_)) + )); + } + + let mut packet = [0_u8; 512]; + assert!(matches!( + peer.recv_from(&mut packet), + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock + )); + } + + #[test] + fn managed_peer_close_affects_only_the_matching_target() { + let peer = active_target_fixture_at("127.0.0.1:2112".parse().unwrap()); + let other = active_target_fixture_at("127.0.0.1:2113".parse().unwrap()); + + { + let mut target = peer.target.lock().expect("target mutex poisoned"); + assert!(process_active_target_packet( + &peer.hub, + &mut target, + &peer.close_reply, + ClientTimestamp::now(), + ReceiveMeta::default(), + )); + assert!(target.runtime.is_peer_closed()); + assert!(target.schedule.is_none()); + assert_eq!( + target.final_reason, + Some(ManagedTargetEndReason::PeerClosed) + ); + } + + let target = other.target.lock().expect("target mutex poisoned"); + assert!(target.runtime.is_open()); + assert!(matches!(target.status, TargetStatus::Active)); + assert!(target.schedule.is_some()); + assert_eq!(target.final_reason, None); + } + + #[test] + fn managed_non_active_states_do_not_attempt_local_close() { + let remote: SocketAddr = "127.0.0.1:2112".parse().unwrap(); + let mut normal_config = group_config( + None, + Duration::from_millis(10), + ManagedGroupPacing::Staggered, + ) + .client; + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + let hub = EventHub::new(); + let mut opening = + TargetState::new(&normal_config, target("opening", remote), 0, Instant::now()).unwrap(); + close_locked_target(&socket, &hub, &mut opening, ManagedTargetEndReason::Removed); + assert_eq!(opening.close_send_attempts, 0); + + normal_config.run_mode = crate::RunMode::NoTest; + let no_test_params = params_from_config(&normal_config).unwrap(); + let mut no_test = + TargetState::new(&normal_config, target("no-test", remote), 0, Instant::now()).unwrap(); + let reply = OpenReply { + flags: FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, + token: 0, + params: no_test_params, + }; + let prepared = no_test + .runtime + .prepare_open_acceptance(reply, ClientTimestamp::now()) + .unwrap(); + no_test.runtime.commit_open(prepared); + close_locked_target( + &socket, + &hub, + &mut no_test, + ManagedTargetEndReason::Cancelled, + ); + assert_eq!(no_test.close_send_attempts, 0); + } + #[test] fn authoritative_status_counts_only_peer_closed_outcomes() { let remote: SocketAddr = "127.0.0.1:2112".parse().unwrap(); diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 451368b..60a2f86 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -446,15 +446,6 @@ impl SessionMachine { commit.event } - pub(crate) fn close_with(&mut self, send: F) -> Result, ClientError> - where - F: FnOnce(&[u8]) -> Result<(), ClientError>, - { - let prepared = self.prepare_close()?; - send(prepared.bytes)?; - Ok(vec![self.commit_local_close(prepared.commit)]) - } - pub(crate) fn pending_is_empty(&self) -> bool { match &self.state { MachineState::Open(session) => session.pending.len() == 0, -- 2.54.0 From 3948e3dec97f5029014e49d843750dc2c57c6a09 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 13:44:33 +0700 Subject: [PATCH 4/4] fix(client): timestamp close events after send --- crates/irtt-client/src/client.rs | 11 +++- crates/irtt-client/src/client/tests/close.rs | 30 ++++++++++- crates/irtt-client/src/managed/group.rs | 57 +++++++++++++++----- crates/irtt-client/src/session/machine.rs | 45 +++++++++++----- 4 files changed, 114 insertions(+), 29 deletions(-) diff --git a/crates/irtt-client/src/client.rs b/crates/irtt-client/src/client.rs index 1d1c85f..c9e9a99 100644 --- a/crates/irtt-client/src/client.rs +++ b/crates/irtt-client/src/client.rs @@ -60,6 +60,7 @@ struct ClientTestHooks { fail_close_send: Cell, close_reported_len: Cell>, close_send_attempts: Cell, + close_sent_at: Cell>, fail_cleanup_send: Cell, recv_buffer_len_override: Cell>, fail_dscp_restore: Cell, @@ -199,7 +200,15 @@ impl Client { return Err(ClientError::Socket(err)); } }; - let event = self.runtime.commit_local_close(prepared.commit); + #[cfg(not(test))] + let sent_at = ClientTimestamp::now(); + #[cfg(test)] + let sent_at = self + .test_hooks + .close_sent_at + .take() + .unwrap_or_else(ClientTimestamp::now); + let event = self.runtime.commit_local_close(prepared.commit, sent_at); self.schedule = None; self.applied_dscp = None; diff --git a/crates/irtt-client/src/client/tests/close.rs b/crates/irtt-client/src/client/tests/close.rs index 656692f..79813cb 100644 --- a/crates/irtt-client/src/client/tests/close.rs +++ b/crates/irtt-client/src/client/tests/close.rs @@ -29,10 +29,19 @@ fn close_sends_one_close_packet_with_negotiated_token() { }); let mut client = Client::connect(default_test_config(server.addr)).unwrap(); assert_open_started(client.open().unwrap()); + let sent_at = ClientTimestamp { + wall: SystemTime::UNIX_EPOCH + Duration::from_secs(1_234), + mono: Instant::now() + Duration::from_secs(5), + }; + client.test_hooks.close_sent_at.set(Some(sent_at)); let events = client.close().unwrap(); assert!(matches!( events.as_slice(), - [ClientEvent::SessionClosed { token: TOKEN, .. }] + [ClientEvent::SessionClosed { + token: TOKEN, + at, + .. + }] if *at == sent_at )); assert!(!client.runtime.is_open()); assert!(client.schedule.is_none()); @@ -127,9 +136,15 @@ fn close_send_failure_leaves_machine_and_schedule_open_for_retry() { }); let mut client = Client::connect(default_test_config(server.addr)).unwrap(); assert_open_started(client.open().unwrap()); + let sent_at = ClientTimestamp { + wall: SystemTime::UNIX_EPOCH + Duration::from_secs(2_345), + mono: Instant::now() + Duration::from_secs(6), + }; + client.test_hooks.close_sent_at.set(Some(sent_at)); client.test_hooks.fail_close_send.set(true); assert!(matches!(client.close(), Err(ClientError::Socket(_)))); + assert_eq!(client.test_hooks.close_sent_at.get(), Some(sent_at)); assert!(client.runtime.is_open()); assert!(client.schedule.is_some()); assert_eq!(client.applied_dscp, Some(0)); @@ -137,8 +152,13 @@ fn close_send_failure_leaves_machine_and_schedule_open_for_retry() { assert!(matches!( client.close().unwrap().as_slice(), - [ClientEvent::SessionClosed { token: TOKEN, .. }] + [ClientEvent::SessionClosed { + token: TOKEN, + at, + .. + }] if *at == sent_at )); + assert_eq!(client.test_hooks.close_sent_at.get(), None); assert_eq!(client.test_hooks.close_send_attempts.get(), 2); assert!(matches!(client.close(), Err(ClientError::AlreadyClosed))); server.join(); @@ -160,6 +180,11 @@ fn short_successful_close_commits_before_length_mismatch() { let mut client = Client::connect(default_test_config(server.addr)).unwrap(); assert_open_started(client.open().unwrap()); let expected = client.runtime.prepare_close().unwrap().bytes.len(); + let sent_at = ClientTimestamp { + wall: SystemTime::UNIX_EPOCH + Duration::from_secs(3_456), + mono: Instant::now() + Duration::from_secs(7), + }; + client.test_hooks.close_sent_at.set(Some(sent_at)); client.test_hooks.close_reported_len.set(Some(expected - 1)); assert!(matches!( @@ -172,6 +197,7 @@ fn short_successful_close_commits_before_length_mismatch() { assert!(!client.runtime.is_open()); assert!(client.schedule.is_none()); assert_eq!(client.applied_dscp, None); + assert_eq!(client.test_hooks.close_sent_at.get(), None); assert!(matches!(client.close(), Err(ClientError::AlreadyClosed))); server.join(); } diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index 9d1145e..0e74ab9 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -733,6 +733,8 @@ struct TargetState { close_send_attempts: usize, #[cfg(test)] close_event_reserve_error: bool, + #[cfg(test)] + close_sent_at: Option, } type ScheduledTarget = (Arc>, Instant); @@ -855,6 +857,8 @@ impl TargetState { close_send_attempts: 0, #[cfg(test)] close_event_reserve_error: false, + #[cfg(test)] + close_sent_at: None, }) } @@ -1872,7 +1876,14 @@ fn close_locked_target( return; } }; - let event = target.runtime.commit_local_close(prepared.commit); + #[cfg(not(test))] + let sent_at = ClientTimestamp::now(); + #[cfg(test)] + let sent_at = target + .close_sent_at + .take() + .unwrap_or_else(ClientTimestamp::now); + let event = target.runtime.commit_local_close(prepared.commit, sent_at); target.schedule = None; #[cfg(test)] let bytes = reported_bytes.unwrap_or(bytes); @@ -3046,8 +3057,13 @@ mod tests { let fixture = active_target_fixture_at(peer.local_addr().unwrap()); let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + let sent_at = ClientTimestamp { + wall: std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(4_567), + mono: Instant::now() + Duration::from_secs(8), + }; { let mut target = fixture.target.lock().expect("target mutex poisoned"); + target.close_sent_at = Some(sent_at); close_locked_target( &socket, &fixture.hub, @@ -3069,19 +3085,20 @@ mod tests { assert_eq!(u64::from_le_bytes(packet[4..12].try_into().unwrap()), TOKEN); assert_eq!(len, 12); let events = drain_available_group_events(&fixture.subscription); - assert_eq!( - events - .iter() - .filter(|event| matches!( - event, - ManagedGroupEvent::Client(TargetEvent { - event: ClientEvent::SessionClosed { token: TOKEN, .. }, - .. - }) - )) - .count(), - 1 - ); + let closes: Vec<_> = events + .iter() + .filter_map(|event| match event { + ManagedGroupEvent::Client(TargetEvent { + event: + ClientEvent::SessionClosed { + token: TOKEN, at, .. + }, + .. + }) => Some(*at), + _ => None, + }) + .collect(); + assert_eq!(closes, [sent_at]); } #[test] @@ -3127,8 +3144,13 @@ mod tests { let fixture = active_target_fixture_at(peer.local_addr().unwrap()); let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + let sent_at = ClientTimestamp { + wall: std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(5_678), + mono: Instant::now() + Duration::from_secs(9), + }; { let mut target = fixture.target.lock().expect("target mutex poisoned"); + target.close_sent_at = Some(sent_at); target.close_send_error = true; close_locked_target( &socket, @@ -3139,6 +3161,7 @@ mod tests { assert!(target.runtime.is_open()); assert!(target.schedule.is_some()); assert_eq!(target.close_send_attempts, 1); + assert_eq!(target.close_sent_at, Some(sent_at)); assert!(matches!( target.final_reason, Some(ManagedTargetEndReason::Failed(_)) @@ -3169,9 +3192,14 @@ mod tests { let fixture = active_target_fixture_at(peer.local_addr().unwrap()); let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + let sent_at = ClientTimestamp { + wall: std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(6_789), + mono: Instant::now() + Duration::from_secs(10), + }; { let mut target = fixture.target.lock().expect("target mutex poisoned"); let expected = target.runtime.prepare_close().unwrap().bytes.len(); + target.close_sent_at = Some(sent_at); target.close_reported_len = Some(expected - 1); close_locked_target( &socket, @@ -3182,6 +3210,7 @@ mod tests { assert!(!target.runtime.is_open()); assert!(target.schedule.is_none()); assert_eq!(target.close_send_attempts, 1); + assert_eq!(target.close_sent_at, None); let failure = target .final_reason .as_ref() diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 60a2f86..5072ced 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -115,7 +115,7 @@ pub(crate) struct PreparedClose<'a> { #[derive(Debug)] pub(crate) struct CloseCommit { packets_sent: u64, - event: ClientEvent, + token: u64, } #[derive(Debug)] @@ -425,16 +425,16 @@ impl SessionMachine { bytes: &session.local_close_packet, commit: CloseCommit { packets_sent: session.packets_sent, - event: ClientEvent::SessionClosed { - remote: self.remote, - token: session.token, - at: ClientTimestamp::now(), - }, + token: session.token, }, }) } - pub(crate) fn commit_local_close(&mut self, commit: CloseCommit) -> ClientEvent { + pub(crate) fn commit_local_close( + &mut self, + commit: CloseCommit, + sent_at: ClientTimestamp, + ) -> ClientEvent { debug_assert!( matches!(self.state, MachineState::Open(_)), "local close commits only from open state" @@ -443,7 +443,11 @@ impl SessionMachine { source: CloseSource::Local, packets_sent: commit.packets_sent, }; - commit.event + ClientEvent::SessionClosed { + remote: self.remote, + token: commit.token, + at: sent_at, + } } pub(crate) fn pending_is_empty(&self) -> bool { @@ -1370,9 +1374,8 @@ mod tests { } #[test] - fn prepared_close_is_stable_inert_and_commits_preserved_state() { + fn prepare_close_is_stable_inert_and_does_not_determine_event_timestamp() { let mut machine = open_machine(4, Duration::from_secs(1)); - let remote = machine.remote; active_mut(&mut machine).packets_sent = 7; let first_bytes = { @@ -1387,6 +1390,8 @@ mod tests { ) .unwrap() ); + assert_eq!(prepared.commit.packets_sent, 7); + assert_eq!(prepared.commit.token, 0x0102_0304_0506_0708); prepared.bytes.to_vec() }; assert!(machine.is_open()); @@ -1394,15 +1399,31 @@ mod tests { let prepared = machine.prepare_close().unwrap(); assert_eq!(prepared.bytes, first_bytes); - let event = machine.commit_local_close(prepared.commit); + assert_eq!(prepared.commit.packets_sent, 7); + assert_eq!(prepared.commit.token, 0x0102_0304_0506_0708); + assert!(machine.is_open()); + } + + #[test] + fn local_close_commit_uses_supplied_exact_timestamp() { + let mut machine = open_machine(4, Duration::from_secs(1)); + let remote = machine.remote; + active_mut(&mut machine).packets_sent = 7; + let prepared = machine.prepare_close().unwrap(); + let sent_at = ClientTimestamp { + wall: UNIX_EPOCH + Duration::from_secs(1_234), + mono: Instant::now() + Duration::from_secs(5), + }; + let event = machine.commit_local_close(prepared.commit, sent_at); assert!(matches!( event, ClientEvent::SessionClosed { remote: event_remote, token: 0x0102_0304_0506_0708, - .. + at, } if event_remote == remote + && at == sent_at )); assert!(matches!( machine.state, -- 2.54.0