README.md

@@ -173,73 +173,51 @@ Add it from a local checkout:
 
 ```toml
 [dependencies]
-irtt-client = { path = "crates/irtt-client" }
+irtt-client = { path = "crates/irtt-client", features = ["tokio"] }
 ```
 
-Minimal managed-client example:
+Minimal managed-client example for a synchronous frontend:
 
 ```rust
 use std::time::Duration;
 
-use irtt_client::{
-    ClientConfig, ClientEvent, ManagedClient, SubscriberConfig,
+use irtt_client::managed::{
+    BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy,
+    ManagedTargetConfig, TargetId,
 };
+use irtt_client::ClientConfig;
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let config = ClientConfig {
-        server_addr: "netperf-eu.bufferbloat.net:2112".to_owned(),
+    let config = ManagedClientConfig {
+        client: ClientConfig {
             duration: Some(Duration::from_secs(10)),
             interval: Duration::from_secs(1),
             ..ClientConfig::default()
+        },
+        completion: ManagedCompletionPolicy::FinishWhenQuiescent,
+        ..ManagedClientConfig::default()
     };
-
-    let (session, events) =
-        ManagedClient::start_with_subscription(
-            config,
-            SubscriberConfig::default(),
-        )?;
-
-    while let Ok(event) = events.recv() {
-        match event {
-            ClientEvent::EchoReply { seq, rtt, .. } => {
-                println!(
-                    "seq={seq} rtt_us={}",
-                    rtt.effective.as_micros()
-                );
-            }
-            ClientEvent::SessionClosed { .. } => break,
-            _ => {}
-        }
-    }
-
-    let outcome = session.join()?;
+    let targets = vec![ManagedTargetConfig::new(
+        TargetId::from("edge"),
+        "netperf-eu.bufferbloat.net:2112",
+    )];
+    let owner = BlockingManagedClient::start(config, targets)?;
+    let outcome = owner.join()?;
     println!("session ended: {:?}", outcome.end_reason);
 
     Ok(())
 }
 ```
 
-The client library emits structured events for session lifecycle, sent probes, successful replies, loss, duplicates, late replies, warnings, and shutdown.
+For presentation events, use `BlockingManagedClient::start_with_subscription`,
+`ManagedEvent`, and `ManagedEventTryRecvError`, while consulting the durable
+`handle.status()` and final outcome for authoritative state. Presentation
+events are bounded and lossy, not reliable history. Dynamic desired-target
+updates are available through `ManagedClientHandle::update_targets`.
 
-For shared-socket multi-target integrations, `ManagedClientGroup` publishes
-`ManagedGroupEvent::TargetFinished` exactly once for every terminal target
-incarnation, including open/runtime failures, removal, cancellation, no-test
-completion, and peer closure. Consumers can use `ManagedTargetEndReason` and
-`ManagedTargetFailureKind` for outcome and exit decisions without parsing
-diagnostic text.
-
-`ManagedGroupCompletionPolicy::AllTargetsComplete` supports finite static
-groups. `ExplicitCancellation` supports long-lived controllers; replacing the
-desired set with an empty vector removes current targets but leaves the group
-idle and ready for a later `update_targets` call. `join()` reports aggregate
-outcome counts and retains only the 256 most recent target outcomes, so
-long-running target churn remains bounded. The aggregate
-`peer_closed_target_outcomes` count includes peer closures omitted from that
-recent snapshot.
-
-Managed event subscriptions remain bounded and nonblocking. When using a
-dropping overflow policy, check `EventSubscription::dropped_events()` before
-trusting a complete statistical summary.
+`BlockingManagedClient` owns a dedicated Tokio runtime and thread for
+synchronous callers. Callers that already own Tokio can drive
+`ManagedClientTask` directly.
 
 ## Binaries and features
 

crates/irtt-cli/Cargo.toml

@@ -30,7 +30,7 @@ required-features = ["tui"]
 
 [features]
 default = ["client", "stats"]
-client-runtime = ["dep:irtt-client", "dep:irtt-proto"]
+client-runtime = ["dep:irtt-client", "irtt-client/tokio", "dep:irtt-proto"]
 client = ["client-runtime"]
 stats = ["client-runtime", "dep:irtt-stats"]
 tui = ["client-runtime", "stats", "dep:ratatui", "dep:crossterm"]

crates/irtt-cli/src/cmd/client/args.rs

@@ -4,8 +4,8 @@ use clap::{Parser, ValueEnum};
 use irtt_client::ClientConfig;
 
 use crate::shared::client::{
-    parse_labelled_target, parse_test_duration, resolved_managed_targets, target_specs,
-    CommonClientArgs, GroupPacingArg, LabelledTargetArg, ResolvedTarget, TargetSpec, TimestampArg,
+    parse_labelled_target, parse_test_duration, prepare_managed_targets, target_specs,
+    CommonClientArgs, GroupPacingArg, LabelledTargetArg, PreparedTarget, TargetSpec, TimestampArg,
 };
 
 pub const DEFAULT_CLIENT_DURATION: Duration = Duration::from_secs(10);
@@ -111,10 +111,8 @@ impl ClientArgs {
         target_specs(&self.targets, &self.labelled_targets)
     }
 
-    pub fn resolved_managed_targets(&self) -> Result<Vec<ResolvedCliTarget>, String> {
-        let specs = self.target_specs()?;
-        let config = self.to_client_config();
-        resolved_managed_targets(specs, &config)
+    pub fn managed_targets(&self) -> Result<Vec<CliManagedTarget>, String> {
+        prepare_managed_targets(self.target_specs()?)
     }
 
     fn primary_target_addr(&self) -> Option<&str> {
@@ -147,7 +145,7 @@ pub enum OutputFormat {
 }
 
 pub type CliTargetSpec = TargetSpec;
-pub type ResolvedCliTarget = ResolvedTarget;
+pub type CliManagedTarget = PreparedTarget;
 
 impl OutputFormat {
     pub fn prints_summary(self) -> bool {
@@ -262,11 +260,9 @@ mod tests {
     }
 
     #[test]
-    fn duplicate_resolved_target_addresses_are_rejected() {
+    fn duplicate_target_endpoints_are_allowed() {
         let args = parse(&["127.0.0.1:2112", "127.0.0.1"]).unwrap();
-        let err = args.resolved_managed_targets().unwrap_err();
-
-        assert!(err.contains("duplicate resolved target address 127.0.0.1:2112"));
+        assert_eq!(args.managed_targets().unwrap().len(), 2);
     }
 
     #[test]

crates/irtt-cli/src/cmd/tui/args.rs

@@ -6,8 +6,8 @@ use irtt_client::ClientConfig;
 #[cfg(test)]
 use crate::shared::client::TimestampArg;
 use crate::shared::client::{
-    parse_labelled_target, parse_test_duration, resolved_managed_targets, target_specs,
-    CommonClientArgs, GroupPacingArg, LabelledTargetArg, ResolvedTarget, TargetSpec,
+    parse_labelled_target, parse_test_duration, prepare_managed_targets, target_specs,
+    CommonClientArgs, GroupPacingArg, LabelledTargetArg, PreparedTarget, TargetSpec,
 };
 
 pub const DEFAULT_TUI_DURATION: Duration = Duration::ZERO;
@@ -72,10 +72,8 @@ impl TuiArgs {
         target_specs(&self.targets, &self.labelled_targets)
     }
 
-    pub fn resolved_managed_targets(&self) -> Result<Vec<ResolvedTuiTarget>, String> {
-        let specs = self.target_specs()?;
-        let config = self.to_client_config();
-        resolved_managed_targets(specs, &config)
+    pub fn managed_targets(&self) -> Result<Vec<TuiManagedTarget>, String> {
+        prepare_managed_targets(self.target_specs()?)
     }
 
     #[cfg(test)]
@@ -93,7 +91,7 @@ impl TuiArgs {
 }
 
 pub type TuiTargetSpec = TargetSpec;
-pub type ResolvedTuiTarget = ResolvedTarget;
+pub type TuiManagedTarget = PreparedTarget;
 
 impl std::ops::Deref for TuiArgs {
     type Target = CommonClientArgs;
@@ -199,11 +197,9 @@ mod tests {
     }
 
     #[test]
-    fn duplicate_resolved_target_addresses_are_rejected() {
+    fn duplicate_target_endpoints_are_allowed() {
         let args = parse(&["127.0.0.1:2112", "127.0.0.1"]).unwrap();
-        let err = args.resolved_managed_targets().unwrap_err();
-
-        assert!(err.contains("duplicate resolved target address 127.0.0.1:2112"));
+        assert_eq!(args.managed_targets().unwrap().len(), 2);
     }
 
     #[test]

crates/irtt-cli/src/cmd/tui/run.rs

@@ -1,33 +1,30 @@
 use std::{
+    collections::HashSet,
     io,
     sync::atomic::{AtomicBool, Ordering},
     thread,
     time::{Duration, Instant},
 };
 
-use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers};
-use irtt_client::{
-    EventSubscription, EventSubscriptionError, ManagedClient, ManagedClientGroup,
-    ManagedClientGroupConfig, ManagedGroupCompletionPolicy, ManagedGroupEndReason,
-    ManagedGroupEvent, SessionEndReason, SubscriberConfig, SubscriberOverflow,
-};
-
+use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal};
 use crate::{
-    cmd::tui::args::{ResolvedTuiTarget, TuiArgs},
+    cmd::tui::args::TuiArgs,
     shared::client::{
         is_shutdown_requested,
         session::{
-            peer_close_run_error, request_group_stop_for_peer_close, request_group_stop_once,
+            drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close,
+            request_managed_stop_once, ManagedDrainState,
         },
     },
 };
-
-use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal};
+use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers};
+use irtt_client::managed::{
+    BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedEndReason,
+    ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError, TargetInstance,
+};
 
 const RENDER_INTERVAL: Duration = Duration::from_millis(250);
 const TUI_WAIT_SLICE: Duration = Duration::from_millis(20);
-const IDLE_SLEEP: Duration = Duration::from_millis(5);
-const GROUP_COMPLETION_GRACE: Duration = Duration::from_secs(1);
 const MANAGED_EVENT_CAPACITY: usize = 16_384;
 
 pub fn run_tui(
@@ -35,7 +32,7 @@ pub fn run_tui(
     shutdown_requested: &AtomicBool,
 ) -> Result<(), Box<dyn std::error::Error>> {
     let targets = args
-        .resolved_managed_targets()
+        .managed_targets()
         .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?;
     let continuous = args.is_continuous();
     let mut terminal = TuiTerminal::enter()?;
@@ -44,329 +41,179 @@ pub fn run_tui(
         targets.iter().map(|target| target.label.clone()),
     );
     let mut next_render = Instant::now();
-
     if is_shutdown_requested(shutdown_requested) {
         return Ok(());
     }
-
     state.set_status(TuiStatus::Opening);
     render_if_due(&mut terminal, &state, &mut next_render, true)?;
-
-    if targets.len() > 1 {
-        return run_group_tui(
-            args,
-            targets,
-            &mut terminal,
-            &mut state,
-            &mut next_render,
-            shutdown_requested,
-        );
-    }
-
-    let (session, events) = match ManagedClient::start_with_subscription(
-        args.to_client_config(),
-        managed_event_subscriber_config(),
+    let target_count = targets.len();
+    let config = ManagedClientConfig {
+        client: args.to_client_config(),
+        pacing: args.pacing.into(),
+        completion: ManagedCompletionPolicy::FinishWhenQuiescent,
+        event_capacity: MANAGED_EVENT_CAPACITY,
+        outcome_history_limit: target_count,
+        max_live_target_generations: target_count,
+        ..ManagedClientConfig::default()
+    };
+    let (owner, mut events) = match BlockingManagedClient::start_with_subscription(
+        config,
+        targets
+            .iter()
+            .map(|target| target.managed.clone())
+            .collect(),
     ) {
-        Ok(session) => session,
-        Err(err) => {
-            state.set_error(err.to_string());
+        Ok(value) => value,
+        Err(error) => {
+            state.set_error(error.to_string());
             render_if_due(&mut terminal, &state, &mut next_render, true)?;
-            return Err(Box::new(err));
+            return Err(Box::new(error));
         }
     };
-
+    let handle = owner.handle();
     let mut interrupted = false;
+    let mut stop_requested = false;
+    let mut terminal_targets = HashSet::new();
+    let mut dropped_events = 0;
+    let mut subscription_closed = false;
     loop {
         if is_shutdown_requested(shutdown_requested) {
             interrupted = true;
-            session.stop();
-            break;
+            if request_managed_stop_once(&mut stop_requested) {
+                drop(handle.stop());
+            }
         }
-
         if handle_input(&mut state, shutdown_requested)? {
             render_if_due(&mut terminal, &state, &mut next_render, true)?;
         }
         if state.quit_requested {
             interrupted = true;
-            session.stop();
-            break;
-        }
-
-        match events.recv_timeout(managed_tui_wait_duration(&next_render, state.paused)) {
-            Ok(Some(event)) => {
-                state.process_event(&event);
-                render_if_due(&mut terminal, &state, &mut next_render, false)?;
-            }
-            Ok(None) => {
-                render_if_due(&mut terminal, &state, &mut next_render, false)?;
-            }
-            Err(EventSubscriptionError::Disconnected) => break,
+            if request_managed_stop_once(&mut stop_requested) {
+                drop(handle.stop());
             }
         }
-    interrupted |= is_shutdown_requested(shutdown_requested);
-
-    if interrupted {
-        state.set_status(TuiStatus::Interrupted);
-        render_if_due(&mut terminal, &state, &mut next_render, true)?;
-    }
-
-    if interrupted {
-        state.set_status(TuiStatus::Draining);
-        drain_single_tui_events(&events, &mut state);
-        render_if_due(&mut terminal, &state, &mut next_render, true)?;
-    }
-
-    state.set_status(TuiStatus::Closing);
-    render_if_due(&mut terminal, &state, &mut next_render, true)?;
-
-    let outcome = session.join();
-    drain_single_tui_events(&events, &mut state);
-    state.mark_dropped_client_events(events.dropped_events());
-    let outcome = match outcome {
-        Ok(outcome) => outcome,
-        Err(err) => {
-            state.set_error(err.to_string());
-            render_if_due(&mut terminal, &state, &mut next_render, true)?;
-            return Err(Box::new(err));
-        }
-    };
-
-    interrupted |= is_shutdown_requested(shutdown_requested);
-    if let Some(error) = peer_close_run_error(
+        if request_managed_stop_for_peer_close(
             continuous,
             interrupted,
-        u64::from(outcome.end_reason == SessionEndReason::PeerClosed),
-    ) {
-        state.set_run_error(error.clone());
-        render_if_due(&mut terminal, &state, &mut next_render, true)?;
-        return Err(error.into());
-    }
-    state.set_status(TuiStatus::Complete);
-    render_if_due(&mut terminal, &state, &mut next_render, true)?;
-    Ok(())
-}
-
-fn run_group_tui(
-    args: TuiArgs,
-    targets: Vec<ResolvedTuiTarget>,
-    terminal: &mut TuiTerminal,
-    state: &mut TuiState,
-    next_render: &mut Instant,
-    shutdown_requested: &AtomicBool,
-) -> Result<(), Box<dyn std::error::Error>> {
-    let mut config = args.to_client_config();
-    let managed_targets = targets
-        .iter()
-        .map(|target| target.managed.clone())
-        .collect::<Vec<_>>();
-    let expected_target_count = managed_targets.len();
-    if let Some(first) = managed_targets.first() {
-        config.server_addr = first.remote.to_string();
-    }
-
-    let group_config = ManagedClientGroupConfig {
-        client: config,
-        pacing: args.pacing.into(),
-        completion: ManagedGroupCompletionPolicy::AllTargetsComplete,
-    };
-    let (session, events) = ManagedClientGroup::start_with_subscription(
-        group_config,
-        managed_targets,
-        SubscriberConfig {
-            capacity: MANAGED_EVENT_CAPACITY,
-            overflow: SubscriberOverflow::DropOldest,
-        },
-    )?;
-
-    let mut interrupted = false;
-    let mut stop_requested = false;
-    let mut peer_close_requested_stop = false;
-    let mut terminal_targets = std::collections::HashSet::new();
-    let mut saw_target_event = false;
-    let mut last_event_at = Instant::now();
-
-    let exit = loop {
-        if is_shutdown_requested(shutdown_requested) {
-            interrupted = true;
-            if request_group_stop_once(&mut stop_requested) {
-                session.stop();
-            }
-        }
-
-        if handle_input(state, shutdown_requested)? {
-            render_if_due(terminal, state, next_render, true)?;
-        }
-        if state.quit_requested {
-            interrupted = true;
-            if request_group_stop_once(&mut stop_requested) {
-                session.stop();
-            }
-        }
-        if request_group_stop_for_peer_close(
-            args.is_continuous(),
-            interrupted,
-            session.peer_closed_target_count(),
-            &mut peer_close_requested_stop,
+            handle.status().peer_closed_target_outcomes,
             &mut stop_requested,
         ) {
-            session.stop();
+            drop(handle.stop());
         }
-
-        match events.try_recv() {
-            Ok(Some(group_event)) => {
-                saw_target_event = true;
-                last_event_at = Instant::now();
-                match group_event {
-                    ManagedGroupEvent::Client(target_event) => {
-                        state.process_target_event(&target_event);
+        let drain_state = drain_tui_events(
+            &mut events,
+            &mut state,
+            &mut terminal_targets,
+            &mut dropped_events,
+        );
+        match drain_state {
+            ManagedDrainState::Empty => {
+                thread::sleep(managed_tui_wait_duration(&next_render, state.paused));
             }
-                    ManagedGroupEvent::TargetFinished(target) => {
-                        terminal_targets.insert(target.id.as_str().to_owned());
-                        state.process_target_outcome(&target);
-                        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,
-                        ) {
-                            session.stop();
+            ManagedDrainState::Closed => subscription_closed = true,
+            ManagedDrainState::BudgetExhausted => {}
         }
+        if handle.status().final_outcome.is_some() || subscription_closed {
+            break;
         }
+        render_if_due(&mut terminal, &state, &mut next_render, false)?;
     }
-                render_if_due(terminal, state, next_render, false)?;
-            }
-            Ok(None) => {
-                if interrupted {
-                    break GroupLoopExit::Interrupted;
-                }
-                if terminal_targets.len() >= expected_target_count {
-                    break GroupLoopExit::AllTargetsTerminal;
-                }
-                if should_join_group_after_idle(&args, saw_target_event, last_event_at) {
-                    break GroupLoopExit::IdleGraceElapsed;
-                }
-                wait_for_tui_activity(None, next_render, state, terminal, shutdown_requested)?;
-                thread::sleep(IDLE_SLEEP);
-            }
-            Err(EventSubscriptionError::Disconnected) => {
-                break GroupLoopExit::SubscriptionDisconnected
-            }
-        }
-    };
-
-    if exit.should_stop_before_join() && request_group_stop_once(&mut stop_requested) {
-        session.stop();
-    }
-
     if interrupted {
         state.set_status(TuiStatus::Interrupted);
-        render_if_due(terminal, state, next_render, true)?;
+        render_if_due(&mut terminal, &state, &mut next_render, true)?;
     }
-
     state.set_status(TuiStatus::Closing);
-    render_if_due(terminal, state, next_render, true)?;
-
-    let outcome = session.join()?;
-    while let Ok(Some(group_event)) = events.try_recv() {
-        match group_event {
-            ManagedGroupEvent::Client(target_event) => state.process_target_event(&target_event),
-            ManagedGroupEvent::TargetFinished(target) => {
-                if terminal_targets.insert(target.id.as_str().to_owned()) {
-                    state.process_target_outcome(&target);
+    render_if_due(&mut terminal, &state, &mut next_render, true)?;
+    let outcome = owner.join()?;
+    drain_final_tui_events(
+        &mut events,
+        &mut state,
+        &mut terminal_targets,
+        &mut dropped_events,
+    );
+    state.mark_dropped_managed_events(dropped_events);
+    if outcome.discarded_target_outcomes != 0 {
+        state.set_run_error(format!(
+            "{} final target outcomes were discarded",
+            outcome.discarded_target_outcomes
+        ));
     }
-            }
-        }
-    }
-    state.mark_dropped_group_events(events.dropped_events());
-    for target in &outcome.targets {
-        if terminal_targets.insert(target.id.as_str().to_owned()) {
+    for target in outcome.recent_target_outcomes.iter() {
+        if terminal_targets.insert(target.target.clone()) {
             state.process_target_outcome(target);
         }
     }
-
     interrupted |= is_shutdown_requested(shutdown_requested);
-    let peer_closed_target_outcomes = outcome
-        .peer_closed_target_outcomes
-        .max(u64::from(peer_close_requested_stop));
-    if let Some(error) = peer_close_run_error(
-        args.is_continuous(),
-        interrupted,
-        peer_closed_target_outcomes,
-    ) {
+    let error = match &outcome.end_reason {
+        ManagedEndReason::DriverFailed(failure) => {
+            Some(format!("managed driver failed: {failure}"))
+        }
+        _ => peer_close_run_error(continuous, interrupted, outcome.peer_closed_target_outcomes)
+            .or_else(|| {
+                (!interrupted
+                    && outcome.successful_target_outcomes == 0
+                    && outcome.failed_target_outcomes > 0)
+                    .then(|| {
+                        format!(
+                            "no managed target completed successfully ({} failed)",
+                            outcome.failed_target_outcomes
+                        )
+                    })
+            }),
+    };
+    if let Some(error) = error {
         state.set_run_error(error.clone());
-        render_if_due(terminal, state, next_render, true)?;
+        render_if_due(&mut terminal, &state, &mut next_render, true)?;
         return Err(error.into());
     }
-    if outcome.end_reason == ManagedGroupEndReason::Cancelled
-        && !interrupted
-        && !peer_close_requested_stop
-    {
-        return match exit {
-            GroupLoopExit::IdleGraceElapsed => {
-                Err("managed client group stayed idle before all targets completed".into())
-            }
-            GroupLoopExit::SubscriptionDisconnected => {
-                Err("managed client group event subscription disconnected before completion".into())
-            }
-            GroupLoopExit::Interrupted | GroupLoopExit::AllTargetsTerminal => {
-                Err("managed client group was cancelled".into())
-            }
-        };
-    }
-    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)?;
-        return Err(
-            format!("no managed target completed successfully ({failed_targets} failed)").into(),
-        );
-    }
-
     state.set_status(TuiStatus::Complete);
-    render_if_due(terminal, state, next_render, true)?;
+    render_if_due(&mut terminal, &state, &mut next_render, true)?;
     Ok(())
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum GroupLoopExit {
-    Interrupted,
-    AllTargetsTerminal,
-    IdleGraceElapsed,
-    SubscriptionDisconnected,
-}
-
-impl GroupLoopExit {
-    fn should_stop_before_join(self) -> bool {
-        matches!(
-            self,
-            Self::Interrupted | Self::IdleGraceElapsed | Self::SubscriptionDisconnected
-        )
+fn process_tui_event(
+    event: ManagedEvent,
+    state: &mut TuiState,
+    terminal_targets: &mut HashSet<TargetInstance>,
+) {
+    match event {
+        ManagedEvent::Client { target, event } => state.process_target_event(&target, &event),
+        ManagedEvent::TargetFinished { outcome } => {
+            terminal_targets.insert(outcome.target.clone());
+            state.process_target_outcome(&outcome);
+        }
+        _ => {}
     }
 }
-
-fn estimated_group_completion_grace(args: &TuiArgs) -> Duration {
-    let open_timeout: Duration = args.to_client_config().open_timeouts.iter().sum();
-    open_timeout
-        .saturating_add(args.duration)
-        .saturating_add(GROUP_COMPLETION_GRACE)
+fn drain_tui_events(
+    events: &mut ManagedEventSubscription,
+    state: &mut TuiState,
+    terminal_targets: &mut HashSet<TargetInstance>,
+    dropped_events: &mut u64,
+) -> ManagedDrainState {
+    drain_managed_events(events, dropped_events, |event| {
+        process_tui_event(event, state, terminal_targets);
+        Ok::<(), std::convert::Infallible>(())
+    })
+    .expect("processing TUI managed events is infallible")
 }
 
-fn should_join_group_after_idle(
-    args: &TuiArgs,
-    saw_target_event: bool,
-    last_event_at: Instant,
-) -> bool {
-    if args.is_continuous() && saw_target_event {
-        return false;
+fn drain_final_tui_events(
+    events: &mut ManagedEventSubscription,
+    state: &mut TuiState,
+    terminal_targets: &mut HashSet<TargetInstance>,
+    dropped_events: &mut u64,
+) {
+    loop {
+        match events.try_recv() {
+            Ok(event) => process_tui_event(event, state, terminal_targets),
+            Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break,
+            Err(ManagedEventTryRecvError::Lagged(count)) => {
+                *dropped_events = dropped_events.saturating_add(count);
+            }
+        }
     }
-    last_event_at.elapsed() > estimated_group_completion_grace(args)
 }
-
 fn handle_input(state: &mut TuiState, shutdown_requested: &AtomicBool) -> io::Result<bool> {
     let mut force_render = false;
     while event::poll(Duration::ZERO)? {
@@ -377,12 +224,10 @@ fn handle_input(state: &mut TuiState, shutdown_requested: &AtomicBool) -> io::Re
             continue;
         }
         match key.code {
-            KeyCode::Char('q') => {
-                state.quit_requested = true;
-                shutdown_requested.store(true, Ordering::Relaxed);
-                force_render = true;
-            }
-            KeyCode::Char('c') if key.modifiers.contains(KeyModifiers::CONTROL) => {
+            KeyCode::Char('q') | KeyCode::Char('c')
+                if key.code == KeyCode::Char('q')
+                    || key.modifiers.contains(KeyModifiers::CONTROL) =>
+            {
                 state.quit_requested = true;
                 shutdown_requested.store(true, Ordering::Relaxed);
                 force_render = true;
@@ -444,7 +289,6 @@ fn handle_input(state: &mut TuiState, shutdown_requested: &AtomicBool) -> io::Re
     }
     Ok(force_render)
 }
-
 fn render_if_due(
     terminal: &mut TuiTerminal,
     state: &TuiState,
@@ -452,27 +296,12 @@ fn render_if_due(
     force: bool,
 ) -> io::Result<()> {
     let now = Instant::now();
-    if !should_render(now, *next_render, state.paused, force) {
-        return Ok(());
-    }
+    if should_render(now, *next_render, state.paused, force) {
         terminal.draw(state)?;
         *next_render = now + RENDER_INTERVAL;
+    }
     Ok(())
 }
-
-fn managed_event_subscriber_config() -> SubscriberConfig {
-    SubscriberConfig {
-        capacity: MANAGED_EVENT_CAPACITY,
-        overflow: SubscriberOverflow::DropOldest,
-    }
-}
-
-fn drain_single_tui_events(events: &EventSubscription, state: &mut TuiState) {
-    while let Ok(Some(event)) = events.try_recv() {
-        state.process_event(&event);
-    }
-}
-
 fn managed_tui_wait_duration(next_render: &Instant, paused: bool) -> Duration {
     let render_wait = if paused {
         TUI_WAIT_SLICE
@@ -481,152 +310,3 @@ fn managed_tui_wait_duration(next_render: &Instant, paused: bool) -> Duration {
     };
     render_wait.min(TUI_WAIT_SLICE)
 }
-
-fn wait_for_tui_activity(
-    next_send_deadline: Option<Instant>,
-    next_render: &mut Instant,
-    state: &mut TuiState,
-    terminal: &mut TuiTerminal,
-    shutdown_requested: &AtomicBool,
-) -> io::Result<()> {
-    let wait_for = tui_wait_duration(next_send_deadline, *next_render, state.paused);
-    if wait_for.is_zero() || !event::poll(wait_for)? {
-        return Ok(());
-    }
-
-    if handle_input(state, shutdown_requested)? {
-        render_if_due(terminal, state, next_render, true)?;
-    }
-    Ok(())
-}
-
-fn tui_wait_duration(
-    next_send_deadline: Option<Instant>,
-    next_render: Instant,
-    paused: bool,
-) -> Duration {
-    let now = Instant::now();
-    let send_wait = next_send_deadline
-        .map(|deadline| deadline.saturating_duration_since(now))
-        .unwrap_or(IDLE_SLEEP);
-    let render_wait = if paused {
-        send_wait
-    } else {
-        next_render.saturating_duration_since(now)
-    };
-    send_wait.min(render_wait).min(TUI_WAIT_SLICE)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use clap::Parser;
-
-    fn parse(args: &[&str]) -> TuiArgs {
-        let mut argv = vec!["irtt-tui"];
-        argv.extend_from_slice(args);
-        TuiArgs::try_parse_from(argv).unwrap()
-    }
-
-    #[test]
-    fn continuous_multi_target_idle_does_not_stop_after_events() {
-        let args = parse(&["127.0.0.1:2112", "127.0.0.2:2112"]);
-        let old_event_at = Instant::now() - estimated_group_completion_grace(&args) - IDLE_SLEEP;
-
-        assert!(!should_join_group_after_idle(&args, true, old_event_at));
-    }
-
-    #[test]
-    fn finite_or_unopened_group_can_leave_after_idle_grace() {
-        let finite = parse(&["--duration", "1s", "127.0.0.1:2112", "127.0.0.2:2112"]);
-        let finite_old_event_at =
-            Instant::now() - estimated_group_completion_grace(&finite) - IDLE_SLEEP;
-        assert!(should_join_group_after_idle(
-            &finite,
-            true,
-            finite_old_event_at
-        ));
-
-        let continuous = parse(&["127.0.0.1:2112", "127.0.0.2:2112"]);
-        let unopened_old_event_at =
-            Instant::now() - estimated_group_completion_grace(&continuous) - IDLE_SLEEP;
-        assert!(should_join_group_after_idle(
-            &continuous,
-            false,
-            unopened_old_event_at
-        ));
-    }
-
-    #[test]
-    fn protective_group_exits_stop_before_joining() {
-        assert!(GroupLoopExit::Interrupted.should_stop_before_join());
-        assert!(GroupLoopExit::IdleGraceElapsed.should_stop_before_join());
-        assert!(GroupLoopExit::SubscriptionDisconnected.should_stop_before_join());
-        assert!(!GroupLoopExit::AllTargetsTerminal.should_stop_before_join());
-    }
-
-    #[test]
-    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,
-            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,
-        ));
-
-        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,
-            1,
-            &mut peer_close_requested_stop,
-            &mut stop_requested,
-        ));
-    }
-
-    #[test]
-    fn single_target_managed_drain_consumes_queued_terminal_event() {
-        let hub = irtt_client::EventHub::new();
-        let events = hub
-            .subscribe(SubscriberConfig {
-                capacity: 4,
-                overflow: SubscriberOverflow::DropOldest,
-            })
-            .unwrap();
-        hub.publish(irtt_client::ClientEvent::SessionClosed {
-            remote: "127.0.0.1:2112".parse().unwrap(),
-            token: 7,
-            at: irtt_client::ClientTimestamp::now(),
-        });
-        hub.disconnect_all();
-        let mut state = TuiState::new(TuiConfig::default());
-
-        drain_single_tui_events(&events, &mut state);
-
-        assert_eq!(state.status(), TuiStatus::Complete);
-        assert_eq!(
-            events.try_recv().unwrap_err(),
-            EventSubscriptionError::Disconnected
-        );
-    }
-}

crates/irtt-cli/src/cmd/tui/ui.rs

@@ -9,10 +9,8 @@ use crossterm::{
     execute,
     terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen},
 };
-use irtt_client::{
-    ClientEvent, ManagedTargetEndReason, ManagedTargetOutcome, NegotiatedParams, SignedDuration,
-    TargetEvent,
-};
+use irtt_client::managed::{ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance};
+use irtt_client::{ClientEvent, NegotiatedParams, SignedDuration};
 use irtt_stats::{Snapshot, StatsCollector, TimeStats};
 use ratatui::{
     backend::CrosstermBackend,
@@ -140,12 +138,8 @@ impl TuiState {
         }
     }
 
-    pub(super) fn process_event(&mut self, event: &ClientEvent) {
-        self.process_event_for_target(0, event);
-    }
-
-    pub(super) fn process_target_event(&mut self, event: &TargetEvent) {
-        let label = event.target.as_str();
+    pub(super) fn process_target_event(&mut self, target: &TargetInstance, event: &ClientEvent) {
+        let label = target.id.as_str();
         let idx = if let Some(idx) = self.target_index.get(label).copied() {
             idx
         } else {
@@ -159,11 +153,11 @@ impl TuiState {
             self.target_index.insert(label.to_owned(), idx);
             idx
         };
-        self.process_event_for_target(idx, &event.event);
+        self.process_event_for_target(idx, event);
     }
 
     pub(super) fn process_target_outcome(&mut self, outcome: &ManagedTargetOutcome) {
-        let label = outcome.id.as_str();
+        let label = outcome.target.id.as_str();
         let idx = if let Some(idx) = self.target_index.get(label).copied() {
             idx
         } else {
@@ -176,7 +170,7 @@ impl TuiState {
             idx
         };
 
-        let (status, recent, warning) = match &outcome.end_reason {
+        let (status, recent, primary_warning) = match &outcome.end_reason {
             ManagedTargetEndReason::TestComplete => {
                 (TargetStatus::Closed, "test completed".to_owned(), None)
             }
@@ -190,15 +184,17 @@ impl TuiState {
                 "no-test negotiation completed".to_owned(),
                 None,
             ),
-            ManagedTargetEndReason::Cancelled => {
-                (TargetStatus::Cancelled, "target cancelled".to_owned(), None)
+            ManagedTargetEndReason::Stopped => {
+                (TargetStatus::Stopped, "target stopped".to_owned(), None)
             }
             ManagedTargetEndReason::Removed => {
                 (TargetStatus::Removed, "target removed".to_owned(), None)
             }
-            ManagedTargetEndReason::OpenFailed(failure)
-            | ManagedTargetEndReason::Failed(failure) => {
-                let message = format!("{}: {}", failure.kind, failure.message);
+            ManagedTargetEndReason::Replaced => {
+                (TargetStatus::Replaced, "target replaced".to_owned(), None)
+            }
+            ManagedTargetEndReason::Failed(failure) => {
+                let message = format!("{:?}: {}", failure.kind, failure.message);
                 (
                     TargetStatus::Failed,
                     format!("target failed: {message}"),
@@ -207,8 +203,21 @@ impl TuiState {
             }
         };
 
+        let cleanup_warning = outcome.cleanup_failure.as_ref().map(|failure| {
+            format!(
+                "cleanup failed ({:?} {:?}): {}",
+                failure.phase, failure.kind, failure.message
+            )
+        });
+        let warning = match (primary_warning, cleanup_warning.as_ref()) {
+            (Some(primary), Some(cleanup)) => Some(format!("{primary}; {cleanup}")),
+            (Some(primary), None) => Some(primary),
+            (None, Some(cleanup)) => Some(cleanup.clone()),
+            (None, None) => None,
+        };
+
         if let Some(target) = self.targets.get_mut(idx) {
-            target.remote = Some(outcome.remote.to_string());
+            target.remote = outcome.remote.map(|remote| remote.to_string());
             target.status = status;
             if let Some(warning) = &warning {
                 target.last_warning = Some(warning.clone());
@@ -218,6 +227,9 @@ impl TuiState {
             self.last_warning = Some(warning);
         }
         self.push_event(format!("{label}: {recent}"));
+        if let Some(cleanup_warning) = cleanup_warning {
+            self.push_event(format!("{label}: {cleanup_warning}"));
+        }
 
         if self
             .targets
@@ -382,17 +394,8 @@ impl TuiState {
         self.status = status;
     }
 
-    #[cfg(test)]
-    pub(super) fn status(&self) -> TuiStatus {
-        self.status
-    }
-
-    pub(super) fn mark_dropped_client_events(&mut self, dropped_events: u64) {
-        self.mark_dropped_events(dropped_events, "managed client");
-    }
-
-    pub(super) fn mark_dropped_group_events(&mut self, dropped_events: u64) {
-        self.mark_dropped_events(dropped_events, "managed group");
+    pub(super) fn mark_dropped_managed_events(&mut self, dropped_events: u64) {
+        self.mark_dropped_events(dropped_events, "managed run");
     }
 
     fn mark_dropped_events(&mut self, dropped_events: u64, source: &str) {
@@ -566,8 +569,9 @@ pub(super) enum TargetStatus {
     Closed,
     Failed,
     NoTest,
-    Cancelled,
+    Stopped,
     Removed,
+    Replaced,
     Unknown,
 }
 
@@ -579,8 +583,9 @@ impl TargetStatus {
             Self::Closed => "closed",
             Self::Failed => "failed",
             Self::NoTest => "no-test",
-            Self::Cancelled => "cancelled",
+            Self::Stopped => "stopped",
             Self::Removed => "removed",
+            Self::Replaced => "replaced",
             Self::Unknown => "unknown",
         }
     }
@@ -588,7 +593,12 @@ impl TargetStatus {
     fn is_terminal(self) -> bool {
         matches!(
             self,
-            Self::Closed | Self::Failed | Self::NoTest | Self::Cancelled | Self::Removed
+            Self::Closed
+                | Self::Failed
+                | Self::NoTest
+                | Self::Stopped
+                | Self::Removed
+                | Self::Replaced
         )
     }
 
@@ -690,7 +700,6 @@ impl Default for TuiConfig {
 pub(super) enum TuiStatus {
     Opening,
     Running,
-    Draining,
     Interrupted,
     Closing,
     Complete,
@@ -702,7 +711,6 @@ impl TuiStatus {
         match self {
             Self::Opening => "opening",
             Self::Running => "running",
-            Self::Draining => "draining",
             Self::Interrupted => "interrupted",
             Self::Closing => "closing",
             Self::Complete => "complete",
@@ -2030,12 +2038,15 @@ fn duration_ns(value: Duration) -> i128 {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use irtt_client::managed::{
+        ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetFailurePhase, TargetId,
+    };
     use irtt_client::{
-        ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, ServerTiming, TargetId,
-        WarningKind,
+        ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, ServerTiming, WarningKind,
     };
     use std::{
         net::{IpAddr, Ipv4Addr, SocketAddr},
+        sync::Arc,
         time::UNIX_EPOCH,
     };
 
@@ -2139,23 +2150,25 @@ mod tests {
         &state.targets[0]
     }
 
-    fn target_event(label: &str, event: ClientEvent) -> TargetEvent {
-        TargetEvent {
-            target: TargetId::from(label),
-            event,
+    fn target_instance(label: &str) -> TargetInstance {
+        TargetInstance {
+            id: TargetId::from(label),
+            generation: 1,
         }
     }
 
     fn target_outcome(label: &str, end_reason: ManagedTargetEndReason) -> ManagedTargetOutcome {
         ManagedTargetOutcome {
-            id: TargetId::from(label),
-            remote: remote(),
+            target: target_instance(label),
+            server_addr: Arc::from("127.0.0.1:2112"),
+            remote: Some(remote()),
             end_reason,
             packets_sent: 0,
             replies_received: 0,
             duplicates: 0,
             late: 0,
             warning_events: 0,
+            cleanup_failure: None,
         }
     }
 
@@ -2173,7 +2186,8 @@ mod tests {
     #[test]
     fn session_started_sets_remote_session_and_running_status() {
         let mut state = TuiState::default();
-        state.process_event(&ClientEvent::SessionStarted {
+        let target = target_instance("target");
+        let event = ClientEvent::SessionStarted {
             remote: remote(),
             token: 0xabc,
             negotiated: NegotiatedParams {
@@ -2181,7 +2195,8 @@ mod tests {
                 restrictions: Vec::new(),
             },
             at: ts(Duration::ZERO),
-        });
+        };
+        state.process_target_event(&target, &event);
 
         let target = primary_target(&state);
         assert_eq!(target.remote.as_deref(), Some("127.0.0.1:2112"));
@@ -2193,8 +2208,10 @@ mod tests {
     #[test]
     fn echo_reply_appends_bounded_primary_history() {
         let mut state = TuiState::default();
+        let target = target_instance("target");
         for seq in 0..(HISTORY_LIMIT as u32 + 3) {
-            state.process_event(&reply(seq, i128::from(seq) * 1_000));
+            let event = reply(seq, i128::from(seq) * 1_000);
+            state.process_target_event(&target, &event);
         }
 
         let target = primary_target(&state);
@@ -2213,13 +2230,15 @@ mod tests {
     #[test]
     fn duplicate_and_late_replies_do_not_append_primary_history() {
         let mut state = TuiState::default();
-        state.process_event(&ClientEvent::DuplicateReply {
+        let target = target_instance("target");
+        let duplicate = ClientEvent::DuplicateReply {
             seq: 7,
             remote: remote(),
             received_at: ts(Duration::from_secs(1)),
             bytes: 64,
-        });
-        state.process_event(&ClientEvent::LateReply {
+        };
+        state.process_target_event(&target, &duplicate);
+        let late = ClientEvent::LateReply {
             seq: 8,
             highest_seen: 9,
             remote: remote(),
@@ -2231,7 +2250,8 @@ mod tests {
             received_stats: None,
             bytes: 64,
             packet_meta: PacketMeta::default(),
-        });
+        };
+        state.process_target_event(&target, &late);
 
         let target = primary_target(&state);
         assert!(target.graph_history.is_empty());
@@ -2258,7 +2278,9 @@ mod tests {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
 
-        state.process_target_event(&target_event("b", reply(11, 2_500_000)));
+        let target = target_instance("b");
+        let event = reply(11, 2_500_000);
+        state.process_target_event(&target, &event);
 
         assert!(state.targets[0].graph_history.is_empty());
         assert_eq!(state.targets[0].stats.snapshot().packets.unique_replies, 0);
@@ -2276,18 +2298,16 @@ mod tests {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
 
-        state.process_target_event(&target_event(
-            "a",
-            ClientEvent::DuplicateReply {
+        let a = target_instance("a");
+        let duplicate = ClientEvent::DuplicateReply {
             seq: 7,
             remote: remote(),
             received_at: ts(Duration::from_secs(1)),
             bytes: 64,
-            },
-        ));
-        state.process_target_event(&target_event(
-            "b",
-            ClientEvent::LateReply {
+        };
+        state.process_target_event(&a, &duplicate);
+        let b = target_instance("b");
+        let late = ClientEvent::LateReply {
             seq: 8,
             highest_seen: 9,
             remote: remote(),
@@ -2299,8 +2319,8 @@ mod tests {
             received_stats: None,
             bytes: 64,
             packet_meta: PacketMeta::default(),
-            },
-        ));
+        };
+        state.process_target_event(&b, &late);
 
         assert!(state.targets[0].graph_history.is_empty());
         assert!(state.targets[1].graph_history.is_empty());
@@ -2314,33 +2334,29 @@ mod tests {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
 
-        state.process_target_event(&target_event(
-            "a",
-            ClientEvent::EchoLoss {
+        let a = target_instance("a");
+        let loss = ClientEvent::EchoLoss {
             seq: 3,
             sent_at: ts(Duration::from_millis(3)),
             timeout_at: Instant::now(),
-            },
-        ));
-        state.process_target_event(&target_event(
-            "b",
-            ClientEvent::Warning {
+        };
+        state.process_target_event(&a, &loss);
+        let b = target_instance("b");
+        let warning = ClientEvent::Warning {
             kind: WarningKind::WrongToken,
             message: "wrong token".to_owned(),
             at: ts(Duration::ZERO),
-            },
-        ));
-        state.process_target_event(&target_event(
-            "a",
-            ClientEvent::NoTestCompleted {
+        };
+        state.process_target_event(&b, &warning);
+        let no_test = ClientEvent::NoTestCompleted {
             remote: remote(),
             negotiated: NegotiatedParams {
                 params: irtt_proto::Params::default(),
                 restrictions: Vec::new(),
             },
             at: ts(Duration::ZERO),
-            },
-        ));
+        };
+        state.process_target_event(&a, &no_test);
 
         assert_eq!(state.targets[0].stats.snapshot().events.loss_events, 1);
         assert_eq!(state.targets[0].status, TargetStatus::NoTest);
@@ -2357,30 +2373,28 @@ mod tests {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
 
-        state.process_target_event(&target_event(
-            "a",
-            ClientEvent::NoTestCompleted {
+        let a = target_instance("a");
+        let no_test = ClientEvent::NoTestCompleted {
             remote: remote(),
             negotiated: NegotiatedParams {
                 params: irtt_proto::Params::default(),
                 restrictions: Vec::new(),
             },
             at: ts(Duration::ZERO),
-            },
-        ));
+        };
+        state.process_target_event(&a, &no_test);
 
         assert_eq!(state.targets[0].status, TargetStatus::NoTest);
         assert_eq!(state.targets[1].status, TargetStatus::Opening);
         assert_eq!(state.status, TuiStatus::Running);
 
-        state.process_target_event(&target_event(
-            "b",
-            ClientEvent::SessionClosed {
+        let b = target_instance("b");
+        let closed = ClientEvent::SessionClosed {
             remote: remote(),
             token: 0xabc,
             at: ts(Duration::ZERO),
-            },
-        ));
+        };
+        state.process_target_event(&b, &closed);
 
         assert_eq!(state.targets[1].status, TargetStatus::Closed);
         assert_eq!(state.status, TuiStatus::Complete);
@@ -2390,14 +2404,15 @@ mod tests {
     fn terminal_failure_marks_the_correct_target_failed() {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
-        let failure = irtt_client::ManagedTargetFailure {
-            kind: irtt_client::ManagedTargetFailureKind::OpeningTimeout,
-            message: "all open requests timed out".to_owned(),
+        let failure = ManagedTargetFailure {
+            phase: ManagedTargetFailurePhase::Opening,
+            kind: ManagedTargetFailureKind::Timeout,
+            message: "all open requests timed out".into(),
         };
 
         state.process_target_outcome(&target_outcome(
             "b",
-            ManagedTargetEndReason::OpenFailed(failure),
+            ManagedTargetEndReason::Failed(failure),
         ));
 
         assert_eq!(state.targets[0].status, TargetStatus::Opening);
@@ -2405,7 +2420,7 @@ mod tests {
         assert!(state.targets[1]
             .last_warning
             .as_deref()
-            .is_some_and(|warning| warning.contains("opening timeout")));
+            .is_some_and(|warning| warning.contains("Timeout")));
     }
 
     #[test]
@@ -2415,9 +2430,10 @@ mod tests {
         state.process_target_outcome(&target_outcome("a", ManagedTargetEndReason::PeerClosed));
         state.process_target_outcome(&target_outcome(
             "b",
-            ManagedTargetEndReason::Failed(irtt_client::ManagedTargetFailure {
-                kind: irtt_client::ManagedTargetFailureKind::RuntimeProtocol,
-                message: "pending probe limit exceeded".to_owned(),
+            ManagedTargetEndReason::Failed(ManagedTargetFailure {
+                phase: ManagedTargetFailurePhase::Receiving,
+                kind: ManagedTargetFailureKind::Protocol,
+                message: "pending probe limit exceeded".into(),
             }),
         ));
 
@@ -2430,6 +2446,79 @@ mod tests {
             .is_some_and(|warning| warning.contains("pending probe limit")));
     }
 
+    #[test]
+    fn cleanup_failure_warns_without_changing_successful_target_status() {
+        let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]);
+        let mut outcome = target_outcome("a", ManagedTargetEndReason::TestComplete);
+        outcome.remote = None;
+        outcome.cleanup_failure = Some(ManagedTargetFailure {
+            phase: ManagedTargetFailurePhase::Closing,
+            kind: ManagedTargetFailureKind::Socket,
+            message: "close datagram failed".into(),
+        });
+
+        state.process_target_outcome(&outcome);
+
+        assert_eq!(state.targets[0].status, TargetStatus::Closed);
+        assert_eq!(state.targets[0].remote, None);
+        assert!(state.targets[0]
+            .last_warning
+            .as_deref()
+            .is_some_and(|warning| warning.contains("cleanup failed (Closing Socket)")));
+        assert!(state
+            .recent_events
+            .iter()
+            .any(|event| event.contains("cleanup failed (Closing Socket)")));
+    }
+
+    #[test]
+    fn primary_and_cleanup_failures_are_both_visible() {
+        let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]);
+        let mut outcome = target_outcome(
+            "a",
+            ManagedTargetEndReason::Failed(ManagedTargetFailure {
+                phase: ManagedTargetFailurePhase::Receiving,
+                kind: ManagedTargetFailureKind::Protocol,
+                message: "primary failure".into(),
+            }),
+        );
+        outcome.cleanup_failure = Some(ManagedTargetFailure {
+            phase: ManagedTargetFailurePhase::Closing,
+            kind: ManagedTargetFailureKind::Socket,
+            message: "cleanup failure".into(),
+        });
+
+        state.process_target_outcome(&outcome);
+
+        assert_eq!(state.targets[0].status, TargetStatus::Failed);
+        assert!(state.targets[0]
+            .last_warning
+            .as_deref()
+            .is_some_and(|warning| warning.contains("primary failure")
+                && warning.contains("cleanup failure")));
+        assert!(state
+            .recent_events
+            .iter()
+            .any(|event| event.contains("target failed: Protocol: primary failure")));
+        assert!(state
+            .recent_events
+            .iter()
+            .any(|event| event.contains("cleanup failed (Closing Socket): cleanup failure")));
+    }
+
+    #[test]
+    fn managed_outcomes_map_stopped_removed_and_replaced_statuses() {
+        for (reason, expected) in [
+            (ManagedTargetEndReason::Stopped, TargetStatus::Stopped),
+            (ManagedTargetEndReason::Removed, TargetStatus::Removed),
+            (ManagedTargetEndReason::Replaced, TargetStatus::Replaced),
+        ] {
+            let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]);
+            state.process_target_outcome(&target_outcome("a", reason));
+            assert_eq!(state.targets[0].status, expected);
+        }
+    }
+
     #[test]
     fn run_error_preserves_peer_closed_target_status() {
         let mut state = TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned()]);
@@ -2454,9 +2543,10 @@ mod tests {
         for label in ["a", "b"] {
             state.process_target_outcome(&target_outcome(
                 label,
-                ManagedTargetEndReason::OpenFailed(irtt_client::ManagedTargetFailure {
-                    kind: irtt_client::ManagedTargetFailureKind::OpeningProtocol,
-                    message: "server returned a zero connection token".to_owned(),
+                ManagedTargetEndReason::Failed(ManagedTargetFailure {
+                    phase: ManagedTargetFailurePhase::Opening,
+                    kind: ManagedTargetFailureKind::Protocol,
+                    message: "server returned a zero connection token".into(),
                 }),
             ));
         }
@@ -2473,8 +2563,12 @@ mod tests {
         let mut state =
             TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
 
-        state.process_target_event(&target_event("a", reply(1, 1_000_000)));
-        state.process_target_event(&target_event("b", reply(2, 2_000_000)));
+        let a = target_instance("a");
+        let a_reply = reply(1, 1_000_000);
+        state.process_target_event(&a, &a_reply);
+        let b = target_instance("b");
+        let b_reply = reply(2, 2_000_000);
+        state.process_target_event(&b, &b_reply);
         state.clear_visible_history();
 
         assert!(state.targets[0].graph_history.is_empty());
@@ -2488,11 +2582,13 @@ mod tests {
     #[test]
     fn warning_updates_recent_events_and_last_warning() {
         let mut state = TuiState::default();
-        state.process_event(&ClientEvent::Warning {
+        let target = target_instance("target");
+        let event = ClientEvent::Warning {
             kind: WarningKind::WrongToken,
             message: "wrong token".to_owned(),
             at: ts(Duration::ZERO),
-        });
+        };
+        state.process_target_event(&target, &event);
 
         assert_eq!(
             state.last_warning.as_deref(),
@@ -2509,7 +2605,7 @@ mod tests {
     fn dropped_events_mark_tui_statistics_incomplete() {
         let mut state = TuiState::default();
 
-        state.mark_dropped_group_events(9);
+        state.mark_dropped_managed_events(9);
 
         assert_eq!(state.dropped_events, 9);
         assert!(state
@@ -2519,20 +2615,20 @@ mod tests {
         assert!(state
             .recent_events
             .back()
-            .is_some_and(|event| event.contains("dropped 9 managed group events")));
+            .is_some_and(|event| event.contains("dropped 9 managed run events")));
     }
 
     #[test]
     fn dropped_single_client_events_are_visible() {
         let mut state = TuiState::default();
 
-        state.mark_dropped_client_events(3);
+        state.mark_dropped_managed_events(3);
 
         assert_eq!(state.dropped_events, 3);
         assert!(state
             .recent_events
             .back()
-            .is_some_and(|event| event.contains("dropped 3 managed client events")));
+            .is_some_and(|event| event.contains("dropped 3 managed run events")));
     }
 
     #[test]
@@ -2811,7 +2907,9 @@ mod tests {
     #[test]
     fn sample_details_preserve_signed_values() {
         let mut state = TuiState::default();
-        state.process_event(&reply(4, -1_250_000));
+        let target = target_instance("target");
+        let event = reply(4, -1_250_000);
+        state.process_target_event(&target, &event);
 
         let target = primary_target(&state);
         let sample = target.last_sample.unwrap();

crates/irtt-cli/src/shared/client/mod.rs

@@ -10,6 +10,6 @@ pub use args::{
 pub use config::{expected_probe_count, DEFAULT_RECV_TIMEOUT};
 pub use session::is_shutdown_requested;
 pub use targets::{
-    parse_labelled_target, resolved_managed_targets, target_specs, GroupPacingArg,
-    LabelledTargetArg, ResolvedTarget, TargetSpec,
+    parse_labelled_target, prepare_managed_targets, target_specs, GroupPacingArg,
+    LabelledTargetArg, PreparedTarget, TargetSpec,
 };

crates/irtt-cli/src/shared/client/session.rs

@@ -1,5 +1,40 @@
 use std::sync::atomic::{AtomicBool, Ordering};
 
+#[cfg(feature = "client-runtime")]
+use irtt_client::managed::{ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError};
+
+#[cfg(feature = "client-runtime")]
+pub(crate) const MANAGED_EVENT_WORK_BUDGET: usize = 128;
+
+#[cfg(feature = "client-runtime")]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum ManagedDrainState {
+    Empty,
+    Closed,
+    BudgetExhausted,
+}
+
+/// Drain a bounded amount of lossy presentation work before returning to the
+/// frontend control loop.
+#[cfg(feature = "client-runtime")]
+pub(crate) fn drain_managed_events<E>(
+    events: &mut ManagedEventSubscription,
+    dropped_events: &mut u64,
+    mut process: impl FnMut(ManagedEvent) -> Result<(), E>,
+) -> Result<ManagedDrainState, E> {
+    for _ in 0..MANAGED_EVENT_WORK_BUDGET {
+        match events.try_recv() {
+            Ok(event) => process(event)?,
+            Err(ManagedEventTryRecvError::Lagged(count)) => {
+                *dropped_events = dropped_events.saturating_add(count);
+            }
+            Err(ManagedEventTryRecvError::Empty) => return Ok(ManagedDrainState::Empty),
+            Err(ManagedEventTryRecvError::Closed) => return Ok(ManagedDrainState::Closed),
+        }
+    }
+    Ok(ManagedDrainState::BudgetExhausted)
+}
+
 pub fn is_shutdown_requested(shutdown_requested: &AtomicBool) -> bool {
     shutdown_requested.load(Ordering::Relaxed)
 }
@@ -27,21 +62,19 @@ pub fn peer_close_run_error(
     ))
 }
 
-pub fn request_group_stop_for_peer_close(
+pub fn request_managed_stop_for_peer_close(
     continuous: bool,
     interrupted: bool,
-    peer_closed_target_count: u64,
-    peer_close_requested_stop: &mut bool,
+    peer_closed_target_outcomes: u64,
     stop_requested: &mut bool,
 ) -> bool {
-    if !continuous || interrupted || peer_closed_target_count == 0 {
+    if !continuous || interrupted || peer_closed_target_outcomes == 0 {
         return false;
     }
-    *peer_close_requested_stop = true;
-    request_group_stop_once(stop_requested)
+    request_managed_stop_once(stop_requested)
 }
 
-pub fn request_group_stop_once(stop_requested: &mut bool) -> bool {
+pub fn request_managed_stop_once(stop_requested: &mut bool) -> bool {
     if *stop_requested {
         return false;
     }
@@ -70,46 +103,38 @@ mod tests {
 
     #[test]
     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_for_peer_close(
+        assert!(!request_managed_stop_for_peer_close(
             false,
             false,
             1,
-            &mut peer_close_requested_stop,
             &mut stop_requested,
         ));
-        assert!(!request_group_stop_for_peer_close(
+        assert!(!request_managed_stop_for_peer_close(
             true,
             true,
             1,
-            &mut peer_close_requested_stop,
             &mut stop_requested,
         ));
-        assert!(!request_group_stop_for_peer_close(
+        assert!(!request_managed_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(
+        assert!(request_managed_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(
+        assert!(!request_managed_stop_for_peer_close(
             true,
             false,
             1,
-            &mut peer_close_requested_stop,
             &mut stop_requested,
         ));
     }

crates/irtt-cli/src/shared/client/targets.rs

@@ -1,7 +1,7 @@
-use std::{collections::HashSet, net::ToSocketAddrs};
+use std::collections::HashSet;
 
 use clap::ValueEnum;
-use irtt_client::{ClientConfig, ManagedGroupPacing, ManagedTargetConfig, TargetId};
+use irtt_client::managed::{ManagedPacing, ManagedTargetConfig, TargetId};
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct LabelledTargetArg {
@@ -16,7 +16,7 @@ pub struct TargetSpec {
 }
 
 #[derive(Debug, Clone)]
-pub struct ResolvedTarget {
+pub struct PreparedTarget {
     pub label: String,
     pub managed: ManagedTargetConfig,
 }
@@ -27,7 +27,7 @@ pub enum GroupPacingArg {
     Burst,
 }
 
-impl From<GroupPacingArg> for ManagedGroupPacing {
+impl From<GroupPacingArg> for ManagedPacing {
     fn from(value: GroupPacingArg) -> Self {
         match value {
             GroupPacingArg::Staggered => Self::Staggered,
@@ -93,118 +93,21 @@ pub fn target_specs(
     Ok(specs)
 }
 
-pub fn resolved_managed_targets(
-    specs: Vec<TargetSpec>,
-    config: &ClientConfig,
-) -> Result<Vec<ResolvedTarget>, String> {
-    let mut remotes = HashSet::new();
+pub fn prepare_managed_targets(specs: Vec<TargetSpec>) -> Result<Vec<PreparedTarget>, String> {
     let mut targets = Vec::with_capacity(specs.len());
     for spec in specs {
-        let remote = resolve_target(&spec.addr, config).map_err(|err| {
-            format!(
-                "failed to resolve target {} ({:?}): {err}",
-                spec.label, spec.addr
-            )
-        })?;
-        if !remotes.insert(remote) {
-            return Err(format!(
-                "duplicate resolved target address {remote} for label {}",
-                spec.label
-            ));
-        }
-        targets.push(ResolvedTarget {
+        targets.push(PreparedTarget {
             label: spec.label.clone(),
-            managed: ManagedTargetConfig {
-                id: TargetId::from(spec.label),
-                remote,
-                auth: None,
-            },
+            managed: ManagedTargetConfig::new(TargetId::from(spec.label), spec.addr),
         });
     }
     Ok(targets)
 }
 
-pub fn resolve_target(addr: &str, config: &ClientConfig) -> Result<std::net::SocketAddr, String> {
-    let normalized = normalize_target_addr(addr);
-    let mut addrs = normalized
-        .to_socket_addrs()
-        .map_err(|_| format!("failed to resolve address {normalized:?}"))?;
-    addrs
-        .find(|addr| {
-            (!config.socket_config.ipv4_only || addr.is_ipv4())
-                && (!config.socket_config.ipv6_only || addr.is_ipv6())
-        })
-        .ok_or_else(|| format!("failed to resolve address {normalized:?}"))
-}
-
-pub fn normalize_target_addr(addr: &str) -> String {
-    const DEFAULT_PORT: u16 = 2112;
-    if addr.parse::<std::net::SocketAddr>().is_ok() {
-        return addr.to_owned();
-    }
-    if addr.starts_with('[') && addr.ends_with(']') {
-        return format!("{addr}:{DEFAULT_PORT}");
-    }
-    if addr.starts_with('[') {
-        return addr.to_owned();
-    }
-    if addr.parse::<std::net::Ipv6Addr>().is_ok() {
-        return format!("[{addr}]:{DEFAULT_PORT}");
-    }
-    if addr
-        .rsplit_once(':')
-        .is_some_and(|(_, port)| port.parse::<u16>().is_ok())
-    {
-        return addr.to_owned();
-    }
-    format!("{addr}:{DEFAULT_PORT}")
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
 
-    #[test]
-    fn target_normalization_matches_single_target_forms() {
-        assert_eq!(normalize_target_addr("127.0.0.1"), "127.0.0.1:2112");
-        assert_eq!(normalize_target_addr("127.0.0.1:1234"), "127.0.0.1:1234");
-        assert_eq!(normalize_target_addr("example.com"), "example.com:2112");
-        assert_eq!(
-            normalize_target_addr("example.com:1234"),
-            "example.com:1234"
-        );
-        assert_eq!(normalize_target_addr("[::1]"), "[::1]:2112");
-        assert_eq!(normalize_target_addr("[::1]:1234"), "[::1]:1234");
-        assert_eq!(normalize_target_addr("::1"), "[::1]:2112");
-    }
-
-    #[test]
-    fn target_resolution_respects_family_filters() {
-        let config = ClientConfig::default();
-        assert_eq!(
-            resolve_target("127.0.0.1", &config).unwrap(),
-            "127.0.0.1:2112".parse().unwrap()
-        );
-        assert_eq!(
-            resolve_target("[::1]", &config).unwrap(),
-            "[::1]:2112".parse().unwrap()
-        );
-        assert_eq!(
-            resolve_target("::1", &config).unwrap(),
-            "[::1]:2112".parse().unwrap()
-        );
-
-        let mut ipv4_only = ClientConfig::default();
-        ipv4_only.socket_config.ipv4_only = true;
-        assert!(resolve_target("127.0.0.1", &ipv4_only).unwrap().is_ipv4());
-        assert!(resolve_target("::1", &ipv4_only).is_err());
-
-        let mut ipv6_only = ClientConfig::default();
-        ipv6_only.socket_config.ipv6_only = true;
-        assert!(resolve_target("::1", &ipv6_only).unwrap().is_ipv6());
-        assert!(resolve_target("127.0.0.1", &ipv6_only).is_err());
-    }
-
     #[test]
     fn target_specs_suffix_repeated_positionals_and_reject_duplicate_labels() {
         let positionals = vec!["host-a:2112".to_owned(), "host-a:2112".to_owned()];
@@ -222,19 +125,19 @@ mod tests {
     }
 
     #[test]
-    fn duplicate_resolved_target_addresses_are_rejected() {
+    fn target_preparation_preserves_hostname_and_allows_duplicate_endpoints() {
         let specs = vec![
             TargetSpec {
                 label: "a".to_owned(),
-                addr: "127.0.0.1:2112".to_owned(),
+                addr: "example.test".to_owned(),
             },
             TargetSpec {
                 label: "b".to_owned(),
-                addr: "127.0.0.1".to_owned(),
+                addr: "example.test".to_owned(),
             },
         ];
-        let err = resolved_managed_targets(specs, &ClientConfig::default()).unwrap_err();
-
-        assert!(err.contains("duplicate resolved target address 127.0.0.1:2112"));
+        let targets = prepare_managed_targets(specs).unwrap();
+        assert_eq!(targets[0].managed.server_addr, "example.test");
+        assert_eq!(targets[1].managed.server_addr, "example.test");
     }
 }

crates/irtt-cli/tests/client_multi_target.rs

@@ -323,7 +323,10 @@ fn continuous_single_target_interruption_drains_final_events_and_succeeds() {
     assert!(stdout
         .lines()
         .any(|line| line.starts_with("session_closed,")));
-    assert!(stderr.contains("interrupted, closing session"), "{stderr}");
+    assert!(
+        stderr.contains("interrupted, closing managed run"),
+        "{stderr}"
+    );
     assert!(!stderr.contains("peer closure"), "{stderr}");
 }
 
@@ -643,7 +646,10 @@ fn explicit_group_interruption_succeeds_without_peer_close_error() {
         stdout.lines().any(|line| line.starts_with("b,")),
         "{stdout}"
     );
-    assert!(stderr.contains("interrupted, closing group"), "{stderr}");
+    assert!(
+        stderr.contains("interrupted, closing managed run"),
+        "{stderr}"
+    );
     assert!(!stderr.contains("peer closure"), "{stderr}");
 }
 

crates/irtt-client/src/client.rs

@@ -31,7 +31,10 @@ use schedule::{instant_abs_diff, ProbeSchedule};
 #[derive(Debug, Clone, Copy)]
 enum ProbeScheduleMode {
     CallerPaced,
-    Managed { scheduled_at: Instant },
+    #[cfg(test)]
+    Managed {
+        scheduled_at: Instant,
+    },
 }
 
 #[derive(Debug)]
@@ -91,8 +94,8 @@ use crate::{
 ///
 /// `Client` exposes the protocol steps directly: connect a UDP socket, open a
 /// session, send probes, receive replies, poll timeouts, and close. Callers
-/// that do not need to own this loop can use [`ManagedClient`](crate::ManagedClient)
-/// instead.
+/// that do not need to own this loop can use the unified managed API in
+/// [`crate::managed`] when the `tokio` feature is enabled.
 #[derive(Debug)]
 pub struct Client {
     runtime: SessionMachine,
@@ -250,23 +253,6 @@ impl Client {
         self.recv_once_inner()
     }
 
-    pub(crate) fn recv_once_until(
-        &mut self,
-        deadline: Option<Instant>,
-        max_wait: Duration,
-    ) -> Result<Vec<ClientEvent>, ClientError> {
-        let Some(timeout) = bounded_receive_timeout(
-            deadline,
-            self.runtime.config().socket_config.recv_timeout,
-            max_wait,
-            Instant::now(),
-        ) else {
-            return Ok(Vec::new());
-        };
-        self.socket.set_read_timeout(Some(timeout))?;
-        self.recv_once_inner()
-    }
-
     fn recv_once_inner(&mut self) -> Result<Vec<ClientEvent>, ClientError> {
         let datagram = match recv_datagram(&self.socket, &mut self.recv_buffer) {
             Ok(datagram) => datagram,
@@ -353,14 +339,7 @@ impl Client {
         self.runtime.is_peer_closed()
     }
 
-    pub(crate) fn has_timed_out_metadata(&self) -> bool {
-        self.runtime.has_timed_out_metadata()
-    }
-
-    pub(crate) fn packets_sent(&self) -> u64 {
-        self.runtime.packets_sent()
-    }
-
+    #[cfg(test)]
     pub(crate) fn send_managed_probe(
         &mut self,
         scheduled_at: Instant,
@@ -400,6 +379,7 @@ impl Client {
             ProbeScheduleMode::CallerPaced => {
                 schedule.preflight_caller_commit(machine_preflight.next_packets_sent)?
             }
+            #[cfg(test)]
             ProbeScheduleMode::Managed { scheduled_at } => {
                 schedule.preflight_managed_commit(scheduled_at, permission_at)?
             }
@@ -763,6 +743,7 @@ impl Client {
     }
 }
 
+#[cfg(test)]
 fn bounded_receive_timeout(
     deadline: Option<Instant>,
     configured_timeout: Option<Duration>,

crates/irtt-client/src/client/schedule.rs

@@ -79,6 +79,7 @@ impl ProbeSchedule {
         Ok(self.finish_commit(scheduled_at, next_send_at))
     }
 
+    #[cfg(any(feature = "tokio", test))]
     pub(crate) fn preflight_managed_commit(
         &self,
         scheduled_at: Instant,
@@ -122,6 +123,7 @@ fn next_probe_deadline(
         .ok_or(ClientError::DurationOverflow)
 }
 
+#[cfg(any(feature = "tokio", test))]
 pub(crate) fn advance_cadence(
     deadline: Instant,
     interval: Duration,
@@ -161,6 +163,7 @@ pub(crate) fn advance_cadence(
     Ok((scheduled_at, next_at, elapsed_slots))
 }
 
+#[cfg(any(feature = "tokio", test))]
 fn duration_from_nanos(nanos: u128) -> Result<Duration, ClientError> {
     const NANOS_PER_SECOND: u128 = 1_000_000_000;
     let seconds =

crates/irtt-client/src/error.rs

@@ -186,27 +186,4 @@ pub enum ClientError {
     /// The reported reason names a value that must be changed before retrying.
     #[error("invalid configuration: {reason}")]
     InvalidConfig { reason: String },
-    /// A managed-session worker thread panicked before producing an outcome.
-    ///
-    /// This is a runtime failure from joining the managed session, not a
-    /// protocol response from the server.
-    #[error("managed client worker thread panicked")]
-    WorkerPanicked,
-}
-
-/// Failure reported while setting up or receiving from an event subscription.
-///
-/// Zero-capacity subscriber queues are rejected when subscriptions are created.
-/// Once a subscription exists, this error type reports delivery failures observed
-/// by [`EventSubscription`](crate::EventSubscription).
-#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
-pub enum EventSubscriptionError {
-    /// The subscription can no longer receive new events.
-    ///
-    /// Already queued events are delivered first. After the queue is drained,
-    /// `recv`, `recv_timeout`, and `try_recv` return this error when the hub
-    /// disconnects the subscriber, the hub is shut down, or the subscription is
-    /// disconnected by its overflow policy.
-    #[error("event subscription is disconnected")]
-    Disconnected,
 }

crates/irtt-client/src/lib.rs

@@ -13,39 +13,10 @@
 //! also signed when the required wall-clock timestamps are available; negative
 //! values usually indicate client/server clock skew.
 //!
-//! A managed session can drive the socket loop on a worker thread and publish
-//! events through a subscription:
-//!
-//! ```no_run
-//! use std::time::Duration;
-//!
-//! use irtt_client::{ClientConfig, ClientEvent, ManagedClient, SubscriberConfig};
-//!
-//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
-//! let config = ClientConfig {
-//!     server_addr: "127.0.0.1:2112".to_owned(),
-//!     duration: Some(Duration::from_secs(10)),
-//!     interval: Duration::from_secs(1),
-//!     ..ClientConfig::default()
-//! };
-//!
-//! let (session, events) =
-//!     ManagedClient::start_with_subscription(config, SubscriberConfig::default())?;
-//!
-//! while let Ok(event) = events.recv() {
-//!     match event {
-//!         ClientEvent::EchoReply { seq, rtt, .. } => {
-//!             println!("seq={seq} effective_rtt_us={}", rtt.effective.as_micros());
-//!         }
-//!         ClientEvent::SessionClosed { .. } => break,
-//!         _ => {}
-//!     }
-//! }
-//!
-//! let _outcome = session.join()?;
-//! # Ok(())
-//! # }
-//! ```
+//! [`Client`] is the runtime-free low-level blocking adapter. With the
+//! `tokio` feature, `AsyncClient` provides the corresponding Tokio adapter;
+//! [`managed`] contains the unified managed task/handle implementation,
+//! including the dedicated-runtime blocking frontend for synchronous callers.
 //!
 #![cfg_attr(
     not(all(target_os = "linux", feature = "ancillary")),
@@ -75,20 +46,10 @@ pub use config::{
     ClientAuthConfig, ClientConfig, NegotiationPolicy, RecvBudget, RunMode, SocketConfig,
     MAX_DSCP_CODEPOINT, MAX_SERVER_FILL_BYTES, MAX_TTL, MAX_UDP_PAYLOAD_LENGTH,
 };
-pub use error::{ClientError, EventSubscriptionError};
+pub use error::ClientError;
 pub use event::{
     ClientEvent, OneWayDelaySample, OpenOutcome, PacketMeta, ReceivedStatsSample, RttSample,
     ServerTiming, SignedDuration, WarningKind,
 };
-pub use managed::legacy::{
-    CancellationToken, EventHub, EventSubscription, ManagedClient, ManagedClientGroup,
-    ManagedClientGroupConfig, ManagedClientGroupSession, ManagedClientSession,
-    ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, ManagedGroupOutcome,
-    ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure,
-    ManagedTargetFailureKind, ManagedTargetOutcome, SessionEndReason, SessionOutcome,
-    SubscriberConfig, SubscriberOverflow, TargetEvent, TargetEventSubscription,
-    MANAGED_GROUP_OUTCOME_HISTORY_LIMIT,
-};
-pub use managed::TargetId;
 pub use session::{NegotiatedParams, NegotiationRestriction};
 pub use timing::ClientTimestamp;

crates/irtt-client/src/managed/blocking.rs

@@ -7,8 +7,8 @@ use thiserror::Error;
 use tokio::runtime::{Builder, Runtime};
 
 use super::{
-    ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedConfigError, ManagedOutcome,
-    ManagedTargetConfig,
+    ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask, ManagedConfigError,
+    ManagedEventSubscription, ManagedOutcome, ManagedSubscribeError, ManagedTargetConfig,
 };
 
 const BLOCKING_RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(100);
@@ -56,6 +56,10 @@ pub enum BlockingManagedStartError {
     #[error(transparent)]
     Config(#[from] ManagedConfigError),
 
+    /// The managed event subscription could not be created before startup.
+    #[error(transparent)]
+    Subscribe(#[from] ManagedSubscribeError),
+
     /// The dedicated Tokio runtime could not be created.
     #[error("failed to create blocking managed Tokio runtime: {source}")]
     Runtime {
@@ -98,6 +102,29 @@ impl BlockingManagedClient {
         targets: Vec<ManagedTargetConfig>,
     ) -> Result<Self, BlockingManagedStartError> {
         let (task, handle) = ManagedClient::task(config, targets)?;
+        Self::start_task(task, handle)
+    }
+
+    /// Validate configuration, subscribe to presentation events, and start a
+    /// dedicated managed worker.
+    ///
+    /// The returned subscription exists before the managed task receives its
+    /// first poll, so it cannot miss startup events because of subscription
+    /// ordering.
+    pub fn start_with_subscription(
+        config: ManagedClientConfig,
+        targets: Vec<ManagedTargetConfig>,
+    ) -> Result<(Self, ManagedEventSubscription), BlockingManagedStartError> {
+        let (task, handle) = ManagedClient::task(config, targets)?;
+        let subscription = handle.subscribe()?;
+        let owner = Self::start_task(task, handle)?;
+        Ok((owner, subscription))
+    }
+
+    fn start_task(
+        task: ManagedClientTask,
+        handle: ManagedClientHandle,
+    ) -> Result<Self, BlockingManagedStartError> {
         let runtime = Builder::new_current_thread()
             .enable_io()
             .enable_time()

crates/irtt-client/src/managed/blocking_tests.rs

@@ -14,7 +14,8 @@ use tokio::{runtime::Builder, time::timeout};
 use super::{
     blocking::{join_worker, WorkerRuntime},
     BlockingManagedClient, BlockingManagedJoinError, ManagedClientConfig, ManagedCompletionPolicy,
-    ManagedEndReason, ManagedLifecycle, ManagedPacing, ManagedTargetConfig, ManagedTargetEndReason,
+    ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedPacing, ManagedTargetConfig,
+    ManagedTargetEndReason,
 };
 use crate::ClientConfig;
 
@@ -183,6 +184,35 @@ fn starts_without_caller_runtime_and_finishes_naturally() {
     server.finish();
 }
 
+#[test]
+fn subscribed_start_captures_events_from_a_short_run() {
+    let server = EchoServer::start();
+    let (owner, mut events) = BlockingManagedClient::start_with_subscription(
+        config(ManagedCompletionPolicy::FinishWhenQuiescent),
+        vec![target(&server)],
+    )
+    .unwrap();
+
+    server.wait_probe();
+    let outcome = owner.join().unwrap();
+    assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete);
+
+    let mut saw_started = false;
+    let mut saw_finished = false;
+    let mut saw_completed = false;
+    while let Ok(event) = events.try_recv() {
+        match event {
+            ManagedEvent::Started => saw_started = true,
+            ManagedEvent::TargetFinished { .. } => saw_finished = true,
+            ManagedEvent::Completed { .. } => saw_completed = true,
+            _ => {}
+        }
+    }
+    assert!(saw_started && saw_finished && saw_completed);
+    server.wait_close();
+    server.finish();
+}
+
 #[test]
 fn external_handle_stop_ends_with_graceful_close() {
     let server = EchoServer::start();

crates/irtt-client/src/managed/cancellation.rs

@@ -1,52 +0,0 @@
-use std::sync::{
-    atomic::{AtomicBool, Ordering},
-    Arc,
-};
-
-/// Cloneable cooperative cancellation flag.
-///
-/// Managed sessions use this internally to ask their worker loop to stop. It is
-/// exported for callers that want the same simple cancellation primitive in
-/// code built around `irtt-client`.
-#[derive(Debug, Clone, Default)]
-pub struct CancellationToken {
-    cancelled: Arc<AtomicBool>,
-}
-
-impl CancellationToken {
-    /// Create a token in the non-cancelled state.
-    pub fn new() -> Self {
-        Self::default()
-    }
-
-    /// Mark the token as cancelled.
-    ///
-    /// Cancellation is idempotent and is visible through all clones of the
-    /// token.
-    pub fn cancel(&self) {
-        self.cancelled.store(true, Ordering::SeqCst);
-    }
-
-    /// Return whether cancellation has been requested.
-    #[must_use]
-    pub fn is_cancelled(&self) -> bool {
-        self.cancelled.load(Ordering::SeqCst)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn cancel_is_idempotent_and_visible_through_clones() {
-        let token = CancellationToken::new();
-        let clone = token.clone();
-
-        token.cancel();
-        token.cancel();
-
-        assert!(token.is_cancelled());
-        assert!(clone.is_cancelled());
-    }
-}

crates/irtt-client/src/managed/hub.rs

@@ -1,606 +0,0 @@
-use std::{
-    collections::{HashMap, VecDeque},
-    sync::{
-        atomic::{AtomicU64, Ordering},
-        Arc, Condvar, Mutex, Weak,
-    },
-    time::{Duration, Instant},
-};
-
-use crate::{error::EventSubscriptionError, event::ClientEvent};
-
-/// Configuration for one managed event subscriber.
-///
-/// Each subscriber has an independent bounded queue. The queue bounds memory
-/// growth when event production is faster than that subscriber's consumer.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub struct SubscriberConfig {
-    /// Maximum number of events buffered for this subscriber.
-    ///
-    /// Must be greater than zero.
-    pub capacity: usize,
-    /// Policy applied when publishing to a full subscriber queue.
-    pub overflow: SubscriberOverflow,
-}
-
-impl Default for SubscriberConfig {
-    fn default() -> Self {
-        Self {
-            capacity: 1024,
-            overflow: SubscriberOverflow::DropNewest,
-        }
-    }
-}
-
-/// Overflow behavior for a bounded event subscription queue.
-///
-/// These policies are applied independently per subscriber when that
-/// subscriber's queue is full.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum SubscriberOverflow {
-    /// Leave the existing queue unchanged and discard the newly published event.
-    DropNewest,
-    /// Remove the oldest queued event and enqueue the newly published event.
-    DropOldest,
-    /// Disconnect the subscriber and clear its queue.
-    Disconnect,
-}
-
-/// Publish/subscribe fan-out for managed event values.
-///
-/// `EventHub` is used by managed sessions internally and is also exported for
-/// callers that want the same bounded subscription behavior around their own
-/// event producer. Publishing clones each event once per subscriber. Once
-/// sealed with [`disconnect_all`](Self::disconnect_all), the hub cannot accept
-/// new subscribers or publish more events.
-#[derive(Debug)]
-pub struct EventHub<T = ClientEvent> {
-    inner: Arc<HubInner<T>>,
-}
-
-#[derive(Debug)]
-struct HubInner<T> {
-    next_id: AtomicU64,
-    state: Mutex<HubState<T>>,
-}
-
-#[derive(Debug)]
-struct HubState<T> {
-    subscribers: HashMap<u64, Arc<SubscriberInner<T>>>,
-    sealed: bool,
-}
-
-/// Handle for receiving managed client events.
-///
-/// Each subscription has its own bounded queue configured by
-/// [`SubscriberConfig`]. The configured [`SubscriberOverflow`] policy applies
-/// when that queue is full. Use [`dropped_events`](Self::dropped_events) to
-/// detect incomplete consumption. Dropping the handle unregisters the
-/// subscriber.
-#[must_use = "dropping the subscription unregisters it"]
-#[derive(Debug)]
-pub struct EventSubscription<T = ClientEvent> {
-    id: u64,
-    hub: Weak<HubInner<T>>,
-    inner: Arc<SubscriberInner<T>>,
-}
-
-#[derive(Debug)]
-struct SubscriberInner<T> {
-    state: Mutex<SubscriberState<T>>,
-    available: Condvar,
-    config: SubscriberConfig,
-}
-
-#[derive(Debug)]
-struct SubscriberState<T> {
-    queue: VecDeque<T>,
-    connected: bool,
-    dropped_events: u64,
-}
-
-impl<T> Clone for EventHub<T> {
-    fn clone(&self) -> Self {
-        Self {
-            inner: self.inner.clone(),
-        }
-    }
-}
-
-impl<T> EventHub<T> {
-    /// Create an empty event hub with no subscribers.
-    pub fn new() -> Self {
-        Self {
-            inner: Arc::new(HubInner {
-                next_id: AtomicU64::new(1),
-                state: Mutex::new(HubState {
-                    subscribers: HashMap::new(),
-                    sealed: false,
-                }),
-            }),
-        }
-    }
-
-    /// Register a new subscriber with its own bounded queue.
-    ///
-    /// Returns an error when `config.capacity` is zero.
-    pub fn subscribe(
-        &self,
-        config: SubscriberConfig,
-    ) -> Result<EventSubscription<T>, crate::ClientError> {
-        if config.capacity == 0 {
-            return Err(crate::ClientError::InvalidConfig {
-                reason: "subscriber capacity must be greater than zero".to_owned(),
-            });
-        }
-
-        let mut hub_state = self.inner.state.lock().expect("event hub mutex poisoned");
-        if hub_state.sealed {
-            return Err(crate::ClientError::AlreadyClosed);
-        }
-
-        let id = self.inner.next_id.fetch_add(1, Ordering::Relaxed);
-        let inner = Arc::new(SubscriberInner {
-            state: Mutex::new(SubscriberState {
-                queue: VecDeque::with_capacity(config.capacity),
-                connected: true,
-                dropped_events: 0,
-            }),
-            available: Condvar::new(),
-            config,
-        });
-        hub_state.subscribers.insert(id, inner.clone());
-
-        Ok(EventSubscription {
-            id,
-            hub: Arc::downgrade(&self.inner),
-            inner,
-        })
-    }
-
-    /// Permanently seal the hub and disconnect all subscribers.
-    ///
-    /// Already queued events remain available to drain. Future subscriptions
-    /// fail with [`crate::ClientError::AlreadyClosed`].
-    pub fn disconnect_all(&self) {
-        let subscribers: Vec<Arc<SubscriberInner<T>>> = {
-            let mut state = self.inner.state.lock().expect("event hub mutex poisoned");
-            state.sealed = true;
-            state
-                .subscribers
-                .drain()
-                .map(|(_, subscriber)| subscriber)
-                .collect()
-        };
-
-        for subscriber in subscribers {
-            subscriber.disconnect();
-        }
-    }
-
-    #[cfg(test)]
-    fn subscriber_count(&self) -> usize {
-        self.inner
-            .state
-            .lock()
-            .expect("event hub mutex poisoned")
-            .subscribers
-            .len()
-    }
-}
-
-impl<T: Clone> EventHub<T> {
-    /// Publish an event to all currently connected subscribers.
-    ///
-    /// Slow subscribers are handled according to their
-    /// [`SubscriberOverflow`] policy. Publishing does not block waiting for a
-    /// subscriber to consume queued events.
-    pub fn publish(&self, event: T) {
-        let subscribers: Vec<(u64, Arc<SubscriberInner<T>>)> = self
-            .inner
-            .state
-            .lock()
-            .expect("event hub mutex poisoned")
-            .subscribers
-            .iter()
-            .map(|(id, subscriber)| (*id, subscriber.clone()))
-            .collect();
-
-        let mut disconnected = Vec::new();
-        for (id, subscriber) in subscribers {
-            // Events are cloned per subscriber; bounded queues keep slow
-            // consumers from causing unbounded memory growth.
-            if !subscriber.publish(event.clone()) {
-                disconnected.push(id);
-            }
-        }
-
-        if disconnected.is_empty() {
-            return;
-        }
-
-        let mut subscribers = self.inner.state.lock().expect("event hub mutex poisoned");
-        for id in disconnected {
-            subscribers.subscribers.remove(&id);
-        }
-    }
-}
-
-impl<T> Default for EventHub<T> {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl<T> EventSubscription<T> {
-    /// Block until the next queued event is available.
-    ///
-    /// If the subscription is disconnected, this returns
-    /// [`EventSubscriptionError::Disconnected`] after any already queued events
-    /// have been drained.
-    pub fn recv(&self) -> Result<T, EventSubscriptionError> {
-        let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
-        loop {
-            if let Some(event) = state.queue.pop_front() {
-                return Ok(event);
-            }
-            if !state.connected {
-                return Err(EventSubscriptionError::Disconnected);
-            }
-            state = self
-                .inner
-                .available
-                .wait(state)
-                .expect("subscriber mutex poisoned");
-        }
-    }
-
-    /// Wait up to `timeout` for the next queued event.
-    ///
-    /// Returns `Ok(None)` when the timeout elapses while the subscription
-    /// remains connected. If the subscription is disconnected, this returns
-    /// [`EventSubscriptionError::Disconnected`] after any already queued events
-    /// have been drained.
-    pub fn recv_timeout(&self, timeout: Duration) -> Result<Option<T>, EventSubscriptionError> {
-        let started_at = Instant::now();
-        let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
-        loop {
-            if let Some(event) = state.queue.pop_front() {
-                return Ok(Some(event));
-            }
-            if !state.connected {
-                return Err(EventSubscriptionError::Disconnected);
-            }
-
-            let Some(wait_for) = timeout.checked_sub(started_at.elapsed()) else {
-                return Ok(None);
-            };
-            if wait_for.is_zero() {
-                return Ok(None);
-            }
-
-            let (next_state, _) = self
-                .inner
-                .available
-                .wait_timeout(state, wait_for)
-                .expect("subscriber mutex poisoned");
-            state = next_state;
-        }
-    }
-
-    /// Try to receive one queued event without blocking.
-    ///
-    /// Returns `Ok(None)` when the subscription is still connected but no event
-    /// is queued. If the subscription is disconnected, this returns
-    /// [`EventSubscriptionError::Disconnected`] after any already queued events
-    /// have been drained.
-    pub fn try_recv(&self) -> Result<Option<T>, EventSubscriptionError> {
-        let mut state = self.inner.state.lock().expect("subscriber mutex poisoned");
-        if let Some(event) = state.queue.pop_front() {
-            return Ok(Some(event));
-        }
-        if !state.connected {
-            return Err(EventSubscriptionError::Disconnected);
-        }
-        Ok(None)
-    }
-
-    /// Return the number of events discarded by this subscription's overflow
-    /// policy.
-    ///
-    /// The count is monotonic for the lifetime of the subscription and remains
-    /// available after the hub is sealed or the subscription disconnects.
-    pub fn dropped_events(&self) -> u64 {
-        self.inner
-            .state
-            .lock()
-            .expect("subscriber mutex poisoned")
-            .dropped_events
-    }
-}
-
-impl<T> Drop for EventSubscription<T> {
-    fn drop(&mut self) {
-        self.inner.disconnect();
-        if let Some(hub) = self.hub.upgrade() {
-            hub.state
-                .lock()
-                .expect("event hub mutex poisoned")
-                .subscribers
-                .remove(&self.id);
-        }
-    }
-}
-
-impl<T> SubscriberInner<T> {
-    fn publish(&self, event: T) -> bool {
-        let mut state = self.state.lock().expect("subscriber mutex poisoned");
-        if !state.connected {
-            return false;
-        }
-
-        if state.queue.len() < self.config.capacity {
-            state.queue.push_back(event);
-            self.available.notify_one();
-            return true;
-        }
-
-        match self.config.overflow {
-            SubscriberOverflow::DropNewest => {
-                state.dropped_events = state.dropped_events.saturating_add(1);
-                true
-            }
-            SubscriberOverflow::DropOldest => {
-                state.queue.pop_front();
-                state.queue.push_back(event);
-                state.dropped_events = state.dropped_events.saturating_add(1);
-                self.available.notify_one();
-                true
-            }
-            SubscriberOverflow::Disconnect => {
-                let discarded = u64::try_from(state.queue.len())
-                    .unwrap_or(u64::MAX)
-                    .saturating_add(1);
-                state.dropped_events = state.dropped_events.saturating_add(discarded);
-                state.queue.clear();
-                state.connected = false;
-                self.available.notify_all();
-                false
-            }
-        }
-    }
-
-    fn disconnect(&self) {
-        let mut state = self.state.lock().expect("subscriber mutex poisoned");
-        state.connected = false;
-        self.available.notify_all();
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::{ClientTimestamp, WarningKind};
-
-    fn event(n: usize) -> ClientEvent {
-        ClientEvent::Warning {
-            kind: WarningKind::UntrackedReply,
-            message: format!("event-{n}"),
-            at: ClientTimestamp::now(),
-        }
-    }
-
-    fn event_message(event: ClientEvent) -> String {
-        match event {
-            ClientEvent::Warning { message, .. } => message,
-            other => panic!("unexpected event: {other:?}"),
-        }
-    }
-
-    fn assert_next(
-        sub: &EventSubscription,
-        expected: Result<Option<&str>, EventSubscriptionError>,
-    ) {
-        match (sub.try_recv(), expected) {
-            (Ok(Some(event)), Ok(Some(message))) => assert_eq!(event_message(event), message),
-            (Ok(None), Ok(None)) => {}
-            (Err(actual), Err(expected)) => assert_eq!(actual, expected),
-            (actual, expected) => panic!("expected {expected:?}, got {actual:?}"),
-        }
-    }
-
-    #[test]
-    fn publishes_to_multiple_subscribers() {
-        let hub = EventHub::new();
-        let a = hub.subscribe(SubscriberConfig::default()).unwrap();
-        let b = hub.subscribe(SubscriberConfig::default()).unwrap();
-
-        hub.publish(event(1));
-
-        assert_eq!(event_message(a.try_recv().unwrap().unwrap()), "event-1");
-        assert_eq!(event_message(b.try_recv().unwrap().unwrap()), "event-1");
-    }
-
-    #[test]
-    fn full_subscriber_overflow_policy_is_applied() {
-        let cases = [
-            (
-                SubscriberOverflow::DropNewest,
-                Ok(Some("event-1")),
-                Ok(None),
-                1,
-            ),
-            (
-                SubscriberOverflow::DropOldest,
-                Ok(Some("event-2")),
-                Ok(None),
-                1,
-            ),
-            (
-                SubscriberOverflow::Disconnect,
-                Err(EventSubscriptionError::Disconnected),
-                Err(EventSubscriptionError::Disconnected),
-                2,
-            ),
-        ];
-
-        for (overflow, first_expected, second_expected, expected_drops) in cases {
-            let hub = EventHub::new();
-            let sub = hub
-                .subscribe(SubscriberConfig {
-                    capacity: 1,
-                    overflow,
-                })
-                .unwrap();
-
-            hub.publish(event(1));
-            hub.publish(event(2));
-
-            assert_next(&sub, first_expected);
-            assert_next(&sub, second_expected);
-            assert_eq!(sub.dropped_events(), expected_drops);
-        }
-    }
-
-    #[test]
-    fn drop_oldest_counter_accumulates_and_survives_hub_seal() {
-        let hub = EventHub::new();
-        let sub = hub
-            .subscribe(SubscriberConfig {
-                capacity: 2,
-                overflow: SubscriberOverflow::DropOldest,
-            })
-            .unwrap();
-
-        for n in 1..=5 {
-            hub.publish(event(n));
-        }
-        hub.disconnect_all();
-
-        assert_eq!(sub.dropped_events(), 3);
-        assert_next(&sub, Ok(Some("event-4")));
-        assert_next(&sub, Ok(Some("event-5")));
-        assert_next(&sub, Err(EventSubscriptionError::Disconnected));
-        assert_eq!(sub.dropped_events(), 3);
-    }
-
-    #[test]
-    fn full_subscriber_does_not_prevent_other_delivery() {
-        let hub = EventHub::new();
-        let slow = hub
-            .subscribe(SubscriberConfig {
-                capacity: 1,
-                overflow: SubscriberOverflow::DropNewest,
-            })
-            .unwrap();
-        let fast = hub
-            .subscribe(SubscriberConfig {
-                capacity: 4,
-                overflow: SubscriberOverflow::DropNewest,
-            })
-            .unwrap();
-
-        hub.publish(event(1));
-        hub.publish(event(2));
-
-        assert_eq!(event_message(slow.try_recv().unwrap().unwrap()), "event-1");
-        assert_eq!(event_message(fast.try_recv().unwrap().unwrap()), "event-1");
-        assert_eq!(event_message(fast.try_recv().unwrap().unwrap()), "event-2");
-    }
-
-    #[test]
-    fn subscribing_after_events_does_not_replay() {
-        let hub = EventHub::new();
-        hub.publish(event(1));
-
-        let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
-
-        assert!(sub.try_recv().unwrap().is_none());
-    }
-
-    #[test]
-    fn dropping_subscription_unregisters_from_hub() {
-        let hub: EventHub = EventHub::new();
-        let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
-        assert_eq!(hub.subscriber_count(), 1);
-
-        drop(sub);
-
-        assert_eq!(hub.subscriber_count(), 0);
-    }
-
-    #[test]
-    fn disconnect_all_wakes_blocking_receivers() {
-        let hub: EventHub = EventHub::new();
-        let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
-        hub.disconnect_all();
-
-        assert_eq!(
-            sub.recv().unwrap_err(),
-            EventSubscriptionError::Disconnected
-        );
-    }
-
-    #[test]
-    fn timed_receive_drains_queued_events_and_observes_disconnect() {
-        let hub = EventHub::new();
-        let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
-
-        assert_eq!(sub.recv_timeout(Duration::ZERO).unwrap(), None);
-        hub.publish(event(1));
-        hub.disconnect_all();
-
-        assert_eq!(
-            event_message(sub.recv_timeout(Duration::from_secs(1)).unwrap().unwrap()),
-            "event-1"
-        );
-        assert_eq!(
-            sub.recv_timeout(Duration::from_secs(1)).unwrap_err(),
-            EventSubscriptionError::Disconnected
-        );
-    }
-
-    #[test]
-    fn sealed_hub_drains_queued_events_then_disconnects() {
-        let hub = EventHub::new();
-        let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
-        hub.publish(event(1));
-
-        hub.disconnect_all();
-        hub.publish(event(2));
-
-        assert_eq!(event_message(sub.recv().unwrap()), "event-1");
-        assert_eq!(
-            sub.recv().unwrap_err(),
-            EventSubscriptionError::Disconnected
-        );
-    }
-
-    #[test]
-    fn subscribing_after_hub_is_sealed_fails() {
-        let hub: EventHub = EventHub::new();
-        hub.disconnect_all();
-        hub.disconnect_all();
-
-        assert!(matches!(
-            hub.subscribe(SubscriberConfig::default()),
-            Err(crate::ClientError::AlreadyClosed)
-        ));
-    }
-
-    #[test]
-    fn session_closed_event_can_be_queued() {
-        let hub = EventHub::new();
-        let sub = hub.subscribe(SubscriberConfig::default()).unwrap();
-        hub.publish(ClientEvent::SessionClosed {
-            remote: "127.0.0.1:1".parse().unwrap(),
-            token: 1,
-            at: ClientTimestamp::now(),
-        });
-
-        assert!(matches!(
-            sub.recv().unwrap(),
-            ClientEvent::SessionClosed { token: 1, .. }
-        ));
-    }
-}

crates/irtt-client/src/managed/mod.rs

@@ -1,29 +1,7 @@
-mod cancellation;
-mod group;
-mod hub;
-mod runner;
 mod target_id;
 
 pub use target_id::TargetId;
 
-/// Temporary exports for the obsolete thread-based managed implementations.
-#[doc(hidden)]
-pub mod legacy {
-    pub use super::cancellation::CancellationToken;
-    pub use super::group::{
-        ManagedClientGroup, ManagedClientGroupConfig, ManagedClientGroupSession,
-        ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent,
-        ManagedGroupOutcome, ManagedGroupPacing, ManagedTargetConfig, ManagedTargetEndReason,
-        ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetOutcome, TargetEvent,
-        TargetEventSubscription, MANAGED_GROUP_OUTCOME_HISTORY_LIMIT,
-    };
-    pub use super::hub::{EventHub, EventSubscription, SubscriberConfig, SubscriberOverflow};
-    pub use super::runner::{
-        ManagedClient, ManagedClientSession, SessionEndReason, SessionOutcome,
-    };
-    pub use super::TargetId;
-}
-
 #[cfg(feature = "tokio")]
 mod blocking;
 #[cfg(feature = "tokio")]

crates/irtt-client/src/managed/model.rs

@@ -173,6 +173,9 @@ pub enum ManagedEvent {
 /// Receiving half of the lossy managed event stream.
 pub type ManagedEventSubscription = tokio::sync::broadcast::Receiver<ManagedEvent>;
 
+/// Immediate result of attempting to receive a managed presentation event.
+pub type ManagedEventTryRecvError = tokio::sync::broadcast::error::TryRecvError;
+
 /// Final authoritative outcome returned by the task.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct ManagedOutcome {

crates/irtt-client/src/managed/runner.rs

@@ -1,883 +0,0 @@
-use std::{
-    thread::{self, JoinHandle},
-    time::{Duration, Instant},
-};
-
-use crate::{
-    config::{ClientConfig, RecvBudget},
-    error::ClientError,
-    event::{ClientEvent, OpenOutcome},
-    Client,
-};
-
-use super::{
-    cancellation::CancellationToken,
-    hub::{EventHub, EventSubscription, SubscriberConfig},
-};
-
-const MANAGED_RECV_TIMEOUT: Duration = Duration::from_millis(20);
-const MANAGED_RECV_BUDGET: RecvBudget = RecvBudget { max_packets: 64 };
-const MANAGED_FINAL_DRAIN: Duration = Duration::from_millis(100);
-const IDLE_SLEEP: Duration = Duration::from_millis(1);
-
-/// Entry point for running a client session on a worker thread.
-///
-/// The managed API owns the lower-level [`Client`](crate::Client) loop: it
-/// opens the session, sends probes at the negotiated interval, receives
-/// datagrams, publishes [`ClientEvent`] values, and closes the session when the
-/// run completes or is cancelled. Delayed workers skip missed pacing slots
-/// instead of sending catch-up bursts.
-#[derive(Debug)]
-pub struct ManagedClient;
-
-/// Running managed client session.
-///
-/// [`ManagedClientSession::join`] waits for the worker and returns the session
-/// outcome or client error. Dropping the session requests cooperative
-/// cancellation; callers that need the outcome should explicitly join instead
-/// of relying on drop.
-#[must_use = "dropping the session cancels the managed client; call join() to wait for completion"]
-#[derive(Debug)]
-pub struct ManagedClientSession {
-    hub: EventHub,
-    cancellation: CancellationToken,
-    worker: Option<JoinHandle<Result<SessionOutcome, ClientError>>>,
-}
-
-/// Outcome returned by a completed managed client session.
-///
-/// These are runner-level lifecycle counters, not statistical summaries. Use
-/// `irtt-stats` with emitted `ClientEvent` values for RTT, loss, IPDV, and
-/// related summaries.
-#[must_use = "managed session outcomes contain completion status and counters"]
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct SessionOutcome {
-    /// Why the managed loop stopped.
-    pub end_reason: SessionEndReason,
-    /// Number of echo requests sent by the client.
-    pub packets_sent: u64,
-    /// Number of first in-window echo replies received.
-    pub replies_received: u64,
-    /// Number of duplicate reply events emitted.
-    pub duplicates: u64,
-    /// Number of late reply events emitted.
-    pub late: u64,
-    /// Number of warning events emitted.
-    pub warning_events: u64,
-}
-
-/// Reason a managed client session ended.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum SessionEndReason {
-    /// The negotiated finite test duration completed and the client closed the
-    /// session normally.
-    TestComplete,
-    /// The peer ended the session with an authenticated close-flagged reply.
-    PeerClosed,
-    /// Cancellation was requested through [`ManagedClientSession::stop`] or by
-    /// dropping the session handle.
-    Cancelled,
-    /// The session was opened in [`RunMode::NoTest`](crate::RunMode) and
-    /// completed after negotiation.
-    NoTestComplete,
-}
-
-impl ManagedClient {
-    /// Start a managed session without creating an initial event subscription.
-    ///
-    /// The returned [`ManagedClientSession`] can still create subscriptions via
-    /// [`ManagedClientSession::subscribe`], but events emitted before a
-    /// subscriber is registered are not replayed.
-    pub fn start(config: ClientConfig) -> Result<ManagedClientSession, ClientError> {
-        Self::start_inner(config, None).map(|(session, _)| session)
-    }
-
-    /// Start a managed session and subscribe to events before the worker runs.
-    ///
-    /// The initial subscription receives the open lifecycle event and
-    /// subsequent session events, subject to its queue capacity and overflow
-    /// policy.
-    pub fn start_with_subscription(
-        config: ClientConfig,
-        subscriber_config: SubscriberConfig,
-    ) -> Result<(ManagedClientSession, EventSubscription), ClientError> {
-        let (session, subscription) = Self::start_inner(config, Some(subscriber_config))?;
-        Ok((
-            session,
-            subscription.expect("initial subscription must be present"),
-        ))
-    }
-
-    fn start_inner(
-        mut config: ClientConfig,
-        subscriber_config: Option<SubscriberConfig>,
-    ) -> Result<(ManagedClientSession, Option<EventSubscription>), ClientError> {
-        if config.socket_config.recv_timeout.is_none()
-            || config.socket_config.recv_timeout > Some(MANAGED_RECV_TIMEOUT)
-        {
-            config.socket_config.recv_timeout = Some(MANAGED_RECV_TIMEOUT);
-        }
-
-        let hub = EventHub::new();
-        let initial_subscription = subscriber_config
-            .map(|config| hub.subscribe(config))
-            .transpose()?;
-
-        let mut client = Client::connect(config)?;
-        let outcome = client.open()?;
-        publish_open_outcome(&hub, &outcome);
-
-        let cancellation = CancellationToken::new();
-        let worker_hub = hub.clone();
-        let worker_cancellation = cancellation.clone();
-        let worker =
-            thread::spawn(move || run_client_with_cleanup(client, worker_hub, worker_cancellation));
-
-        Ok((
-            ManagedClientSession {
-                hub,
-                cancellation,
-                worker: Some(worker),
-            },
-            initial_subscription,
-        ))
-    }
-}
-
-impl ManagedClientSession {
-    /// Add another event subscriber to this running managed session.
-    ///
-    /// The subscription receives only events published after it is registered.
-    pub fn subscribe(&self, config: SubscriberConfig) -> Result<EventSubscription, ClientError> {
-        self.hub.subscribe(config)
-    }
-
-    /// Request cooperative cancellation of the managed session.
-    ///
-    /// Cancellation is observed by the worker loop between socket receive,
-    /// timeout polling, and send scheduling steps. Call [`join`](Self::join) to
-    /// wait for the worker and obtain the final [`SessionOutcome`].
-    pub fn stop(&self) {
-        self.cancellation.cancel();
-    }
-
-    /// Wait for the managed worker thread to finish.
-    ///
-    /// On success, returns lifecycle counters for the completed session. If the
-    /// worker panicked, this returns [`ClientError::WorkerPanicked`]. Joining
-    /// also disconnects all event subscriptions while leaving already queued
-    /// events available to drain.
-    pub fn join(mut self) -> Result<SessionOutcome, ClientError> {
-        let worker = self
-            .worker
-            .take()
-            .expect("ManagedClientSession invariant violated: worker handle missing before join");
-        match worker.join() {
-            Ok(outcome) => {
-                self.hub.disconnect_all();
-                outcome
-            }
-            Err(_) => {
-                self.hub.disconnect_all();
-                Err(ClientError::WorkerPanicked)
-            }
-        }
-    }
-}
-
-impl Drop for ManagedClientSession {
-    fn drop(&mut self) {
-        self.cancellation.cancel();
-    }
-}
-
-fn publish_open_outcome(hub: &EventHub, outcome: &OpenOutcome) {
-    match outcome {
-        OpenOutcome::Started { event, .. } | OpenOutcome::NoTestCompleted { event, .. } => {
-            hub.publish(event.clone());
-        }
-    }
-}
-
-fn run_client(
-    mut client: Client,
-    hub: EventHub,
-    cancellation: CancellationToken,
-) -> Result<SessionOutcome, ClientError> {
-    if client.is_run_complete() {
-        return Ok(SessionOutcome {
-            end_reason: SessionEndReason::NoTestComplete,
-            packets_sent: 0,
-            replies_received: 0,
-            duplicates: 0,
-            late: 0,
-            warning_events: 0,
-        });
-    }
-
-    let mut counters = OutcomeCounters::default();
-    let mut cancelled = false;
-
-    loop {
-        if cancellation.is_cancelled() {
-            cancelled = true;
-            publish_events(
-                &hub,
-                &mut counters,
-                client.recv_available(MANAGED_RECV_BUDGET)?,
-            );
-            if !client.is_peer_closed() {
-                publish_events(&hub, &mut counters, client.poll_timeouts()?);
-            }
-            break;
-        }
-
-        let send_deadline = client.next_send_deadline();
-        if let Some(deadline) = send_deadline.filter(|deadline| *deadline <= Instant::now()) {
-            let events = client.send_managed_probe(deadline)?;
-            publish_events(&hub, &mut counters, events);
-            continue;
-        }
-
-        publish_events(
-            &hub,
-            &mut counters,
-            client.recv_once_until(send_deadline, MANAGED_RECV_TIMEOUT)?,
-        );
-        if client.is_peer_closed() {
-            break;
-        }
-        publish_events(&hub, &mut counters, client.poll_timeouts()?);
-
-        if client.is_run_complete() {
-            break;
-        }
-    }
-
-    if !cancelled {
-        drain_final_late_replies(&mut client, &hub, &mut counters)?;
-    }
-
-    let packets_sent = client.packets_sent();
-    if !client.is_peer_closed() {
-        let close_events = client.close()?;
-        publish_events(&hub, &mut counters, close_events);
-    }
-
-    let end_reason = if cancelled {
-        SessionEndReason::Cancelled
-    } else if client.is_peer_closed() {
-        SessionEndReason::PeerClosed
-    } else {
-        SessionEndReason::TestComplete
-    };
-
-    Ok(SessionOutcome {
-        end_reason,
-        packets_sent,
-        replies_received: counters.replies_received,
-        duplicates: counters.duplicates,
-        late: counters.late,
-        warning_events: counters.warning_events,
-    })
-}
-
-fn run_client_with_cleanup(
-    client: Client,
-    hub: EventHub,
-    cancellation: CancellationToken,
-) -> Result<SessionOutcome, ClientError> {
-    let outcome = run_client(client, hub.clone(), cancellation);
-    hub.disconnect_all();
-    outcome
-}
-
-fn drain_final_late_replies(
-    client: &mut Client,
-    hub: &EventHub,
-    counters: &mut OutcomeCounters,
-) -> Result<(), ClientError> {
-    if !client.has_timed_out_metadata() {
-        return Ok(());
-    }
-
-    let deadline = Instant::now() + MANAGED_FINAL_DRAIN;
-    while Instant::now() < deadline && client.has_timed_out_metadata() {
-        if client.is_peer_closed() {
-            break;
-        }
-
-        let mut published = false;
-
-        let events = client.recv_available(MANAGED_RECV_BUDGET)?;
-        published |= !events.is_empty();
-        publish_events(hub, counters, events);
-
-        if client.is_peer_closed() {
-            break;
-        }
-
-        let events = client.poll_timeouts()?;
-        published |= !events.is_empty();
-        publish_events(hub, counters, events);
-
-        if !published {
-            thread::sleep(IDLE_SLEEP);
-        }
-    }
-    Ok(())
-}
-
-fn publish_events(hub: &EventHub, counters: &mut OutcomeCounters, events: Vec<ClientEvent>) {
-    for event in events {
-        counters.observe(&event);
-        hub.publish(event);
-    }
-}
-
-#[derive(Debug, Default)]
-struct OutcomeCounters {
-    replies_received: u64,
-    duplicates: u64,
-    late: u64,
-    warning_events: u64,
-}
-
-impl OutcomeCounters {
-    fn observe(&mut self, event: &ClientEvent) {
-        match event {
-            ClientEvent::EchoReply { .. } => self.replies_received += 1,
-            ClientEvent::DuplicateReply { .. } => self.duplicates += 1,
-            ClientEvent::LateReply { .. } => self.late += 1,
-            ClientEvent::Warning { .. } => self.warning_events += 1,
-            _ => {}
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::{config::NegotiationPolicy, EventSubscriptionError, SubscriberOverflow};
-    use irtt_proto::{
-        echo_packet_len,
-        flags::{self, FLAG_OPEN, FLAG_REPLY},
-        layout::PacketLayout,
-        Clock, Params, ReceivedStats, StampAt, TimestampFields, MAGIC, PROTOCOL_VERSION,
-    };
-    use std::{
-        net::{SocketAddr, UdpSocket},
-        sync::mpsc,
-    };
-
-    const TOKEN: u64 = 0x1234_5678_90ab_cdef;
-    struct FakeServer {
-        addr: SocketAddr,
-        done: JoinHandle<()>,
-    }
-
-    impl FakeServer {
-        fn join(self) {
-            self.done.join().unwrap();
-        }
-    }
-
-    fn test_echo_packet_len(hmac: bool, params: &Params) -> usize {
-        echo_packet_len(hmac, params)
-            .expect("managed runner test params must have a non-negative packet length")
-    }
-
-    fn test_params(duration: Option<Duration>, interval: Duration) -> Params {
-        Params {
-            protocol_version: PROTOCOL_VERSION,
-            duration_ns: duration.map_or(0, test_duration_ns_i64),
-            interval_ns: test_duration_ns_i64(interval),
-            length: 0,
-            received_stats: ReceivedStats::Both,
-            stamp_at: StampAt::Both,
-            clock: Clock::Both,
-            dscp: 0,
-            server_fill: None,
-        }
-    }
-
-    fn test_duration_ns_i64(duration: Duration) -> i64 {
-        i64::try_from(duration.as_nanos()).expect("test duration fits i64 nanoseconds")
-    }
-
-    fn config(addr: SocketAddr, duration: Option<Duration>) -> ClientConfig {
-        ClientConfig {
-            server_addr: addr.to_string(),
-            duration,
-            interval: Duration::from_millis(10),
-            negotiation_policy: NegotiationPolicy::Strict,
-            open_timeouts: vec![Duration::from_millis(200)],
-            probe_timeout: Duration::from_millis(40),
-            ..ClientConfig::default()
-        }
-    }
-
-    fn start_echo_server(params: Params) -> FakeServer {
-        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
-        let addr = socket.local_addr().unwrap();
-        let done = thread::spawn(move || {
-            let (_, peer) = recv_request(&socket);
-            socket
-                .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
-                .unwrap();
-            socket
-                .set_read_timeout(Some(Duration::from_millis(500)))
-                .unwrap();
-
-            while let Some((packet, peer)) = recv_request_timeout(&socket) {
-                if packet[3] & flags::FLAG_CLOSE != 0 {
-                    break;
-                }
-
-                let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap());
-                let ts = TimestampFields {
-                    recv_wall: Some(1_000_000_000),
-                    recv_mono: Some(100_000),
-                    send_wall: Some(1_000_000_000),
-                    send_mono: Some(100_000),
-                    ..Default::default()
-                };
-                socket
-                    .send_to(&echo_reply_packet(TOKEN, seq, &params, &ts), peer)
-                    .unwrap();
-            }
-        });
-        FakeServer { addr, done }
-    }
-
-    fn start_delayed_reply_server(params: Params, delay: Duration) -> FakeServer {
-        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
-        let addr = socket.local_addr().unwrap();
-        let done = thread::spawn(move || {
-            let (_, peer) = recv_request(&socket);
-            socket
-                .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
-                .unwrap();
-            socket
-                .set_read_timeout(Some(Duration::from_millis(500)))
-                .unwrap();
-
-            let Some((packet, peer)) = recv_request_timeout(&socket) else {
-                return;
-            };
-            assert_eq!(packet[3] & flags::FLAG_CLOSE, 0);
-            assert!(
-                packet.len() >= 16,
-                "echo request too short: {}",
-                packet.len()
-            );
-            let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap());
-            thread::sleep(delay);
-            let ts = TimestampFields {
-                recv_wall: Some(1_000_000_000),
-                recv_mono: Some(100_000),
-                send_wall: Some(1_001_000_000),
-                send_mono: Some(1_100_000),
-                ..Default::default()
-            };
-            socket
-                .send_to(&echo_reply_packet(TOKEN, seq, &params, &ts), peer)
-                .unwrap();
-
-            while let Some((packet, _)) = recv_request_timeout(&socket) {
-                if packet[3] & flags::FLAG_CLOSE != 0 {
-                    break;
-                }
-            }
-        });
-        FakeServer { addr, done }
-    }
-
-    fn start_peer_close_server(params: Params) -> FakeServer {
-        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
-        let addr = socket.local_addr().unwrap();
-        let done = thread::spawn(move || {
-            let (_, peer) = recv_request(&socket);
-            socket
-                .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
-                .unwrap();
-            socket
-                .set_read_timeout(Some(Duration::from_millis(500)))
-                .unwrap();
-
-            let Some((packet, peer)) = recv_request_timeout(&socket) else {
-                return;
-            };
-            let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap());
-            let ts = TimestampFields {
-                recv_wall: Some(1_000_000_000),
-                recv_mono: Some(100_000),
-                send_wall: Some(1_000_000_000),
-                send_mono: Some(100_000),
-                ..Default::default()
-            };
-            socket
-                .send_to(
-                    &echo_reply_packet_with_flags(
-                        TOKEN,
-                        seq,
-                        &params,
-                        &ts,
-                        FLAG_REPLY | flags::FLAG_CLOSE,
-                    ),
-                    peer,
-                )
-                .unwrap();
-
-            if let Some((packet, _)) = recv_request_timeout(&socket) {
-                panic!(
-                    "managed cleanup must not send any packet after peer close; flags={:?} len={}",
-                    packet.get(3).copied(),
-                    packet.len()
-                );
-            }
-        });
-        FakeServer { addr, done }
-    }
-
-    fn recv_request(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
-        let mut buf = [0_u8; 2048];
-        let (size, peer) = socket.recv_from(&mut buf).unwrap();
-        (buf[..size].to_vec(), peer)
-    }
-
-    fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec<u8>, SocketAddr)> {
-        let mut buf = [0_u8; 2048];
-        socket
-            .recv_from(&mut buf)
-            .ok()
-            .map(|(size, peer)| (buf[..size].to_vec(), peer))
-    }
-
-    fn open_reply(flags: u8, token: u64, params: &Params) -> Vec<u8> {
-        let mut packet = Vec::new();
-        packet.extend_from_slice(&MAGIC);
-        packet.push(flags);
-        packet.extend_from_slice(&token.to_le_bytes());
-        packet.extend_from_slice(&params.encode());
-        packet
-    }
-
-    fn echo_reply_packet(
-        token: u64,
-        seq: u32,
-        params: &Params,
-        timestamps: &TimestampFields,
-    ) -> Vec<u8> {
-        echo_reply_packet_with_flags(token, seq, params, timestamps, FLAG_REPLY)
-    }
-
-    fn echo_reply_packet_with_flags(
-        token: u64,
-        seq: u32,
-        params: &Params,
-        timestamps: &TimestampFields,
-        flags: u8,
-    ) -> Vec<u8> {
-        let layout = PacketLayout::echo(false, params);
-        let packet_len = test_echo_packet_len(false, params);
-        let mut packet = Vec::with_capacity(packet_len);
-
-        packet.extend_from_slice(&MAGIC);
-        packet.push(flags);
-        packet.extend_from_slice(&token.to_le_bytes());
-        packet.extend_from_slice(&seq.to_le_bytes());
-
-        if layout.recv_count {
-            packet.extend_from_slice(&42_u32.to_le_bytes());
-        }
-        if layout.recv_window {
-            packet.extend_from_slice(&0x07_u64.to_le_bytes());
-        }
-        if layout.recv_wall {
-            packet.extend_from_slice(&timestamps.recv_wall.unwrap_or(0).to_le_bytes());
-        }
-        if layout.recv_mono {
-            packet.extend_from_slice(&timestamps.recv_mono.unwrap_or(0).to_le_bytes());
-        }
-        if layout.midpoint_wall {
-            packet.extend_from_slice(&timestamps.midpoint_wall.unwrap_or(0).to_le_bytes());
-        }
-        if layout.midpoint_mono {
-            packet.extend_from_slice(&timestamps.midpoint_mono.unwrap_or(0).to_le_bytes());
-        }
-        if layout.send_wall {
-            packet.extend_from_slice(&timestamps.send_wall.unwrap_or(0).to_le_bytes());
-        }
-        if layout.send_mono {
-            packet.extend_from_slice(&timestamps.send_mono.unwrap_or(0).to_le_bytes());
-        }
-
-        packet.resize(packet_len, 0);
-        packet
-    }
-
-    fn recv_event_with_timeout(sub: &EventSubscription) -> ClientEvent {
-        let deadline = Instant::now() + Duration::from_secs(2);
-        loop {
-            match sub.try_recv() {
-                Ok(Some(event)) => return event,
-                Ok(None) if Instant::now() < deadline => {
-                    thread::sleep(Duration::from_millis(1));
-                }
-                Ok(None) => panic!("timed out waiting for managed client event"),
-                Err(err) => panic!("subscription ended while waiting for event: {err}"),
-            }
-        }
-    }
-
-    fn collect_until_closed(sub: &EventSubscription) -> Vec<ClientEvent> {
-        let mut events = Vec::new();
-        let deadline = Instant::now() + Duration::from_secs(2);
-        while Instant::now() < deadline {
-            match sub.try_recv() {
-                Ok(Some(event)) => {
-                    let closed = matches!(event, ClientEvent::SessionClosed { .. });
-                    events.push(event);
-                    if closed {
-                        return events;
-                    }
-                }
-                Ok(None) => thread::sleep(Duration::from_millis(1)),
-                Err(_) => return events,
-            }
-        }
-        panic!("timed out waiting for session close event");
-    }
-
-    #[test]
-    fn stop_is_idempotent() {
-        let server = start_echo_server(test_params(None, Duration::from_millis(10)));
-        let session = ManagedClient::start(config(server.addr, None)).unwrap();
-        session.stop();
-        session.stop();
-        let outcome = session.join().unwrap();
-        server.join();
-
-        assert_eq!(outcome.end_reason, SessionEndReason::Cancelled);
-    }
-
-    #[test]
-    fn finite_managed_run_emits_session_probe_and_close_events() {
-        let duration = Duration::from_millis(35);
-        let server = start_echo_server(test_params(Some(duration), Duration::from_millis(10)));
-        let (session, sub) = ManagedClient::start_with_subscription(
-            config(server.addr, Some(duration)),
-            SubscriberConfig {
-                capacity: 16,
-                overflow: SubscriberOverflow::DropNewest,
-            },
-        )
-        .unwrap();
-
-        let events = collect_until_closed(&sub);
-        let outcome = session.join().unwrap();
-        server.join();
-
-        assert_eq!(outcome.end_reason, SessionEndReason::TestComplete);
-        assert!(outcome.packets_sent > 0);
-        assert!(outcome.replies_received > 0);
-        assert!(events
-            .iter()
-            .any(|event| matches!(event, ClientEvent::SessionStarted { .. })));
-        assert!(events
-            .iter()
-            .any(|event| matches!(event, ClientEvent::EchoReply { .. })));
-        assert!(events
-            .iter()
-            .any(|event| matches!(event, ClientEvent::SessionClosed { .. })));
-    }
-
-    #[test]
-    fn joining_before_consuming_preserves_queued_final_events() {
-        let duration = Duration::from_millis(25);
-        let server = start_echo_server(test_params(Some(duration), Duration::from_millis(10)));
-        let (session, sub) = ManagedClient::start_with_subscription(
-            config(server.addr, Some(duration)),
-            SubscriberConfig {
-                capacity: 32,
-                overflow: SubscriberOverflow::DropOldest,
-            },
-        )
-        .unwrap();
-
-        let outcome = session.join().unwrap();
-        let mut events = Vec::new();
-        loop {
-            match sub.try_recv() {
-                Ok(Some(event)) => events.push(event),
-                Ok(None) => panic!("joined managed subscription remained connected"),
-                Err(EventSubscriptionError::Disconnected) => break,
-            }
-        }
-        server.join();
-
-        assert_eq!(outcome.end_reason, SessionEndReason::TestComplete);
-        assert_eq!(sub.dropped_events(), 0);
-        assert!(matches!(
-            events.first(),
-            Some(ClientEvent::SessionStarted { .. })
-        ));
-        assert!(matches!(
-            events.last(),
-            Some(ClientEvent::SessionClosed { .. })
-        ));
-    }
-
-    #[test]
-    fn finite_managed_run_drains_late_reply_after_timeout_before_close() {
-        let interval = Duration::from_millis(100);
-        let duration = interval;
-        let params = test_params(Some(duration), interval);
-        let server = start_delayed_reply_server(params, Duration::from_millis(60));
-        let mut cfg = config(server.addr, Some(duration));
-        cfg.interval = interval;
-        cfg.probe_timeout = Duration::from_millis(20);
-
-        let (session, sub) = ManagedClient::start_with_subscription(
-            cfg,
-            SubscriberConfig {
-                capacity: 16,
-                overflow: SubscriberOverflow::DropNewest,
-            },
-        )
-        .unwrap();
-
-        let events = collect_until_closed(&sub);
-        let outcome = session.join().unwrap();
-        server.join();
-
-        assert_eq!(outcome.end_reason, SessionEndReason::TestComplete);
-        assert_eq!(outcome.packets_sent, 1);
-        assert_eq!(outcome.replies_received, 0);
-        assert_eq!(outcome.late, 1);
-        assert!(events
-            .iter()
-            .any(|event| matches!(event, ClientEvent::EchoLoss { seq: 0, .. })));
-        assert!(events.iter().any(|event| matches!(
-            event,
-            ClientEvent::LateReply {
-                seq: 0,
-                sent_at: Some(_),
-                rtt: Some(_),
-                ..
-            }
-        )));
-        let late_before_close = events
-            .iter()
-            .position(|event| matches!(event, ClientEvent::LateReply { rtt: Some(_), .. }));
-        let close = events
-            .iter()
-            .position(|event| matches!(event, ClientEvent::SessionClosed { .. }));
-        let late_before_close = late_before_close.expect("missing stats-eligible LateReply");
-        let close = close.expect("missing SessionClosed");
-        assert!(late_before_close < close);
-    }
-
-    #[test]
-    fn peer_close_during_managed_run_is_successful() {
-        let server = start_peer_close_server(test_params(None, Duration::from_millis(10)));
-        let (session, sub) = ManagedClient::start_with_subscription(
-            config(server.addr, None),
-            SubscriberConfig {
-                capacity: 16,
-                overflow: SubscriberOverflow::DropNewest,
-            },
-        )
-        .unwrap();
-
-        let events = collect_until_closed(&sub);
-        let outcome = session.join().unwrap();
-        server.join();
-
-        assert_eq!(outcome.end_reason, SessionEndReason::PeerClosed);
-        assert_eq!(outcome.packets_sent, 1);
-        assert_eq!(outcome.replies_received, 1);
-        assert!(events
-            .iter()
-            .any(|event| matches!(event, ClientEvent::EchoReply { .. })));
-        assert!(events
-            .iter()
-            .any(|event| matches!(event, ClientEvent::SessionClosed { .. })));
-    }
-
-    #[test]
-    fn continuous_managed_run_can_be_stopped_cleanly() {
-        let server = start_echo_server(test_params(None, Duration::from_millis(10)));
-        let session = ManagedClient::start(config(server.addr, None)).unwrap();
-        let sub = session
-            .subscribe(SubscriberConfig {
-                capacity: 16,
-                overflow: SubscriberOverflow::DropNewest,
-            })
-            .unwrap();
-
-        loop {
-            if matches!(recv_event_with_timeout(&sub), ClientEvent::EchoReply { .. }) {
-                break;
-            }
-        }
-        session.stop();
-        let events = collect_until_closed(&sub);
-        let outcome = session.join().unwrap();
-        server.join();
-
-        assert_eq!(outcome.end_reason, SessionEndReason::Cancelled);
-        assert!(outcome.packets_sent > 0);
-        assert!(outcome.replies_received > 0);
-        assert!(events
-            .iter()
-            .any(|event| matches!(event, ClientEvent::SessionClosed { .. })));
-    }
-
-    #[test]
-    fn join_reports_worker_panic() {
-        let hub = EventHub::new();
-        let cancellation = CancellationToken::new();
-        let worker = thread::spawn(|| -> Result<SessionOutcome, ClientError> {
-            panic!("intentional managed worker panic")
-        });
-        let session = ManagedClientSession {
-            hub,
-            cancellation,
-            worker: Some(worker),
-        };
-
-        assert!(matches!(session.join(), Err(ClientError::WorkerPanicked)));
-    }
-
-    #[test]
-    fn no_test_managed_run_returns_no_test_outcome() {
-        use crate::RunMode;
-
-        let params = test_params(Some(Duration::from_millis(10)), Duration::from_millis(10));
-        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
-        let addr = socket.local_addr().unwrap();
-        let (tx, rx) = mpsc::channel();
-        let done = thread::spawn(move || {
-            let (request, peer) = recv_request(&socket);
-            tx.send(request[3] & flags::FLAG_CLOSE != 0).unwrap();
-            socket
-                .send_to(
-                    &open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, &params),
-                    peer,
-                )
-                .unwrap();
-        });
-
-        let mut cfg = config(addr, Some(Duration::from_millis(10)));
-        cfg.run_mode = RunMode::NoTest;
-        let session = ManagedClient::start(cfg).unwrap();
-        assert!(rx.recv_timeout(Duration::from_secs(2)).unwrap());
-        let outcome = session.join().unwrap();
-        done.join().unwrap();
-
-        assert_eq!(outcome.end_reason, SessionEndReason::NoTestComplete);
-        assert_eq!(outcome.packets_sent, 0);
-    }
-}

crates/irtt-client/src/probe.rs

@@ -145,6 +145,7 @@ impl TimedOutMap {
         self.insertion_order.clear();
     }
 
+    #[cfg(test)]
     pub fn len(&self) -> usize {
         self.map.len()
     }

crates/irtt-client/src/receive.rs

@@ -15,14 +15,6 @@ pub(crate) struct ReceivedDatagram {
     pub(crate) meta: ReceiveMeta,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub(crate) struct ReceivedDatagramFrom {
-    pub(crate) len: usize,
-    pub(crate) source: SocketAddr,
-    pub(crate) received_at: ClientTimestamp,
-    pub(crate) meta: ReceiveMeta,
-}
-
 #[cfg(not(all(target_os = "linux", feature = "ancillary")))]
 pub(crate) fn recv_datagram(
     socket: &UdpSocket,
@@ -38,22 +30,6 @@ pub(crate) fn recv_datagram(
     })
 }
 
-#[cfg(not(all(target_os = "linux", feature = "ancillary")))]
-pub(crate) fn recv_datagram_from(
-    socket: &UdpSocket,
-    buf: &mut [u8],
-) -> Result<ReceivedDatagramFrom, io::Error> {
-    let (len, source) = socket.recv_from(buf)?;
-    let received_at = ClientTimestamp::now();
-
-    Ok(ReceivedDatagramFrom {
-        len,
-        source,
-        received_at,
-        meta: ReceiveMeta::default(),
-    })
-}
-
 #[cfg(all(target_os = "linux", feature = "ancillary"))]
 pub(crate) fn recv_datagram(
     socket: &UdpSocket,
@@ -88,14 +64,6 @@ pub(crate) fn try_recv_tokio_datagram(
     linux::try_recv_tokio_datagram(socket, buf)
 }
 
-#[cfg(all(target_os = "linux", feature = "ancillary"))]
-pub(crate) fn recv_datagram_from(
-    socket: &UdpSocket,
-    buf: &mut [u8],
-) -> Result<ReceivedDatagramFrom, io::Error> {
-    linux::recv_datagram_from(socket, buf)
-}
-
 #[cfg(not(all(target_os = "linux", feature = "ancillary")))]
 pub(crate) fn configure_receive_metadata(
     _socket: &UdpSocket,

crates/irtt-client/src/receive/linux.rs

@@ -1,17 +1,13 @@
 #![allow(unsafe_code)]
 use std::{
     io, mem,
-    net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6, UdpSocket},
+    net::{SocketAddr, UdpSocket},
     os::fd::{AsRawFd, RawFd},
     ptr,
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
-use crate::{
-    metadata::ReceiveMeta,
-    receive::{ReceivedDatagram, ReceivedDatagramFrom},
-    timing::ClientTimestamp,
-};
+use crate::{metadata::ReceiveMeta, receive::ReceivedDatagram, timing::ClientTimestamp};
 
 const CONTROL_LEN: usize = 128;
 
@@ -109,61 +105,6 @@ fn recv_datagram_fd(
     })
 }
 
-pub(crate) fn recv_datagram_from(
-    socket: &UdpSocket,
-    buf: &mut [u8],
-) -> Result<ReceivedDatagramFrom, io::Error> {
-    let mut control = ControlBuffer::new();
-    let mut name: libc::sockaddr_storage = unsafe { mem::zeroed() };
-    let mut iov = libc::iovec {
-        iov_base: buf.as_mut_ptr().cast(),
-        iov_len: buf.len(),
-    };
-    let mut msg: libc::msghdr = unsafe { mem::zeroed() };
-    msg.msg_name = (&mut name as *mut libc::sockaddr_storage).cast();
-    msg.msg_namelen = mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
-    msg.msg_iov = &mut iov;
-    msg.msg_iovlen = 1;
-    msg.msg_control = control.as_mut_ptr().cast();
-    msg.msg_controllen = control.len();
-
-    let len = unsafe {
-        // SAFETY: `msg` points to one writable iovec backed by `buf`, a
-        // writable source-address buffer, and a writable control buffer. All
-        // buffers live until `recvmsg` returns. The socket file descriptor is
-        // borrowed from a valid `UdpSocket`.
-        libc::recvmsg(socket.as_raw_fd(), &mut msg, 0)
-    };
-    if len < 0 {
-        return Err(io::Error::last_os_error());
-    }
-    let len = usize::try_from(len).map_err(|_| {
-        io::Error::new(
-            io::ErrorKind::InvalidData,
-            "recvmsg returned an unrepresentable datagram length",
-        )
-    })?;
-    let received_at = ClientTimestamp::now();
-    let source = unsafe {
-        // SAFETY: `name` and `msg_namelen` were initialized by a successful
-        // `recvmsg` call above.
-        socket_addr_from_storage(&name, msg.msg_namelen)?
-    };
-    let meta = unsafe {
-        // SAFETY: `msg` was initialized by a successful `recvmsg` call. The
-        // parser only reads cmsghdr entries within `msg_controllen` and ignores
-        // short or unrelated control messages.
-        parse_receive_meta(&msg)
-    };
-
-    Ok(ReceivedDatagramFrom {
-        len,
-        source,
-        received_at,
-        meta,
-    })
-}
-
 unsafe fn parse_receive_meta(msg: &libc::msghdr) -> ReceiveMeta {
     let mut meta = ReceiveMeta::default();
     let mut cmsg = libc::CMSG_FIRSTHDR(msg);
@@ -185,52 +126,6 @@ unsafe fn parse_receive_meta(msg: &libc::msghdr) -> ReceiveMeta {
     meta
 }
 
-unsafe fn socket_addr_from_storage(
-    storage: &libc::sockaddr_storage,
-    len: libc::socklen_t,
-) -> io::Result<SocketAddr> {
-    match i32::from(storage.ss_family) {
-        libc::AF_INET => {
-            if usize::try_from(len).unwrap_or(0) < mem::size_of::<libc::sockaddr_in>() {
-                return Err(io::Error::new(
-                    io::ErrorKind::InvalidData,
-                    "short IPv4 source address from recvmsg",
-                ));
-            }
-            let addr = ptr::read_unaligned(
-                (storage as *const libc::sockaddr_storage).cast::<libc::sockaddr_in>(),
-            );
-            let ip = Ipv4Addr::from(u32::from_be(addr.sin_addr.s_addr));
-            let port = u16::from_be(addr.sin_port);
-            Ok(SocketAddr::V4(SocketAddrV4::new(ip, port)))
-        }
-        libc::AF_INET6 => {
-            if usize::try_from(len).unwrap_or(0) < mem::size_of::<libc::sockaddr_in6>() {
-                return Err(io::Error::new(
-                    io::ErrorKind::InvalidData,
-                    "short IPv6 source address from recvmsg",
-                ));
-            }
-            let addr = ptr::read_unaligned(
-                (storage as *const libc::sockaddr_storage).cast::<libc::sockaddr_in6>(),
-            );
-            let ip = Ipv6Addr::from(addr.sin6_addr.s6_addr);
-            let port = u16::from_be(addr.sin6_port);
-            let flowinfo = u32::from_be(addr.sin6_flowinfo);
-            Ok(SocketAddr::V6(SocketAddrV6::new(
-                ip,
-                port,
-                flowinfo,
-                addr.sin6_scope_id,
-            )))
-        }
-        family => Err(io::Error::new(
-            io::ErrorKind::InvalidData,
-            format!("unsupported source address family from recvmsg: {family}"),
-        )),
-    }
-}
-
 fn is_traffic_class_cmsg(cmsg: &libc::cmsghdr) -> bool {
     (cmsg.cmsg_level == libc::IPPROTO_IP && cmsg.cmsg_type == libc::IP_TOS)
         || (cmsg.cmsg_level == libc::IPPROTO_IPV6 && cmsg.cmsg_type == libc::IPV6_TCLASS)

crates/irtt-client/src/session/machine.rs

@@ -480,6 +480,7 @@ impl SessionMachine {
         )
     }
 
+    #[cfg(test)]
     pub(crate) fn has_timed_out_metadata(&self) -> bool {
         matches!(
             &self.state,
@@ -487,6 +488,7 @@ impl SessionMachine {
         )
     }
 
+    #[cfg(any(feature = "tokio", test))]
     pub(crate) fn packets_sent(&self) -> u64 {
         match &self.state {
             MachineState::Open(session) => session.packets_sent,

crates/irtt-client/src/socket.rs

@@ -164,44 +164,6 @@ pub(crate) fn connect_tokio_udp_socket(
     Ok(tokio::net::UdpSocket::from_std(socket)?)
 }
 
-pub(crate) fn bind_unconnected_udp_socket(
-    config: &SocketConfig,
-    family_remote: SocketAddr,
-) -> Result<UdpSocket, ClientError> {
-    let domain = if family_remote.is_ipv4() {
-        Domain::IPV4
-    } else {
-        Domain::IPV6
-    };
-    let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
-
-    if config.ipv6_only && family_remote.is_ipv6() {
-        socket.set_only_v6(true)?;
-    }
-    let bind_addr = config.bind_addr.unwrap_or_else(|| {
-        if family_remote.is_ipv4() {
-            SocketAddr::from(([0, 0, 0, 0], 0))
-        } else {
-            SocketAddr::from(([0_u16; 8], 0))
-        }
-    });
-    socket.bind(&bind_addr.into())?;
-
-    let socket: UdpSocket = socket.into();
-    configure_receive_metadata(&socket, family_remote).map_err(|source| {
-        ClientError::SocketOption {
-            operation: "enable receive metadata",
-            remote: family_remote,
-            source,
-        }
-    })?;
-    if let Some(ttl) = config.ttl {
-        apply_ttl_to_socket(&socket, family_remote, ttl)?;
-    }
-    socket.set_read_timeout(config.recv_timeout)?;
-    Ok(socket)
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;