From 8e58c0f04bbde76b8fa5cae06d78678272ae3d79 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 28 Jul 2026 00:15:59 +0700 Subject: [PATCH] fix(managed): keep empty dynamic groups idle Track desired membership separately from active target state and make natural completion an explicit group policy. Empty authoritative updates now remove all targets without terminating the scheduler, while static CLI groups retain all-target completion. Document that construction still requires an address-family-defining target and add regressions for empty replacement, later re-addition, idle control responsiveness, idle cancellation, and finite static completion. --- crates/irtt-cli/src/cmd/client/run.rs | 5 +- crates/irtt-cli/src/cmd/tui/run.rs | 6 +- crates/irtt-client/src/lib.rs | 8 +- crates/irtt-client/src/managed/group.rs | 224 +++++++++++++++++++++--- crates/irtt-client/src/managed/mod.rs | 8 +- 5 files changed, 215 insertions(+), 36 deletions(-) diff --git a/crates/irtt-cli/src/cmd/client/run.rs b/crates/irtt-cli/src/cmd/client/run.rs index 1eedf2f..bce5523 100644 --- a/crates/irtt-cli/src/cmd/client/run.rs +++ b/crates/irtt-cli/src/cmd/client/run.rs @@ -10,8 +10,8 @@ use std::{ use irtt_client::{ ClientEvent, EventSubscriptionError, ManagedClientGroup, ManagedClientGroupConfig, - ManagedGroupEndReason, ManagedGroupEvent, ManagedTargetOutcome, SubscriberConfig, - SubscriberOverflow, TargetEvent, + ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, ManagedTargetOutcome, + SubscriberConfig, SubscriberOverflow, TargetEvent, }; #[cfg(all(test, feature = "stats"))] use irtt_client::{ClientTimestamp, PacketMeta, RttSample, SignedDuration}; @@ -279,6 +279,7 @@ fn run_group_stream( let group_config = ManagedClientGroupConfig { client: config, pacing: args.pacing.into(), + completion: ManagedGroupCompletionPolicy::AllTargetsComplete, }; let (session, events) = ManagedClientGroup::start_with_subscription( diff --git a/crates/irtt-cli/src/cmd/tui/run.rs b/crates/irtt-cli/src/cmd/tui/run.rs index 5d68f12..b40d112 100644 --- a/crates/irtt-cli/src/cmd/tui/run.rs +++ b/crates/irtt-cli/src/cmd/tui/run.rs @@ -7,8 +7,9 @@ use std::{ use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers}; use irtt_client::{ - EventSubscriptionError, ManagedClientGroup, ManagedClientGroupConfig, ManagedGroupEndReason, - ManagedGroupEvent, SubscriberConfig, SubscriberOverflow, + EventSubscriptionError, ManagedClientGroup, ManagedClientGroupConfig, + ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, SubscriberConfig, + SubscriberOverflow, }; use crate::{ @@ -154,6 +155,7 @@ fn run_group_tui( let group_config = ManagedClientGroupConfig { client: config, pacing: args.pacing.into(), + completion: ManagedGroupCompletionPolicy::AllTargetsComplete, }; let (session, events) = ManagedClientGroup::start_with_subscription( group_config, diff --git a/crates/irtt-client/src/lib.rs b/crates/irtt-client/src/lib.rs index 9db8cde..a302407 100644 --- a/crates/irtt-client/src/lib.rs +++ b/crates/irtt-client/src/lib.rs @@ -78,10 +78,10 @@ pub use event::{ pub use managed::{ CancellationToken, EventHub, EventSubscription, ManagedClient, ManagedClientGroup, ManagedClientGroupConfig, ManagedClientGroupSession, ManagedClientSession, - ManagedGroupEndReason, ManagedGroupEvent, ManagedGroupOutcome, ManagedGroupPacing, - ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure, ManagedTargetFailureKind, - ManagedTargetOutcome, SessionEndReason, SessionOutcome, SubscriberConfig, SubscriberOverflow, - TargetEvent, TargetEventSubscription, TargetId, + ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, ManagedGroupOutcome, + ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure, + ManagedTargetFailureKind, ManagedTargetOutcome, SessionEndReason, SessionOutcome, + SubscriberConfig, SubscriberOverflow, TargetEvent, TargetEventSubscription, TargetId, }; 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 54e084a..16565e6 100644 --- a/crates/irtt-client/src/managed/group.rs +++ b/crates/irtt-client/src/managed/group.rs @@ -104,6 +104,20 @@ pub enum ManagedGroupPacing { Burst, } +/// Policy controlling when a managed group completes naturally. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum ManagedGroupCompletionPolicy { + /// Complete after a non-empty desired target set reaches terminal outcomes. + /// + /// Replacing the desired set with an empty set leaves the group alive and + /// idle so a later [`ManagedClientGroupSession::update_targets`] call can + /// resume probing. + #[default] + AllTargetsComplete, + /// Remain alive until explicit cancellation, including after all targets finish. + ExplicitCancellation, +} + /// Configuration for a managed multi-target client group. #[derive(Debug, Clone, PartialEq, Eq, Default)] pub struct ManagedClientGroupConfig { @@ -116,6 +130,8 @@ pub struct ManagedClientGroupConfig { pub client: ClientConfig, /// Coordinated group pacing mode. pub pacing: ManagedGroupPacing, + /// Policy controlling natural group completion. + pub completion: ManagedGroupCompletionPolicy, } /// Target-scoped managed event. @@ -388,6 +404,7 @@ impl ManagedClientGroup { (target.id.clone(), target.remote) }; registry.remotes.insert(remote, id.clone()); + registry.desired.insert(id.clone()); registry.targets.insert(id, state); } @@ -458,7 +475,8 @@ impl ManagedClientGroupSession { /// The update is authoritative. Existing targets with the same /// [`TargetId`], remote address, and auth setting remain running. In v1, /// changing the remote address or auth for an existing target id is - /// rejected instead of mutating the active session in place. + /// rejected instead of mutating the active session in place. An empty set + /// removes every target but leaves the group alive and idle. pub fn update_targets(&self, targets: Vec) -> Result<(), ClientError> { let (reply_tx, reply_rx) = mpsc::channel(); self.control_tx @@ -538,6 +556,7 @@ impl Drop for GroupSchedulerCleanup { #[derive(Debug, Default)] struct TargetRegistry { + desired: HashSet, targets: HashMap>>, remotes: HashMap, } @@ -677,7 +696,7 @@ fn run_group_scheduler( finish_completed_targets(&socket, &shared, now); collect_finished_targets(&shared, &mut records); - if registry_is_empty(&shared) { + if group_should_complete(&shared, config.completion) { return Ok(ManagedGroupOutcome { end_reason: ManagedGroupEndReason::AllTargetsComplete, targets: records, @@ -908,11 +927,9 @@ fn apply_target_update( records: &mut Vec, ) -> Result<(), ClientError> { validate_open_timeouts(&config.open_timeouts)?; - if targets.is_empty() { - remove_all_targets(socket, shared, records, ManagedTargetEndReason::Removed); - return Ok(()); + if !targets.is_empty() { + validate_target_configs(&targets, Some(shared.family_remote))?; } - validate_target_configs(&targets, Some(shared.family_remote))?; let desired_ids: HashSet = targets.iter().map(|target| target.id.clone()).collect(); let mut additions = Vec::new(); @@ -980,6 +997,7 @@ fn apply_target_update( registry.remotes.insert(remote, id.clone()); registry.targets.insert(id, target); } + registry.desired = desired_ids; removed }; @@ -1243,19 +1261,6 @@ fn cancel_remaining_targets( } } -fn remove_all_targets( - socket: &UdpSocket, - shared: &GroupShared, - records: &mut Vec, - reason: ManagedTargetEndReason, -) { - let targets = drain_registry(shared); - for target in targets { - let outcome = close_target(socket, &shared.hub, target, reason.clone()); - record_target_outcome(shared, records, outcome); - } -} - fn record_target_outcome( shared: &GroupShared, records: &mut Vec, @@ -1344,13 +1349,15 @@ fn all_targets(shared: &GroupShared) -> Vec>> { .collect() } -fn registry_is_empty(shared: &GroupShared) -> bool { - shared +fn group_should_complete(shared: &GroupShared, completion: ManagedGroupCompletionPolicy) -> bool { + if completion == ManagedGroupCompletionPolicy::ExplicitCancellation { + return false; + } + let registry = shared .registry .lock() - .expect("target registry mutex poisoned") - .targets - .is_empty() + .expect("target registry mutex poisoned"); + !registry.desired.is_empty() && registry.targets.is_empty() } #[derive(Debug)] @@ -1584,6 +1591,7 @@ mod tests { ..ClientConfig::default() }, pacing, + completion: ManagedGroupCompletionPolicy::AllTargetsComplete, } } @@ -2453,6 +2461,174 @@ mod tests { && target.end_reason == ManagedTargetEndReason::Removed)); } + #[test] + fn start_rejects_empty_target_set_with_documented_error() { + let err = ManagedClientGroup::start( + group_config( + None, + Duration::from_millis(20), + ManagedGroupPacing::Staggered, + ), + Vec::new(), + ) + .unwrap_err(); + + assert!(matches!(err, ClientError::InvalidConfig { .. })); + assert!(err + .to_string() + .contains("managed client group requires at least one target")); + } + + #[test] + fn dynamic_group_can_remove_to_empty_and_add_targets_again() { + let interval = Duration::from_millis(20); + let params = test_params(None, interval); + let a = start_echo_server(params.clone(), Duration::ZERO); + let b = start_echo_server(params, Duration::ZERO); + let mut config = group_config(None, interval, ManagedGroupPacing::Burst); + config.completion = ManagedGroupCompletionPolicy::ExplicitCancellation; + + let (session, sub) = ManagedClientGroup::start_with_subscription( + config, + vec![target("a", a.addr)], + SubscriberConfig { + capacity: 128, + overflow: SubscriberOverflow::DropNewest, + }, + ) + .unwrap(); + + while { + let event = recv_event_with_timeout(&sub); + event.target.as_str() != "a" || !matches!(event.event, ClientEvent::EchoReply { .. }) + } {} + + session.update_targets(Vec::new()).unwrap(); + let removal_deadline = Instant::now() + Duration::from_secs(2); + loop { + match sub.try_recv() { + Ok(Some(ManagedGroupEvent::TargetFinished(outcome))) + if outcome.id.as_str() == "a" + && outcome.end_reason == ManagedTargetEndReason::Removed => + { + break; + } + Ok(Some(_)) | Ok(None) if Instant::now() < removal_deadline => { + thread::sleep(Duration::from_millis(1)); + } + result => panic!("timed out waiting for removed outcome: {result:?}"), + } + } + + session.update_targets(vec![target("b", b.addr)]).unwrap(); + while { + let event = recv_event_with_timeout(&sub); + event.target.as_str() != "b" || !matches!(event.event, ClientEvent::EchoReply { .. }) + } {} + + session.stop(); + let events = collect_group_events_until_disconnected(&sub); + let outcome = session.join().unwrap(); + a.join(); + b.join(); + + assert_eq!(outcome.end_reason, ManagedGroupEndReason::Cancelled); + assert!(events.iter().any(|event| matches!( + event, + ManagedGroupEvent::TargetFinished(ManagedTargetOutcome { + id, + end_reason: ManagedTargetEndReason::Cancelled, + .. + }) if id.as_str() == "b" + ))); + } + + #[test] + fn removing_final_target_keeps_control_channel_responsive() { + let interval = Duration::from_millis(20); + let server = start_echo_server(test_params(None, interval), Duration::ZERO); + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered); + config.completion = ManagedGroupCompletionPolicy::ExplicitCancellation; + let session = ManagedClientGroup::start(config, vec![target("only", server.addr)]).unwrap(); + + session.update_targets(Vec::new()).unwrap(); + session.update_targets(Vec::new()).unwrap(); + let idle_subscription = session + .subscribe(SubscriberConfig { + capacity: 4, + overflow: SubscriberOverflow::DropNewest, + }) + .expect("an idle dynamic group must still accept subscribers"); + + session.stop(); + let outcome = session.join().unwrap(); + server.join(); + + assert_eq!(outcome.end_reason, ManagedGroupEndReason::Cancelled); + assert_eq!( + idle_subscription.try_recv(), + Err(EventSubscriptionError::Disconnected) + ); + } + + #[test] + fn explicit_cancellation_while_idle_publishes_removal_and_disconnects() { + let interval = Duration::from_millis(20); + let server = start_echo_server(test_params(None, interval), Duration::ZERO); + let mut config = group_config(None, interval, ManagedGroupPacing::Staggered); + config.completion = ManagedGroupCompletionPolicy::ExplicitCancellation; + let (session, sub) = ManagedClientGroup::start_with_subscription( + config, + vec![target("idle", server.addr)], + SubscriberConfig { + capacity: 16, + overflow: SubscriberOverflow::DropNewest, + }, + ) + .unwrap(); + + session.update_targets(Vec::new()).unwrap(); + session.stop(); + let events = collect_group_events_until_disconnected(&sub); + let outcome = session.join().unwrap(); + server.join(); + + assert_eq!(outcome.end_reason, ManagedGroupEndReason::Cancelled); + assert!(events.iter().any(|event| matches!( + event, + ManagedGroupEvent::TargetFinished(ManagedTargetOutcome { + id, + end_reason: ManagedTargetEndReason::Removed, + .. + }) if id.as_str() == "idle" + ))); + } + + #[test] + fn finite_static_group_still_completes_naturally() { + let interval = Duration::from_millis(10); + let duration = Duration::from_millis(30); + let server = start_echo_server(test_params(Some(duration), interval), Duration::ZERO); + let session = ManagedClientGroup::start( + group_config(Some(duration), interval, ManagedGroupPacing::Staggered), + vec![target("finite", server.addr)], + ) + .unwrap(); + + let outcome = session.join().unwrap(); + server.join(); + + assert_eq!( + outcome.end_reason, + ManagedGroupEndReason::AllTargetsComplete + ); + assert_eq!(outcome.targets.len(), 1); + assert_eq!( + outcome.targets[0].end_reason, + ManagedTargetEndReason::TestComplete + ); + } + #[test] fn stop_join_cleans_up_group_threads() { let interval = Duration::from_millis(20); diff --git a/crates/irtt-client/src/managed/mod.rs b/crates/irtt-client/src/managed/mod.rs index 09d15b4..0cbf73c 100644 --- a/crates/irtt-client/src/managed/mod.rs +++ b/crates/irtt-client/src/managed/mod.rs @@ -5,10 +5,10 @@ mod runner; pub use cancellation::CancellationToken; pub use group::{ - ManagedClientGroup, ManagedClientGroupConfig, ManagedClientGroupSession, ManagedGroupEndReason, - ManagedGroupEvent, ManagedGroupOutcome, ManagedGroupPacing, ManagedTargetConfig, - ManagedTargetEndReason, ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetOutcome, - TargetEvent, TargetEventSubscription, TargetId, + ManagedClientGroup, ManagedClientGroupConfig, ManagedClientGroupSession, + ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, ManagedGroupOutcome, + ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure, + ManagedTargetFailureKind, ManagedTargetOutcome, TargetEvent, TargetEventSubscription, TargetId, }; pub use hub::{EventHub, EventSubscription, SubscriberConfig, SubscriberOverflow}; pub use runner::{ManagedClient, ManagedClientSession, SessionEndReason, SessionOutcome};