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

@@ -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,
+        peer_close_run_error, request_group_stop_for_peer_close, request_group_stop_once,
-        should_stop_group_on_peer_close,
+        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,19 +432,21 @@ 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,
+                            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();
                         }
                     }
                 }
             }
-            }
             Ok(None) => {
                 if interrupted {
                     break;

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,18 +223,20 @@ 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,
+                            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();
                         }
                     }
                 }
-                }
                 render_if_due(terminal, state, next_render, false)?;
             }
             Ok(None) => {
@@ -553,21 +566,41 @@ mod tests {
     }
 
     #[test]
-    fn continuous_group_peer_close_requests_stop_unless_interrupted() {
+    fn continuous_group_reliable_peer_close_status_requests_stop_once() {
-        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(
             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,
+            1,
-            &irtt_client::ManagedTargetEndReason::PeerClosed
+            &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,
         ));
     }
 

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() {
+    fn grouped_peer_close_stop_policy_is_run_mode_and_one_shot() {
-        assert!(should_stop_group_on_peer_close(
+        let mut peer_close_requested_stop = false;
-            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
-        ));
-
         let mut stop_requested = false;
-        assert!(request_group_stop_once(&mut stop_requested));
+        assert!(!request_group_stop_for_peer_close(
-        assert!(!request_group_stop_once(&mut stop_requested));
+            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,
+        ));
     }
 }

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<ManagedGroupEvent>,
     control_tx: mpsc::Sender<ControlMessage>,
     cancellation: CancellationToken,
+    peer_closed_target_count: Arc<AtomicU64>,
     scheduler: Option<JoinHandle<Result<ManagedGroupOutcome, ClientError>>>,
     receiver: Option<JoinHandle<Result<(), ClientError>>>,
 }
@@ -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<ManagedGroupEvent>,
     cancellation: CancellationToken,
     control_tx: mpsc::Sender<ControlMessage>,
+    peer_closed_target_count: Arc<AtomicU64>,
     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)
             );
         }
-
+        assert_eq!(shared.peer_closed_target_count.load(Ordering::Acquire), 0);
-        let shared = shared_with_target(
-            &fixture.client_config,
-            fixture.hub.clone(),
-            fixture.target.clone(),
-        );
         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();