Merge branch 'feature/async-client-frontend'

Cargo.lock

@@ -317,6 +317,7 @@ dependencies = [
  "crossterm_winapi",
  "derive_more",
  "document-features",
+ "futures-core",
  "mio",
  "parking_lot",
  "rustix",
@@ -753,10 +754,12 @@ dependencies = [
  "clap",
  "crossterm",
  "ctrlc",
+ "futures-core",
  "irtt-client",
  "irtt-proto",
  "irtt-server",
  "irtt-stats",
+ "nix 0.31.3",
  "ratatui",
  "tokio",
 ]

crates/irtt-app/Cargo.toml

@@ -42,25 +42,29 @@ doc = false
 
 [features]
 default = ["client", "server", "tui"]
-client = ["dep:irtt-client", "irtt-client/ancillary", "irtt-client/tokio", "dep:irtt-proto", "dep:irtt-stats"]
+client = ["dep:tokio", "dep:irtt-client", "irtt-client/ancillary", "irtt-client/tokio", "dep:irtt-proto", "dep:irtt-stats"]
 server = ["dep:irtt-server", "dep:tokio"]
-tui =["dep:irtt-client", "irtt-client/ancillary", "irtt-client/tokio", "dep:irtt-proto", "dep:irtt-stats", "dep:ratatui", "dep:crossterm"]
+tui =["dep:tokio", "dep:futures-core", "dep:irtt-client", "irtt-client/ancillary", "irtt-client/tokio", "dep:irtt-proto", "dep:irtt-stats", "dep:ratatui", "dep:crossterm"]
 
 [dependencies]
 clap = { version = "4.6.7", features = ["derive", "env"] }
-crossterm = { version = "0.29.0", optional = true }
+crossterm = { version = "0.29.0", features = ["event-stream"], optional = true }
+futures-core = { version = "0.3.34", default-features = false, optional = true }
 ctrlc = { version = "3.5.2", features = ["termination"] }
 irtt-client = { workspace = true, optional = true }
 irtt-proto = { workspace = true, optional = true }
 irtt-server = { workspace = true, optional = true }
 irtt-stats = { workspace = true, optional = true }
 ratatui = { version = "0.30.2", optional = true }
-tokio = { version = "1.53.2", default-features = false, features = ["rt", "time"], optional = true }
+tokio = { version = "1.53.2", default-features = false, features = ["rt", "time", "sync", "macros"], optional = true }
 
 [dev-dependencies]
 irtt-server.workspace = true
 tokio = { version = "1.53.2", default-features = false, features = ["rt", "sync"] }
 
+[target.'cfg(unix)'.dev-dependencies]
+nix = { version = "0.31.3", features = ["term", "poll", "signal", "process"] }
+
 [profile.release]
 opt-level = "z"
 lto = true

crates/irtt-app/src/cmd/client/run.rs

@@ -1,15 +1,13 @@
 use std::{
     collections::{BTreeMap, BTreeSet, HashSet, VecDeque},
     io::{self, BufRead, Write},
-    sync::{atomic::AtomicBool, Arc, Mutex},
+    sync::Arc,
     thread,
-    time::Duration,
 };
 
 use irtt_client::{
     managed::{
-        BlockingManagedClient, ManagedClientHandle, ManagedCommandApplyError, ManagedEndReason,
+        ManagedClient, ManagedCommandApplyError, ManagedEndReason, ManagedEvent, ManagedStatus,
-        ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError, ManagedStatus,
         ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance,
     },
     ClientEvent,
@@ -21,11 +19,12 @@ use super::{
 };
 
 use crate::shared::client::{
-    expected_probe_count, is_shutdown_requested, parse_stdin_target_set,
+    expected_probe_count, parse_stdin_target_set,
     session::{
-        drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close,
+        drain_final_events, peer_close_run_error, request_managed_stop_for_peer_close,
-        request_managed_stop_once, should_print_final_summary, ManagedDrainState,
+        request_managed_stop_once, should_print_final_summary,
     },
+    worker::ManagedWorker,
     ManagedRunSetup, STDIN_MAX_DESIRED_TARGETS, STDIN_OUTCOME_HISTORY_LIMIT,
 };
 
@@ -40,12 +39,11 @@ const FINITE_STATS_MEMORY_WARNING_BYTES: u64 = 128 * MIB;
 const FINITE_STATS_MEMORY_STRONG_WARNING_BYTES: u64 = 512 * MIB;
 
 const FINITE_STATS_MEMORY_VERY_STRONG_WARNING_BYTES: u64 = GIB;
-const MANAGED_EVENT_WAIT_SLICE: Duration = Duration::from_millis(20);
 const MAX_STDIN_RECORD_BYTES: usize = 64 * 1024;
 
-pub fn run_stream(
+pub async fn run_stream(
     args: ClientArgs,
-    shutdown_requested: &AtomicBool,
+    mut shutdown: tokio::sync::watch::Receiver<bool>,
 ) -> Result<(), Box<dyn std::error::Error>> {
     if args.list_columns {
         print!("{}", OutputConfig::list_columns());
@@ -63,7 +61,7 @@ pub fn run_stream(
     )
     .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?;
     if setup.stdin_controlled {
-        return run_stdin_stream(setup, output_config, shutdown_requested);
+        return run_stdin_stream(setup, output_config, shutdown).await;
     }
     let continuous = args.is_continuous();
     let target_count = setup.target_count();
@@ -71,14 +69,13 @@ pub fn run_stream(
     if let Some(warning) = finite_stats_memory_warning(&args, target_count) {
         eprintln!("{warning}");
     }
-    if is_shutdown_requested(shutdown_requested) {
+    if *shutdown.borrow() {
         return Ok(());
     }
-    let (owner, mut events) = BlockingManagedClient::start_with_subscription(
+    let (task, handle) = ManagedClient::task(setup.managed_config(), setup.managed_targets())?;
-        setup.managed_config(),
+    let mut events = handle.subscribe()?;
-        setup.managed_targets(),
+    let mut status = handle.subscribe_status();
-    )?;
+    let owner = ManagedWorker::start(task, handle.clone())?;
-    let handle = owner.handle();
     let mut stdout = io::LineWriter::new(io::stdout().lock());
     let mut stream_output = StreamOutput {
         config: output_config,
@@ -102,9 +99,9 @@ pub fn run_stream(
     let mut dropped_events = 0_u64;
     let mut interrupted = false;
     let mut stop_requested = false;
-    let mut subscription_closed = false;
     loop {
-        if is_shutdown_requested(shutdown_requested) {
+        let snapshot = status.borrow_and_update().clone();
+        if *shutdown.borrow() {
             interrupted = true;
             if request_managed_stop_once(&mut stop_requested) {
                 drop(handle.stop());
@@ -113,38 +110,33 @@ pub fn run_stream(
         if request_managed_stop_for_peer_close(
             continuous,
             interrupted,
-            handle.status().peer_closed_target_outcomes,
+            snapshot.peer_closed_target_outcomes,
             &mut stop_requested,
         ) {
             drop(handle.stop());
         }
-        let drain_state = drain_events(
+        if snapshot.final_outcome.is_some() {
-            &mut events,
-            &mut stream_output,
-            &mut stats,
-            &mut terminal_targets,
-            &mut dropped_events,
-        )?;
-        match drain_state {
-            ManagedDrainState::Empty => thread::sleep(MANAGED_EVENT_WAIT_SLICE),
-            ManagedDrainState::Closed => subscription_closed = true,
-            ManagedDrainState::BudgetExhausted => {}
-        }
-        if handle.status().final_outcome.is_some() || subscription_closed {
             break;
         }
+        tokio::select! {
+            _ = shutdown.changed(), if !interrupted => {}
+            changed = status.changed() => { if changed.is_err() { break; } }
+            event = events.recv() => match event {
+                Ok(event) => process_event(event, &mut stream_output, &mut stats, &mut terminal_targets)?,
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => {
+                    dropped_events = dropped_events.saturating_add(count);
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+            }
+        }
     }
     if interrupted {
         eprintln!("interrupted, closing managed run...");
     }
-    let outcome = owner.join()?;
+    let outcome = owner.join().await?;
-    drain_final_events(
+    drain_final_events(&mut events, &mut dropped_events, |event| {
-        &mut events,
+        process_event(event, &mut stream_output, &mut stats, &mut terminal_targets)
-        &mut stream_output,
+    })?;
-        &mut stats,
-        &mut terminal_targets,
-        &mut dropped_events,
-    )?;
     if let Some(warning) = dropped_event_warning(dropped_events) {
         eprintln!("{warning}");
     }
@@ -159,7 +151,7 @@ pub fn run_stream(
             report_target_failure(target);
         }
     }
-    interrupted |= is_shutdown_requested(shutdown_requested);
+    interrupted |= *shutdown.borrow();
     let terminal_error = match &outcome.end_reason {
         ManagedEndReason::DriverFailed(failure) => {
             Some(format!("managed driver failed: {failure}"))
@@ -201,98 +193,9 @@ pub fn run_stream(
 }
 
 #[derive(Clone)]
-struct StdinTargetSet {
+enum StdinUpdate {
-    revision: u64,
+    Targets(Vec<ManagedTargetConfig>),
-    targets: Vec<ManagedTargetConfig>,
+    Stop(StdinStop),
-}
-
-#[derive(Default)]
-struct StdinMailboxState {
-    next_revision: u64,
-    latest: Option<StdinTargetSet>,
-    eof: bool,
-    fatal: Option<String>,
-}
-
-#[derive(Default)]
-struct StdinMailbox {
-    state: Mutex<StdinMailboxState>,
-}
-
-impl StdinMailbox {
-    fn publish(&self, targets: Vec<ManagedTargetConfig>) {
-        let mut state = self
-            .state
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        state.next_revision = state.next_revision.saturating_add(1);
-        state.latest = Some(StdinTargetSet {
-            revision: state.next_revision,
-            targets,
-        });
-    }
-
-    fn finish(&self) {
-        self.state
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .eof = true;
-    }
-
-    fn fail(&self, error: String) {
-        let mut state = self
-            .state
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        state.fatal = Some(error);
-        state.latest = None;
-    }
-
-    fn latest(&self) -> Option<StdinTargetSet> {
-        self.state
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .latest
-            .clone()
-    }
-
-    fn clear_if_current(&self, revision: u64) {
-        let mut state = self
-            .state
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        if state
-            .latest
-            .as_ref()
-            .is_some_and(|set| set.revision == revision)
-        {
-            state.latest = None;
-        }
-    }
-
-    fn is_current(&self, revision: u64) -> bool {
-        self.state
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .latest
-            .as_ref()
-            .is_some_and(|set| set.revision == revision)
-    }
-
-    fn fatal(&self) -> Option<String> {
-        self.state
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .fatal
-            .clone()
-    }
-
-    fn is_finished(&self) -> bool {
-        self.state
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .eof
-    }
 }
 
 fn read_stdin_record<R: BufRead>(
@@ -371,7 +274,10 @@ fn read_stdin_record<R: BufRead>(
     }
 }
 
-fn read_stdin_target_sets<R: BufRead>(reader: &mut R, mailbox: &StdinMailbox) {
+fn read_stdin_target_sets<R: BufRead>(
+    reader: &mut R,
+    updates: &tokio::sync::watch::Sender<StdinUpdate>,
+) {
     let mut line = 0_u64;
     let mut record = Vec::with_capacity(MAX_STDIN_RECORD_BYTES + 1);
     loop {
@@ -383,101 +289,43 @@ fn read_stdin_target_sets<R: BufRead>(reader: &mut R, mailbox: &StdinMailbox) {
                 }
                 match parse_stdin_target_set(&record, STDIN_MAX_DESIRED_TARGETS) {
                     Ok(targets) => {
-                        mailbox.publish(targets.into_iter().map(|target| target.managed).collect())
+                        updates.send_replace(StdinUpdate::Targets(
+                            targets.into_iter().map(|target| target.managed).collect(),
+                        ));
                     }
                     Err(error) => {
-                        mailbox.fail(format!("invalid --targets-stdin line {line}: {error}"));
+                        updates.send_replace(StdinUpdate::Stop(StdinStop::Fatal(format!(
+                            "invalid --targets-stdin line {line}: {error}"
+                        ))));
                         return;
                     }
                 }
             }
             Ok(None) => {
-                mailbox.finish();
+                updates.send_replace(StdinUpdate::Stop(StdinStop::Eof));
                 return;
             }
             Err(_) => {
-                mailbox.fail(format!("failed to read --targets-stdin line {}", line + 1));
+                updates.send_replace(StdinUpdate::Stop(StdinStop::Fatal(format!(
+                    "failed to read --targets-stdin line {}",
+                    line + 1
+                ))));
                 return;
             }
         }
     }
 }
 
-fn spawn_stdin_target_reader(mailbox: Arc<StdinMailbox>) -> io::Result<()> {
+fn spawn_stdin_target_reader(updates: tokio::sync::watch::Sender<StdinUpdate>) -> io::Result<()> {
     thread::Builder::new()
         .name("irtt-targets-stdin".to_owned())
         .spawn(move || {
             let stdin = io::stdin();
-            read_stdin_target_sets(&mut stdin.lock(), &mailbox);
+            read_stdin_target_sets(&mut stdin.lock(), &updates);
         })
         .map(|_| ())
 }
 
-#[derive(Clone)]
-struct RetryAfterStatus {
-    revision: u64,
-    status: Arc<ManagedStatus>,
-}
-
-fn retry_waits_for_status_change(
-    retry_after: &mut Option<RetryAfterStatus>,
-    target_set: &StdinTargetSet,
-    status: &Arc<ManagedStatus>,
-) -> bool {
-    let Some(retry) = retry_after else {
-        return false;
-    };
-    if retry.revision != target_set.revision {
-        *retry_after = None;
-        return false;
-    }
-    Arc::ptr_eq(&retry.status, status)
-}
-
-fn retry_after_capacity_rejection(
-    mailbox: &StdinMailbox,
-    target_set: &StdinTargetSet,
-    status: Arc<ManagedStatus>,
-) -> Option<RetryAfterStatus> {
-    mailbox
-        .is_current(target_set.revision)
-        .then_some(RetryAfterStatus {
-            revision: target_set.revision,
-            status,
-        })
-}
-
-fn apply_latest_stdin_target_set(
-    handle: &ManagedClientHandle,
-    mailbox: &StdinMailbox,
-    retry_after: &mut Option<RetryAfterStatus>,
-) -> Result<(), String> {
-    let Some(target_set) = mailbox.latest() else {
-        *retry_after = None;
-        return Ok(());
-    };
-    let status = handle.status();
-    if retry_waits_for_status_change(retry_after, &target_set, &status) {
-        return Ok(());
-    }
-
-    let receipt = handle
-        .update_targets(target_set.targets.clone())
-        .map_err(|_| "failed to submit --targets-stdin update".to_owned())?;
-    match receipt.blocking_wait() {
-        Ok(_) => {
-            mailbox.clear_if_current(target_set.revision);
-            *retry_after = None;
-            Ok(())
-        }
-        Err(ManagedCommandApplyError::LiveGenerationLimitExceeded { .. }) => {
-            *retry_after = retry_after_capacity_rejection(mailbox, &target_set, status);
-            Ok(())
-        }
-        Err(_) => Err("--targets-stdin update was rejected".to_owned()),
-    }
-}
-
 struct BoundedTargetSet {
     limit: usize,
     order: VecDeque<TargetInstance>,
@@ -542,37 +390,6 @@ fn process_stdin_event<W: Write>(
     Ok(())
 }
 
-fn drain_stdin_events<W: Write>(
-    events: &mut ManagedEventSubscription,
-    stream_output: &mut StreamOutput<'_, W>,
-    stats: &mut BTreeMap<TargetInstance, StatsCollector>,
-    terminal_targets: &mut BoundedTargetSet,
-    dropped_events: &mut u64,
-) -> io::Result<ManagedDrainState> {
-    drain_managed_events(events, dropped_events, |event| {
-        process_stdin_event(event, stream_output, stats, terminal_targets)
-    })
-}
-
-fn drain_final_stdin_events<W: Write>(
-    events: &mut ManagedEventSubscription,
-    stream_output: &mut StreamOutput<'_, W>,
-    stats: &mut BTreeMap<TargetInstance, StatsCollector>,
-    terminal_targets: &mut BoundedTargetSet,
-    dropped_events: &mut u64,
-) -> io::Result<()> {
-    loop {
-        match events.try_recv() {
-            Ok(event) => process_stdin_event(event, stream_output, stats, terminal_targets)?,
-            Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break,
-            Err(ManagedEventTryRecvError::Lagged(count)) => {
-                *dropped_events = dropped_events.saturating_add(count);
-            }
-        }
-    }
-    Ok(())
-}
-
 fn reconcile_stdin_stats(
     stats: &mut BTreeMap<TargetInstance, StatsCollector>,
     status: &ManagedStatus,
@@ -613,36 +430,20 @@ enum StdinStop {
     Fatal(String),
 }
 
-fn stdin_stop_request(
+async fn run_stdin_stream(
-    shutdown_requested: &AtomicBool,
-    mailbox: &StdinMailbox,
-) -> Option<StdinStop> {
-    if is_shutdown_requested(shutdown_requested) {
-        Some(StdinStop::Interrupted)
-    } else if let Some(error) = mailbox.fatal() {
-        Some(StdinStop::Fatal(error))
-    } else if mailbox.is_finished() {
-        Some(StdinStop::Eof)
-    } else {
-        None
-    }
-}
-
-fn run_stdin_stream(
     setup: ManagedRunSetup,
     output_config: OutputConfig,
-    shutdown_requested: &AtomicBool,
+    mut shutdown: tokio::sync::watch::Receiver<bool>,
 ) -> Result<(), Box<dyn std::error::Error>> {
-    if is_shutdown_requested(shutdown_requested) {
+    if *shutdown.borrow() {
         return Ok(());
     }
-    let (owner, mut events) = BlockingManagedClient::start_with_subscription(
+    let (task, handle) = ManagedClient::task(setup.managed_config(), setup.managed_targets())?;
-        setup.managed_config(),
+    let mut events = handle.subscribe()?;
-        setup.managed_targets(),
+    let mut status = handle.subscribe_status();
-    )?;
+    let owner = ManagedWorker::start(task, handle.clone())?;
-    let handle = owner.handle();
+    let (updates, mut stdin) = tokio::sync::watch::channel(StdinUpdate::Targets(Vec::new()));
-    let mailbox = Arc::new(StdinMailbox::default());
+    spawn_stdin_target_reader(updates)?;
-    spawn_stdin_target_reader(Arc::clone(&mailbox))?;
 
     let mut stdout = io::LineWriter::new(io::stdout().lock());
     let mut stream_output = StreamOutput {
@@ -654,59 +455,103 @@ fn run_stdin_stream(
     };
     let mut stats = BTreeMap::new();
     let mut terminal_targets = BoundedTargetSet::new(STDIN_OUTCOME_HISTORY_LIMIT);
+    let mut stdin_changed = false;
+    let mut desired = None;
+    let mut pending = None;
+    let mut submitted = Vec::new();
     let mut retry_after = None;
     let mut dropped_events = 0_u64;
     let mut stop_requested = false;
     let mut stop = None;
     let mut summary_targets = None;
-    let mut subscription_closed = false;
-
     loop {
-        let status = handle.status();
+        let snapshot = status.borrow_and_update().clone();
         if stop.is_none() {
-            if let Some(requested_stop) = stdin_stop_request(shutdown_requested, &mailbox) {
+            if *shutdown.borrow() {
-                summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status));
+                stop = Some(StdinStop::Interrupted);
-                stop = Some(requested_stop);
+            } else if stdin_changed || stdin.has_changed().unwrap_or(true) {
-                if request_managed_stop_once(&mut stop_requested) {
+                stdin_changed = false;
-                    drop(handle.stop());
+                match stdin.borrow_and_update().clone() {
+                    StdinUpdate::Targets(targets) => {
+                        desired = Some(targets);
+                        retry_after = None;
+                    }
+                    StdinUpdate::Stop(reason) => stop = Some(reason),
                 }
-            } else if let Err(error) =
+            }
-                apply_latest_stdin_target_set(&handle, &mailbox, &mut retry_after)
+            if stop.is_none()
+                && pending.is_none()
+                && !retry_after
+                    .as_ref()
+                    .is_some_and(|previous| Arc::ptr_eq(previous, &snapshot))
             {
-                summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status));
+                if let Some(targets) = desired.take() {
-                stop = Some(StdinStop::Fatal(error));
+                    submitted = targets;
+                    match handle.update_targets(submitted.clone()) {
+                        Ok(receipt) => {
+                            pending = Some(receipt);
+                            retry_after = Some(snapshot.clone());
+                        }
+                        Err(_) => {
+                            stop = Some(StdinStop::Fatal(
+                                "failed to submit --targets-stdin update".to_owned(),
+                            ))
+                        }
+                    }
+                }
+            }
+            if stop.is_some() {
+                summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &snapshot));
                 if request_managed_stop_once(&mut stop_requested) {
                     drop(handle.stop());
                 }
             }
         }
-
+        reconcile_stdin_stats(&mut stats, &snapshot, summary_targets.as_ref());
-        let drain_state = drain_stdin_events(
+        if snapshot.final_outcome.is_some() {
-            &mut events,
-            &mut stream_output,
-            &mut stats,
-            &mut terminal_targets,
-            &mut dropped_events,
-        )?;
-        reconcile_stdin_stats(&mut stats, &handle.status(), summary_targets.as_ref());
-        match drain_state {
-            ManagedDrainState::Empty => thread::sleep(MANAGED_EVENT_WAIT_SLICE),
-            ManagedDrainState::Closed => subscription_closed = true,
-            ManagedDrainState::BudgetExhausted => {}
-        }
-        if handle.status().final_outcome.is_some() || subscription_closed {
             break;
         }
+        tokio::select! {
+            _ = shutdown.changed(), if stop.is_none() => {}
+            _ = stdin.changed(), if stop.is_none() => { stdin_changed = true; }
+            changed = status.changed() => { if changed.is_err() { break; } }
+            result = async { pending.as_mut().expect("guarded pending receipt").await }, if pending.is_some() && stop.is_none() => {
+                pending = None;
+                match result {
+                    Ok(_) => retry_after = None,
+                    Err(ManagedCommandApplyError::LiveGenerationLimitExceeded { .. }) => {
+                        // A newer desired set supersedes this rejected set. Otherwise,
+                        // retry only after the driver's status advances from submission.
+                        if desired.is_none() {
+                            desired = Some(std::mem::take(&mut submitted));
+                        } else {
+                            retry_after = None;
+                        }
+                    }
+                    Err(_) => {
+                        stop = Some(StdinStop::Fatal("--targets-stdin update was rejected".to_owned()));
+                        summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status.borrow()));
+                        if request_managed_stop_once(&mut stop_requested) { drop(handle.stop()); }
+                    }
+                }
+            }
+            event = events.recv() => match event {
+                Ok(event) => {
+                    process_stdin_event(event, &mut stream_output, &mut stats, &mut terminal_targets)?;
+                    reconcile_stdin_stats(&mut stats, &status.borrow(), summary_targets.as_ref());
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(count)) => {
+                    dropped_events = dropped_events.saturating_add(count);
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+            }
+        }
     }
 
-    let outcome = owner.join()?;
+    let outcome = owner.join().await?;
-    drain_final_stdin_events(
+    drain_final_events(&mut events, &mut dropped_events, |event| {
-        &mut events,
+        process_stdin_event(event, &mut stream_output, &mut stats, &mut terminal_targets)
-        &mut stream_output,
+    })?;
-        &mut stats,
-        &mut terminal_targets,
-        &mut dropped_events,
-    )?;
     if let Some(warning) = dropped_event_warning(dropped_events) {
         eprintln!("{warning}");
     }
@@ -782,37 +627,6 @@ fn process_event<W: Write>(
     Ok(())
 }
 
-fn drain_events<W: Write>(
-    events: &mut ManagedEventSubscription,
-    stream_output: &mut StreamOutput<'_, W>,
-    stats: &mut BTreeMap<String, StatsCollector>,
-    terminal_targets: &mut HashSet<TargetInstance>,
-    dropped_events: &mut u64,
-) -> io::Result<ManagedDrainState> {
-    drain_managed_events(events, dropped_events, |event| {
-        process_event(event, stream_output, stats, terminal_targets)
-    })
-}
-
-fn drain_final_events<W: Write>(
-    events: &mut ManagedEventSubscription,
-    stream_output: &mut StreamOutput<'_, W>,
-    stats: &mut BTreeMap<String, StatsCollector>,
-    terminal_targets: &mut HashSet<TargetInstance>,
-    dropped_events: &mut u64,
-) -> io::Result<()> {
-    loop {
-        match events.try_recv() {
-            Ok(event) => process_event(event, stream_output, stats, terminal_targets)?,
-            Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break,
-            Err(ManagedEventTryRecvError::Lagged(count)) => {
-                *dropped_events = dropped_events.saturating_add(count);
-            }
-        }
-    }
-    Ok(())
-}
-
 fn report_target_failure(target: &ManagedTargetOutcome) {
     for message in target_failure_messages(target) {
         eprintln!("{message}");

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

@@ -1,133 +1,168 @@
-use std::{
+use std::{collections::HashSet, future::poll_fn, io, time::Duration};
-    collections::HashSet,
-    io,
-    sync::atomic::{AtomicBool, Ordering},
-    thread,
-    time::{Duration, Instant},
-};
 
-use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal};
+use super::ui::{TuiConfig, TuiState, TuiStatus, TuiTerminal};
 use crate::{
     cmd::tui::args::TuiArgs,
     shared::client::{
-        is_shutdown_requested,
         session::{
-            drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close,
+            drain_final_events, peer_close_run_error, request_managed_stop_for_peer_close,
-            request_managed_stop_once, ManagedDrainState,
+            request_managed_stop_once,
         },
+        worker::ManagedWorker,
     },
 };
 use crossterm::{
-    event::{self, Event, KeyCode, KeyEventKind, KeyModifiers},
+    event::{Event, EventStream, KeyCode, KeyEventKind, KeyModifiers},
     terminal,
 };
-use irtt_client::managed::{
+use futures_core::Stream;
-    BlockingManagedClient, ManagedEndReason, ManagedEvent, ManagedEventSubscription,
+use irtt_client::{
-    ManagedEventTryRecvError, TargetInstance,
+    managed::{ManagedClient, ManagedEndReason, ManagedEvent, TargetInstance},
+    ClientEvent,
 };
 use ratatui::layout::Rect;
+use tokio::{
+    sync::{broadcast::error::RecvError, watch},
+    time::{sleep, Instant},
+};
 
 const RENDER_INTERVAL: Duration = Duration::from_millis(250);
-const TUI_WAIT_SLICE: Duration = Duration::from_millis(20);
-const INPUT_EVENT_WORK_BUDGET: usize = 128;
 
-pub fn run_tui(
+pub async fn run_tui(
     args: TuiArgs,
-    shutdown_requested: &AtomicBool,
+    mut shutdown: watch::Receiver<bool>,
 ) -> Result<(), Box<dyn std::error::Error>> {
     let setup = args
         .prepare()
         .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?;
     let continuous = args.is_continuous();
+    if *shutdown.borrow() {
+        return Ok(());
+    }
     let mut terminal = TuiTerminal::enter()?;
     let mut state = TuiState::with_target_labels(
         TuiConfig::from_args(&args, &setup),
         setup.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)?;
+    terminal.draw(&state)?;
-    let (owner, mut events) = match BlockingManagedClient::start_with_subscription(
+    let (task, handle) = match ManagedClient::task(setup.managed_config(), setup.managed_targets())
-        setup.managed_config(),
+    {
-        setup.managed_targets(),
-    ) {
         Ok(value) => value,
         Err(error) => {
             state.set_error(error.to_string());
-            render_if_due(&mut terminal, &state, &mut next_render, true)?;
+            terminal.draw(&state)?;
             return Err(Box::new(error));
         }
     };
-    let handle = owner.handle();
+    // Both subscriptions exist before the measurement task can be polled.
+    let mut events = handle.subscribe()?;
+    let mut status = handle.subscribe_status();
+    let worker = match ManagedWorker::start(task, handle.clone()) {
+        Ok(worker) => worker,
+        Err(error) => {
+            state.set_run_error(error.to_string());
+            terminal.draw(&state)?;
+            return Err(error.into());
+        }
+    };
+    let completion = worker.join();
+    tokio::pin!(completion);
+    // Input and rendering stay on the frontend executor. EventStream's drop
+    // wakes its reader; no terminal task can outlive the terminal guard.
+    let mut input = EventStream::new();
+    let render = sleep(RENDER_INTERVAL);
+    tokio::pin!(render);
     let mut interrupted = false;
     let mut stop_requested = false;
     let mut terminal_targets = HashSet::new();
-    let mut dropped_events = 0;
+    let mut dropped_events = 0_u64;
-    let mut subscription_closed = false;
+    let mut events_closed = false;
-    loop {
+    let mut status_closed = false;
-        if is_shutdown_requested(shutdown_requested) {
+    let outcome = loop {
-            interrupted = true;
+        let mut force_render = false;
-            if request_managed_stop_once(&mut stop_requested) {
+        // Default select fairness lets input, status and rendering progress
-                drop(handle.stop());
+        // even when measurement events are continuously ready.
+        tokio::select! {
+            outcome = &mut completion => break outcome,
+            _ = shutdown.changed(), if !interrupted => {
+                interrupted = true;
+                state.set_status(TuiStatus::Interrupted);
+                force_render = true;
             }
-        }
+            changed = status.changed(), if !status_closed => {
-        if handle_input(&mut state, shutdown_requested)? {
+                status_closed = changed.is_err();
-            render_if_due(&mut terminal, &state, &mut next_render, true)?;
+                let snapshot = status.borrow_and_update().clone();
-        }
+                if request_managed_stop_for_peer_close(
-        if state.quit_requested {
+                    continuous, interrupted, snapshot.peer_closed_target_outcomes,
-            interrupted = true;
+                    &mut stop_requested,
-            if request_managed_stop_once(&mut stop_requested) {
+                ) {
-                drop(handle.stop());
+                    drop(handle.stop());
+                }
+                force_render = true;
             }
+            event = events.recv(), if !events_closed => match event {
+                Ok(event) => force_render = process_tui_event(
+                    event, &mut state, &mut terminal_targets,
+                ),
+                Err(RecvError::Lagged(count)) => {
+                    dropped_events = dropped_events.saturating_add(count);
+                    state.mark_dropped_managed_events(dropped_events);
+                    force_render = true;
+                }
+                Err(RecvError::Closed) => events_closed = true,
+            },
+            event = poll_fn(|cx| std::pin::Pin::new(&mut input).poll_next(cx)) => {
+                let event = event.ok_or_else(|| io::Error::new(
+                    io::ErrorKind::UnexpectedEof, "terminal input closed",
+                ))??;
+                match event {
+                    Event::Resize(_, _) => force_render = true,
+                    Event::Key(key) if key.kind != KeyEventKind::Release => {
+                        match key.code {
+                            KeyCode::Char('q') | KeyCode::Char('c')
+                                if key.code == KeyCode::Char('q')
+                                    || key.modifiers.contains(KeyModifiers::CONTROL) => {
+                                interrupted = true;
+                                state.set_status(TuiStatus::Interrupted);
+                                force_render = true;
+                            }
+                            _ => {
+                                let (width, height) = terminal::size()?;
+                                force_render = state.handle_key(
+                                    key.code, Rect::new(0, 0, width, height),
+                                );
+                            }
+                        }
+                    }
+                    _ => {}
+                }
+            }
+            _ = &mut render, if !state.paused => force_render = true,
         }
-        if request_managed_stop_for_peer_close(
+        if interrupted && request_managed_stop_once(&mut stop_requested) {
-            continuous,
-            interrupted,
-            handle.status().peer_closed_target_outcomes,
-            &mut stop_requested,
-        ) {
             drop(handle.stop());
         }
-        let previous_dropped = dropped_events;
+        if force_render {
-        let drain_state = drain_tui_events(
+            terminal.draw(&state)?;
-            &mut events,
+            // Schedule from the completed draw, never catch up missed frames.
-            &mut state,
+            render.as_mut().reset(Instant::now() + RENDER_INTERVAL);
-            &mut terminal_targets,
-            &mut dropped_events,
-        );
-        if dropped_events != previous_dropped {
-            // Data loss changes the meaning of displayed statistics, even while
-            // ordinary measurement refreshes are paused.
-            render_if_due(&mut terminal, &state, &mut next_render, true)?;
         }
-        match drain_state {
+    };
-            ManagedDrainState::Empty => {
-                thread::sleep(managed_tui_wait_duration(&next_render, state.paused));
-            }
-            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)?;
-    }
-    if interrupted {
-        state.set_status(TuiStatus::Interrupted);
-        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)?;
+    terminal.draw(&state)?;
-    let outcome = owner.join()?;
+    let outcome = match outcome {
-    drain_final_tui_events(
+        Ok(outcome) => outcome,
-        &mut events,
+        Err(error) => {
-        &mut state,
+            state.set_run_error(error.to_string());
-        &mut terminal_targets,
+            terminal.draw(&state)?;
-        &mut dropped_events,
+            return Err(error);
-    );
+        }
+    };
+    drain_final_events(&mut events, &mut dropped_events, |event| {
+        process_tui_event(event, &mut state, &mut terminal_targets);
+        Ok::<(), std::convert::Infallible>(())
+    })?;
     state.mark_dropped_managed_events(dropped_events);
     if outcome.discarded_target_outcomes != 0 {
         state.set_run_error(format!(
@@ -140,7 +175,7 @@ pub fn run_tui(
             state.process_target_outcome(target);
         }
     }
-    interrupted |= is_shutdown_requested(shutdown_requested);
+    interrupted |= *shutdown.borrow();
     let error = match &outcome.end_reason {
         ManagedEndReason::DriverFailed(failure) => {
             Some(format!("managed driver failed: {failure}"))
@@ -160,11 +195,11 @@ pub fn run_tui(
     };
     if let Some(error) = error {
         state.set_run_error(error.clone());
-        render_if_due(&mut terminal, &state, &mut next_render, true)?;
+        terminal.draw(&state)?;
         return Err(error.into());
     }
     state.set_status(TuiStatus::Complete);
-    render_if_due(&mut terminal, &state, &mut next_render, true)?;
+    terminal.draw(&state)?;
     Ok(())
 }
 
@@ -172,145 +207,24 @@ fn process_tui_event(
     event: ManagedEvent,
     state: &mut TuiState,
     terminal_targets: &mut HashSet<TargetInstance>,
-) {
+) -> bool {
     match event {
-        ManagedEvent::Client { target, event } => state.process_target_event(&target, &event),
+        ManagedEvent::Client { target, event } => {
+            let force_render = matches!(
+                event,
+                ClientEvent::SessionStarted(_)
+                    | ClientEvent::NoTestCompleted(_)
+                    | ClientEvent::SessionClosed { .. }
+                    | ClientEvent::Warning { .. }
+            );
+            state.process_target_event(&target, &event);
+            force_render
+        }
         ManagedEvent::TargetFinished { outcome } => {
             terminal_targets.insert(outcome.target.clone());
             state.process_target_outcome(&outcome);
+            true
         }
-        _ => {}
+        _ => false,
-    }
-}
-fn drain_tui_events(
-    events: &mut ManagedEventSubscription,
-    state: &mut TuiState,
-    terminal_targets: &mut HashSet<TargetInstance>,
-    dropped_events: &mut u64,
-) -> ManagedDrainState {
-    let previous_dropped = *dropped_events;
-    let drained = 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");
-    if *dropped_events != previous_dropped {
-        state.mark_dropped_managed_events(*dropped_events);
-    }
-    drained
-}
-
-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);
-            }
-        }
-    }
-}
-fn handle_input(state: &mut TuiState, shutdown_requested: &AtomicBool) -> io::Result<bool> {
-    let mut force_render = false;
-    for _ in 0..INPUT_EVENT_WORK_BUDGET {
-        if !event::poll(Duration::ZERO)? {
-            break;
-        }
-        let key = match event::read()? {
-            Event::Key(key) => key,
-            Event::Resize(_, _) => {
-                force_render = true;
-                continue;
-            }
-            _ => continue,
-        };
-        if key.kind == KeyEventKind::Release {
-            continue;
-        }
-        match key.code {
-            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;
-                break;
-            }
-            _ => {
-                let (width, height) = terminal::size()?;
-                force_render |= state.handle_key(key.code, Rect::new(0, 0, width, height));
-            }
-        }
-    }
-    Ok(force_render)
-}
-fn render_if_due(
-    terminal: &mut TuiTerminal,
-    state: &TuiState,
-    next_render: &mut Instant,
-    force: bool,
-) -> io::Result<()> {
-    let now = Instant::now();
-    if should_render(now, *next_render, state.paused, force) {
-        terminal.draw(state)?;
-        *next_render = now + RENDER_INTERVAL;
-    }
-    Ok(())
-}
-fn managed_tui_wait_duration(next_render: &Instant, paused: bool) -> Duration {
-    let render_wait = if paused {
-        TUI_WAIT_SLICE
-    } else {
-        next_render.saturating_duration_since(Instant::now())
-    };
-    render_wait.min(TUI_WAIT_SLICE)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use ratatui::{backend::TestBackend, Terminal};
-
-    #[test]
-    fn subscription_lag_keeps_incomplete_statistics_visible_while_paused() {
-        for paused in [false, true] {
-            // The production subscription is a broadcast receiver. Overflow the
-            // real channel rather than synthesizing a lag result or UI state.
-            let (sender, mut events) = tokio::sync::broadcast::channel(1);
-            sender.send(ManagedEvent::Started).unwrap();
-            sender.send(ManagedEvent::Started).unwrap();
-            let mut state = TuiState::default();
-            if paused {
-                state.toggle_pause();
-            }
-            let mut dropped = 0;
-            let mut terminal_targets = HashSet::new();
-            drain_tui_events(&mut events, &mut state, &mut terminal_targets, &mut dropped);
-            assert_eq!(dropped, 1);
-            let mut terminal = Terminal::new(TestBackend::new(200, 40)).unwrap();
-            for _ in 0..2 {
-                // Empty drains must not clear the warning on later redraws.
-                drain_tui_events(&mut events, &mut state, &mut terminal_targets, &mut dropped);
-                terminal
-                    .draw(|frame| super::super::ui::draw_dashboard(frame, &state))
-                    .unwrap();
-                let text = terminal
-                    .backend()
-                    .buffer()
-                    .content()
-                    .iter()
-                    .map(|cell| cell.symbol())
-                    .collect::<String>();
-                assert!(text.contains("incomplete:dropped=1"));
-                assert_eq!(text.contains("display paused"), paused);
-            }
-        }
     }
 }

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

@@ -106,7 +106,6 @@ pub(super) struct TuiState {
     details_open: bool,
     details_scroll: (u16, u16),
     pub(super) paused: bool,
-    pub(super) quit_requested: bool,
 }
 
 impl TuiState {
@@ -143,7 +142,6 @@ impl TuiState {
             details_open: false,
             details_scroll: (0, 0),
             paused: false,
-            quit_requested: false,
         }
     }
 
@@ -939,10 +937,6 @@ pub(super) fn draw_dashboard(frame: &mut Frame<'_>, state: &TuiState) {
     frame.render_widget(status_line(state), rows[3]);
 }
 
-pub(super) fn should_render(now: Instant, next_render: Instant, paused: bool, force: bool) -> bool {
-    force || (!paused && now >= next_render)
-}
-
 fn dashboard_layout(area: Rect, state: &TuiState) -> [Rect; 4] {
     let footer = 3 + u16::from(state.last_warning.is_some());
     let target_rows = state.targets.len().min(6) as u16;
@@ -1564,229 +1558,3 @@ fn format_optional_hex(value: Option<u64>) -> String {
 fn duration_ns(value: Duration) -> i128 {
     i128::try_from(value.as_nanos()).unwrap_or(i128::MAX)
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use irtt_client::{ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, ServerTiming};
-    use ratatui::backend::TestBackend;
-    use std::time::SystemTime;
-
-    fn measure(state: &mut TuiState, label: &str, seq: u32, at: Instant, effective_ns: i128) {
-        let target = TargetInstance {
-            id: label.into(),
-            generation: 1,
-        };
-        let received_at = ClientTimestamp {
-            mono: at,
-            wall: SystemTime::UNIX_EPOCH + Duration::from_secs(1),
-        };
-        let sent_at = ClientTimestamp {
-            mono: at - Duration::from_millis(10),
-            wall: received_at.wall - Duration::from_millis(10),
-        };
-        let remote = "127.0.0.1:2112".parse().unwrap();
-        state.process_target_event(
-            &target,
-            &ClientEvent::EchoSent {
-                seq,
-                remote,
-                scheduled_at: None,
-                sent_at,
-                bytes: 64,
-                send_call: Duration::ZERO,
-                timer_error: None,
-            },
-        );
-        state.process_target_event(
-            &target,
-            &ClientEvent::EchoReply {
-                seq,
-                remote,
-                sent_at,
-                received_at,
-                rtt: RttSample {
-                    raw: Duration::from_millis(10),
-                    adjusted: Some(SignedDuration::from_nanos(effective_ns)),
-                    effective: SignedDuration::from_nanos(effective_ns),
-                },
-                server_timing: Some(ServerTiming {
-                    receive_wall_ns: None,
-                    receive_mono_ns: None,
-                    send_wall_ns: None,
-                    send_mono_ns: None,
-                    midpoint_wall_ns: None,
-                    midpoint_mono_ns: None,
-                    processing: Some(Duration::from_nanos((10_000_000 - effective_ns) as u64)),
-                }),
-                one_way: Some(OneWayDelaySample {
-                    client_to_server: Some(SignedDuration::from_nanos(-3_000_000)),
-                    server_to_client: Some(SignedDuration::from_nanos(effective_ns + 3_000_000)),
-                }),
-                received_stats: None,
-                bytes: 64,
-                packet_meta: PacketMeta::default(),
-            },
-        );
-    }
-
-    fn screen(state: &TuiState, width: u16, height: u16) -> String {
-        let mut terminal = Terminal::new(TestBackend::new(width, height)).unwrap();
-        terminal.draw(|frame| draw_dashboard(frame, state)).unwrap();
-        terminal
-            .backend()
-            .buffer()
-            .content()
-            .chunks(usize::from(width))
-            .map(|row| row.iter().map(|cell| cell.symbol()).collect::<String>())
-            .collect::<Vec<_>>()
-            .join("\n")
-    }
-
-    #[test]
-    fn target_focus_and_details_remain_usable_across_sizes() {
-        let mut state = TuiState::with_target_labels(
-            TuiConfig::default(),
-            (0..8).map(|i| format!("target{i}")),
-        );
-        let now = Instant::now();
-        measure(
-            &mut state,
-            "target7",
-            0,
-            now - Duration::from_secs(5),
-            8_000_000,
-        );
-        measure(&mut state, "target7", 1, now, 5_000_000);
-        for (width, height) in [(56, 18), (80, 24), (120, 40)] {
-            let area = Rect::new(0, 0, width, height);
-            state.handle_key(KeyCode::Home, area);
-            state.handle_key(KeyCode::BackTab, area);
-            let text = screen(&state, width, height);
-            assert!(text.contains(">target7"), "{text}");
-            assert!(
-                text.contains("jitter σ") && text.contains("loss %"),
-                "{text}"
-            );
-            assert!(
-                text.contains(&format_optional_ns_i128(Some(-3_000_000)))
-                    && text.contains(&format_optional_ns_i128(Some(5_000_000))),
-                "{text}"
-            );
-            state.handle_key(KeyCode::Char('d'), area);
-            assert!(screen(&state, width, height).contains("target: target7"));
-            state.handle_key(KeyCode::End, area);
-            assert!(screen(&state, width, height).contains("target7: reply"));
-            state.handle_key(KeyCode::Tab, area);
-            // Selecting another target resets detail scrolling and exposes its own data.
-            assert!(screen(&state, width, height).contains("target: target0"));
-            state.handle_key(KeyCode::Esc, area);
-        }
-        assert!(screen(&state, 40, 10).contains("terminal too small"));
-    }
-
-    #[test]
-    fn history_reset_keeps_latest_summary_and_statistics_while_paused() {
-        let mut state = TuiState::default();
-        let area = Rect::new(0, 0, 100, 40);
-        state.handle_key(KeyCode::Char('p'), area);
-        measure(&mut state, "target", 0, Instant::now(), 8_000_000);
-        state.handle_key(KeyCode::Char('r'), area);
-        let text = screen(&state, 100, 40);
-        assert!(
-            text.contains(&format_optional_ns_i128(Some(8_000_000)))
-                && text.contains("display paused")
-        );
-        assert!(text.contains("waiting for primary replies"));
-        state.handle_key(KeyCode::Char('d'), area);
-        let text = screen(&state, 100, 40);
-        assert!(text.contains("sent 1  received 1  unique 1"));
-        assert!(text.contains(&format!(
-            "effective RTT: {}",
-            format_optional_ns_i128(Some(8_000_000))
-        )));
-        state.handle_key(KeyCode::Esc, area);
-        measure(&mut state, "target", 1, Instant::now(), 5_000_000);
-        assert!(!screen(&state, 100, 40).contains("waiting for primary replies"));
-    }
-
-    #[test]
-    fn graph_controls_preserve_signed_metrics_and_history_range() {
-        let mut state =
-            TuiState::with_target_labels(TuiConfig::default(), ["a".to_owned(), "b".to_owned()]);
-        let area = Rect::new(0, 0, 100, 40);
-        let now = Instant::now();
-        for label in ["a", "b"] {
-            measure(
-                &mut state,
-                label,
-                0,
-                now - Duration::from_secs(120),
-                -2_000_000,
-            );
-            measure(
-                &mut state,
-                label,
-                1,
-                now - Duration::from_secs(60),
-                -2_000_000,
-            );
-            measure(&mut state, label, 2, now, -2_000_000);
-        }
-        state.handle_key(KeyCode::Home, area);
-        let viewport = state
-            .graph_viewport
-            .range(now, state.newest_graph_sample_time());
-        assert_eq!(viewport.start, now - Duration::from_secs(120));
-        assert_eq!(viewport.end, now - Duration::from_secs(60));
-        for metric in [
-            GraphMetric::EffectiveRtt,
-            GraphMetric::RawRtt,
-            GraphMetric::AdjustedRtt,
-            GraphMetric::ClientToServer,
-            GraphMetric::ServerToClient,
-            GraphMetric::ServerProcessing,
-        ] {
-            assert_eq!(state.graph_metric, metric);
-            let series = state
-                .targets
-                .iter()
-                .enumerate()
-                .map(|(i, target)| target_metric_series(target, i, viewport, metric).unwrap())
-                .collect::<Vec<_>>();
-            // Both endpoints belong to the viewport, for both targets.
-            assert!(series.iter().all(|series| series.data.len() == 2));
-            if matches!(
-                metric,
-                GraphMetric::EffectiveRtt | GraphMetric::AdjustedRtt | GraphMetric::ClientToServer
-            ) {
-                assert!(series
-                    .iter()
-                    .all(|series| series.data.iter().all(|(_, y)| *y < 0.0)));
-                assert!(chart_y_bounds(&series, metric.axis_kind()).0 < 0.0);
-            }
-            state.handle_key(KeyCode::Char('m'), area);
-        }
-        state.handle_key(KeyCode::PageDown, area);
-        assert_eq!(state.graph_viewport.range(now, Some(now)).end, now);
-        state.handle_key(KeyCode::Left, area);
-        assert_eq!(
-            state.graph_viewport.range(now, Some(now)).end,
-            now - Duration::from_secs(15)
-        );
-        state.handle_key(KeyCode::Right, area);
-        state.handle_key(KeyCode::PageUp, area);
-        assert_eq!(
-            state.graph_viewport.range(now, Some(now)).end,
-            now - Duration::from_secs(60)
-        );
-        state.handle_key(KeyCode::Char('+'), area);
-        assert_eq!(state.graph_viewport.window, Duration::from_secs(40));
-        state.handle_key(KeyCode::Char('-'), area);
-        assert_eq!(state.graph_viewport.window, DEFAULT_GRAPH_WINDOW);
-        state.handle_key(KeyCode::Char('0'), area);
-        assert!(!state.graph_viewport.range(now, Some(now)).is_live);
-        state.handle_key(KeyCode::End, area);
-        assert!(state.graph_viewport.range(now, Some(now)).is_live);
-    }
-}

crates/irtt-app/src/entry.rs

@@ -1,9 +1,9 @@
-#[cfg(any(feature = "client", feature = "server", feature = "tui"))]
+#[cfg(feature = "server")]
 use std::sync::{atomic::AtomicBool, Arc};
 use std::{env, ffi::OsString, process::ExitCode};
 
 use crate::applet::{dispatch_from_argv, AppletDispatch, RequestedApplet};
-#[cfg(any(feature = "client", feature = "server", feature = "tui"))]
+#[cfg(feature = "server")]
 use crate::signal::install_signal_handler;
 
 /// Entry point for the `irtt-rs` multicall dispatcher binary.
@@ -75,9 +75,9 @@ fn run_dispatcher_from_env() -> Result<(), Box<dyn std::error::Error>> {
     }
 }
 
-/// Installs the shutdown signal handler shared by every applet and hands the
+/// Installs the shutdown signal handler for synchronous applets and hands the
 /// resulting flag to `f`.
-#[cfg(any(feature = "client", feature = "server", feature = "tui"))]
+#[cfg(feature = "server")]
 fn with_shutdown_flag(
     f: impl FnOnce(&AtomicBool) -> Result<(), Box<dyn std::error::Error>>,
 ) -> Result<(), Box<dyn std::error::Error>> {
@@ -91,10 +91,13 @@ fn with_shutdown_flag(
 fn run_client_applet(argv: Vec<OsString>) -> Result<(), Box<dyn std::error::Error>> {
     use clap::Parser;
 
-    with_shutdown_flag(|shutdown_requested| {
+    let shutdown = crate::signal::install_async_signal_handler()
-        let args = crate::cmd::client::ClientArgs::parse_from(argv);
+        .map_err(|err| format!("failed to install signal handler: {err}"))?;
-        crate::cmd::client::run_stream(args, shutdown_requested)
+    let args = crate::cmd::client::ClientArgs::parse_from(argv);
-    })
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?
+        .block_on(crate::cmd::client::run_stream(args, shutdown))
 }
 
 #[cfg(not(feature = "client"))]
@@ -121,10 +124,13 @@ fn run_server_applet(_argv: Vec<OsString>) -> Result<(), Box<dyn std::error::Err
 fn run_tui_applet(argv: Vec<OsString>) -> Result<(), Box<dyn std::error::Error>> {
     use clap::Parser;
 
-    with_shutdown_flag(|shutdown_requested| {
+    let shutdown = crate::signal::install_async_signal_handler()
-        let args = crate::cmd::tui::TuiArgs::parse_from(argv);
+        .map_err(|err| format!("failed to install signal handler: {err}"))?;
-        crate::cmd::tui::run_tui(args, shutdown_requested)
+    let args = crate::cmd::tui::TuiArgs::parse_from(argv);
-    })
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?
+        .block_on(crate::cmd::tui::run_tui(args, shutdown))
 }
 
 #[cfg(not(feature = "tui"))]

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

@@ -3,6 +3,7 @@ pub mod config;
 pub mod prepare;
 pub mod session;
 pub mod targets;
+pub(crate) mod worker;
 
 pub use args::{
     parse_dscp, parse_duration, parse_length, parse_server_fill, parse_test_duration, parse_ttl,
@@ -13,7 +14,6 @@ pub use prepare::{
     prepare_managed_run, ManagedRunSetup, TargetSelection, MANAGED_EVENT_CAPACITY,
     STDIN_MAX_DESIRED_TARGETS, STDIN_MAX_LIVE_TARGET_GENERATIONS, STDIN_OUTCOME_HISTORY_LIMIT,
 };
-pub use session::is_shutdown_requested;
 pub use targets::{
     parse_stdin_target_set, parse_target, prepare_managed_targets, target_specs, GroupPacingArg,
     PreparedTarget, TargetArg, TargetSpec,

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

@@ -1,42 +1,20 @@
-use std::sync::atomic::{AtomicBool, Ordering};
-
-#[cfg(any(feature = "client", feature = "tui"))]
 use irtt_client::managed::{ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError};
 
-#[cfg(any(feature = "client", feature = "tui"))]
+pub(crate) fn drain_final_events<E>(
-pub(crate) const MANAGED_EVENT_WORK_BUDGET: usize = 128;
-
-#[cfg(any(feature = "client", feature = "tui"))]
-#[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(any(feature = "client", feature = "tui"))]
-pub(crate) fn drain_managed_events<E>(
     events: &mut ManagedEventSubscription,
     dropped_events: &mut u64,
     mut process: impl FnMut(ManagedEvent) -> Result<(), E>,
-) -> Result<ManagedDrainState, E> {
+) -> Result<(), E> {
-    for _ in 0..MANAGED_EVENT_WORK_BUDGET {
+    loop {
         match events.try_recv() {
             Ok(event) => process(event)?,
+            Err(ManagedEventTryRecvError::Empty | ManagedEventTryRecvError::Closed) => break,
             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)
+    Ok(())
-}
-
-pub fn is_shutdown_requested(shutdown_requested: &AtomicBool) -> bool {
-    shutdown_requested.load(Ordering::Relaxed)
 }
 
 pub fn should_print_final_summary(continuous: bool, interrupted: bool) -> bool {

crates/irtt-app/src/shared/client/worker.rs

@@ -0,0 +1,67 @@
+use std::{io, thread, time::Duration};
+
+use irtt_client::managed::{ManagedClientHandle, ManagedClientTask, ManagedOutcome};
+use tokio::sync::oneshot;
+
+/// Keeps measurement scheduling separate from frontend output and statistics.
+/// Drop also stops and joins on frontend errors or cancellation.
+pub(crate) struct ManagedWorker {
+    handle: ManagedClientHandle,
+    thread: Option<thread::JoinHandle<()>>,
+    completion: oneshot::Receiver<io::Result<ManagedOutcome>>,
+}
+
+impl ManagedWorker {
+    pub(crate) fn start(task: ManagedClientTask, handle: ManagedClientHandle) -> io::Result<Self> {
+        let (completed, completion) = oneshot::channel();
+        let thread = thread::Builder::new()
+            .name("irtt-managed".to_owned())
+            .spawn(move || {
+                // Construct and destroy the runtime on its owning thread, including
+                // startup failures; never drop a runtime inside the frontend executor.
+                let outcome = (|| {
+                    let runtime = tokio::runtime::Builder::new_current_thread()
+                        .enable_all()
+                        .build()?;
+                    let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                        runtime.block_on(task)
+                    }));
+                    runtime.shutdown_timeout(Duration::from_millis(100));
+                    match outcome {
+                        Ok(outcome) => Ok(outcome),
+                        Err(panic) => std::panic::resume_unwind(panic),
+                    }
+                })();
+                let _ = completed.send(outcome);
+            })?;
+        Ok(Self {
+            handle,
+            thread: Some(thread),
+            completion,
+        })
+    }
+
+    pub(crate) async fn join(mut self) -> Result<ManagedOutcome, Box<dyn std::error::Error>> {
+        let outcome = (&mut self.completion).await;
+        self.join_thread()?;
+        Ok(outcome
+            .map_err(|_| io::Error::other("managed worker completed without an outcome"))??)
+    }
+
+    fn join_thread(&mut self) -> io::Result<()> {
+        self.thread
+            .take()
+            .expect("managed worker is joined once")
+            .join()
+            .map_err(|_| io::Error::other("managed worker thread panicked"))
+    }
+}
+
+impl Drop for ManagedWorker {
+    fn drop(&mut self) {
+        if self.thread.is_some() {
+            drop(self.handle.stop());
+            let _ = self.join_thread();
+        }
+    }
+}

crates/irtt-app/src/signal.rs

@@ -1,10 +1,22 @@
+#[cfg(feature = "server")]
 use std::sync::{
     atomic::{AtomicBool, Ordering},
     Arc,
 };
 
+#[cfg(feature = "server")]
 pub fn install_signal_handler(shutdown_requested: Arc<AtomicBool>) -> Result<(), ctrlc::Error> {
     ctrlc::set_handler(move || {
         shutdown_requested.store(true, Ordering::Relaxed);
     })
 }
+
+/// Preserve ctrlc's SIGINT/SIGTERM policy while waking the async frontends.
+#[cfg(any(feature = "client", feature = "tui"))]
+pub fn install_async_signal_handler() -> Result<tokio::sync::watch::Receiver<bool>, ctrlc::Error> {
+    let (shutdown, receiver) = tokio::sync::watch::channel(false);
+    ctrlc::set_handler(move || {
+        shutdown.send_replace(true);
+    })?;
+    Ok(receiver)
+}

crates/irtt-app/tests/cli.rs

@@ -1,487 +0,0 @@
-#[cfg(feature = "client")]
-#[path = "support/in_tree_server.rs"]
-mod in_tree_server;
-
-#[cfg(feature = "client")]
-use irtt_server::ServerConfig;
-use std::process::Command;
-
-#[cfg(feature = "client")]
-struct ClientProcess(std::process::Child);
-
-#[cfg(feature = "client")]
-impl Drop for ClientProcess {
-    fn drop(&mut self) {
-        let _ = self.0.kill();
-        let _ = self.0.wait();
-    }
-}
-
-#[cfg(feature = "client")]
-impl ClientProcess {
-    fn wait(&mut self) -> std::process::ExitStatus {
-        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
-        loop {
-            if let Some(status) = self.0.try_wait().unwrap() {
-                return status;
-            }
-            assert!(
-                std::time::Instant::now() < deadline,
-                "client did not terminate within five seconds"
-            );
-            std::thread::sleep(std::time::Duration::from_millis(10));
-        }
-    }
-}
-
-// Dispatch/help/availability belong to the application and run in every build.
-#[test]
-fn dispatcher_reports_and_runs_the_enabled_applets() {
-    let dispatcher = env!("CARGO_BIN_EXE_irtt-rs");
-    let help = Command::new(dispatcher).arg("--help").output().unwrap();
-    assert!(help.status.success());
-    let help = String::from_utf8(help.stdout).unwrap();
-    for (applet, enabled) in [
-        ("client", cfg!(feature = "client")),
-        ("server", cfg!(feature = "server")),
-        ("tui", cfg!(feature = "tui")),
-    ] {
-        let row = help
-            .lines()
-            .find(|line| line.trim_start().starts_with(applet))
-            .unwrap();
-        assert_eq!(row.contains("not available"), !enabled);
-        let output = Command::new(dispatcher)
-            .args([applet, "--help"])
-            .output()
-            .unwrap();
-        assert_eq!(output.status.success(), enabled, "{applet}");
-        if !enabled {
-            assert!(String::from_utf8(output.stderr)
-                .unwrap()
-                .contains("not available"));
-        }
-    }
-    let binaries = [
-        dispatcher,
-        #[cfg(feature = "client")]
-        env!("CARGO_BIN_EXE_irtt-client"),
-        #[cfg(feature = "server")]
-        env!("CARGO_BIN_EXE_irtt-server"),
-        #[cfg(feature = "tui")]
-        env!("CARGO_BIN_EXE_irtt-tui"),
-    ];
-    for binary in binaries {
-        assert!(Command::new(binary)
-            .arg("--help")
-            .output()
-            .unwrap()
-            .status
-            .success());
-        assert!(!Command::new(binary)
-            .arg("--invalid-option")
-            .output()
-            .unwrap()
-            .status
-            .success());
-    }
-}
-
-// The default presentation must reach a meaningful packet summary for a finite run.
-#[cfg(feature = "client")]
-#[test]
-fn default_client_output_emits_a_packet_summary() {
-    use std::{io::Read, process::Stdio, thread};
-    let server = in_tree_server::InTreeServer::start(ServerConfig::default());
-    let mut client = ClientProcess(
-        Command::new(env!("CARGO_BIN_EXE_irtt-client"))
-            .env_clear()
-            .args([
-                "--duration",
-                "100ms",
-                "--interval",
-                "20ms",
-                &server.addr.to_string(),
-            ])
-            .stdin(Stdio::null())
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped())
-            .spawn()
-            .unwrap(),
-    );
-    let mut stdout = client.0.stdout.take().unwrap();
-    let stdout = thread::spawn(move || {
-        let mut text = String::new();
-        stdout.read_to_string(&mut text).unwrap();
-        text
-    });
-    let mut stderr = client.0.stderr.take().unwrap();
-    let stderr = thread::spawn(move || {
-        let mut text = String::new();
-        stderr.read_to_string(&mut text).unwrap();
-        text
-    });
-    let status = client.wait();
-    let stdout = stdout.join().unwrap();
-    let stderr = stderr.join().unwrap();
-    assert!(status.success(), "{stderr}");
-    assert!(stdout.contains("irtt-rs summary"), "{stdout}");
-    let packets = stdout
-        .lines()
-        .find(|line| line.starts_with("packets:"))
-        .expect("default output must include the packet summary");
-    for field in ["sent=", "received="] {
-        let count = packets
-            .split_whitespace()
-            .find_map(|word| word.strip_prefix(field))
-            .unwrap()
-            .parse::<u64>()
-            .unwrap();
-        assert!(
-            count > 0,
-            "summary must report {field} a positive count: {packets}"
-        );
-    }
-}
-
-#[cfg(feature = "client")]
-#[test]
-fn target_authentication_inherits_overrides_or_disables_the_global_key() {
-    use clap::Parser;
-    use irtt_app::cmd::client::args::ClientArgs;
-    use irtt_client::{managed::TargetAuth, Authentication, HmacKey};
-    let args = ClientArgs::try_parse_from([
-        "irtt-client",
-        "--hmac",
-        "CONSPICUOUS_GLOBAL_HMAC_KEY",
-        "inherit=127.0.0.1:2112",
-        "override=127.0.0.1:2113@hmac=CONSPICUOUS_TARGET_HMAC_KEY",
-        "disable=127.0.0.1:2114@hmac=",
-    ])
-    .unwrap();
-    let setup = args.prepare().unwrap();
-    assert_eq!(
-        setup.managed_config().client.auth,
-        Authentication::Hmac(HmacKey::new(b"CONSPICUOUS_GLOBAL_HMAC_KEY".as_slice()))
-    );
-    let targets = setup.managed_targets();
-    let authentication: Vec<_> = targets.iter().map(|target| target.auth.clone()).collect();
-    assert_eq!(
-        authentication,
-        [
-            TargetAuth::Inherit,
-            TargetAuth::Override(Authentication::Hmac(HmacKey::new(
-                b"CONSPICUOUS_TARGET_HMAC_KEY".as_slice()
-            ))),
-            TargetAuth::Override(Authentication::Unauthenticated)
-        ]
-    );
-    for output in [format!("{args:?}"), format!("{setup:?}")] {
-        assert!(!output.contains("CONSPICUOUS_GLOBAL_HMAC_KEY"));
-        assert!(!output.contains("CONSPICUOUS_TARGET_HMAC_KEY"));
-    }
-}
-
-// A real authenticated server detects broken client argument-to-config mapping.
-#[cfg(feature = "client")]
-#[test]
-fn client_arguments_export_a_real_client_run() {
-    use std::{io::Read, process::Stdio, thread};
-    let dispatcher = env!("CARGO_BIN_EXE_irtt-rs");
-    for option in ["--dscp", "--ttl"] {
-        assert!(!Command::new(dispatcher)
-            .args(["client", option, "256", "127.0.0.1:9"])
-            .output()
-            .unwrap()
-            .status
-            .success());
-    }
-    let server =
-        in_tree_server::InTreeServer::start(ServerConfig::default().with_hmac_key(b"cli-test"));
-    let mut client = ClientProcess(
-        Command::new(dispatcher)
-            .args([
-                "client",
-                "--duration",
-                "100ms",
-                "--interval",
-                "20ms",
-                "--hmac",
-                "cli-test",
-                "--length",
-                "128",
-                "--sfill",
-                "pattern:ab",
-                "--clock",
-                "wall",
-                "--tstamp",
-                "both",
-                "--stats",
-                "both",
-                "--ttl",
-                "64",
-                "--format",
-                "jsonl",
-                "--columns",
-                "target,event,bytes",
-                &format!("edge={}", server.addr),
-            ])
-            .stdin(Stdio::null())
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped())
-            .spawn()
-            .unwrap(),
-    );
-    // Drain both pipes while waiting so output cannot block client termination.
-    let mut stdout = client.0.stdout.take().unwrap();
-    let stdout = thread::spawn(move || {
-        let mut bytes = Vec::new();
-        stdout.read_to_end(&mut bytes).unwrap();
-        bytes
-    });
-    let mut stderr = client.0.stderr.take().unwrap();
-    let stderr = thread::spawn(move || {
-        let mut bytes = Vec::new();
-        stderr.read_to_end(&mut bytes).unwrap();
-        bytes
-    });
-    let status = client.wait();
-    let stdout = stdout.join().unwrap();
-    let stderr = stderr.join().unwrap();
-    assert!(status.success(), "{}", String::from_utf8_lossy(&stderr));
-    let stdout = String::from_utf8(stdout).unwrap();
-    let rows: Vec<_> = stdout.lines().collect();
-    assert!(!rows.is_empty());
-    assert!(rows.iter().all(|row| row.starts_with('{')
-        && row.ends_with('}')
-        && row.contains("\"target\":\"edge\"")));
-    assert!(rows
-        .iter()
-        .any(|row| row.contains("\"event\":\"echo_reply\"") && row.contains("\"bytes\":128")));
-    assert!(rows
-        .iter()
-        .any(|row| row.contains("\"event\":\"session_closed\"")));
-}
-
-#[cfg(any(feature = "client", feature = "tui"))]
-#[test]
-fn malformed_target_diagnostics_do_not_expose_authentication_keys() {
-    let secret = "malformed-target-secret-92817";
-    for applet in [
-        #[cfg(feature = "client")]
-        "client",
-        #[cfg(feature = "tui")]
-        "tui",
-    ] {
-        let output = Command::new(env!("CARGO_BIN_EXE_irtt-rs"))
-            .args([applet, &format!(r"edge=127.0.0.1:9@hmac={secret}\q")])
-            .output()
-            .unwrap();
-        assert!(!output.status.success());
-        let stderr = String::from_utf8(output.stderr).unwrap();
-        assert!(
-            !stderr.trim().is_empty(),
-            "malformed target must report a diagnostic"
-        );
-        assert!(!stderr.contains(secret), "target authentication key leaked");
-        assert!(!String::from_utf8_lossy(&output.stdout).contains(secret));
-    }
-}
-
-#[cfg(feature = "server")]
-#[test]
-fn server_arguments_resolve_binds_and_map_policy() {
-    use clap::{CommandFactory, FromArgMatches, Parser};
-    use irtt_app::cmd::server::ServerArgs;
-    use irtt_server::TimestampAllowance;
-    use std::time::Duration;
-    // Isolate zero-argument defaults from the test runner's environment without
-    // mutating process-global environment in this parallel test process.
-    let defaults = ServerArgs::from_arg_matches(
-        &ServerArgs::command()
-            .mut_args(|arg| arg.env(None::<&str>))
-            .try_get_matches_from(["irtt-server"])
-            .unwrap(),
-    )
-    .unwrap();
-    assert_eq!(
-        defaults.resolve_binds(),
-        vec![
-            "[::]:2112".parse().unwrap(),
-            "0.0.0.0:2112".parse().unwrap()
-        ]
-    );
-    assert_eq!(
-        defaults.server_config(),
-        irtt_server::ServerConfig::default()
-    );
-    let args = ServerArgs::try_parse_from([
-        "irtt-server",
-        "--bind",
-        "127.0.0.1:2113",
-        "--bind",
-        "[::1]:2114",
-        "--hmac",
-        "server-cli-test",
-        "--max-sessions",
-        "3",
-        "--max-packet-length",
-        "96",
-        "--min-interval",
-        "25ms",
-        "--burst",
-        "2",
-        "--idle-timeout",
-        "8s",
-        "--max-duration",
-        "1m",
-        "--timestamp-allowance",
-        "single",
-        "--no-dscp",
-    ])
-    .unwrap();
-    assert_eq!(
-        args.resolve_binds(),
-        vec![
-            "127.0.0.1:2113".parse().unwrap(),
-            "[::1]:2114".parse().unwrap()
-        ]
-    );
-    let config = args.server_config();
-    assert_eq!(config.hmac_key(), Some(b"server-cli-test".as_slice()));
-    assert_eq!(config.max_sessions(), 3);
-    assert_eq!(config.max_packet_length(), 96);
-    assert_eq!(config.min_send_interval(), Duration::from_millis(25));
-    assert_eq!(config.burst_allowance(), 2);
-    assert_eq!(config.idle_timeout(), Duration::from_secs(8));
-    assert_eq!(config.max_test_duration(), Some(Duration::from_secs(60)));
-    assert_eq!(config.timestamp_allowance(), TimestampAllowance::Single);
-    assert!(!config.dscp_allowed());
-}
-
-#[cfg(feature = "server")]
-#[test]
-fn server_command_line_binds_replace_the_environment_bind() {
-    use std::net::UdpSocket;
-    // Occupied ports give bounded, observable startup failures, without starting
-    // a daemon or needing a process/signal harness merely to inspect arguments.
-    let environment = UdpSocket::bind("127.0.0.1:0").unwrap();
-    let explicit = UdpSocket::bind("127.0.0.1:0").unwrap();
-    let environment_addr = environment.local_addr().unwrap().to_string();
-    let explicit_addr = explicit.local_addr().unwrap().to_string();
-    for use_explicit in [false, true] {
-        let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-server"));
-        command
-            .env_clear()
-            .env("IRTT_SERVER_BIND", &environment_addr)
-            .env("IRTT_SERVER_MAX_SESSIONS", "invalid")
-            .args(["--max-sessions", "3"]);
-        if use_explicit {
-            command.args(["--bind", "127.0.0.1:0", "--bind", &explicit_addr]);
-        }
-        let output = command.output().unwrap();
-        assert!(!output.status.success());
-        let stderr = String::from_utf8(output.stderr).unwrap();
-        let expected = if use_explicit {
-            &explicit_addr
-        } else {
-            &environment_addr
-        };
-        assert!(
-            stderr.contains(expected),
-            "startup must attempt the selected bind: {stderr}"
-        );
-        assert!(
-            !stderr.contains("listening on"),
-            "partial startup must not announce listeners"
-        );
-    }
-}
-
-// Exercise stdin replacement and EOF through the actual application process.
-#[cfg(feature = "client")]
-#[test]
-fn targets_stdin_supersedes_the_active_set_and_stops_on_eof() {
-    use std::{
-        io::{BufRead, BufReader, Read, Write},
-        process::Stdio,
-        sync::mpsc,
-        thread,
-        time::{Duration, Instant},
-    };
-    let server = in_tree_server::InTreeServer::start(ServerConfig::default());
-    let mut client = ClientProcess(
-        Command::new(env!("CARGO_BIN_EXE_irtt-client"))
-            .env_clear()
-            .args([
-                "--duration",
-                "0",
-                "--targets-stdin",
-                "--interval",
-                "20ms",
-                "--format",
-                "jsonl",
-                "--columns",
-                "target,event",
-            ])
-            .stdin(Stdio::piped())
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped())
-            .spawn()
-            .unwrap(),
-    );
-    let stdout = client.0.stdout.take().unwrap();
-    let (tx, rx) = mpsc::channel();
-    let reader = thread::spawn(move || {
-        for line in BufReader::new(stdout).lines() {
-            if tx.send(line.unwrap()).is_err() {
-                break;
-            }
-        }
-    });
-    let mut stdin = client.0.stdin.take().unwrap();
-    let mut rows = Vec::new();
-    for target in ["first", "second"] {
-        writeln!(stdin, "{target}={}", server.addr).unwrap();
-        stdin.flush().unwrap();
-        let deadline = Instant::now() + Duration::from_secs(5);
-        loop {
-            let row = rx
-                .recv_timeout(deadline.saturating_duration_since(Instant::now()))
-                .expect("stdin target set did not produce a reply");
-            let replied = row.contains(&format!("\"target\":\"{target}\""))
-                && row.contains("\"event\":\"echo_reply\"");
-            rows.push(row);
-            if replied {
-                break;
-            }
-        }
-    }
-    // Retirement can finish after the newer target's first reply.
-    let deadline = Instant::now() + Duration::from_secs(5);
-    while !rows.iter().any(|row| {
-        row.contains("\"target\":\"first\"") && row.contains("\"event\":\"session_closed\"")
-    }) {
-        rows.push(
-            rx.recv_timeout(deadline.saturating_duration_since(Instant::now()))
-                .expect("superseded stdin target did not retire"),
-        );
-    }
-    drop(stdin);
-    let status = client.wait();
-    reader.join().unwrap();
-    rows.extend(rx.try_iter());
-    let mut stderr = String::new();
-    client
-        .0
-        .stderr
-        .take()
-        .unwrap()
-        .read_to_string(&mut stderr)
-        .unwrap();
-    assert!(status.success(), "{stderr}");
-    assert!(rows
-        .iter()
-        .any(|row| row.contains("\"target\":\"second\"")
-            && row.contains("\"event\":\"session_closed\"")));
-}

crates/irtt-app/tests/support/in_tree_server.rs

@@ -1,74 +0,0 @@
-use std::{
-    net::{IpAddr, Ipv4Addr, SocketAddr},
-    sync::mpsc,
-    thread::{self, JoinHandle},
-    time::Duration,
-};
-
-use irtt_server::{Server, ServerConfig};
-use tokio::sync::oneshot;
-
-const STARTUP_TIMEOUT: Duration = Duration::from_secs(2);
-
-pub struct InTreeServer {
-    pub addr: SocketAddr,
-    shutdown: Option<oneshot::Sender<()>>,
-    done: Option<JoinHandle<Result<(), irtt_server::ServerRuntimeError>>>,
-}
-
-impl InTreeServer {
-    pub fn start(config: ServerConfig) -> Self {
-        let (startup_tx, startup_rx) = mpsc::sync_channel(1);
-        let (shutdown_tx, shutdown_rx) = oneshot::channel();
-        let done = thread::spawn(move || {
-            let runtime = tokio::runtime::Builder::new_current_thread()
-                .enable_all()
-                .build()
-                .expect("failed to build in-tree test server runtime");
-            runtime.block_on(async move {
-                let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0);
-                let mut server = Server::bind(bind_addr, config).await?;
-                startup_tx
-                    .send(server.local_addr()?)
-                    .expect("test stopped waiting for in-tree server startup");
-                server
-                    .run(async {
-                        let _ = shutdown_rx.await;
-                    })
-                    .await
-            })
-        });
-
-        let addr = match startup_rx.recv_timeout(STARTUP_TIMEOUT) {
-            Ok(addr) => addr,
-            Err(error) => {
-                let _ = shutdown_tx.send(());
-                let result = done.join().expect("in-tree test server thread panicked");
-                panic!("in-tree test server failed before startup ({error}): {result:?}");
-            }
-        };
-
-        Self {
-            addr,
-            shutdown: Some(shutdown_tx),
-            done: Some(done),
-        }
-    }
-
-    fn stop_inner(&mut self) {
-        if let Some(shutdown) = self.shutdown.take() {
-            let _ = shutdown.send(());
-        }
-        if let Some(done) = self.done.take() {
-            done.join()
-                .expect("in-tree test server thread panicked")
-                .expect("in-tree test server stopped with an error");
-        }
-    }
-}
-
-impl Drop for InTreeServer {
-    fn drop(&mut self) {
-        self.stop_inner();
-    }
-}