From ef19c45c1f1cfc9186cbf4253e8f345e261a64fd Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 4 Aug 2026 13:08:38 +0700 Subject: [PATCH] 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,