diff --git a/services/flysim/crates/fly-session/src/cli.rs b/services/flysim/crates/fly-session/src/cli.rs index ae24bff..1f7bc68 100644 --- a/services/flysim/crates/fly-session/src/cli.rs +++ b/services/flysim/crates/fly-session/src/cli.rs @@ -21,7 +21,9 @@ use std::process::ExitCode; use crate::agent::AgentFaults; use crate::environment::EnvironmentFaults; -use crate::launcher::{AgentLaunch, EnvironmentLaunch, ExecutionMode, Started, serve_one}; +use crate::launcher::{ + AgentLaunch, EnvironmentLaunch, ExecutionMode, Started, flags, serve_one, +}; use crate::types::*; const USAGE: &str = "\ @@ -69,9 +71,12 @@ pub fn main() -> ExitCode { let command = command.to_string_lossy().into_owned(); let rest: Vec = args.map(|a| a.to_string_lossy().into_owned()).collect(); let result = match command.as_str() { - "agent" | "environment" => Options::parse(&rest).and_then(|o| serve(&command, &o)), - "measure" => Options::parse(&rest).and_then(|o| measure(&o)), - "measure-row" => Options::parse(&rest).and_then(|o| measure_row(&o)), + "agent" => Options::parse(&rest, &[flags::COMMON, flags::AGENT_ONLY]) + .and_then(|o| serve(&command, &o)), + "environment" => Options::parse(&rest, &[flags::COMMON, flags::ENVIRONMENT_ONLY]) + .and_then(|o| serve(&command, &o)), + "measure" => Options::parse(&rest, &[flags::MEASURE]).and_then(|o| measure(&o)), + "measure-row" => Options::parse(&rest, &[flags::MEASURE]).and_then(|o| measure_row(&o)), "--help" | "-h" | "help" => { print!("{USAGE}"); return ExitCode::SUCCESS; @@ -92,13 +97,22 @@ pub fn main() -> ExitCode { struct Options(BTreeMap); impl Options { - fn parse(args: &[String]) -> Result { + /// Reads the options of one command, refusing any flag that command does not have. + /// + /// `allowed` comes from [`crate::launcher::flags`], the same constants the launcher + /// writes the argv from. An unknown flag is an error naming it rather than a value that + /// is quietly ignored: a renamed option must fail the launch, not turn into a no-op that + /// no test notices. + fn parse(args: &[String], allowed: &[&[&str]]) -> Result { let mut out = BTreeMap::new(); let mut iter = args.iter(); while let Some(flag) = iter.next() { let Some(name) = flag.strip_prefix("--") else { return Err(format!("expected an option, found {flag:?}")); }; + if !allowed.iter().any(|set| set.contains(&name)) { + return Err(format!("unknown option --{name} for this command")); + } let value = iter .next() .ok_or_else(|| format!("option --{name} needs a value"))?; @@ -158,51 +172,53 @@ impl Options { /// Serves one worker until `Worker.Shutdown`, then exits. fn serve(role: &str, options: &Options) -> Result<(), String> { - let socket = options.path("socket")?; - let store_root = options.path("store-root")?; - let client_id = options.required("client-id")?.to_owned(); - let service = options.required("service")?.to_owned(); - let threads = options.usize("threads", 1)?; + let socket = options.path(flags::SOCKET)?; + let store_root = options.path(flags::STORE_ROOT)?; + let client_id = options.required(flags::CLIENT_ID)?.to_owned(); + let service = options.required(flags::SERVICE)?.to_owned(); + let threads = options.usize(flags::THREADS, 1)?; if threads == 0 { return Err("--threads must be at least 1".to_owned()); } - let session_id = options.id("session")?; - let incarnation_id = options.id("incarnation")?; + let session_id = options.id(flags::SESSION)?; + let incarnation_id = options.id(flags::INCARNATION)?; let what = match role { "agent" => Started::Agent(AgentLaunch { session_id, - agent_id: options.id("agent")?, - port_id: options.id("port")?, + agent_id: options.id(flags::AGENT)?, + port_id: options.id(flags::PORT)?, incarnation_id, - tick_duration: options.rational("tick-numerator", "tick-denominator")?, - warmup_ticks: options.u64("warmup-ticks", 0)?, + tick_duration: options.rational(flags::TICK_NUMERATOR, flags::TICK_DENOMINATOR)?, + warmup_ticks: options.u64(flags::WARMUP_TICKS, 0)?, worker_threads: threads, // This process's own log. The supervisor reads what crosses the bus, not this. sensors: crate::media::SensorLog::new(), faults: AgentFaults { - fail_commit_at_step: options.opt_u64("fail-commit-at-step")?, - prepare_delay_ms: options.u64("prepare-delay-ms", 0)?, - commit_delay_ms: options.u64("commit-delay-ms", 0)?, + fail_commit_at_step: options.opt_u64(flags::FAIL_COMMIT_AT_STEP)?, + prepare_delay_ms: options.u64(flags::PREPARE_DELAY_MS, 0)?, + commit_delay_ms: options.u64(flags::COMMIT_DELAY_MS, 0)?, }, client_id: client_id.clone(), service: service.clone(), }), _ => Started::Environment(EnvironmentLaunch { session_id, - worker_id: options.id("worker")?, + worker_id: options.id(flags::WORKER)?, incarnation_id, - step_duration: options.rational("step-numerator", "step-denominator")?, - ports: parse_ports(options.required("ports")?)?, + step_duration: options.rational(flags::STEP_NUMERATOR, flags::STEP_DENOMINATOR)?, + ports: parse_ports(options.required(flags::PORTS)?)?, worker_threads: threads, - observation_delay_steps: options.u64("observation-delay-steps", 0)?, + observation_delay_steps: options.u64(flags::OBSERVATION_DELAY_STEPS, 0)?, renders: crate::media::RenderCounter::new(), faults: EnvironmentFaults { - advance_delay_ms: options.u64("advance-delay-ms", 0)?, - omit_view_at_boundary: options.opt_u64("omit-view-at-boundary")?, - stale_view_at_boundary: options.opt_u64("stale-view-at-boundary")?, - truncated_view_at_boundary: options.opt_u64("truncated-view-at-boundary")?, - omit_audio_at_boundary: options.opt_u64("omit-audio-at-boundary")?, - overlapping_audio_at_boundary: options.opt_u64("overlapping-audio-at-boundary")?, + advance_delay_ms: options.u64(flags::ADVANCE_DELAY_MS, 0)?, + omit_view_at_boundary: options.opt_u64(flags::OMIT_VIEW_AT_BOUNDARY)?, + stale_view_at_boundary: options.opt_u64(flags::STALE_VIEW_AT_BOUNDARY)?, + truncated_view_at_boundary: options + .opt_u64(flags::TRUNCATED_VIEW_AT_BOUNDARY)?, + omit_audio_at_boundary: options.opt_u64(flags::OMIT_AUDIO_AT_BOUNDARY)?, + overlapping_audio_at_boundary: options + .opt_u64(flags::OVERLAPPING_AUDIO_AT_BOUNDARY)?, }, client_id: client_id.clone(), service: service.clone(), @@ -288,3 +304,34 @@ fn measure_row(options: &Options) -> Result<(), String> { println!("{}", row.to_json()); Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + /// An unknown flag is refused by name. A renamed option must fail the launch rather than + /// be accepted and ignored, which would turn a fault or a delay into a no-op. + #[test] + fn an_unknown_option_is_refused_by_name() { + let args: Vec = ["--session", "demo", "--stale-view-at-boundry", "2"] + .iter() + .map(|s| (*s).to_owned()) + .collect(); + let error = Options::parse(&args, &[flags::COMMON, flags::ENVIRONMENT_ONLY]) + .expect_err("an unknown option is refused"); + assert!(error.contains("--stale-view-at-boundry"), "{error}"); + } + + /// A flag that belongs to another command is refused too: an agent has no render delay. + #[test] + fn an_option_of_another_command_is_refused() { + let args: Vec = ["--observation-delay-steps", "2"] + .iter() + .map(|s| (*s).to_owned()) + .collect(); + Options::parse(&args, &[flags::COMMON, flags::AGENT_ONLY]) + .expect_err("the agent command has no render delay"); + Options::parse(&args, &[flags::COMMON, flags::ENVIRONMENT_ONLY]) + .expect("the environment command does"); + } +} diff --git a/services/flysim/crates/fly-session/src/launcher.rs b/services/flysim/crates/fly-session/src/launcher.rs index b0097ed..c80551e 100644 --- a/services/flysim/crates/fly-session/src/launcher.rs +++ b/services/flysim/crates/fly-session/src/launcher.rs @@ -1204,6 +1204,97 @@ fn launch_error(worker_id: &Id, e: &flybus::BusError) -> DomainError { ) } + +// ------------------------------------------------------------------------------------------- +// The command line between a launcher and a participant in its own process + +/// Every `--flag` a launched participant or a measurement child accepts, named once. +/// +/// The launcher writes these and [`crate::cli`] reads them. Naming them in two places by +/// convention would let a rename turn an option into a silent no-op -- a fault that never +/// fires, a delay that is never applied -- so both sides use these constants and the parser +/// refuses any flag outside the set for the command it is parsing. +pub(crate) mod flags { + pub const SOCKET: &str = "socket"; + pub const STORE_ROOT: &str = "store-root"; + pub const CLIENT_ID: &str = "client-id"; + pub const SERVICE: &str = "service"; + pub const THREADS: &str = "threads"; + pub const SESSION: &str = "session"; + pub const INCARNATION: &str = "incarnation"; + + pub const AGENT: &str = "agent"; + pub const PORT: &str = "port"; + pub const TICK_NUMERATOR: &str = "tick-numerator"; + pub const TICK_DENOMINATOR: &str = "tick-denominator"; + pub const WARMUP_TICKS: &str = "warmup-ticks"; + pub const PREPARE_DELAY_MS: &str = "prepare-delay-ms"; + pub const COMMIT_DELAY_MS: &str = "commit-delay-ms"; + pub const FAIL_COMMIT_AT_STEP: &str = "fail-commit-at-step"; + + pub const WORKER: &str = "worker"; + pub const PORTS: &str = "ports"; + pub const STEP_NUMERATOR: &str = "step-numerator"; + pub const STEP_DENOMINATOR: &str = "step-denominator"; + pub const ADVANCE_DELAY_MS: &str = "advance-delay-ms"; + pub const OMIT_VIEW_AT_BOUNDARY: &str = "omit-view-at-boundary"; + pub const OBSERVATION_DELAY_STEPS: &str = "observation-delay-steps"; + pub const STALE_VIEW_AT_BOUNDARY: &str = "stale-view-at-boundary"; + pub const TRUNCATED_VIEW_AT_BOUNDARY: &str = "truncated-view-at-boundary"; + pub const OMIT_AUDIO_AT_BOUNDARY: &str = "omit-audio-at-boundary"; + pub const OVERLAPPING_AUDIO_AT_BOUNDARY: &str = "overlapping-audio-at-boundary"; + + pub const MODE: &str = "mode"; + pub const AGENTS: &str = "agents"; + pub const STEPS: &str = "steps"; + pub const WARMUP_STEPS: &str = "warmup-steps"; + pub const WORKER_THREADS: &str = "worker-threads"; + pub const MODES: &str = "modes"; + + /// What every launched worker is given. + pub const COMMON: &[&str] = &[ + SOCKET, + STORE_ROOT, + CLIENT_ID, + SERVICE, + THREADS, + SESSION, + INCARNATION, + ]; + /// What only an agent is given. + pub const AGENT_ONLY: &[&str] = &[ + AGENT, + PORT, + TICK_NUMERATOR, + TICK_DENOMINATOR, + WARMUP_TICKS, + PREPARE_DELAY_MS, + COMMIT_DELAY_MS, + FAIL_COMMIT_AT_STEP, + ]; + /// What only the environment is given, media options included. + pub const ENVIRONMENT_ONLY: &[&str] = &[ + WORKER, + PORTS, + STEP_NUMERATOR, + STEP_DENOMINATOR, + ADVANCE_DELAY_MS, + OMIT_VIEW_AT_BOUNDARY, + OBSERVATION_DELAY_STEPS, + STALE_VIEW_AT_BOUNDARY, + TRUNCATED_VIEW_AT_BOUNDARY, + OMIT_AUDIO_AT_BOUNDARY, + OVERLAPPING_AUDIO_AT_BOUNDARY, + ]; + /// What a measurement run or one of its row children is given. + pub const MEASURE: &[&str] = &[MODE, AGENTS, STEPS, WARMUP_STEPS, WORKER_THREADS, MODES]; +} + +/// One `--flag value` pair, so the flag name is written once and never spelled inline. +fn arg(name: &str, value: impl std::fmt::Display) -> (String, String) { + (format!("--{name}"), value.to_string()) +} + // ------------------------------------------------------------------------------------------- // What a participant is @@ -1227,66 +1318,49 @@ impl Started { match self { Started::Agent(spec) => { let mut args = vec![ - ("--session".to_owned(), spec.session_id.clone()), - ("--agent".to_owned(), spec.agent_id.clone()), - ("--port".to_owned(), spec.port_id.clone()), - ("--incarnation".to_owned(), spec.incarnation_id.clone()), - ("--tick-numerator".to_owned(), spec.tick_duration.numerator.to_string()), - ( - "--tick-denominator".to_owned(), - spec.tick_duration.denominator.to_string(), - ), - ("--warmup-ticks".to_owned(), spec.warmup_ticks.to_string()), - ( - "--prepare-delay-ms".to_owned(), - spec.faults.prepare_delay_ms.to_string(), - ), - ( - "--commit-delay-ms".to_owned(), - spec.faults.commit_delay_ms.to_string(), - ), + arg(flags::SESSION, &spec.session_id), + arg(flags::AGENT, &spec.agent_id), + arg(flags::PORT, &spec.port_id), + arg(flags::INCARNATION, &spec.incarnation_id), + arg(flags::TICK_NUMERATOR, spec.tick_duration.numerator), + arg(flags::TICK_DENOMINATOR, spec.tick_duration.denominator), + arg(flags::WARMUP_TICKS, spec.warmup_ticks), + arg(flags::PREPARE_DELAY_MS, spec.faults.prepare_delay_ms), + arg(flags::COMMIT_DELAY_MS, spec.faults.commit_delay_ms), ]; if let Some(step) = spec.faults.fail_commit_at_step { - args.push(("--fail-commit-at-step".to_owned(), step.to_string())); + args.push(arg(flags::FAIL_COMMIT_AT_STEP, step)); } args } Started::Environment(spec) => { let mut args = vec![ - ("--session".to_owned(), spec.session_id.clone()), - ("--worker".to_owned(), spec.worker_id.clone()), - ("--incarnation".to_owned(), spec.incarnation_id.clone()), - ("--step-numerator".to_owned(), spec.step_duration.numerator.to_string()), - ( - "--step-denominator".to_owned(), - spec.step_duration.denominator.to_string(), - ), - ("--ports".to_owned(), spec.ports.join(",")), - ( - "--advance-delay-ms".to_owned(), - spec.faults.advance_delay_ms.to_string(), - ), + arg(flags::SESSION, &spec.session_id), + arg(flags::WORKER, &spec.worker_id), + arg(flags::INCARNATION, &spec.incarnation_id), + arg(flags::STEP_NUMERATOR, spec.step_duration.numerator), + arg(flags::STEP_DENOMINATOR, spec.step_duration.denominator), + arg(flags::PORTS, spec.ports.join(",")), + arg(flags::ADVANCE_DELAY_MS, spec.faults.advance_delay_ms), + // The media options a world in another process needs to be exactly this + // world. Its render counter and its agents' sensor logs stay there. + arg(flags::OBSERVATION_DELAY_STEPS, spec.observation_delay_steps), ]; - if let Some(boundary) = spec.faults.omit_view_at_boundary { - args.push(("--omit-view-at-boundary".to_owned(), boundary.to_string())); - } - // The media options a world in another process needs to be exactly this - // world. Its render counter and its agents' sensor logs stay in that process. - args.push(( - "--observation-delay-steps".to_owned(), - spec.observation_delay_steps.to_string(), - )); for (flag, boundary) in [ - ("--stale-view-at-boundary", spec.faults.stale_view_at_boundary), - ("--truncated-view-at-boundary", spec.faults.truncated_view_at_boundary), - ("--omit-audio-at-boundary", spec.faults.omit_audio_at_boundary), + (flags::OMIT_VIEW_AT_BOUNDARY, spec.faults.omit_view_at_boundary), + (flags::STALE_VIEW_AT_BOUNDARY, spec.faults.stale_view_at_boundary), ( - "--overlapping-audio-at-boundary", + flags::TRUNCATED_VIEW_AT_BOUNDARY, + spec.faults.truncated_view_at_boundary, + ), + (flags::OMIT_AUDIO_AT_BOUNDARY, spec.faults.omit_audio_at_boundary), + ( + flags::OVERLAPPING_AUDIO_AT_BOUNDARY, spec.faults.overlapping_audio_at_boundary, ), ] { if let Some(boundary) = boundary { - args.push((flag.to_owned(), boundary.to_string())); + args.push(arg(flag, boundary)); } } args @@ -1371,3 +1445,98 @@ pub(crate) async fn register_with_retry( } } } + +#[cfg(test)] +mod flag_tests { + use super::*; + + fn every_media_fault() -> EnvironmentFaults { + EnvironmentFaults { + advance_delay_ms: 3, + omit_view_at_boundary: Some(1), + stale_view_at_boundary: Some(2), + truncated_view_at_boundary: Some(3), + omit_audio_at_boundary: Some(4), + overlapping_audio_at_boundary: Some(5), + } + } + + fn environment_launch() -> EnvironmentLaunch { + EnvironmentLaunch { + session_id: id("demo"), + worker_id: id("arena"), + incarnation_id: id("arena-inc-1"), + step_duration: RationalNs::new(1, 60).expect("a cadence"), + ports: vec![id("p1"), id("p2")], + worker_threads: 1, + observation_delay_steps: 2, + renders: crate::media::RenderCounter::new(), + faults: every_media_fault(), + client_id: "environment".to_owned(), + service: "env.arena".to_owned(), + } + } + + fn agent_launch() -> AgentLaunch { + AgentLaunch { + session_id: id("demo"), + agent_id: id("fly-a"), + port_id: id("p1"), + incarnation_id: id("fly-a-inc-1"), + tick_duration: RationalNs::new(1, 1_000).expect("a tick"), + warmup_ticks: 10, + worker_threads: 1, + sensors: crate::media::SensorLog::new(), + faults: AgentFaults { + fail_commit_at_step: Some(2), + prepare_delay_ms: 1, + commit_delay_ms: 2, + }, + client_id: "worker-fly-a".to_owned(), + service: "agent.fly-a".to_owned(), + } + } + + /// Both halves of the command line name the same constants, and this proves it for every + /// argument a launch can produce: a flag the launcher writes that the parser does not + /// accept would be a silently ignored option, which is what the parser now refuses. + #[test] + fn every_flag_a_launch_writes_is_one_its_command_accepts() { + for (started, allowed) in [ + ( + Started::Environment(environment_launch()), + [flags::COMMON, flags::ENVIRONMENT_ONLY], + ), + (Started::Agent(agent_launch()), [flags::COMMON, flags::AGENT_ONLY]), + ] { + let arguments = started.arguments(); + assert!(!arguments.is_empty()); + for (flag, _value) in &arguments { + let name = flag.strip_prefix("--").expect("every argument is a --flag"); + assert!( + allowed.iter().any(|set| set.contains(&name)), + "{} writes --{name}, which its command does not accept", + started.subcommand() + ); + } + } + } + + /// Every media option reaches the argv when it is set, so a world in another process is + /// exactly the world the composition asked for. + #[test] + fn the_media_options_are_all_written_for_a_separate_process() { + let started = Started::Environment(environment_launch()); + let written: Vec = started.arguments().into_iter().map(|(f, _)| f).collect(); + for flag in [ + flags::OBSERVATION_DELAY_STEPS, + flags::OMIT_VIEW_AT_BOUNDARY, + flags::STALE_VIEW_AT_BOUNDARY, + flags::TRUNCATED_VIEW_AT_BOUNDARY, + flags::OMIT_AUDIO_AT_BOUNDARY, + flags::OVERLAPPING_AUDIO_AT_BOUNDARY, + ] { + assert!(written.contains(&format!("--{flag}")), "--{flag} is not written"); + } + } +} diff --git a/services/flysim/crates/fly-session/src/measure.rs b/services/flysim/crates/fly-session/src/measure.rs index 963139f..ed79c6a 100644 --- a/services/flysim/crates/fly-session/src/measure.rs +++ b/services/flysim/crates/fly-session/src/measure.rs @@ -24,7 +24,7 @@ use serde_json::{Value, json}; use crate::coordinator::DispatchOrder; use crate::harness::{AgentSpec, HarnessConfig, SessionHarness, Via}; -use crate::launcher::ExecutionMode; +use crate::launcher::{ExecutionMode, flags}; use crate::metrics::{Percentiles, physical_cores}; /// What to compare. @@ -241,15 +241,15 @@ fn row_in_a_child( ) -> Result { let output = std::process::Command::new(program) .arg("measure-row") - .arg("--mode") + .arg(format!("--{}", flags::MODE)) .arg(mode.label()) - .arg("--agents") + .arg(format!("--{}", flags::AGENTS)) .arg(agents.to_string()) - .arg("--steps") + .arg(format!("--{}", flags::STEPS)) .arg(config.steps.to_string()) - .arg("--warmup-steps") + .arg(format!("--{}", flags::WARMUP_STEPS)) .arg(config.warmup_steps.to_string()) - .arg("--worker-threads") + .arg(format!("--{}", flags::WORKER_THREADS)) .arg(config.worker_threads.to_string()) .stdin(std::process::Stdio::null()) .output() diff --git a/services/flysim/crates/fly-session/tests/media.rs b/services/flysim/crates/fly-session/tests/media.rs index 6315627..220221c 100644 --- a/services/flysim/crates/fly-session/tests/media.rs +++ b/services/flysim/crates/fly-session/tests/media.rs @@ -45,7 +45,10 @@ both_transports!( a_cadence_that_does_not_divide_the_sample_rate_still_lands_on_whole_samples, ); -all_modes!(the_media_path_works_in_every_execution_mode); +all_modes!( + the_media_path_works_in_every_execution_mode, + every_media_fault_fires_in_every_execution_mode, +); const STEPS: u64 = 3; @@ -606,6 +609,77 @@ async fn the_media_path_works_in_every_execution_mode(mode: ExecutionMode) { f.shutdown().await; } +/// Every media fault fires wherever the world runs, which is what proves the wiring rather +/// than assuming it. +/// +/// A fault reaches a world in another process only as a `--flag` on its command line, so a +/// renamed or dropped flag would make these faults silently stop firing. Here each one has to +/// fail its transition in every execution mode; the parser refuses an unknown flag, so a +/// mismatch fails the launch instead of turning into a no-op. +async fn every_media_fault_fires_in_every_execution_mode(mode: ExecutionMode) { + let cases: [(&str, EnvironmentFaults, u64); 4] = [ + ( + "an extra-delayed view", + EnvironmentFaults { + stale_view_at_boundary: Some(2), + ..EnvironmentFaults::default() + }, + 2, + ), + ( + "a frame of the wrong length", + EnvironmentFaults { + truncated_view_at_boundary: Some(1), + ..EnvironmentFaults::default() + }, + 1, + ), + ( + "a missing audio chunk", + EnvironmentFaults { + omit_audio_at_boundary: Some(2), + ..EnvironmentFaults::default() + }, + 2, + ), + ( + "an overlapping audio chunk", + EnvironmentFaults { + overlapping_audio_at_boundary: Some(2), + ..EnvironmentFaults::default() + }, + 2, + ), + ]; + for (what, faults, steps) in cases { + let config = HarnessConfig { + environment_faults: faults, + ..HarnessConfig::default() + }; + let mut f = mode_fixture(mode, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let outcome = within("run", f.harness.coordinator.run(steps)).await; + let failure = match outcome { + Err(failure) => failure, + Ok(reports) => { + panic!("{what} did not fail the transition in {mode:?} mode: {reports:?}") + } + }; + assert_eq!( + failure.error.code, + ErrorCode::BufferInvalid, + "{what} in {mode:?} mode: {failure}" + ); + assert_eq!(f.harness.coordinator.phase(), Phase::Failed); + assert_eq!( + f.harness.coordinator.stats().advances, + steps - 1, + "{what} in {mode:?} mode committed the boundaries before it and no more" + ); + f.shutdown().await; + } +} + /// The retention table: a required agent input is retained through encoding and Commit with no /// coalescing, while a spectator's snapshots are a latest subscription with finite credits. async fn required_agent_input_is_never_coalesced_while_spectator_snapshots_are(via: Via) {