session: an unknown launch option is an error, not a no-op

The command line between a launcher and a participant in its own process
was coupled by string convention on both sides and checked by neither: the
parser accepted any --flag it did not know, so a renamed or dropped option
would have become a fault that never fires or a delay that is never
applied, with nothing failing.

Every flag now has one name, in launcher::flags, written by the launcher's
argv and read by the parser. Options::parse takes the sets its command
allows and refuses anything outside them by name, so a mismatch fails the
launch. measure and measure-row use the same constants for their own
flags.

Three tests hold the two sides together: every flag a launch writes is one
its command accepts, every media option is written for a separate process,
and an unknown or wrong-command flag is refused by name. The four media
faults now also run end to end in all three execution modes, each failing
its own transition, so the wiring across a process boundary is proved
rather than assumed.
This commit is contained in:
acamilo 2026-09-22 16:28:59 +00:00
parent f456fe9522
commit 66ec2e5c6b
4 changed files with 372 additions and 82 deletions

View file

@ -21,7 +21,9 @@ use std::process::ExitCode;
use crate::agent::AgentFaults; use crate::agent::AgentFaults;
use crate::environment::EnvironmentFaults; 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::*; use crate::types::*;
const USAGE: &str = "\ const USAGE: &str = "\
@ -69,9 +71,12 @@ pub fn main() -> ExitCode {
let command = command.to_string_lossy().into_owned(); let command = command.to_string_lossy().into_owned();
let rest: Vec<String> = args.map(|a| a.to_string_lossy().into_owned()).collect(); let rest: Vec<String> = args.map(|a| a.to_string_lossy().into_owned()).collect();
let result = match command.as_str() { let result = match command.as_str() {
"agent" | "environment" => Options::parse(&rest).and_then(|o| serve(&command, &o)), "agent" => Options::parse(&rest, &[flags::COMMON, flags::AGENT_ONLY])
"measure" => Options::parse(&rest).and_then(|o| measure(&o)), .and_then(|o| serve(&command, &o)),
"measure-row" => Options::parse(&rest).and_then(|o| measure_row(&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" => { "--help" | "-h" | "help" => {
print!("{USAGE}"); print!("{USAGE}");
return ExitCode::SUCCESS; return ExitCode::SUCCESS;
@ -92,13 +97,22 @@ pub fn main() -> ExitCode {
struct Options(BTreeMap<String, String>); struct Options(BTreeMap<String, String>);
impl Options { impl Options {
fn parse(args: &[String]) -> Result<Options, String> { /// 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<Options, String> {
let mut out = BTreeMap::new(); let mut out = BTreeMap::new();
let mut iter = args.iter(); let mut iter = args.iter();
while let Some(flag) = iter.next() { while let Some(flag) = iter.next() {
let Some(name) = flag.strip_prefix("--") else { let Some(name) = flag.strip_prefix("--") else {
return Err(format!("expected an option, found {flag:?}")); 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 let value = iter
.next() .next()
.ok_or_else(|| format!("option --{name} needs a value"))?; .ok_or_else(|| format!("option --{name} needs a value"))?;
@ -158,51 +172,53 @@ impl Options {
/// Serves one worker until `Worker.Shutdown`, then exits. /// Serves one worker until `Worker.Shutdown`, then exits.
fn serve(role: &str, options: &Options) -> Result<(), String> { fn serve(role: &str, options: &Options) -> Result<(), String> {
let socket = options.path("socket")?; let socket = options.path(flags::SOCKET)?;
let store_root = options.path("store-root")?; let store_root = options.path(flags::STORE_ROOT)?;
let client_id = options.required("client-id")?.to_owned(); let client_id = options.required(flags::CLIENT_ID)?.to_owned();
let service = options.required("service")?.to_owned(); let service = options.required(flags::SERVICE)?.to_owned();
let threads = options.usize("threads", 1)?; let threads = options.usize(flags::THREADS, 1)?;
if threads == 0 { if threads == 0 {
return Err("--threads must be at least 1".to_owned()); return Err("--threads must be at least 1".to_owned());
} }
let session_id = options.id("session")?; let session_id = options.id(flags::SESSION)?;
let incarnation_id = options.id("incarnation")?; let incarnation_id = options.id(flags::INCARNATION)?;
let what = match role { let what = match role {
"agent" => Started::Agent(AgentLaunch { "agent" => Started::Agent(AgentLaunch {
session_id, session_id,
agent_id: options.id("agent")?, agent_id: options.id(flags::AGENT)?,
port_id: options.id("port")?, port_id: options.id(flags::PORT)?,
incarnation_id, incarnation_id,
tick_duration: options.rational("tick-numerator", "tick-denominator")?, tick_duration: options.rational(flags::TICK_NUMERATOR, flags::TICK_DENOMINATOR)?,
warmup_ticks: options.u64("warmup-ticks", 0)?, warmup_ticks: options.u64(flags::WARMUP_TICKS, 0)?,
worker_threads: threads, worker_threads: threads,
// This process's own log. The supervisor reads what crosses the bus, not this. // This process's own log. The supervisor reads what crosses the bus, not this.
sensors: crate::media::SensorLog::new(), sensors: crate::media::SensorLog::new(),
faults: AgentFaults { faults: AgentFaults {
fail_commit_at_step: options.opt_u64("fail-commit-at-step")?, fail_commit_at_step: options.opt_u64(flags::FAIL_COMMIT_AT_STEP)?,
prepare_delay_ms: options.u64("prepare-delay-ms", 0)?, prepare_delay_ms: options.u64(flags::PREPARE_DELAY_MS, 0)?,
commit_delay_ms: options.u64("commit-delay-ms", 0)?, commit_delay_ms: options.u64(flags::COMMIT_DELAY_MS, 0)?,
}, },
client_id: client_id.clone(), client_id: client_id.clone(),
service: service.clone(), service: service.clone(),
}), }),
_ => Started::Environment(EnvironmentLaunch { _ => Started::Environment(EnvironmentLaunch {
session_id, session_id,
worker_id: options.id("worker")?, worker_id: options.id(flags::WORKER)?,
incarnation_id, incarnation_id,
step_duration: options.rational("step-numerator", "step-denominator")?, step_duration: options.rational(flags::STEP_NUMERATOR, flags::STEP_DENOMINATOR)?,
ports: parse_ports(options.required("ports")?)?, ports: parse_ports(options.required(flags::PORTS)?)?,
worker_threads: threads, 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(), renders: crate::media::RenderCounter::new(),
faults: EnvironmentFaults { faults: EnvironmentFaults {
advance_delay_ms: options.u64("advance-delay-ms", 0)?, advance_delay_ms: options.u64(flags::ADVANCE_DELAY_MS, 0)?,
omit_view_at_boundary: options.opt_u64("omit-view-at-boundary")?, omit_view_at_boundary: options.opt_u64(flags::OMIT_VIEW_AT_BOUNDARY)?,
stale_view_at_boundary: options.opt_u64("stale-view-at-boundary")?, stale_view_at_boundary: options.opt_u64(flags::STALE_VIEW_AT_BOUNDARY)?,
truncated_view_at_boundary: options.opt_u64("truncated-view-at-boundary")?, truncated_view_at_boundary: options
omit_audio_at_boundary: options.opt_u64("omit-audio-at-boundary")?, .opt_u64(flags::TRUNCATED_VIEW_AT_BOUNDARY)?,
overlapping_audio_at_boundary: options.opt_u64("overlapping-audio-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(), client_id: client_id.clone(),
service: service.clone(), service: service.clone(),
@ -288,3 +304,34 @@ fn measure_row(options: &Options) -> Result<(), String> {
println!("{}", row.to_json()); println!("{}", row.to_json());
Ok(()) 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<String> = ["--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<String> = ["--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");
}
}

View file

@ -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 // What a participant is
@ -1227,66 +1318,49 @@ impl Started {
match self { match self {
Started::Agent(spec) => { Started::Agent(spec) => {
let mut args = vec![ let mut args = vec![
("--session".to_owned(), spec.session_id.clone()), arg(flags::SESSION, &spec.session_id),
("--agent".to_owned(), spec.agent_id.clone()), arg(flags::AGENT, &spec.agent_id),
("--port".to_owned(), spec.port_id.clone()), arg(flags::PORT, &spec.port_id),
("--incarnation".to_owned(), spec.incarnation_id.clone()), arg(flags::INCARNATION, &spec.incarnation_id),
("--tick-numerator".to_owned(), spec.tick_duration.numerator.to_string()), arg(flags::TICK_NUMERATOR, spec.tick_duration.numerator),
( arg(flags::TICK_DENOMINATOR, spec.tick_duration.denominator),
"--tick-denominator".to_owned(), arg(flags::WARMUP_TICKS, spec.warmup_ticks),
spec.tick_duration.denominator.to_string(), arg(flags::PREPARE_DELAY_MS, spec.faults.prepare_delay_ms),
), arg(flags::COMMIT_DELAY_MS, spec.faults.commit_delay_ms),
("--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(),
),
]; ];
if let Some(step) = spec.faults.fail_commit_at_step { 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 args
} }
Started::Environment(spec) => { Started::Environment(spec) => {
let mut args = vec![ let mut args = vec![
("--session".to_owned(), spec.session_id.clone()), arg(flags::SESSION, &spec.session_id),
("--worker".to_owned(), spec.worker_id.clone()), arg(flags::WORKER, &spec.worker_id),
("--incarnation".to_owned(), spec.incarnation_id.clone()), arg(flags::INCARNATION, &spec.incarnation_id),
("--step-numerator".to_owned(), spec.step_duration.numerator.to_string()), arg(flags::STEP_NUMERATOR, spec.step_duration.numerator),
( arg(flags::STEP_DENOMINATOR, spec.step_duration.denominator),
"--step-denominator".to_owned(), arg(flags::PORTS, spec.ports.join(",")),
spec.step_duration.denominator.to_string(), arg(flags::ADVANCE_DELAY_MS, spec.faults.advance_delay_ms),
),
("--ports".to_owned(), spec.ports.join(",")),
(
"--advance-delay-ms".to_owned(),
spec.faults.advance_delay_ms.to_string(),
),
];
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 // 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. // world. Its render counter and its agents' sensor logs stay there.
args.push(( arg(flags::OBSERVATION_DELAY_STEPS, spec.observation_delay_steps),
"--observation-delay-steps".to_owned(), ];
spec.observation_delay_steps.to_string(),
));
for (flag, boundary) in [ for (flag, boundary) in [
("--stale-view-at-boundary", spec.faults.stale_view_at_boundary), (flags::OMIT_VIEW_AT_BOUNDARY, spec.faults.omit_view_at_boundary),
("--truncated-view-at-boundary", spec.faults.truncated_view_at_boundary), (flags::STALE_VIEW_AT_BOUNDARY, spec.faults.stale_view_at_boundary),
("--omit-audio-at-boundary", spec.faults.omit_audio_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, spec.faults.overlapping_audio_at_boundary,
), ),
] { ] {
if let Some(boundary) = boundary { if let Some(boundary) = boundary {
args.push((flag.to_owned(), boundary.to_string())); args.push(arg(flag, boundary));
} }
} }
args 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<String> = 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");
}
}
}

View file

@ -24,7 +24,7 @@ use serde_json::{Value, json};
use crate::coordinator::DispatchOrder; use crate::coordinator::DispatchOrder;
use crate::harness::{AgentSpec, HarnessConfig, SessionHarness, Via}; use crate::harness::{AgentSpec, HarnessConfig, SessionHarness, Via};
use crate::launcher::ExecutionMode; use crate::launcher::{ExecutionMode, flags};
use crate::metrics::{Percentiles, physical_cores}; use crate::metrics::{Percentiles, physical_cores};
/// What to compare. /// What to compare.
@ -241,15 +241,15 @@ fn row_in_a_child(
) -> Result<Row, String> { ) -> Result<Row, String> {
let output = std::process::Command::new(program) let output = std::process::Command::new(program)
.arg("measure-row") .arg("measure-row")
.arg("--mode") .arg(format!("--{}", flags::MODE))
.arg(mode.label()) .arg(mode.label())
.arg("--agents") .arg(format!("--{}", flags::AGENTS))
.arg(agents.to_string()) .arg(agents.to_string())
.arg("--steps") .arg(format!("--{}", flags::STEPS))
.arg(config.steps.to_string()) .arg(config.steps.to_string())
.arg("--warmup-steps") .arg(format!("--{}", flags::WARMUP_STEPS))
.arg(config.warmup_steps.to_string()) .arg(config.warmup_steps.to_string())
.arg("--worker-threads") .arg(format!("--{}", flags::WORKER_THREADS))
.arg(config.worker_threads.to_string()) .arg(config.worker_threads.to_string())
.stdin(std::process::Stdio::null()) .stdin(std::process::Stdio::null())
.output() .output()

View file

@ -45,7 +45,10 @@ both_transports!(
a_cadence_that_does_not_divide_the_sample_rate_still_lands_on_whole_samples, 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; const STEPS: u64 = 3;
@ -606,6 +609,77 @@ async fn the_media_path_works_in_every_execution_mode(mode: ExecutionMode) {
f.shutdown().await; 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 /// 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. /// 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) { async fn required_agent_input_is_never_coalesced_while_spectator_snapshots_are(via: Via) {