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};