From 7cc55bd619f718157ad85aa53a3e8ae66189dcd8 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 28 Jul 2026 15:49:53 +0700 Subject: [PATCH] fix(managed): decouple peer-close control from event delivery Add a scheduler-owned saturating peer-close counter shared with the group session. Update it before best-effort TargetFinished publication so bounded subscriber overflow cannot hide the live shutdown condition, while retaining the existing events, bounded history, and aggregate accounting. Make grouped CLI and TUI supervision consult the durable count on each existing activity loop and funnel both count and event observations through one-shot stop policy. Preserve interruption suppression, queued-event draining, structured failures, drop reporting, and peer-close diagnostic precedence. Cover scheduler-only accounting, non-peer exclusions, receiver and removal races, deterministic capacity-one DropOldest eviction with an active target, frontend one-shot policy, and the existing process-level peer-close and interruption behavior. --- crates/irtt-cli/src/cmd/client/run.rs | 27 ++- crates/irtt-cli/src/cmd/tui/run.rs | 63 ++++-- crates/irtt-cli/src/shared/client/session.rs | 80 +++++--- crates/irtt-client/src/managed/group.rs | 205 ++++++++++++++++++- 4 files changed, 316 insertions(+), 59 deletions(-) diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs index 4d47efb..bbdfae7 100644 --- a/crates/irtt-cli/src/cmd/client/run.rs +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -26,8 +26,8 @@ use crate::shared::client::expected_probe_count; use crate::shared::client::{ is_shutdown_requested, session::{ - peer_close_run_error, request_group_stop_once, should_print_final_summary, - should_stop_group_on_peer_close, + peer_close_run_error, request_group_stop_for_peer_close, request_group_stop_once, + should_print_final_summary, }, }; @@ -396,6 +396,15 @@ fn run_group_stream( session.stop(); } } + if request_group_stop_for_peer_close( + continuous, + interrupted, + session.peer_closed_target_count(), + &mut peer_close_requested_stop, + &mut stop_requested, + ) { + session.stop(); + } match events.try_recv() { Ok(Some(group_event)) => { @@ -423,15 +432,17 @@ fn run_group_stream( ManagedGroupEvent::TargetFinished(target) => { report_target_failure(&target); terminal_targets.insert(target.id.as_str().to_owned()); - if should_stop_group_on_peer_close( + if request_group_stop_for_peer_close( continuous, interrupted, - &target.end_reason, + u64::from(matches!( + &target.end_reason, + irtt_client::ManagedTargetEndReason::PeerClosed + )), + &mut peer_close_requested_stop, + &mut stop_requested, ) { - peer_close_requested_stop = true; - if request_group_stop_once(&mut stop_requested) { - session.stop(); - } + session.stop(); } } } diff --git a/crates/irtt-cli/src/cmd/tui/run.rs b/crates/irtt-cli/src/cmd/tui/run.rs index 99d8b77..6846406 100644 --- a/crates/irtt-cli/src/cmd/tui/run.rs +++ b/crates/irtt-cli/src/cmd/tui/run.rs @@ -16,7 +16,9 @@ use crate::{ cmd::tui::args::{ResolvedTuiTarget, TuiArgs}, shared::client::{ is_shutdown_requested, - session::{peer_close_run_error, request_group_stop_once, should_stop_group_on_peer_close}, + session::{ + peer_close_run_error, request_group_stop_for_peer_close, request_group_stop_once, + }, }, }; @@ -200,6 +202,15 @@ fn run_group_tui( session.stop(); } } + if request_group_stop_for_peer_close( + args.is_continuous(), + interrupted, + session.peer_closed_target_count(), + &mut peer_close_requested_stop, + &mut stop_requested, + ) { + session.stop(); + } match events.try_recv() { Ok(Some(group_event)) => { @@ -212,15 +223,17 @@ fn run_group_tui( ManagedGroupEvent::TargetFinished(target) => { terminal_targets.insert(target.id.as_str().to_owned()); state.process_target_outcome(&target); - if should_stop_group_on_peer_close( + if request_group_stop_for_peer_close( args.is_continuous(), interrupted, - &target.end_reason, + u64::from(matches!( + &target.end_reason, + irtt_client::ManagedTargetEndReason::PeerClosed + )), + &mut peer_close_requested_stop, + &mut stop_requested, ) { - peer_close_requested_stop = true; - if request_group_stop_once(&mut stop_requested) { - session.stop(); - } + session.stop(); } } } @@ -553,21 +566,41 @@ mod tests { } #[test] - fn continuous_group_peer_close_requests_stop_unless_interrupted() { - assert!(should_stop_group_on_peer_close( + fn continuous_group_reliable_peer_close_status_requests_stop_once() { + let mut peer_close_requested_stop = false; + let mut stop_requested = false; + assert!(request_group_stop_for_peer_close( true, false, - &irtt_client::ManagedTargetEndReason::PeerClosed + 1, + &mut peer_close_requested_stop, + &mut stop_requested, )); - assert!(!should_stop_group_on_peer_close( + assert!(peer_close_requested_stop); + assert!(stop_requested); + assert!(!request_group_stop_for_peer_close( + true, false, - false, - &irtt_client::ManagedTargetEndReason::PeerClosed + 1, + &mut peer_close_requested_stop, + &mut stop_requested, )); - assert!(!should_stop_group_on_peer_close( + + let mut peer_close_requested_stop = false; + let mut stop_requested = false; + assert!(!request_group_stop_for_peer_close( + false, + false, + 1, + &mut peer_close_requested_stop, + &mut stop_requested, + )); + assert!(!request_group_stop_for_peer_close( true, true, - &irtt_client::ManagedTargetEndReason::PeerClosed + 1, + &mut peer_close_requested_stop, + &mut stop_requested, )); } diff --git a/crates/irtt-cli/src/shared/client/session.rs b/crates/irtt-cli/src/shared/client/session.rs index 79bce61..6792bf2 100644 --- a/crates/irtt-cli/src/shared/client/session.rs +++ b/crates/irtt-cli/src/shared/client/session.rs @@ -1,7 +1,5 @@ use std::sync::atomic::{AtomicBool, Ordering}; -use irtt_client::ManagedTargetEndReason; - pub fn is_shutdown_requested(shutdown_requested: &AtomicBool) -> bool { shutdown_requested.load(Ordering::Relaxed) } @@ -29,12 +27,18 @@ pub fn peer_close_run_error( )) } -pub fn should_stop_group_on_peer_close( +pub fn request_group_stop_for_peer_close( continuous: bool, interrupted: bool, - end_reason: &ManagedTargetEndReason, + peer_closed_target_count: u64, + peer_close_requested_stop: &mut bool, + stop_requested: &mut bool, ) -> bool { - continuous && !interrupted && matches!(end_reason, ManagedTargetEndReason::PeerClosed) + if !continuous || interrupted || peer_closed_target_count == 0 { + return false; + } + *peer_close_requested_stop = true; + request_group_stop_once(stop_requested) } pub fn request_group_stop_once(stop_requested: &mut bool) -> bool { @@ -65,30 +69,48 @@ mod tests { } #[test] - fn grouped_peer_close_stop_policy_is_run_mode_and_interruption_aware() { - assert!(should_stop_group_on_peer_close( - true, - false, - &ManagedTargetEndReason::PeerClosed - )); - assert!(!should_stop_group_on_peer_close( - false, - false, - &ManagedTargetEndReason::PeerClosed - )); - assert!(!should_stop_group_on_peer_close( - true, - true, - &ManagedTargetEndReason::PeerClosed - )); - assert!(!should_stop_group_on_peer_close( - true, - false, - &ManagedTargetEndReason::TestComplete - )); - + fn grouped_peer_close_stop_policy_is_run_mode_and_one_shot() { + let mut peer_close_requested_stop = false; let mut stop_requested = false; - assert!(request_group_stop_once(&mut stop_requested)); - assert!(!request_group_stop_once(&mut stop_requested)); + assert!(!request_group_stop_for_peer_close( + false, + false, + 1, + &mut peer_close_requested_stop, + &mut stop_requested, + )); + assert!(!request_group_stop_for_peer_close( + true, + true, + 1, + &mut peer_close_requested_stop, + &mut stop_requested, + )); + assert!(!request_group_stop_for_peer_close( + true, + false, + 0, + &mut peer_close_requested_stop, + &mut stop_requested, + )); + assert!(!peer_close_requested_stop); + assert!(!stop_requested); + + assert!(request_group_stop_for_peer_close( + true, + false, + 1, + &mut peer_close_requested_stop, + &mut stop_requested, + )); + assert!(peer_close_requested_stop); + assert!(stop_requested); + assert!(!request_group_stop_for_peer_close( + true, + false, + 1, + &mut peer_close_requested_stop, + &mut stop_requested, + )); } } diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index a634ee1..21315c6 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -3,7 +3,10 @@ use std::{ fmt, hash::{Hash, Hasher}, net::{SocketAddr, UdpSocket}, - sync::{mpsc, Arc, Mutex}, + sync::{ + atomic::{AtomicU64, Ordering}, + mpsc, Arc, Mutex, + }, thread::{self, JoinHandle}, time::{Duration, Instant}, }; @@ -193,6 +196,7 @@ pub struct ManagedClientGroupSession { hub: EventHub, control_tx: mpsc::Sender, cancellation: CancellationToken, + peer_closed_target_count: Arc, scheduler: Option>>, receiver: Option>>, } @@ -453,12 +457,14 @@ impl ManagedClientGroup { .transpose()?; let cancellation = CancellationToken::new(); + let peer_closed_target_count = Arc::new(AtomicU64::new(0)); let (control_tx, control_rx) = mpsc::channel(); let shared = Arc::new(GroupShared { registry: Mutex::new(registry), hub: hub.clone(), cancellation: cancellation.clone(), control_tx: control_tx.clone(), + peer_closed_target_count: peer_closed_target_count.clone(), requested_interval_ns: requested.interval_ns, requested_dscp: requested.dscp, family_remote, @@ -489,6 +495,7 @@ impl ManagedClientGroup { hub, control_tx, cancellation, + peer_closed_target_count, scheduler: Some(scheduler), receiver: Some(receiver), }, @@ -530,6 +537,16 @@ impl ManagedClientGroupSession { let _ = self.control_tx.send(ControlMessage::Wake); } + /// Return the number of scheduler-recorded peer-closed target incarnations. + /// + /// The count is monotonic and saturating for the lifetime of this group. + /// It becomes visible before the corresponding lossy + /// [`ManagedGroupEvent::TargetFinished`] publication and includes outcomes + /// no longer retained in the bounded recent outcome history. + pub fn peer_closed_target_count(&self) -> u64 { + self.peer_closed_target_count.load(Ordering::Acquire) + } + /// Wait for the scheduler and receive threads to finish. /// /// The returned outcome contains aggregate counts plus a bounded recent @@ -578,6 +595,7 @@ struct GroupShared { hub: EventHub, cancellation: CancellationToken, control_tx: mpsc::Sender, + peer_closed_target_count: Arc, requested_interval_ns: i64, requested_dscp: i64, family_remote: SocketAddr, @@ -1364,6 +1382,13 @@ fn record_target_outcome( records: &mut TargetOutcomeHistory, outcome: ManagedTargetOutcome, ) { + if matches!(&outcome.end_reason, ManagedTargetEndReason::PeerClosed) { + let _ = shared.peer_closed_target_count.fetch_update( + Ordering::Release, + Ordering::Relaxed, + |count| Some(count.saturating_add(1)), + ); + } shared .hub .publish(ManagedGroupEvent::TargetFinished(outcome.clone())); @@ -1842,6 +1867,7 @@ mod tests { hub, cancellation: CancellationToken::new(), control_tx, + peer_closed_target_count: Arc::new(AtomicU64::new(0)), requested_interval_ns: requested.interval_ns, requested_dscp: requested.dscp, family_remote: remote, @@ -1866,6 +1892,23 @@ mod tests { } } + fn target_outcome( + id: &str, + remote: SocketAddr, + end_reason: ManagedTargetEndReason, + ) -> ManagedTargetOutcome { + ManagedTargetOutcome { + id: TargetId::from(id), + remote, + end_reason, + packets_sent: 0, + replies_received: 0, + duplicates: 0, + late: 0, + warning_events: 0, + } + } + fn start_echo_server(params: Params, open_delay: Duration) -> FakeServer { let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); let addr = socket.local_addr().unwrap(); @@ -2307,6 +2350,11 @@ mod tests { #[test] fn receiver_peer_close_wins_send_and_removal_races() { let fixture = active_target_fixture(); + let shared = shared_with_target( + &fixture.client_config, + fixture.hub.clone(), + fixture.target.clone(), + ); { let mut target = fixture.target.lock().expect("target mutex poisoned"); assert!(process_active_target_packet( @@ -2322,12 +2370,7 @@ mod tests { Some(ManagedTargetEndReason::PeerClosed) ); } - - let shared = shared_with_target( - &fixture.client_config, - fixture.hub.clone(), - fixture.target.clone(), - ); + assert_eq!(shared.peer_closed_target_count.load(Ordering::Acquire), 0); let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); send_echo_to_target(&socket, &shared, fixture.target.clone(), Instant::now()); @@ -2339,6 +2382,7 @@ mod tests { .final_reason, Some(ManagedTargetEndReason::PeerClosed) ); + assert_eq!(shared.peer_closed_target_count.load(Ordering::Acquire), 0); let mut next_order = 1; let mut records = TargetOutcomeHistory::default(); @@ -2351,7 +2395,9 @@ mod tests { &mut records, ) .unwrap(); + assert_eq!(shared.peer_closed_target_count.load(Ordering::Acquire), 1); collect_finished_targets(&shared, &mut records); + assert_eq!(shared.peer_closed_target_count.load(Ordering::Acquire), 1); let events = drain_available_group_events(&fixture.subscription); let terminal_sequence: Vec<&str> = events @@ -2384,6 +2430,151 @@ mod tests { assert_eq!(outcome.targets[0].replies_received, 1); } + #[test] + fn authoritative_status_counts_only_peer_closed_outcomes() { + let remote: SocketAddr = "127.0.0.1:2112".parse().unwrap(); + let config = group_config( + None, + Duration::from_millis(10), + ManagedGroupPacing::Staggered, + ) + .client; + let target = Arc::new(Mutex::new( + TargetState::new(&config, target("active", remote), 0, Instant::now()).unwrap(), + )); + let hub = EventHub::new(); + let shared = shared_with_target(&config, hub.clone(), target); + let mut records = TargetOutcomeHistory::default(); + + hub.publish(ManagedGroupEvent::Client(TargetEvent { + target: TargetId::from("active"), + event: ClientEvent::EchoLoss { + seq: 0, + sent_at: ClientTimestamp::now(), + timeout_at: Instant::now(), + }, + })); + assert_eq!(shared.peer_closed_target_count.load(Ordering::Acquire), 0); + + let failure = ManagedTargetFailure { + kind: ManagedTargetFailureKind::RuntimeProtocol, + message: "test failure".to_owned(), + }; + for (index, reason) in [ + ManagedTargetEndReason::TestComplete, + ManagedTargetEndReason::NoTestComplete, + ManagedTargetEndReason::Removed, + ManagedTargetEndReason::Cancelled, + ManagedTargetEndReason::OpenFailed(failure.clone()), + ManagedTargetEndReason::Failed(failure), + ] + .into_iter() + .enumerate() + { + record_target_outcome( + &shared, + &mut records, + target_outcome(&format!("non-peer-{index}"), remote, reason), + ); + assert_eq!(shared.peer_closed_target_count.load(Ordering::Acquire), 0); + } + + record_target_outcome( + &shared, + &mut records, + target_outcome("peer", remote, ManagedTargetEndReason::PeerClosed), + ); + assert_eq!(shared.peer_closed_target_count.load(Ordering::Acquire), 1); + + let outcome = records.into_group_outcome(ManagedGroupEndReason::AllTargetsComplete); + assert_eq!(outcome.total_target_outcomes, 7); + assert_eq!(outcome.peer_closed_target_outcomes, 1); + } + + #[test] + fn drop_oldest_can_evict_peer_close_event_without_status_loss() { + let interval = Duration::from_millis(10); + let config = group_config(None, interval, ManagedGroupPacing::Staggered).client; + let peer_remote: SocketAddr = "127.0.0.1:2112".parse().unwrap(); + let healthy_remote: SocketAddr = "127.0.0.1:2113".parse().unwrap(); + let peer = Arc::new(Mutex::new( + TargetState::new(&config, target("peer", peer_remote), 0, Instant::now()).unwrap(), + )); + peer.lock() + .expect("target mutex poisoned") + .mark_finished(ManagedTargetEndReason::PeerClosed); + let healthy = Arc::new(Mutex::new( + TargetState::new( + &config, + target("healthy", healthy_remote), + 1, + Instant::now(), + ) + .unwrap(), + )); + healthy.lock().expect("target mutex poisoned").status = TargetStatus::Active; + + let hub = EventHub::new(); + let subscription = hub + .subscribe(SubscriberConfig { + capacity: 1, + overflow: SubscriberOverflow::DropOldest, + }) + .unwrap(); + let shared = shared_with_target(&config, hub.clone(), peer); + { + let mut registry = shared + .registry + .lock() + .expect("target registry mutex poisoned"); + registry.desired.insert(TargetId::from("healthy")); + registry + .remotes + .insert(healthy_remote, TargetId::from("healthy")); + registry + .targets + .insert(TargetId::from("healthy"), healthy.clone()); + } + + let mut records = TargetOutcomeHistory::default(); + collect_finished_targets(&shared, &mut records); + assert_eq!(shared.peer_closed_target_count.load(Ordering::Acquire), 1); + assert!(matches!( + healthy.lock().expect("target mutex poisoned").status, + TargetStatus::Active + )); + + hub.publish(ManagedGroupEvent::Client(TargetEvent { + target: TargetId::from("healthy"), + event: ClientEvent::EchoLoss { + seq: 7, + sent_at: ClientTimestamp::now(), + timeout_at: Instant::now(), + }, + })); + + assert_eq!(subscription.dropped_events(), 1); + assert!(matches!( + subscription.try_recv(), + Ok(Some(ManagedGroupEvent::Client(TargetEvent { + target, + event: ClientEvent::EchoLoss { seq: 7, .. }, + }))) if target.as_str() == "healthy" + )); + assert_eq!(subscription.try_recv(), Ok(None)); + assert_eq!(shared.peer_closed_target_count.load(Ordering::Acquire), 1); + assert_eq!(records.peer_closed, 1); + assert_eq!( + shared + .registry + .lock() + .expect("target registry mutex poisoned") + .targets + .len(), + 1 + ); + } + #[test] fn receiver_non_close_reply_leaves_target_active() { let fixture = active_target_fixture();