diff --git a/crates/irtt-app/src/cmd/client/args.rs b/crates/irtt-app/src/cmd/client/args.rs index 7fda52c..ba3294c 100644 --- a/crates/irtt-app/src/cmd/client/args.rs +++ b/crates/irtt-app/src/cmd/client/args.rs @@ -23,7 +23,7 @@ pub struct ClientArgs { /// 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." + 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, @@ -41,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, diff --git a/crates/irtt-app/src/cmd/client/run.rs b/crates/irtt-app/src/cmd/client/run.rs index f5c47fa..c6780e2 100644 --- a/crates/irtt-app/src/cmd/client/run.rs +++ b/crates/irtt-app/src/cmd/client/run.rs @@ -943,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/docs/irtt-client.md b/docs/irtt-client.md index 6238d47..2a49159 100644 --- a/docs/irtt-client.md +++ b/docs/irtt-client.md @@ -35,8 +35,21 @@ irtt-client --hmac default-secret eu=host-a:2112@hmac=eu-secret public=host-b:21 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. A literal `@` immediately before `hmac=` in -the target portion is written as `\@`. +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; @@ -64,20 +77,37 @@ 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 `@hmac=KEY` and `@hmac=`: +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 -printf '%s\n' \ - 'ams=ams.example:2112,sg=sg.example:2112@hmac=sg-secret' \ - 'sg=sg.example:2112@hmac=sg-secret,nyc=nyc.example:2112' | - irtt-client --duration 0 --targets-stdin --hmac default-secret +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 requests a -graceful stop. The stdin-controlled mode bounds each desired set to 128 -targets and retains at most 256 live target generations while replacements -drain. +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 @@ -88,8 +118,9 @@ drain. (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