feat(app): add stdin-managed client target sets

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

@@ -20,6 +20,13 @@ pub struct ClientArgs {
     )]
     pub targets: Vec<TargetArg>,
 
+    /// Read complete desired target sets from standard input in continuous mode.
+    #[arg(
+        long,
+        long_help = "Read one complete desired target set per stdin line in continuous mode. Commas separate targets and backslash escapes commas within a target. [] selects an empty desired set. EOF stops the client gracefully."
+    )]
+    pub targets_stdin: bool,
+
     /// Managed group pacing for multi-target mode.
     #[arg(
         long,
@@ -86,12 +93,16 @@ impl ClientArgs {
     /// Targets are only required here. `--list-columns` returns before this is
     /// called, so listing columns still needs no target.
     pub fn prepare(&self) -> Result<ManagedRunSetup, String> {
+        if self.targets_stdin && !self.is_continuous() {
+            return Err("--targets-stdin requires --duration 0".to_owned());
+        }
         prepare_managed_run(
             &self.common,
             self.duration,
             TargetSelection {
                 targets: &self.targets,
                 pacing: self.pacing,
+                stdin_controlled: self.targets_stdin,
             },
         )
     }
@@ -166,6 +177,7 @@ mod tests {
         assert_eq!(args.format, OutputFormat::Table);
         assert_eq!(args.columns, None);
         assert_eq!(args.header, HeaderMode::Auto);
+        assert!(!args.targets_stdin);
         assert!(!args.is_continuous());
         assert_eq!(args.timestamp_mode(), TimestampArg::Both);
 
@@ -308,6 +320,21 @@ mod tests {
         assert_eq!(config.hmac_key, Some(b"secret".to_vec()));
     }
 
+    #[test]
+    fn targets_stdin_requires_continuous_mode_and_allows_empty_initial_set() {
+        assert!(parse(&["--targets-stdin"]).unwrap().prepare().is_err());
+        let args = parse(&["--duration", "0", "--targets-stdin"]).unwrap();
+        let setup = args.prepare().unwrap();
+        assert!(setup.targets.is_empty());
+        assert!(setup.stdin_controlled);
+        let config = setup.managed_config();
+        assert_eq!(
+            config.completion,
+            irtt_client::managed::ManagedCompletionPolicy::ExplicitStop
+        );
+        assert_eq!(config.max_live_target_generations, 256);
+    }
+
     #[test]
     fn output_format_summary_policy_is_table_only() {
         assert!(OutputFormat::Table.prints_summary());

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

@@ -1,16 +1,16 @@
-use std::collections::BTreeMap;
 use std::{
-    collections::HashSet,
-    io::{self, Write},
-    sync::atomic::AtomicBool,
+    collections::{BTreeMap, BTreeSet, HashSet, VecDeque},
+    io::{self, BufRead, Write},
+    sync::{atomic::AtomicBool, Arc, Mutex},
     thread,
     time::Duration,
 };
 
 use irtt_client::{
     managed::{
-        BlockingManagedClient, ManagedEndReason, ManagedEvent, ManagedEventSubscription,
-        ManagedEventTryRecvError, ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance,
+        BlockingManagedClient, ManagedClientHandle, ManagedCommandApplyError, ManagedEndReason,
+        ManagedEvent, ManagedEventSubscription, ManagedEventTryRecvError, ManagedStatus,
+        ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetOutcome, TargetInstance,
     },
     ClientEvent,
 };
@@ -20,13 +20,13 @@ use super::{
     output::{EventRenderStats, OutputConfig},
 };
 
-use crate::shared::client::expected_probe_count;
 use crate::shared::client::{
-    is_shutdown_requested,
+    expected_probe_count, is_shutdown_requested, parse_stdin_target_set,
     session::{
         drain_managed_events, peer_close_run_error, request_managed_stop_for_peer_close,
         request_managed_stop_once, should_print_final_summary, ManagedDrainState,
     },
+    ManagedRunSetup, STDIN_MAX_DESIRED_TARGETS, STDIN_OUTCOME_HISTORY_LIMIT,
 };
 
 use irtt_stats::{StatsCollector, StatsConfig};
@@ -41,6 +41,7 @@ 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(
     args: ClientArgs,
@@ -61,6 +62,9 @@ pub fn run_stream(
         args.verbose,
     )
     .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?;
+    if setup.stdin_controlled {
+        return run_stdin_stream(setup, output_config, shutdown_requested);
+    }
     let continuous = args.is_continuous();
     let target_count = setup.target_count();
 
@@ -196,6 +200,561 @@ pub fn run_stream(
     Ok(())
 }
 
+#[derive(Clone)]
+struct StdinTargetSet {
+    revision: u64,
+    targets: Vec<ManagedTargetConfig>,
+}
+
+#[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>(
+    reader: &mut R,
+    buffer: &mut Vec<u8>,
+) -> io::Result<Option<String>> {
+    buffer.clear();
+    loop {
+        let available = reader.fill_buf()?;
+        if available.is_empty() {
+            if buffer.is_empty() {
+                return Ok(None);
+            }
+            if buffer.len() > MAX_STDIN_RECORD_BYTES {
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidData,
+                    "stdin target record exceeds the maximum size",
+                ));
+            }
+            return std::str::from_utf8(buffer)
+                .map(str::to_owned)
+                .map(Some)
+                .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "stdin is not UTF-8"));
+        }
+
+        if buffer.len() == MAX_STDIN_RECORD_BYTES + 1 {
+            if available.first() == Some(&b'\n') && buffer.last() == Some(&b'\r') {
+                reader.consume(1);
+                buffer.pop();
+                return std::str::from_utf8(buffer)
+                    .map(str::to_owned)
+                    .map(Some)
+                    .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "stdin is not UTF-8"));
+            }
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "stdin target record exceeds the maximum size",
+            ));
+        }
+
+        if let Some(newline) = available.iter().position(|byte| *byte == b'\n') {
+            let before_newline = &available[..newline];
+            if buffer.len().saturating_add(before_newline.len()) > MAX_STDIN_RECORD_BYTES + 1 {
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidData,
+                    "stdin target record exceeds the maximum size",
+                ));
+            }
+            buffer.extend_from_slice(before_newline);
+            reader.consume(newline + 1);
+            if buffer.last() == Some(&b'\r') {
+                buffer.pop();
+            }
+            if buffer.len() > MAX_STDIN_RECORD_BYTES {
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidData,
+                    "stdin target record exceeds the maximum size",
+                ));
+            }
+            return std::str::from_utf8(buffer)
+                .map(str::to_owned)
+                .map(Some)
+                .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "stdin is not UTF-8"));
+        }
+
+        let remaining = MAX_STDIN_RECORD_BYTES + 1 - buffer.len();
+        let take = available.len().min(remaining);
+        buffer.extend_from_slice(&available[..take]);
+        reader.consume(take);
+        if buffer.len() > MAX_STDIN_RECORD_BYTES && buffer.last() != Some(&b'\r') {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "stdin target record exceeds the maximum size",
+            ));
+        }
+    }
+}
+
+fn read_stdin_target_sets<R: BufRead>(reader: &mut R, mailbox: &StdinMailbox) {
+    let mut line = 0_u64;
+    let mut record = Vec::with_capacity(MAX_STDIN_RECORD_BYTES + 1);
+    loop {
+        match read_stdin_record(reader, &mut record) {
+            Ok(Some(record)) => {
+                line = line.saturating_add(1);
+                if record.is_empty() {
+                    continue;
+                }
+                match parse_stdin_target_set(&record, STDIN_MAX_DESIRED_TARGETS) {
+                    Ok(targets) => {
+                        mailbox.publish(targets.into_iter().map(|target| target.managed).collect())
+                    }
+                    Err(error) => {
+                        mailbox.fail(format!("invalid --targets-stdin line {line}: {error}"));
+                        return;
+                    }
+                }
+            }
+            Ok(None) => {
+                mailbox.finish();
+                return;
+            }
+            Err(_) => {
+                mailbox.fail(format!("failed to read --targets-stdin line {}", line + 1));
+                return;
+            }
+        }
+    }
+}
+
+fn spawn_stdin_target_reader(mailbox: Arc<StdinMailbox>) -> 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);
+        })
+        .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>,
+    members: HashSet<TargetInstance>,
+}
+
+impl BoundedTargetSet {
+    fn new(limit: usize) -> Self {
+        Self {
+            limit,
+            order: VecDeque::with_capacity(limit),
+            members: HashSet::with_capacity(limit),
+        }
+    }
+
+    fn insert(&mut self, target: TargetInstance) -> bool {
+        if !self.members.insert(target.clone()) {
+            return false;
+        }
+        if self.order.len() == self.limit {
+            let evicted = self
+                .order
+                .pop_front()
+                .expect("bounded target order is nonempty");
+            self.members.remove(&evicted);
+        }
+        self.order.push_back(target);
+        true
+    }
+}
+
+fn process_stdin_event<W: Write>(
+    event: ManagedEvent,
+    stream_output: &mut StreamOutput<'_, W>,
+    stats: &mut BTreeMap<TargetInstance, StatsCollector>,
+    terminal_targets: &mut BoundedTargetSet,
+) -> io::Result<()> {
+    match event {
+        ManagedEvent::TargetStateChanged { target, .. } => {
+            stats
+                .entry(target)
+                .or_insert_with(|| StatsCollector::new(stats_config(true)));
+        }
+        ManagedEvent::Client { target, event } => {
+            let collector = stats
+                .entry(target.clone())
+                .or_insert_with(|| StatsCollector::new(stats_config(true)));
+            print_events_with_stats(
+                stream_output,
+                std::slice::from_ref(&event),
+                Some(target.id.as_str()),
+                collector,
+            )?;
+        }
+        ManagedEvent::TargetFinished { outcome }
+            if terminal_targets.insert(outcome.target.clone()) =>
+        {
+            report_target_failure(&outcome);
+        }
+        _ => {}
+    }
+    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,
+    final_summary_targets: Option<&BTreeSet<TargetInstance>>,
+) {
+    let live = status
+        .targets
+        .iter()
+        .map(|target| target.target.clone())
+        .collect::<HashSet<_>>();
+    stats.retain(|target, _| {
+        live.contains(target)
+            || final_summary_targets.is_some_and(|targets| targets.contains(target))
+    });
+}
+
+fn snapshot_stdin_summary_targets(
+    stats: &mut BTreeMap<TargetInstance, StatsCollector>,
+    status: &ManagedStatus,
+) -> BTreeSet<TargetInstance> {
+    let targets = status
+        .targets
+        .iter()
+        .map(|target| target.target.clone())
+        .collect::<BTreeSet<_>>();
+    for target in &targets {
+        stats
+            .entry(target.clone())
+            .or_insert_with(|| StatsCollector::new(stats_config(true)));
+    }
+    targets
+}
+
+#[derive(Clone)]
+enum StdinStop {
+    Interrupted,
+    Eof,
+    Fatal(String),
+}
+
+fn stdin_stop_request(
+    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,
+) -> Result<(), Box<dyn std::error::Error>> {
+    if is_shutdown_requested(shutdown_requested) {
+        return Ok(());
+    }
+    let (owner, mut events) = BlockingManagedClient::start_with_subscription(
+        setup.managed_config(),
+        setup.managed_targets(),
+    )?;
+    let handle = owner.handle();
+    let mailbox = Arc::new(StdinMailbox::default());
+    spawn_stdin_target_reader(Arc::clone(&mailbox))?;
+
+    let mut stdout = io::LineWriter::new(io::stdout().lock());
+    let mut stream_output = StreamOutput {
+        config: output_config,
+        header_printed: false,
+        print_final_summary: false,
+        show_running_only_summary_note: false,
+        out: &mut stdout,
+    };
+    let mut stats = BTreeMap::new();
+    let mut terminal_targets = BoundedTargetSet::new(STDIN_OUTCOME_HISTORY_LIMIT);
+    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();
+        if stop.is_none() {
+            if let Some(requested_stop) = stdin_stop_request(shutdown_requested, &mailbox) {
+                summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status));
+                stop = Some(requested_stop);
+                if request_managed_stop_once(&mut stop_requested) {
+                    drop(handle.stop());
+                }
+            } else if let Err(error) =
+                apply_latest_stdin_target_set(&handle, &mailbox, &mut retry_after)
+            {
+                summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status));
+                stop = Some(StdinStop::Fatal(error));
+                if request_managed_stop_once(&mut stop_requested) {
+                    drop(handle.stop());
+                }
+            }
+        }
+
+        let drain_state = drain_stdin_events(
+            &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;
+        }
+    }
+
+    let outcome = owner.join()?;
+    drain_final_stdin_events(
+        &mut events,
+        &mut stream_output,
+        &mut stats,
+        &mut terminal_targets,
+        &mut dropped_events,
+    )?;
+    if let Some(warning) = dropped_event_warning(dropped_events) {
+        eprintln!("{warning}");
+    }
+    for target in outcome.recent_target_outcomes.iter() {
+        if terminal_targets.insert(target.target.clone()) {
+            report_target_failure(target);
+        }
+    }
+    if outcome.discarded_target_outcomes != 0 {
+        eprintln!(
+            "irtt-rs: warning: {} final target outcomes were discarded",
+            outcome.discarded_target_outcomes
+        );
+    }
+
+    let print_summary = matches!(stop, Some(StdinStop::Interrupted | StdinStop::Eof));
+    stream_output.print_final_summary = print_summary;
+    stream_output.show_running_only_summary_note = print_summary;
+    if let Some(summary_targets) = summary_targets {
+        let multi_target = summary_targets.len() > 1;
+        for target in summary_targets {
+            let stats = stats
+                .get(&target)
+                .expect("summary snapshot retains every selected target collector");
+            if multi_target && stream_output.config.prints_summary() {
+                writeln!(stream_output.out)?;
+                writeln!(
+                    stream_output.out,
+                    "target: {} (generation {})",
+                    target.id, target.generation
+                )?;
+            }
+            stream_output.print_summary(stats)?;
+        }
+    }
+    stream_output.out.flush()?;
+
+    match stop {
+        Some(StdinStop::Fatal(error)) => Err(error.into()),
+        _ => match outcome.end_reason {
+            ManagedEndReason::DriverFailed(failure) => {
+                Err(format!("managed driver failed: {failure}").into())
+            }
+            _ => Ok(()),
+        },
+    }
+}
+
 fn process_event<W: Write>(
     event: ManagedEvent,
     stream_output: &mut StreamOutput<'_, W>,

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

@@ -58,6 +58,7 @@ impl TuiArgs {
             TargetSelection {
                 targets: &self.targets,
                 pacing: self.pacing,
+                stdin_controlled: false,
             },
         )
     }