diff --git a/crates/irtt-app/src/cmd/client/args.rs b/crates/irtt-app/src/cmd/client/args.rs index f2e542a..ba3294c 100644 --- a/crates/irtt-app/src/cmd/client/args.rs +++ b/crates/irtt-app/src/cmd/client/args.rs @@ -11,15 +11,22 @@ pub const DEFAULT_CLIENT_DURATION: Duration = Duration::from_secs(10); #[derive(Debug, Clone, Parser)] #[command(name = "irtt-client", about = "Minimal IRTT-compatible stream client")] pub struct ClientArgs { - /// Server address/host, optionally prefixed with LABEL=. Repeat for multi-target mode. + /// Server address/host, optionally prefixed with LABEL= and suffixed with @hmac=KEY. Repeat for multi-target mode. #[arg( value_name = "TARGET", num_args = 0.., value_parser = parse_target, - long_help = "Server address/host, optionally prefixed with LABEL=. Repeat for multi-target mode.\n\nExamples:\n irtt-client host-a:2112 host-b:2112\n irtt-client eu=host-a:2112 us=host-b:2112\n irtt-client host-a:2112 eu=host-b:2112" + long_help = "Server address/host, optionally prefixed with LABEL= and suffixed with @hmac=KEY. Repeat for multi-target mode.\n\nExamples:\n irtt-client host-a:2112 host-b:2112\n irtt-client eu=host-a:2112 us=host-b:2112\n irtt-client ams=host-a:2112@hmac=secret public=host-b:2112@hmac=\n\nA target without @hmac= inherits --hmac. @hmac= explicitly disables it for that target." )] pub targets: Vec, + /// 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. Records are limited to 64 KiB. EOF stops the client gracefully and may discard a revision that has not yet been acknowledged as applied." + )] + pub targets_stdin: bool, + /// Managed group pacing for multi-target mode. #[arg( long, @@ -34,7 +41,7 @@ pub struct ClientArgs { default_value = "10s", value_parser = parse_test_duration, help = "Test duration; use 0 for continuous mode", - long_help = "Test duration; use 0 for continuous mode.\n\nFinite runs retain exact statistics for final summaries. Continuous mode uses bounded-memory running statistics and prints a final summary only when interrupted." + long_help = "Test duration; use 0 for continuous mode.\n\nFinite runs retain exact statistics for final summaries. Continuous mode uses bounded-memory running statistics and prints a final summary when interrupted; --targets-stdin also prints the final retained summary on EOF." )] pub duration: Duration, @@ -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 { + 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, }, ) } @@ -159,14 +170,14 @@ mod tests { let args = parse(&["127.0.0.1:2112"]).unwrap(); assert_eq!(args.targets.len(), 1); - assert_eq!(args.targets[0].label, None); - assert_eq!(args.targets[0].addr, "127.0.0.1:2112"); + assert_eq!(args.prepare().unwrap().targets[0].label, "127.0.0.1:2112"); assert_eq!(args.pacing, GroupPacingArg::Staggered); assert_eq!(args.duration, DEFAULT_CLIENT_DURATION); assert_eq!(args.interval, Duration::from_secs(1)); 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); @@ -254,8 +265,8 @@ mod tests { #[test] fn invalid_labelled_target_syntax_is_rejected() { - assert!(parse(&["=127.0.0.1:2112"]).is_err()); - assert!(parse(&["label="]).is_err()); + assert!(parse(&["=127.0.0.1:2112"]).unwrap().prepare().is_err()); + assert!(parse(&["label="]).unwrap().prepare().is_err()); } #[test] @@ -309,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()); diff --git a/crates/irtt-app/src/cmd/client/run.rs b/crates/irtt-app/src/cmd/client/run.rs index 7fa385d..c6780e2 100644 --- a/crates/irtt-app/src/cmd/client/run.rs +++ b/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, +} + +#[derive(Default)] +struct StdinMailboxState { + next_revision: u64, + latest: Option, + eof: bool, + fatal: Option, +} + +#[derive(Default)] +struct StdinMailbox { + state: Mutex, +} + +impl StdinMailbox { + fn publish(&self, targets: Vec) { + 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 { + 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 { + 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( + reader: &mut R, + buffer: &mut Vec, +) -> io::Result> { + 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(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) -> 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, +} + +fn retry_waits_for_status_change( + retry_after: &mut Option, + target_set: &StdinTargetSet, + status: &Arc, +) -> 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, +) -> Option { + 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, +) -> 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, + members: HashSet, +} + +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( + event: ManagedEvent, + stream_output: &mut StreamOutput<'_, W>, + stats: &mut BTreeMap, + 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( + events: &mut ManagedEventSubscription, + stream_output: &mut StreamOutput<'_, W>, + stats: &mut BTreeMap, + terminal_targets: &mut BoundedTargetSet, + dropped_events: &mut u64, +) -> io::Result { + drain_managed_events(events, dropped_events, |event| { + process_stdin_event(event, stream_output, stats, terminal_targets) + }) +} + +fn drain_final_stdin_events( + events: &mut ManagedEventSubscription, + stream_output: &mut StreamOutput<'_, W>, + stats: &mut BTreeMap, + 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, + status: &ManagedStatus, + final_summary_targets: Option<&BTreeSet>, +) { + let live = status + .targets + .iter() + .map(|target| target.target.clone()) + .collect::>(); + stats.retain(|target, _| { + live.contains(target) + || final_summary_targets.is_some_and(|targets| targets.contains(target)) + }); +} + +fn snapshot_stdin_summary_targets( + stats: &mut BTreeMap, + status: &ManagedStatus, +) -> BTreeSet { + let targets = status + .targets + .iter() + .map(|target| target.target.clone()) + .collect::>(); + 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 { + 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> { + 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( event: ManagedEvent, stream_output: &mut StreamOutput<'_, W>, @@ -384,9 +943,218 @@ fn finite_stats_memory_warning(args: &ClientArgs, target_count: usize) -> Option mod tests { use super::*; use irtt_client::managed::{ - ManagedTargetFailure, ManagedTargetFailureKind, ManagedTargetFailurePhase, TargetId, + ManagedLifecycle, ManagedTargetFailure, ManagedTargetFailureKind, + ManagedTargetFailurePhase, ManagedTargetLifecycle, ManagedTargetStatus, TargetId, }; - use std::sync::Arc; + use std::{io::BufReader, io::Cursor, sync::Arc}; + + #[test] + fn stdin_record_reader_removes_only_line_framing() { + let mut reader = BufReader::with_capacity(2, Cursor::new(b" first \r\n\nthird\r")); + let mut buffer = Vec::with_capacity(MAX_STDIN_RECORD_BYTES + 1); + + assert_eq!( + read_stdin_record(&mut reader, &mut buffer).unwrap(), + Some(" first ".to_owned()) + ); + assert_eq!( + read_stdin_record(&mut reader, &mut buffer).unwrap(), + Some(String::new()) + ); + assert_eq!( + read_stdin_record(&mut reader, &mut buffer).unwrap(), + Some("third\r".to_owned()) + ); + assert_eq!(read_stdin_record(&mut reader, &mut buffer).unwrap(), None); + } + + #[test] + fn stdin_record_reader_rejects_over_limit_unterminated_record() { + let input = vec![b'x'; MAX_STDIN_RECORD_BYTES + 1]; + let mut reader = BufReader::with_capacity(17, Cursor::new(input)); + let mut buffer = Vec::with_capacity(MAX_STDIN_RECORD_BYTES + 1); + + let error = read_stdin_record(&mut reader, &mut buffer).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + assert!(error.to_string().contains("maximum size")); + assert!(buffer.len() <= MAX_STDIN_RECORD_BYTES + 1); + } + + #[test] + fn newer_mailbox_value_survives_an_earlier_submitted_acknowledgement() { + let mailbox = StdinMailbox::default(); + mailbox.publish(vec![ManagedTargetConfig::new("a", "127.0.0.1:2112")]); + let submitted = mailbox.latest().unwrap(); + mailbox.publish(vec![ManagedTargetConfig::new("b", "127.0.0.1:2113")]); + let pending = mailbox.latest().unwrap(); + + mailbox.clear_if_current(submitted.revision); + + assert_eq!(mailbox.latest().unwrap().revision, pending.revision); + } + + #[test] + fn rejected_update_retries_after_status_publishes_during_acknowledgement() { + let mailbox = StdinMailbox::default(); + mailbox.publish(Vec::new()); + let target_set = mailbox.latest().unwrap(); + let saturated = Arc::new(status_with_targets(&[TargetInstance { + id: TargetId::from("old"), + generation: 1, + }])); + let capacity_freed = Arc::new(status_with_targets(&[])); + // Capacity becomes available after the transaction rejected but before + // its acknowledgement reaches the controller. The rejection path must + // retain the pre-submission status, not this later snapshot. + let mut retry_after = + retry_after_capacity_rejection(&mailbox, &target_set, Arc::clone(&saturated)); + + assert!(retry_waits_for_status_change( + &mut retry_after, + &target_set, + &saturated + )); + assert!(!retry_waits_for_status_change( + &mut retry_after, + &target_set, + &capacity_freed + )); + } + + #[test] + fn later_desired_set_supersedes_a_capacity_rejected_candidate_before_retry() { + let mailbox = StdinMailbox::default(); + mailbox.publish(vec![ManagedTargetConfig::new("a", "127.0.0.1:2112")]); + let applied = mailbox.latest().unwrap(); + mailbox.clear_if_current(applied.revision); + + mailbox.publish(vec![ManagedTargetConfig::new("b", "127.0.0.1:2113")]); + let rejected = mailbox.latest().unwrap(); + let saturated = Arc::new(status_with_targets(&[TargetInstance { + id: TargetId::from("old"), + generation: 1, + }])); + let mut retry_after = retry_after_capacity_rejection(&mailbox, &rejected, saturated); + mailbox.publish(vec![ManagedTargetConfig::new("c", "127.0.0.1:2114")]); + let replacement = mailbox.latest().unwrap(); + + assert!(!retry_waits_for_status_change( + &mut retry_after, + &replacement, + &Arc::new(status_with_targets(&[TargetInstance { + id: TargetId::from("old"), + generation: 1, + }])) + )); + assert!(retry_after.is_none()); + assert_eq!(replacement.targets[0].id.as_str(), "c"); + mailbox.clear_if_current(replacement.revision); + assert!(mailbox.latest().is_none()); + } + + #[test] + fn eof_and_shutdown_stop_without_waiting_for_the_stdin_reader() { + let mailbox = StdinMailbox::default(); + let shutdown_requested = AtomicBool::new(false); + assert!(stdin_stop_request(&shutdown_requested, &mailbox).is_none()); + + mailbox.finish(); + assert!(matches!( + stdin_stop_request(&shutdown_requested, &mailbox), + Some(StdinStop::Eof) + )); + + let mailbox = StdinMailbox::default(); + shutdown_requested.store(true, std::sync::atomic::Ordering::Relaxed); + assert!(matches!( + stdin_stop_request(&shutdown_requested, &mailbox), + Some(StdinStop::Interrupted) + )); + } + + fn status_with_targets(targets: &[TargetInstance]) -> ManagedStatus { + ManagedStatus { + lifecycle: ManagedLifecycle::Running, + stop_requested: false, + applied_command_sequence: 0, + desired_target_count: targets.len(), + connecting_target_count: 0, + opening_target_count: 0, + active_target_count: 0, + draining_target_count: 0, + closing_target_count: 0, + terminal_target_count: 0, + total_target_outcomes: 0, + successful_target_outcomes: 0, + failed_target_outcomes: 0, + peer_closed_target_outcomes: 0, + discarded_target_outcomes: 0, + targets: Arc::from( + targets + .iter() + .cloned() + .map(|target| ManagedTargetStatus { + target, + desired: true, + lifecycle: ManagedTargetLifecycle::Active, + server_addr: Arc::from("127.0.0.1:2112"), + remote: None, + }) + .collect::>() + .into_boxed_slice(), + ), + recent_target_outcomes: Arc::from([]), + final_outcome: None, + } + } + + #[test] + fn dynamic_accounting_prunes_retired_generations_and_retains_only_shutdown_snapshot() { + let retired = TargetInstance { + id: TargetId::from("edge"), + generation: 1, + }; + let replacement = TargetInstance { + id: TargetId::from("edge"), + generation: 2, + }; + let live_status = status_with_targets(std::slice::from_ref(&replacement)); + let empty_status = status_with_targets(&[]); + let mut stats = BTreeMap::from([ + (retired.clone(), StatsCollector::new(stats_config(true))), + (replacement.clone(), StatsCollector::new(stats_config(true))), + ]); + + reconcile_stdin_stats(&mut stats, &live_status, None); + assert_eq!(stats.keys().cloned().collect::>(), [replacement]); + + let summary_targets = snapshot_stdin_summary_targets(&mut stats, &live_status); + stats.insert(retired, StatsCollector::new(stats_config(true))); + reconcile_stdin_stats(&mut stats, &empty_status, Some(&summary_targets)); + + assert_eq!( + stats.keys().cloned().collect::>(), + [TargetInstance { + id: TargetId::from("edge"), + generation: 2, + }] + ); + } + + #[test] + fn terminal_deduplication_is_bounded_under_target_churn() { + let mut terminal_targets = BoundedTargetSet::new(2); + let target = |generation| TargetInstance { + id: TargetId::from("edge"), + generation, + }; + + assert!(terminal_targets.insert(target(1))); + assert!(terminal_targets.insert(target(2))); + assert!(terminal_targets.insert(target(3))); + assert!(terminal_targets.insert(target(1))); + assert!(!terminal_targets.insert(target(3))); + } fn long_finite_args() -> ClientArgs { use clap::Parser; diff --git a/crates/irtt-app/src/cmd/tui/args.rs b/crates/irtt-app/src/cmd/tui/args.rs index 8e1320c..5c971e0 100644 --- a/crates/irtt-app/src/cmd/tui/args.rs +++ b/crates/irtt-app/src/cmd/tui/args.rs @@ -14,13 +14,13 @@ pub const DEFAULT_TUI_DURATION: Duration = Duration::ZERO; #[derive(Debug, Clone, Parser)] #[command(name = "irtt-tui", about = "Minimal IRTT-compatible TUI client")] pub struct TuiArgs { - /// Server address/host, optionally prefixed with LABEL=. Repeat for multi-target mode. + /// Server address/host, optionally prefixed with LABEL= and suffixed with @hmac=KEY. Repeat for multi-target mode. #[arg( value_name = "TARGET", num_args = 1.., required = true, value_parser = parse_target, - long_help = "Server address/host, optionally prefixed with LABEL=. Repeat for multi-target mode.\n\nExplicit labels are used in the legend and status table.\n\nExamples:\n irtt-tui host.example\n irtt-tui eu=host.example\n irtt-tui eu=host-a.example us=host-b.example" + long_help = "Server address/host, optionally prefixed with LABEL= and suffixed with @hmac=KEY. Repeat for multi-target mode.\n\nExplicit labels are used in the legend and status table. A target without @hmac= inherits --hmac; @hmac= disables it for that target.\n\nExamples:\n irtt-tui host.example\n irtt-tui eu=host.example\n irtt-tui eu=host-a.example@hmac=secret us=host-b.example" )] pub targets: Vec, @@ -58,6 +58,7 @@ impl TuiArgs { TargetSelection { targets: &self.targets, pacing: self.pacing, + stdin_controlled: false, }, ) } @@ -107,8 +108,7 @@ mod tests { fn tui_parser_defaults_to_continuous_and_has_no_output_option() { let args = parse(&["127.0.0.1:2112"]).unwrap(); assert_eq!(args.targets.len(), 1); - assert_eq!(args.targets[0].label, None); - assert_eq!(args.targets[0].addr, "127.0.0.1:2112"); + assert_eq!(args.prepare().unwrap().targets[0].label, "127.0.0.1:2112"); assert_eq!(args.pacing, GroupPacingArg::Staggered); assert_eq!(args.duration, DEFAULT_TUI_DURATION); assert!(args.is_continuous()); @@ -181,8 +181,8 @@ mod tests { #[test] fn invalid_labelled_target_syntax_is_rejected() { - assert!(parse(&["=127.0.0.1:2112"]).is_err()); - assert!(parse(&["label="]).is_err()); + assert!(parse(&["=127.0.0.1:2112"]).unwrap().prepare().is_err()); + assert!(parse(&["label="]).unwrap().prepare().is_err()); } #[test] diff --git a/crates/irtt-app/src/shared/client/mod.rs b/crates/irtt-app/src/shared/client/mod.rs index 12e99aa..27a2cf4 100644 --- a/crates/irtt-app/src/shared/client/mod.rs +++ b/crates/irtt-app/src/shared/client/mod.rs @@ -9,9 +9,12 @@ pub use args::{ ClockArg, CommonClientArgs, ReceivedStatsArg, TimestampArg, }; pub use config::{expected_probe_count, DEFAULT_RECV_TIMEOUT}; -pub use prepare::{prepare_managed_run, ManagedRunSetup, TargetSelection, MANAGED_EVENT_CAPACITY}; +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_target, prepare_managed_targets, target_specs, GroupPacingArg, PreparedTarget, TargetArg, - TargetSpec, + parse_stdin_target_set, parse_target, prepare_managed_targets, target_specs, GroupPacingArg, + PreparedTarget, TargetArg, TargetSpec, }; diff --git a/crates/irtt-app/src/shared/client/prepare.rs b/crates/irtt-app/src/shared/client/prepare.rs index a91d026..9af920f 100644 --- a/crates/irtt-app/src/shared/client/prepare.rs +++ b/crates/irtt-app/src/shared/client/prepare.rs @@ -22,6 +22,9 @@ use super::{ /// Capacity of the lossy managed presentation event channel. pub const MANAGED_EVENT_CAPACITY: usize = 16_384; +pub const STDIN_MAX_DESIRED_TARGETS: usize = 128; +pub const STDIN_MAX_LIVE_TARGET_GENERATIONS: usize = 256; +pub const STDIN_OUTCOME_HISTORY_LIMIT: usize = 256; /// Target and pacing arguments common to the client applets. /// @@ -31,6 +34,7 @@ pub const MANAGED_EVENT_CAPACITY: usize = 16_384; pub struct TargetSelection<'a> { pub targets: &'a [TargetArg], pub pacing: GroupPacingArg, + pub stdin_controlled: bool, } /// Everything a managed run needs, prepared and validated together. @@ -46,6 +50,8 @@ pub struct ManagedRunSetup { pub client: ClientConfig, /// Send coordination across active targets. pub pacing: ManagedPacing, + /// Whether this stream may replace its desired set from stdin. + pub stdin_controlled: bool, } impl ManagedRunSetup { @@ -70,7 +76,7 @@ impl ManagedRunSetup { /// Managed driver configuration for this run. pub fn managed_config(&self) -> ManagedClientConfig { let target_count = self.target_count(); - ManagedClientConfig { + let mut config = ManagedClientConfig { client: self.client.clone(), pacing: self.pacing, completion: ManagedCompletionPolicy::FinishWhenQuiescent, @@ -78,7 +84,13 @@ impl ManagedRunSetup { outcome_history_limit: target_count, max_live_target_generations: target_count, ..ManagedClientConfig::default() + }; + if self.stdin_controlled { + config.completion = ManagedCompletionPolicy::ExplicitStop; + config.outcome_history_limit = STDIN_OUTCOME_HISTORY_LIMIT; + config.max_live_target_generations = STDIN_MAX_LIVE_TARGET_GENERATIONS; } + config } } @@ -91,12 +103,22 @@ pub fn prepare_managed_run( duration: Duration, selection: TargetSelection<'_>, ) -> Result { - let specs = target_specs(selection.targets)?; + let specs = if selection.stdin_controlled { + if selection.targets.len() > STDIN_MAX_DESIRED_TARGETS { + return Err(format!( + "initial target set exceeds the {STDIN_MAX_DESIRED_TARGETS}-target limit" + )); + } + super::targets::target_specs_with_empty(selection.targets, true)? + } else { + target_specs(selection.targets)? + }; let targets = prepare_managed_targets(specs)?; Ok(ManagedRunSetup { targets, client: common.to_client_config(duration), pacing: selection.pacing.into(), + stdin_controlled: selection.stdin_controlled, }) } @@ -119,17 +141,11 @@ mod tests { } fn unlabeled(addr: &str) -> TargetArg { - TargetArg { - label: None, - addr: addr.to_owned(), - } + TargetArg::new(addr) } fn labeled(label: &str, addr: &str) -> TargetArg { - TargetArg { - label: Some(label.to_owned()), - addr: addr.to_owned(), - } + TargetArg::new(format!("{label}={addr}")) } fn prepare( @@ -137,7 +153,15 @@ mod tests { pacing: GroupPacingArg, duration: Duration, ) -> Result { - prepare_managed_run(&common(&[]), duration, TargetSelection { targets, pacing }) + prepare_managed_run( + &common(&[]), + duration, + TargetSelection { + targets, + pacing, + stdin_controlled: false, + }, + ) } #[test] @@ -212,6 +236,7 @@ mod tests { TargetSelection { targets: &targets, pacing: GroupPacingArg::Staggered, + stdin_controlled: false, }, ) .unwrap(); diff --git a/crates/irtt-app/src/shared/client/targets.rs b/crates/irtt-app/src/shared/client/targets.rs index 35c2d09..392c4d3 100644 --- a/crates/irtt-app/src/shared/client/targets.rs +++ b/crates/irtt-app/src/shared/client/targets.rs @@ -1,26 +1,99 @@ -use std::collections::HashSet; +use std::{collections::HashSet, fmt}; use clap::ValueEnum; -use irtt_client::managed::{ManagedPacing, ManagedTargetConfig, TargetId}; +use irtt_client::{ + managed::{ManagedPacing, ManagedTargetConfig, TargetId}, + ClientAuthConfig, +}; -#[derive(Debug, Clone, PartialEq, Eq)] +/// One raw positional target captured by Clap. +/// +/// Parsing happens during run preparation so invalid target syntax never makes +/// Clap include the original argument (which may carry an HMAC key) in an +/// error diagnostic. +#[derive(Clone, PartialEq, Eq)] pub struct TargetArg { - pub label: Option, - pub addr: String, + input: String, } -#[derive(Debug, Clone, PartialEq, Eq)] +impl TargetArg { + pub fn new(input: impl Into) -> Self { + Self { + input: input.into(), + } + } +} + +impl fmt::Debug for TargetArg { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("TargetArg()") + } +} + +/// Capture one positional target without exposing it through Clap diagnostics. +pub fn parse_target(input: &str) -> Result { + Ok(TargetArg::new(input)) +} + +#[derive(Clone, PartialEq, Eq)] +pub enum TargetAuth { + Inherit, + Override(Vec), + Disable, +} + +impl fmt::Debug for TargetAuth { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Inherit => f.write_str("Inherit"), + Self::Override(_) => f.write_str("Override()"), + Self::Disable => f.write_str("Disable"), + } + } +} + +#[derive(Clone, PartialEq, Eq)] pub struct TargetSpec { pub label: String, pub addr: String, + pub auth: TargetAuth, } -#[derive(Debug, Clone)] +impl fmt::Debug for TargetSpec { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TargetSpec") + .field("label", &self.label) + .field("addr", &self.addr) + .field("auth", &self.auth) + .finish() + } +} + +#[derive(Clone)] pub struct PreparedTarget { pub label: String, pub managed: ManagedTargetConfig, } +impl fmt::Debug for PreparedTarget { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("PreparedTarget") + .field("label", &self.label) + .field("server_addr", &self.managed.server_addr) + .field( + "auth", + &match &self.managed.auth { + None => TargetAuth::Inherit, + Some(ClientAuthConfig { hmac_key: Some(_) }) => { + TargetAuth::Override(Vec::new()) + } + Some(ClientAuthConfig { hmac_key: None }) => TargetAuth::Disable, + }, + ) + .finish() + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] pub enum GroupPacingArg { Staggered, @@ -36,191 +109,383 @@ impl From for ManagedPacing { } } -/// Parse one positional target argument, as `TARGET` or `LABEL=TARGET`. -pub fn parse_target(input: &str) -> Result { - match input.split_once('=') { - None => Ok(TargetArg { - label: None, - addr: input.to_owned(), - }), - Some((label, addr)) => { - if label.is_empty() { - return Err("target label must not be empty".to_owned()); - } - if addr.is_empty() { - return Err("target address must not be empty".to_owned()); - } - Ok(TargetArg { - label: Some(label.to_owned()), - addr: addr.to_owned(), - }) - } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum TargetParseError { + EmptyLabel, + EmptyAddress, + InvalidEscape, + EmptyTarget, +} + +impl fmt::Display for TargetParseError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let message = match self { + Self::EmptyLabel => "target label must not be empty", + Self::EmptyAddress => "target address must not be empty", + Self::InvalidEscape => "invalid target escape sequence", + Self::EmptyTarget => "target must not be empty", + }; + f.write_str(message) } } -pub fn target_specs(targets: &[TargetArg]) -> Result, String> { +fn split_unescaped(input: &str, delimiter: char) -> Result, TargetParseError> { + let mut parts = Vec::new(); + let mut start = 0; + let mut escaped = false; + for (index, ch) in input.char_indices() { + if escaped { + escaped = false; + } else if ch == '\\' { + escaped = true; + } else if ch == delimiter { + parts.push(&input[start..index]); + start = index + ch.len_utf8(); + } + } + if escaped { + return Err(TargetParseError::InvalidEscape); + } + parts.push(&input[start..]); + Ok(parts) +} + +fn first_unescaped(input: &str, delimiter: char) -> Result, TargetParseError> { + let mut escaped = false; + for (index, ch) in input.char_indices() { + if escaped { + escaped = false; + } else if ch == '\\' { + escaped = true; + } else if ch == delimiter { + return Ok(Some(index)); + } + } + if escaped { + return Err(TargetParseError::InvalidEscape); + } + Ok(None) +} + +fn unescape(input: &str) -> Result { + let mut output = String::with_capacity(input.len()); + let mut escaped = false; + for ch in input.chars() { + if escaped { + if !matches!(ch, '\\' | '=' | ';' | ',' | '@') { + return Err(TargetParseError::InvalidEscape); + } + output.push(ch); + escaped = false; + } else if ch == '\\' { + escaped = true; + } else { + output.push(ch); + } + } + if escaped { + return Err(TargetParseError::InvalidEscape); + } + Ok(output) +} + +fn hmac_modifier(input: &str) -> Result, TargetParseError> { + let mut escaped = false; + for (index, ch) in input.char_indices() { + if escaped { + escaped = false; + } else if ch == '\\' { + escaped = true; + } else if input[index..].starts_with("@hmac=") { + return Ok(Some(index)); + } + } + if escaped { + return Err(TargetParseError::InvalidEscape); + } + Ok(None) +} + +fn parse_target_syntax( + input: &str, +) -> Result<(Option, String, TargetAuth), TargetParseError> { + if input.is_empty() { + return Err(TargetParseError::EmptyTarget); + } + let (main, auth) = match hmac_modifier(input)? { + Some(index) => { + let value = unescape(&input[index + "@hmac=".len()..])?; + let auth = if value.is_empty() { + TargetAuth::Disable + } else { + TargetAuth::Override(value.into_bytes()) + }; + (&input[..index], auth) + } + None => (input, TargetAuth::Inherit), + }; + let (label, address) = match first_unescaped(main, '=')? { + None => (None, unescape(main)?), + Some(index) => { + let label = unescape(&main[..index])?; + let address = unescape(&main[index + '='.len_utf8()..])?; + (Some(label), address) + } + }; + if label.as_deref().is_some_and(str::is_empty) { + return Err(TargetParseError::EmptyLabel); + } + if address.is_empty() { + return Err(TargetParseError::EmptyAddress); + } + Ok((label, address, auth)) +} + +pub(crate) fn target_specs_with_empty( + targets: &[TargetArg], + allow_empty: bool, +) -> Result, String> { let mut specs = Vec::with_capacity(targets.len()); - let mut unlabeled_counts = std::collections::HashMap::<&str, usize>::new(); - for target in targets { - let label = match &target.label { - Some(label) => label.clone(), + let mut unlabeled_counts = std::collections::HashMap::::new(); + for (index, target) in targets.iter().enumerate() { + let (explicit_label, addr, auth) = parse_target_syntax(&target.input) + .map_err(|error| format!("invalid target {}: {error}", index + 1))?; + let label = match explicit_label { + Some(label) => label, None => { - let count = unlabeled_counts.entry(target.addr.as_str()).or_default(); + let count = unlabeled_counts.entry(addr.clone()).or_default(); *count += 1; if *count == 1 { - target.addr.clone() + addr.clone() } else { - format!("{}#{}", target.addr, *count) + format!("{}#{}", addr, *count) } } }; - specs.push(TargetSpec { - label, - addr: target.addr.clone(), - }); + specs.push(TargetSpec { label, addr, auth }); } - if specs.is_empty() { + if specs.is_empty() && !allow_empty { return Err("at least one target is required unless --list-columns is set".to_owned()); } let mut labels = HashSet::new(); for spec in &specs { if !labels.insert(spec.label.clone()) { - return Err(format!("duplicate target label {:?}", spec.label)); + return Err("duplicate target label".to_owned()); } } Ok(specs) } +pub fn target_specs(targets: &[TargetArg]) -> Result, String> { + target_specs_with_empty(targets, false) +} + pub fn prepare_managed_targets(specs: Vec) -> Result, String> { let mut targets = Vec::with_capacity(specs.len()); for spec in specs { + let mut managed = ManagedTargetConfig::new(TargetId::from(spec.label.clone()), spec.addr); + managed.auth = match spec.auth { + TargetAuth::Inherit => None, + TargetAuth::Override(hmac_key) => Some(ClientAuthConfig { + hmac_key: Some(hmac_key), + }), + TargetAuth::Disable => Some(ClientAuthConfig { hmac_key: None }), + }; targets.push(PreparedTarget { - label: spec.label.clone(), - managed: ManagedTargetConfig::new(TargetId::from(spec.label), spec.addr), + label: spec.label, + managed, }); } Ok(targets) } +/// Parse one complete stdin target set after its line terminator was removed. +/// +/// Commas frame stdin elements only; positional target arguments do not use +/// this framing and may therefore contain raw commas. +pub fn parse_stdin_target_set( + record: &str, + maximum_targets: usize, +) -> Result, String> { + if record == "[]" { + return Ok(Vec::new()); + } + let elements = split_unescaped(record, ',').map_err(|error| error.to_string())?; + if elements.len() > maximum_targets { + return Err(format!( + "target set exceeds the {maximum_targets}-target limit" + )); + } + let args = elements.into_iter().map(TargetArg::new).collect::>(); + prepare_managed_targets(target_specs_with_empty(&args, true)?) +} + #[cfg(test)] mod tests { use super::*; - fn unlabeled(addr: &str) -> TargetArg { - TargetArg { - label: None, - addr: addr.to_owned(), - } + fn target(input: &str) -> TargetArg { + TargetArg::new(input) } - fn labeled(label: &str, addr: &str) -> TargetArg { - TargetArg { - label: Some(label.to_owned()), - addr: addr.to_owned(), - } + fn specs(inputs: &[&str]) -> Result, String> { + target_specs(&inputs.iter().map(|input| target(input)).collect::>()) } #[test] - fn parse_target_without_equals_is_unlabeled() { - let target = parse_target("host.example").unwrap(); - assert_eq!(target.label, None); - assert_eq!(target.addr, "host.example"); - - let target = parse_target("host.example:2112").unwrap(); - assert_eq!(target.label, None); - assert_eq!(target.addr, "host.example:2112"); - - let target = parse_target("[::1]:2112").unwrap(); - assert_eq!(target.label, None); - assert_eq!(target.addr, "[::1]:2112"); + fn legacy_target_syntax_and_ipv6_remain_supported() { + let parsed = specs(&[ + "host.example", + "host.example:2112", + "[::1]:2112", + "eu=host.example", + ]) + .unwrap(); + assert_eq!(parsed[0].label, "host.example"); + assert_eq!(parsed[1].label, "host.example:2112"); + assert_eq!(parsed[2].label, "[::1]:2112"); + assert_eq!(parsed[3].label, "eu"); + assert!(parsed + .iter() + .all(|target| matches!(target.auth, TargetAuth::Inherit))); } #[test] - fn parse_target_with_equals_splits_on_first() { - let target = parse_target("eu=host.example").unwrap(); - assert_eq!(target.label, Some("eu".to_owned())); - assert_eq!(target.addr, "host.example"); - - let target = parse_target("eu=host.example:2112").unwrap(); - assert_eq!(target.label, Some("eu".to_owned())); - assert_eq!(target.addr, "host.example:2112"); + fn target_hmac_supports_inherit_override_disable_and_base64_padding() { + let parsed = specs(&[ + "default=default.example", + "private=private.example@hmac=YWJjZA==", + "public=public.example@hmac=", + ]) + .unwrap(); + assert!(matches!(parsed[0].auth, TargetAuth::Inherit)); + assert_eq!(parsed[1].auth, TargetAuth::Override(b"YWJjZA==".to_vec())); + assert!(matches!(parsed[2].auth, TargetAuth::Disable)); } #[test] - fn parse_target_rejects_empty_label_or_address() { - assert!(parse_target("=host.example").is_err()); - assert!(parse_target("eu=").is_err()); + fn target_delimiters_are_escaped_contextually() { + let parsed = specs(&["eu\\=west=host\\@hmac=literal@hmac=a\\@b\\,c\\=d\\\\e"]).unwrap(); + assert_eq!(parsed[0].label, "eu=west"); + assert_eq!(parsed[0].addr, "host@hmac=literal"); + assert_eq!(parsed[0].auth, TargetAuth::Override(b"a@b,c=d\\e".to_vec())); + } + + #[test] + fn positional_commas_are_not_reserved() { + let parsed = specs(&["edge=host,region@hmac=key,part"]).unwrap(); + assert_eq!(parsed[0].addr, "host,region"); + assert_eq!(parsed[0].auth, TargetAuth::Override(b"key,part".to_vec())); + } + + #[test] + fn parser_diagnostics_and_debug_do_not_reveal_hmac_values() { + let secret = "very-secret-value"; + let error = specs(&[&format!("target=host@hmac={secret}\\q")]).unwrap_err(); + assert!(!error.contains(secret)); + assert!(!format!("{:?}", target(&format!("target=host@hmac={secret}"))).contains(secret)); + assert!( + !format!("{:?}", TargetAuth::Override(secret.as_bytes().to_vec())).contains(secret) + ); } #[test] fn target_specs_suffix_repeated_unlabeled_and_reject_duplicate_labels() { - let targets = vec![unlabeled("host-a:2112"), unlabeled("host-a:2112")]; - let specs = target_specs(&targets).unwrap(); - - assert_eq!(specs[0].label, "host-a:2112"); - assert_eq!(specs[1].label, "host-a:2112#2"); - - let targets = vec![ - unlabeled("host-a:2112"), - labeled("host-a:2112", "host-b:2112"), - ]; - let err = target_specs(&targets).unwrap_err(); - assert!(err.contains("duplicate target label")); + let parsed = specs(&["host-a:2112", "host-a:2112"]).unwrap(); + assert_eq!(parsed[0].label, "host-a:2112"); + assert_eq!(parsed[1].label, "host-a:2112#2"); + assert_eq!( + specs(&["host-a:2112", "host-a:2112=host-b:2112"]).unwrap_err(), + "duplicate target label" + ); } #[test] - fn target_specs_preserve_argument_order_with_mixed_labels() { - let targets = vec![ - unlabeled("local"), - labeled("eu", "eu.example"), - unlabeled("backup"), - labeled("us", "us.example"), - ]; - let specs = target_specs(&targets).unwrap(); - - let labels: Vec<_> = specs.iter().map(|spec| spec.label.as_str()).collect(); - assert_eq!(labels, ["local", "eu", "backup", "us"]); - let addrs: Vec<_> = specs.iter().map(|spec| spec.addr.as_str()).collect(); - assert_eq!(addrs, ["local", "eu.example", "backup", "us.example"]); - } - - #[test] - fn target_specs_count_unlabeled_occurrences_independently_of_explicit_labels() { - let targets = vec![unlabeled("foo"), labeled("eu", "foo"), unlabeled("foo")]; - let specs = target_specs(&targets).unwrap(); - - let labels: Vec<_> = specs.iter().map(|spec| spec.label.as_str()).collect(); - assert_eq!(labels, ["foo", "eu", "foo#2"]); + fn target_specs_preserve_argument_order_and_unlabeled_counts() { + let parsed = specs(&["local", "eu=eu.example", "foo", "us=us.example", "foo"]).unwrap(); + let labels: Vec<_> = parsed.iter().map(|target| target.label.as_str()).collect(); + assert_eq!(labels, ["local", "eu", "foo", "us", "foo#2"]); + let addresses: Vec<_> = parsed.iter().map(|target| target.addr.as_str()).collect(); + assert_eq!( + addresses, + ["local", "eu.example", "foo", "us.example", "foo"] + ); } #[test] fn target_specs_reject_generated_suffix_collisions() { - let targets = vec![ - unlabeled("host.example"), - unlabeled("host.example"), - labeled("host.example#2", "other.example"), - ]; - let err = target_specs(&targets).unwrap_err(); - assert!(err.contains("duplicate target label")); + assert_eq!( + specs(&[ + "host.example", + "host.example", + "host.example#2=other.example" + ]) + .unwrap_err(), + "duplicate target label" + ); } #[test] - fn target_preparation_preserves_hostname_and_allows_duplicate_endpoints() { - let specs = vec![ - TargetSpec { - label: "a".to_owned(), - addr: "example.test".to_owned(), - }, - TargetSpec { - label: "b".to_owned(), - addr: "example.test".to_owned(), - }, - ]; - let targets = prepare_managed_targets(specs).unwrap(); - assert_eq!(targets[0].managed.server_addr, "example.test"); - assert_eq!(targets[1].managed.server_addr, "example.test"); + fn malformed_target_syntax_is_rejected_without_echoing_input() { + for input in ["=host", "label=", "host@hmac=value\\q", "host\\x"] { + assert!(specs(&[input]).is_err(), "{input}"); + } + } + + #[test] + fn target_preparation_maps_auth_to_managed_config() { + let prepared = + prepare_managed_targets(specs(&["a=one@hmac=key", "b=two@hmac="]).unwrap()).unwrap(); + assert_eq!( + prepared[0] + .managed + .auth + .as_ref() + .unwrap() + .hmac_key + .as_deref(), + Some(b"key".as_slice()) + ); + assert_eq!(prepared[1].managed.auth.as_ref().unwrap().hmac_key, None); + } + + #[test] + fn target_preparation_preserves_duplicate_endpoints() { + let prepared = + prepare_managed_targets(specs(&["a=example.test", "b=example.test"]).unwrap()).unwrap(); + assert_eq!(prepared[0].managed.server_addr, "example.test"); + assert_eq!(prepared[1].managed.server_addr, "example.test"); + } + + #[test] + fn stdin_target_sets_frame_commas_without_trimming_payload() { + let parsed = parse_stdin_target_set("a=one@hmac=x\\,y,b=two", 128).unwrap(); + assert_eq!( + parsed[0].managed.auth.as_ref().unwrap().hmac_key.as_deref(), + Some(b"x,y".as_slice()) + ); + assert_eq!(parsed[1].label, "b"); + assert_eq!( + parse_stdin_target_set(" []", 128).unwrap()[0] + .managed + .server_addr, + " []" + ); + assert_eq!( + parse_stdin_target_set("[] ", 128).unwrap()[0] + .managed + .server_addr, + "[] " + ); + assert!(parse_stdin_target_set("", 128).is_err()); + assert!(parse_stdin_target_set("a=one,b=two", 1).is_err()); + } + + #[test] + fn stdin_empty_set_is_exactly_bracket_pair() { + assert!(parse_stdin_target_set("[]", 128).unwrap().is_empty()); } } diff --git a/crates/irtt-client/src/managed/blocking_tests.rs b/crates/irtt-client/src/managed/blocking_tests.rs index d3df883..743eae8 100644 --- a/crates/irtt-client/src/managed/blocking_tests.rs +++ b/crates/irtt-client/src/managed/blocking_tests.rs @@ -250,15 +250,11 @@ fn dynamic_update_reaches_background_worker() { .unwrap(); let handle = owner.handle(); - let acknowledgement = test_runtime().block_on(async { - timeout( - Duration::from_secs(2), - handle.update_targets(vec![target(&server)]).unwrap(), - ) - .await + let acknowledgement = handle + .update_targets(vec![target(&server)]) .unwrap() - .unwrap() - }); + .blocking_wait() + .unwrap(); assert_eq!(acknowledgement.sequence, 1); server.wait_probe(); test_runtime().block_on(async { diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index 3f190ae..98c8bbe 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -301,6 +301,19 @@ pub struct ManagedCommandReceipt { receiver: oneshot::Receiver>, } +impl ManagedCommandReceipt { + /// Block until the target-set transaction is applied or rejected. + /// + /// This is for synchronous callers such as [`BlockingManagedClient`]. It + /// must not be called from an asynchronous runtime; use the [`Future`] + /// implementation there instead. + pub fn blocking_wait(self) -> Result { + self.receiver + .blocking_recv() + .unwrap_or(Err(ManagedCommandApplyError::AcknowledgementDisconnected)) + } +} + impl Future for ManagedCommandReceipt { type Output = Result; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index 6c13daa..99b323e 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -1576,6 +1576,37 @@ fn authenticated_peer_close_outcome() { assert!(!has_close(&records)); } +#[test] +fn peer_close_leaves_other_explicitly_managed_targets_running() { + let closing = start_server(ServerBehavior::PeerClose, None); + let continuing = start_server(ServerBehavior::Echo, None); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (task, handle) = ManagedClient::task( + managed, + vec![ + target("closing", closing.addr), + target("continuing", continuing.addr), + ], + ) + .unwrap(); + let runtime = runtime(); + let mut task = Box::pin(task); + + drive_task_until(&runtime, &mut task, |task| { + handle.status().peer_closed_target_outcomes == 1 + && task.targets[1].counters.packets_sent != 0 + }); + assert_eq!(handle.status().lifecycle, ManagedLifecycle::Running); + + let stop = handle.stop(); + runtime.block_on(task); + runtime.block_on(stop); + closing.finish(); + assert!(!probes(&continuing.finish()).is_empty()); +} + #[test] fn peer_close_during_stop_drain_remains_authoritative() { let key = b"managed-drain-peer-close".to_vec(); diff --git a/docs/irtt-client.md b/docs/irtt-client.md index 9cd011c..2a49159 100644 --- a/docs/irtt-client.md +++ b/docs/irtt-client.md @@ -6,7 +6,7 @@ irtt-client - IRTT-compatible stream client ## SYNOPSIS -`irtt-client` [*OPTIONS*] [*[LABEL=]TARGET*]... +`irtt-client` [*OPTIONS*] [*[LABEL=]TARGET[@hmac=KEY]*]... `irtt-client` `--list-columns` @@ -18,17 +18,39 @@ eligible runs. ## TARGETS -Every target argument accepts `[LABEL=]TARGET`. An optional `LABEL=` prefix -assigns the logical target name used in output; without it, the target -string itself is used as the label. Labeled and unlabeled targets can be -freely mixed in one argument list. +Every target argument accepts `[LABEL=]TARGET[@hmac=KEY]`. An optional +`LABEL=` prefix assigns the logical target name used in output; without it, +the target string itself is used as the label. Labeled and unlabeled targets +can be freely mixed in one argument list. Prefer explicit labels for +long-running dynamic target sets: the label is the stable target identity +across replacements. ```sh irtt-client host-a:2112 irtt-client eu=host-a:2112 us=host-b:2112 irtt-client host-a:2112 eu=host-b:2112 +irtt-client --hmac default-secret eu=host-a:2112@hmac=eu-secret public=host-b:2112@hmac= ``` +Targets without `@hmac=` inherit the global `--hmac` value. `@hmac=KEY` +overrides it for that target, and `@hmac=` explicitly disables HMAC for that +target. The modifier uses `@`, not a shell command separator; quote an argument +if its key needs shell quoting. + +Target syntax supports these backslash escapes in labels, addresses, and HMAC +values: `\\` for a literal backslash, `\@` for a literal `@`, `\=` for a +literal `=`, and `\;` for a literal semicolon. Escaping `@` is necessary in a +label or address when it would otherwise begin `@hmac=`; escaping the first +`=` is how an unlabeled target or a label can contain that character. An `=` in +an HMAC value, including Base64 padding, does not need escaping. Positional +target arguments may contain a raw comma; stdin target sets have the additional +`\,` escape described below. + +Shells process unquoted backslashes and characters such as semicolons before +the client sees them. Single-quote a target when it contains parser escapes, +for example `'west\=edge=host\@hmac=literal@hmac=a\;b\\c'`, so the +backslashes reach `irtt-client` unchanged. + Multiple targets are probed concurrently. By default, a multi-target session uses **staggered** pacing, spacing active targets across the probe interval; pass `--pacing burst` to send one probe to every active target back-to-back @@ -36,6 +58,57 @@ once per interval instead. Final per-target summaries print in the order targets were supplied on the command line, not the order in which they finish or their labels sort alphabetically. +## DYNAMIC TARGET SETS FROM STANDARD INPUT + +`--targets-stdin` is available only with continuous mode (`--duration 0`). +Positional targets remain an initial desired set, and may be omitted to +start empty: + +~~~sh +irtt-client --duration 0 --targets-stdin +irtt-client --duration 0 --targets-stdin ams=ams.example:2112 +~~~ + +Each non-empty stdin record declares the complete desired target set, rather +than a delta. A later record replaces an earlier set: targets absent from it +are retired, unchanged target configurations retain their generation, and a +changed address or HMAC setting creates a fresh generation. Under transient +live-generation backpressure, only the latest unapplied desired set is kept. + +Commas frame targets in stdin records; escape a literal comma within one +target as `\,`. Target parsing inside each element is otherwise the same as +for positional arguments, including the other escapes above, `@hmac=KEY`, and +`@hmac=`. Use a persistent producer or keep a pipe/FIFO writer open while the +client is being controlled. For example, in one terminal: + +~~~sh +mkfifo /tmp/irtt-targets +irtt-client --duration 0 --targets-stdin --hmac default-secret \ + < /tmp/irtt-targets +~~~ + +Then open the FIFO once from another terminal and keep that descriptor open +while publishing complete desired sets: + +~~~sh +exec 3> /tmp/irtt-targets +printf '%s\n' 'ams=ams.example:2112,sg=sg.example:2112@hmac=sg-secret' >&3 +printf '%s\n' 'sg=sg.example:2112@hmac=sg-secret,nyc=nyc.example:2112' >&3 +# Publish more sets as needed, then send EOF and stop the client: +exec 3>&- +~~~ + +The exact record `[]` selects an empty desired set. Empty records are +ignored; whitespace is otherwise payload, not trimming. EOF is immediate loss +or closure of the control plane and requests a graceful stop. A desired-state +revision that was read but has not yet been acknowledged as applied may be +discarded when EOF is observed; EOF does not wait for a capacity-rejected +revision to become applicable. + +The maximum stdin record payload is 64 KiB. Stdin-controlled mode bounds each +desired set to 128 targets and retains at most 256 live target generations +while replacements drain. + ## FINITE VERSUS CONTINUOUS OPERATION `--duration` controls the run length: @@ -45,8 +118,9 @@ finish or their labels sort alphabetically. (see [OUTPUT](#output) below — CSV, TSV, and JSON Lines never print one). - `--duration 0` is **continuous**: the client runs until interrupted (`Ctrl-C`). For table output, a final summary is printed only if the run - was interrupted; an uninterrupted continuous run (one that exits for - another reason) does not get one. + was interrupted. Stdin-controlled mode is the exception: EOF also requests + a graceful stop and prints the final retained summary. An uninterrupted + continuous run that exits for another reason does not get one. See [MEASUREMENTS AND MEMORY](#measurements-and-memory) below for why this distinction also determines what statistics can be computed and how much @@ -192,6 +266,10 @@ the client exits nonzero — so a process supervisor can restart it; an interrupted run (`Ctrl-C`) never reports peer closure as an error, whichever mode it was in. +With `--targets-stdin`, peer closure is target-local: other desired targets +and the stdin controller continue running. Including that target in a later +complete set starts a fresh generation. + ## EXIT STATUS `0` on success (including a finite run's ordinary completion, and an