From 5381cfc992f0ef68b42c0af7b3b894944d60478f Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 28 Jul 2026 00:18:16 +0700 Subject: [PATCH] fix(managed): bound completed target retention Retain only the 256 most recent target outcomes in the final group snapshot while tracking total, successful, failed, and discarded outcome counts. Publish each authoritative TargetFinished event before inserting into the bounded history. Update CLI/TUI exit decisions to use aggregate counts and add a dynamic churn regression proving every outcome remains observable while join retention stays bounded. --- crates/irtt-cli/src/cmd/client/run.rs | 12 +-- crates/irtt-cli/src/cmd/tui/run.rs | 12 +-- crates/irtt-client/src/lib.rs | 1 + crates/irtt-client/src/managed/group.rs | 137 ++++++++++++++++++++---- crates/irtt-client/src/managed/mod.rs | 1 + 5 files changed, 125 insertions(+), 38 deletions(-) diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs index bce5523..3af8947 100644 --- a/crates/irtt-cli/src/cmd/client/run.rs +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -408,16 +408,8 @@ fn run_group_stream( report_target_failure(target); } } - let successful_targets = outcome - .targets - .iter() - .filter(|target| target.is_success()) - .count(); - let failed_targets = outcome - .targets - .iter() - .filter(|target| target.end_reason.failure().is_some()) - .count(); + let successful_targets = outcome.successful_target_outcomes; + let failed_targets = outcome.failed_target_outcomes; if !interrupted && successful_targets == 0 && failed_targets > 0 { return Err( format!("no managed target completed successfully ({failed_targets} failed)").into(), diff --git a/crates/irtt-cli/src/cmd/tui/run.rs b/crates/irtt-cli/src/cmd/tui/run.rs index b40d112..bc53bea 100644 --- a/crates/irtt-cli/src/cmd/tui/run.rs +++ b/crates/irtt-cli/src/cmd/tui/run.rs @@ -261,16 +261,8 @@ fn run_group_tui( } }; } - let successful_targets = outcome - .targets - .iter() - .filter(|target| target.is_success()) - .count(); - let failed_targets = outcome - .targets - .iter() - .filter(|target| target.end_reason.failure().is_some()) - .count(); + let successful_targets = outcome.successful_target_outcomes; + let failed_targets = outcome.failed_target_outcomes; if !interrupted && successful_targets == 0 && failed_targets > 0 { state.set_status(TuiStatus::Error); render_if_due(terminal, state, next_render, true)?; diff --git a/crates/irtt-client/src/lib.rs b/crates/irtt-client/src/lib.rs index a302407..cbe71b6 100644 --- a/crates/irtt-client/src/lib.rs +++ b/crates/irtt-client/src/lib.rs @@ -82,6 +82,7 @@ pub use managed::{ ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetOutcome, SessionEndReason, SessionOutcome, SubscriberConfig, SubscriberOverflow, TargetEvent, TargetEventSubscription, TargetId, + MANAGED_GROUP_OUTCOME_HISTORY_LIMIT, }; pub use session::{NegotiatedParams, NegotiationRestriction}; pub use timing::ClientTimestamp; diff --git a/crates/irtt-client/src/managed/group.rs b/crates/irtt-client/src/managed/group.rs index 16565e6..3734983 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -1,5 +1,5 @@ use std::{ - collections::{HashMap, HashSet}, + collections::{HashMap, HashSet, VecDeque}, fmt, hash::{Hash, Hasher}, net::{SocketAddr, UdpSocket}, @@ -29,6 +29,8 @@ const GROUP_FINAL_DRAIN: Duration = Duration::from_millis(100); const IDLE_SLEEP: Duration = Duration::from_millis(1); const MAX_SLEEP: Duration = Duration::from_millis(20); const RECV_BUFFER_SIZE: usize = 65_536; +/// Maximum number of recent target outcomes retained in [`ManagedGroupOutcome`]. +pub const MANAGED_GROUP_OUTCOME_HISTORY_LIMIT: usize = 256; /// Caller-owned target identity for managed multi-target probing. #[derive(Clone, Eq)] @@ -193,8 +195,21 @@ pub struct ManagedClientGroupSession { pub struct ManagedGroupOutcome { /// Why the group scheduler stopped. pub end_reason: ManagedGroupEndReason, - /// Per-target lifecycle records. These are not RTT/loss summaries. + /// Recent per-target lifecycle records in completion order. + /// + /// This snapshot retains at most [`MANAGED_GROUP_OUTCOME_HISTORY_LIMIT`] + /// entries. Every outcome is published as + /// [`ManagedGroupEvent::TargetFinished`] before an older snapshot entry can + /// be evicted. pub targets: Vec, + /// Total number of target incarnations that reached a terminal outcome. + pub total_target_outcomes: u64, + /// Number of target outcomes classified as successful. + pub successful_target_outcomes: u64, + /// Number of target outcomes carrying structured failure details. + pub failed_target_outcomes: u64, + /// Number of older outcomes omitted from [`Self::targets`]. + pub discarded_target_outcomes: u64, } /// Reason the managed group stopped. @@ -595,6 +610,43 @@ struct TargetCounters { warning_events: u64, } +#[derive(Debug, Default)] +struct TargetOutcomeHistory { + recent: VecDeque, + total: u64, + successful: u64, + failed: u64, + discarded: u64, +} + +impl TargetOutcomeHistory { + fn record(&mut self, outcome: ManagedTargetOutcome) { + self.total = self.total.saturating_add(1); + if outcome.is_success() { + self.successful = self.successful.saturating_add(1); + } + if outcome.end_reason.failure().is_some() { + self.failed = self.failed.saturating_add(1); + } + if self.recent.len() == MANAGED_GROUP_OUTCOME_HISTORY_LIMIT { + self.recent.pop_front(); + self.discarded = self.discarded.saturating_add(1); + } + self.recent.push_back(outcome); + } + + fn into_group_outcome(self, end_reason: ManagedGroupEndReason) -> ManagedGroupOutcome { + ManagedGroupOutcome { + end_reason, + targets: self.recent.into_iter().collect(), + total_target_outcomes: self.total, + successful_target_outcomes: self.successful, + failed_target_outcomes: self.failed, + discarded_target_outcomes: self.discarded, + } + } +} + impl TargetState { fn new( group_config: &ClientConfig, @@ -667,7 +719,7 @@ fn run_group_scheduler( control_rx: mpsc::Receiver, mut next_order: u64, ) -> Result { - let mut records = Vec::new(); + let mut records = TargetOutcomeHistory::default(); let mut pacing = PacingRuntime::new(config.pacing); let mut pending_control = None; @@ -684,10 +736,7 @@ fn run_group_scheduler( if shared.cancellation.is_cancelled() { cancel_remaining_targets(&socket, &shared, &mut records); - return Ok(ManagedGroupOutcome { - end_reason: ManagedGroupEndReason::Cancelled, - targets: records, - }); + return Ok(records.into_group_outcome(ManagedGroupEndReason::Cancelled)); } let now = Instant::now(); @@ -697,10 +746,7 @@ fn run_group_scheduler( collect_finished_targets(&shared, &mut records); if group_should_complete(&shared, config.completion) { - return Ok(ManagedGroupOutcome { - end_reason: ManagedGroupEndReason::AllTargetsComplete, - targets: records, - }); + return Ok(records.into_group_outcome(ManagedGroupEndReason::AllTargetsComplete)); } let active = active_targets(&shared); @@ -891,7 +937,7 @@ fn drain_control_messages( control_rx: &mpsc::Receiver, first_message: Option, next_order: &mut u64, - records: &mut Vec, + records: &mut TargetOutcomeHistory, ) { if let Some(message) = first_message { handle_control_message(config, socket, shared, next_order, records, message); @@ -906,7 +952,7 @@ fn handle_control_message( socket: &UdpSocket, shared: &GroupShared, next_order: &mut u64, - records: &mut Vec, + records: &mut TargetOutcomeHistory, message: ControlMessage, ) { match message { @@ -924,7 +970,7 @@ fn apply_target_update( shared: &GroupShared, targets: Vec, next_order: &mut u64, - records: &mut Vec, + records: &mut TargetOutcomeHistory, ) -> Result<(), ClientError> { validate_open_timeouts(&config.open_timeouts)?; if !targets.is_empty() { @@ -1207,7 +1253,7 @@ fn send_echo_to_target( } } -fn collect_finished_targets(shared: &GroupShared, records: &mut Vec) { +fn collect_finished_targets(shared: &GroupShared, records: &mut TargetOutcomeHistory) { let finished = { let mut registry = shared .registry @@ -1247,7 +1293,7 @@ fn collect_finished_targets(shared: &GroupShared, records: &mut Vec, + records: &mut TargetOutcomeHistory, ) { let targets = drain_registry(shared); for target in targets { @@ -1263,13 +1309,13 @@ fn cancel_remaining_targets( fn record_target_outcome( shared: &GroupShared, - records: &mut Vec, + records: &mut TargetOutcomeHistory, outcome: ManagedTargetOutcome, ) { shared .hub .publish(ManagedGroupEvent::TargetFinished(outcome.clone())); - records.push(outcome); + records.record(outcome); } fn fail_all_targets(shared: &GroupShared, failure: ManagedTargetFailure) { @@ -2604,6 +2650,61 @@ mod tests { ))); } + #[test] + fn target_churn_retains_bounded_history_after_publishing_every_outcome() { + let sink = UdpSocket::bind("127.0.0.1:0").unwrap(); + let remote = sink.local_addr().unwrap(); + let mut config = group_config( + None, + Duration::from_millis(20), + ManagedGroupPacing::Staggered, + ); + config.completion = ManagedGroupCompletionPolicy::ExplicitCancellation; + config.client.open_timeouts = vec![Duration::from_secs(5)]; + let churn_count = MANAGED_GROUP_OUTCOME_HISTORY_LIMIT + 16; + let (session, sub) = ManagedClientGroup::start_with_subscription( + config, + vec![target("churn", remote)], + SubscriberConfig { + capacity: churn_count + 8, + overflow: SubscriberOverflow::DropNewest, + }, + ) + .unwrap(); + + session.update_targets(Vec::new()).unwrap(); + for _ in 0..churn_count { + session + .update_targets(vec![target("churn", remote)]) + .unwrap(); + session.update_targets(Vec::new()).unwrap(); + } + + session.stop(); + let events = collect_group_events_until_disconnected(&sub); + let outcome = session.join().unwrap(); + drop(sink); + + let expected_total = u64::try_from(churn_count + 1).unwrap(); + let published = events + .iter() + .filter(|event| matches!(event, ManagedGroupEvent::TargetFinished(_))) + .count(); + assert_eq!(published, churn_count + 1); + assert_eq!(outcome.total_target_outcomes, expected_total); + assert_eq!(outcome.successful_target_outcomes, 0); + assert_eq!(outcome.failed_target_outcomes, 0); + assert_eq!(outcome.targets.len(), MANAGED_GROUP_OUTCOME_HISTORY_LIMIT); + assert_eq!( + outcome.discarded_target_outcomes, + expected_total - u64::try_from(MANAGED_GROUP_OUTCOME_HISTORY_LIMIT).unwrap() + ); + assert!(outcome + .targets + .iter() + .all(|target| target.end_reason == ManagedTargetEndReason::Removed)); + } + #[test] fn finite_static_group_still_completes_naturally() { let interval = Duration::from_millis(10); diff --git a/crates/irtt-client/src/managed/mod.rs b/crates/irtt-client/src/managed/mod.rs index 0cbf73c..db04c9e 100644 --- a/crates/irtt-client/src/managed/mod.rs +++ b/crates/irtt-client/src/managed/mod.rs @@ -9,6 +9,7 @@ pub use group::{ ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, ManagedGroupOutcome, ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetOutcome, TargetEvent, TargetEventSubscription, TargetId, + MANAGED_GROUP_OUTCOME_HISTORY_LIMIT, }; pub use hub::{EventHub, EventSubscription, SubscriberConfig, SubscriberOverflow}; pub use runner::{ManagedClient, ManagedClientSession, SessionEndReason, SessionOutcome};