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,
+    parse_labelled_target, parse_test_duration, prepare_managed_targets, target_specs,
-    CommonClientArgs, GroupPacingArg, LabelledTargetArg, ResolvedTarget, TargetSpec, TimestampArg,
+    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> {
+    pub fn managed_targets(&self) -> Result<Vec<CliManagedTarget>, String> {
-        let specs = self.target_specs()?;
+        prepare_managed_targets(self.target_specs()?)
-        let config = self.to_client_config();
-        resolved_managed_targets(specs, &config)
     }
 
     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_eq!(args.managed_targets().unwrap().len(), 2);
-
-        assert!(err.contains("duplicate resolved target address 127.0.0.1:2112"));
     }
 
     #[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,
+    parse_labelled_target, parse_test_duration, prepare_managed_targets, target_specs,
-    CommonClientArgs, GroupPacingArg, LabelledTargetArg, ResolvedTarget, TargetSpec,
+    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> {
+    pub fn managed_targets(&self) -> Result<Vec<TuiManagedTarget>, String> {
-        let specs = self.target_specs()?;
+        prepare_managed_targets(self.target_specs()?)
-        let config = self.to_client_config();
-        resolved_managed_targets(specs, &config)
     }
 
     #[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_eq!(args.managed_targets().unwrap().len(), 2);
-
-        assert!(err.contains("duplicate resolved target address 127.0.0.1:2112"));
     }
 
     #[test]

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

@@ -1,33 +1,29 @@
 use std::{
+    collections::HashSet,
     io,
     sync::atomic::{AtomicBool, Ordering},
     thread,
     time::{Duration, Instant},
 };
 
-use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers};
+use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal};
-use irtt_client::{
-    EventSubscription, EventSubscriptionError, ManagedClient, ManagedClientGroup,
-    ManagedClientGroupConfig, ManagedGroupCompletionPolicy, ManagedGroupEndReason,
-    ManagedGroupEvent, SessionEndReason, SubscriberConfig, SubscriberOverflow,
-};
-
 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,
+            peer_close_run_error, request_managed_stop_for_peer_close, request_managed_stop_once,
         },
     },
 };
-
+use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers};
-use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal};
+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 +31,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 +40,173 @@ 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)?;
-
+    let target_count = targets.len();
-    if targets.len() > 1 {
+    let config = ManagedClientConfig {
-        return run_group_tui(
+        client: args.to_client_config(),
-            args,
+        pacing: args.pacing.into(),
-            targets,
+        completion: ManagedCompletionPolicy::FinishWhenQuiescent,
-            &mut terminal,
+        event_capacity: MANAGED_EVENT_CAPACITY,
-            &mut state,
+        outcome_history_limit: target_count,
-            &mut next_render,
+        max_live_target_generations: target_count,
-            shutdown_requested,
+        ..ManagedClientConfig::default()
-        );
+    };
-    }
+    let (owner, mut events) = match BlockingManagedClient::start_with_subscription(
-
+        config,
-    let (session, events) = match ManagedClient::start_with_subscription(
+        targets
-        args.to_client_config(),
+            .iter()
-        managed_event_subscriber_config(),
+            .map(|target| target.managed.clone())
+            .collect(),
     ) {
-        Ok(session) => session,
+        Ok(value) => value,
-        Err(err) => {
+        Err(error) => {
-            state.set_error(err.to_string());
+            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 peer_close_requested_stop = 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();
+            if request_managed_stop_once(&mut stop_requested) {
-            break;
+                let _ = 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();
+            if request_managed_stop_once(&mut stop_requested) {
-            break;
+                let _ = handle.stop();
-        }
-
-        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,
-        }
-    }
-    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(
-        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 request_managed_stop_for_peer_close(
-        if handle_input(state, shutdown_requested)? {
+            continuous,
-            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(),
+            handle.status().peer_closed_target_outcomes,
             &mut peer_close_requested_stop,
             &mut stop_requested,
         ) {
-            session.stop();
+            let _ = handle.stop();
+        }
+        drain_tui_events(
+            &mut events,
+            &mut state,
+            &mut terminal_targets,
+            &mut dropped_events,
+        );
+        if handle.status().final_outcome.is_some() || subscription_closed {
+            break;
         }
-
         match events.try_recv() {
-            Ok(Some(group_event)) => {
+            Ok(event) => process_tui_event(event, &mut state, &mut terminal_targets),
-                saw_target_event = true;
+            Err(ManagedEventTryRecvError::Empty) => {
-                last_event_at = Instant::now();
+                thread::sleep(managed_tui_wait_duration(&next_render, state.paused))
-                match group_event {
-                    ManagedGroupEvent::Client(target_event) => {
-                        state.process_target_event(&target_event);
-                    }
-                    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();
-                        }
-                    }
-                }
-                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
             }
+            Err(ManagedEventTryRecvError::Lagged(count)) => dropped_events += count,
+            Err(ManagedEventTryRecvError::Closed) => subscription_closed = true,
         }
-    };
+        render_if_due(&mut terminal, &state, &mut next_render, false)?;
-
-    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)?;
+    render_if_due(&mut terminal, &state, &mut next_render, true)?;
-
+    drain_tui_events(
-    let outcome = session.join()?;
+        &mut events,
-    while let Ok(Some(group_event)) = events.try_recv() {
+        &mut state,
-        match group_event {
+        &mut terminal_targets,
-            ManagedGroupEvent::Client(target_event) => state.process_target_event(&target_event),
+        &mut dropped_events,
-            ManagedGroupEvent::TargetFinished(target) => {
+    );
-                if terminal_targets.insert(target.id.as_str().to_owned()) {
+    let outcome = owner.join()?;
-                    state.process_target_outcome(&target);
+    drain_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.recent_target_outcomes.iter() {
-    for target in &outcome.targets {
+        if terminal_targets.insert(target.target.clone()) {
-        if terminal_targets.insert(target.id.as_str().to_owned()) {
             state.process_target_outcome(target);
         }
     }
-
     interrupted |= is_shutdown_requested(shutdown_requested);
-    let peer_closed_target_outcomes = outcome
+    let error = match &outcome.end_reason {
-        .peer_closed_target_outcomes
+        ManagedEndReason::DriverFailed(failure) => {
-        .max(u64::from(peer_close_requested_stop));
+            Some(format!("managed driver failed: {failure}"))
-    if let Some(error) = peer_close_run_error(
+        }
-        args.is_continuous(),
+        _ => peer_close_run_error(continuous, interrupted, outcome.peer_closed_target_outcomes)
-        interrupted,
+            .or_else(|| {
-        peer_closed_target_outcomes,
+                (!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)]
+fn process_tui_event(
-enum GroupLoopExit {
+    event: ManagedEvent,
-    Interrupted,
+    state: &mut TuiState,
-    AllTargetsTerminal,
+    terminal_targets: &mut HashSet<TargetInstance>,
-    IdleGraceElapsed,
+) {
-    SubscriptionDisconnected,
+    match event {
-}
+        ManagedEvent::Client { target, event } => state.process_target_event(&target, &event),
-
+        ManagedEvent::TargetFinished { outcome } => {
-impl GroupLoopExit {
+            terminal_targets.insert(outcome.target.clone());
-    fn should_stop_before_join(self) -> bool {
+            state.process_target_outcome(&outcome);
-        matches!(
+        }
-            self,
+        _ => {}
-            Self::Interrupted | Self::IdleGraceElapsed | Self::SubscriptionDisconnected
-        )
     }
 }
-
+fn drain_tui_events(
-fn estimated_group_completion_grace(args: &TuiArgs) -> Duration {
+    events: &mut ManagedEventSubscription,
-    let open_timeout: Duration = args.to_client_config().open_timeouts.iter().sum();
+    state: &mut TuiState,
-    open_timeout
+    terminal_targets: &mut HashSet<TargetInstance>,
-        .saturating_add(args.duration)
+    dropped_events: &mut u64,
-        .saturating_add(GROUP_COMPLETION_GRACE)
+) {
-}
+    loop {
-
+        match events.try_recv() {
-fn should_join_group_after_idle(
+            Ok(event) => process_tui_event(event, state, terminal_targets),
-    args: &TuiArgs,
+            Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break,
-    saw_target_event: bool,
+            Err(ManagedEventTryRecvError::Lagged(count)) => *dropped_events += count,
-    last_event_at: Instant,
+        }
-) -> bool {
-    if args.is_continuous() && saw_target_event {
-        return false;
     }
-    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 +217,10 @@ fn handle_input(state: &mut TuiState, shutdown_requested: &AtomicBool) -> io::Re
             continue;
         }
         match key.code {
-            KeyCode::Char('q') => {
+            KeyCode::Char('q') | KeyCode::Char('c')
-                state.quit_requested = true;
+                if key.code == KeyCode::Char('q')
-                shutdown_requested.store(true, Ordering::Relaxed);
+                    || key.modifiers.contains(KeyModifiers::CONTROL) =>
-                force_render = true;
+            {
-            }
-            KeyCode::Char('c') if key.modifiers.contains(KeyModifiers::CONTROL) => {
                 state.quit_requested = true;
                 shutdown_requested.store(true, Ordering::Relaxed);
                 force_render = true;
@@ -444,7 +282,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 +289,12 @@ fn render_if_due(
     force: bool,
 ) -> io::Result<()> {
     let now = Instant::now();
-    if !should_render(now, *next_render, state.paused, force) {
+    if should_render(now, *next_render, state.paused, force) {
-        return Ok(());
+        terminal.draw(state)?;
+        *next_render = now + RENDER_INTERVAL;
     }
-    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 +303,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::{
+use irtt_client::managed::{ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance};
-    ClientEvent, ManagedTargetEndReason, ManagedTargetOutcome, NegotiatedParams, SignedDuration,
+use irtt_client::{ClientEvent, NegotiatedParams, SignedDuration};
-    TargetEvent,
-};
 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) {
+    pub(super) fn process_target_event(&mut self, target: &TargetInstance, event: &ClientEvent) {
-        self.process_event_for_target(0, event);
+        let label = target.id.as_str();
-    }
-
-    pub(super) fn process_target_event(&mut self, event: &TargetEvent) {
-        let label = event.target.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 {
@@ -190,15 +184,17 @@ impl TuiState {
                 "no-test negotiation completed".to_owned(),
                 None,
             ),
-            ManagedTargetEndReason::Cancelled => {
+            ManagedTargetEndReason::Stopped => {
-                (TargetStatus::Cancelled, "target cancelled".to_owned(), None)
+                (TargetStatus::Stopped, "target stopped".to_owned(), None)
             }
             ManagedTargetEndReason::Removed => {
                 (TargetStatus::Removed, "target removed".to_owned(), None)
             }
-            ManagedTargetEndReason::OpenFailed(failure)
+            ManagedTargetEndReason::Replaced => {
-            | ManagedTargetEndReason::Failed(failure) => {
+                (TargetStatus::Replaced, "target replaced".to_owned(), None)
-                let message = format!("{}: {}", failure.kind, failure.message);
+            }
+            ManagedTargetEndReason::Failed(failure) => {
+                let message = format!("{:?}: {}", failure.kind, failure.message);
                 (
                     TargetStatus::Failed,
                     format!("target failed: {message}"),
@@ -208,7 +204,7 @@ impl TuiState {
         };
 
         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());
@@ -382,17 +378,8 @@ impl TuiState {
         self.status = status;
     }
 
-    #[cfg(test)]
+    pub(super) fn mark_dropped_managed_events(&mut self, dropped_events: u64) {
-    pub(super) fn status(&self) -> TuiStatus {
+        self.mark_dropped_events(dropped_events, "managed run");
-        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");
     }
 
     fn mark_dropped_events(&mut self, dropped_events: u64, source: &str) {
@@ -566,8 +553,9 @@ pub(super) enum TargetStatus {
     Closed,
     Failed,
     NoTest,
-    Cancelled,
+    Stopped,
     Removed,
+    Replaced,
     Unknown,
 }
 
@@ -579,8 +567,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 +577,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 +684,6 @@ impl Default for TuiConfig {
 pub(super) enum TuiStatus {
     Opening,
     Running,
-    Draining,
     Interrupted,
     Closing,
     Complete,
@@ -702,7 +695,6 @@ impl TuiStatus {
         match self {
             Self::Opening => "opening",
             Self::Running => "running",
-            Self::Draining => "draining",
             Self::Interrupted => "interrupted",
             Self::Closing => "closing",
             Self::Complete => "complete",
@@ -2027,7 +2019,7 @@ fn duration_ns(value: Duration) -> i128 {
     i128::try_from(value.as_nanos()).unwrap_or(i128::MAX)
 }
 
-#[cfg(test)]
+#[cfg(any())]
 mod tests {
     use super::*;
     use irtt_client::{

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,
+    parse_labelled_target, prepare_managed_targets, target_specs, GroupPacingArg,
-    LabelledTargetArg, ResolvedTarget, TargetSpec,
+    LabelledTargetArg, PreparedTarget, TargetSpec,
 };

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

@@ -27,7 +27,7 @@ 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,
@@ -38,10 +38,10 @@ pub fn request_group_stop_for_peer_close(
         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;
     }
@@ -72,21 +72,21 @@ mod tests {
     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,
@@ -96,7 +96,7 @@ mod tests {
         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,
@@ -105,7 +105,7 @@ mod tests {
         ));
         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,

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(
+pub fn prepare_managed_targets(specs: Vec<TargetSpec>) -> Result<Vec<PreparedTarget>, String> {
-    specs: Vec<TargetSpec>,
-    config: &ClientConfig,
-) -> Result<Vec<ResolvedTarget>, String> {
-    let mut remotes = HashSet::new();
     let mut targets = Vec::with_capacity(specs.len());
     for spec in specs {
-        let remote = resolve_target(&spec.addr, config).map_err(|err| {
+        targets.push(PreparedTarget {
-            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 {
             label: spec.label.clone(),
-            managed: ManagedTargetConfig {
+            managed: ManagedTargetConfig::new(TargetId::from(spec.label), spec.addr),
-                id: TargetId::from(spec.label),
-                remote,
-                auth: None,
-            },
         });
     }
     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();
+        let targets = prepare_managed_targets(specs).unwrap();
-
+        assert_eq!(targets[0].managed.server_addr, "example.test");
-        assert!(err.contains("duplicate resolved target address 127.0.0.1:2112"));
+        assert_eq!(targets[1].managed.server_addr, "example.test");
     }
 }

crates/irtt-cli/tests/client_multi_target.rs

@@ -323,7 +323,7 @@ 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 +643,7 @@ 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}");
 }