From 7b7ebcdf28994c1bb1f47fef1867abfb7cceb363 Mon Sep 17 00:00:00 2001 From: acamilo Date: Tue, 22 Sep 2026 14:33:25 +0000 Subject: [PATCH] session: parallel processes, a launcher and the fault behaviour SESSION-02: one agent process per fly and one environment process under the coordinator over the Unix-socket transport, compared against the in-process composition and a dedicated-thread variant. The mode is the only thing that changes; the composition, the coordinator, the workers and the router are the same code in all three. The launcher is the configured supervisor. It owns a total thread budget with one allocation per participant, refused as BUSY before anything starts when the total cannot cover it; the configured client, service, worker and port identities, proved in Worker.Hello before the coordinator pins a registration; Worker.Status health on the supervisor's own monotonic clock at the ipc-v1 section 6 budgets; and reaping, where Worker.Shutdown is the request and the operating system is the guarantee. The worker executable is a subcommand of this crate's one binary, which is what implementation.md section 2 allows in place of a separate worker crate. The coordinator's fault behaviour: every failure names the participant it is attributed to, every domain call has a caller-side deadline so a dead participant is a diagnosed outcome rather than a hang, and failing fences the epoch -- the boundary stops, the handles drop, and no further transition or publication is allowed. Agent.Initialize now carries the launcher's allocation, and an agent refuses one asking for more. tests/processes.rs proves every acceptance bullet once per execution mode, and the two section 4 rows SESSION-01 could not reach in one process: a router restart during a world advance, and an old worker's reply after a restart. measure compares the three modes at one, two and four agents; its table is in the crate README, and it is not a capacity claim. --- services/flysim/crates/fly-session/Cargo.toml | 11 +- services/flysim/crates/fly-session/README.md | 101 +- .../crates/fly-session/examples/processes.rs | 75 + .../flysim/crates/fly-session/src/agent.rs | 15 + .../crates/fly-session/src/bin/fly-session.rs | 8 + services/flysim/crates/fly-session/src/cli.rs | 251 ++++ .../crates/fly-session/src/coordinator.rs | 230 ++- .../flysim/crates/fly-session/src/harness.rs | 358 +++-- .../flysim/crates/fly-session/src/launcher.rs | 1325 +++++++++++++++++ services/flysim/crates/fly-session/src/lib.rs | 9 +- .../flysim/crates/fly-session/src/measure.rs | 331 ++++ .../flysim/crates/fly-session/src/metrics.rs | 176 +++ .../flysim/crates/fly-session/src/worker.rs | 16 + .../crates/fly-session/tests/common/mod.rs | 54 +- .../crates/fly-session/tests/processes.rs | 600 ++++++++ .../crates/fly-session/tests/session.rs | 7 +- 16 files changed, 3382 insertions(+), 185 deletions(-) create mode 100644 services/flysim/crates/fly-session/examples/processes.rs create mode 100644 services/flysim/crates/fly-session/src/bin/fly-session.rs create mode 100644 services/flysim/crates/fly-session/src/cli.rs create mode 100644 services/flysim/crates/fly-session/src/launcher.rs create mode 100644 services/flysim/crates/fly-session/src/measure.rs create mode 100644 services/flysim/crates/fly-session/src/metrics.rs create mode 100644 services/flysim/crates/fly-session/tests/processes.rs diff --git a/services/flysim/crates/fly-session/Cargo.toml b/services/flysim/crates/fly-session/Cargo.toml index f2f110a..25d9a76 100644 --- a/services/flysim/crates/fly-session/Cargo.toml +++ b/services/flysim/crates/fly-session/Cargo.toml @@ -11,6 +11,13 @@ description = "The lockstep session coordinator, its phase machine and a synthet name = "fly_session" path = "src/lib.rs" +# One binary, one role per subcommand. `implementation.md` section 2 allows worker +# executables to be subcommands of one binary rather than separate crates, and the launcher +# starts this one with `agent` or `environment` for a participant in its own process. +[[bin]] +name = "fly-session" +path = "src/bin/fly-session.rs" + [dependencies] # The domain contract (scalars, payloads, canonical digests, the trace format) and the bus. # Everything else this crate needs is std or Tokio. @@ -18,7 +25,9 @@ fly-session-types = { path = "../fly-session-types" } flybus = { path = "../flybus" } serde_json = { workspace = true } -tokio = { version = "1", features = ["rt", "sync", "time", "macros"] } +# `rt-multi-thread` is not only for the tests: a worker process and a dedicated-thread +# worker each build their own runtime sized to the launcher's thread allocation. +tokio = { version = "1", features = ["rt", "rt-multi-thread", "sync", "time", "macros", "io-util"] } [dev-dependencies] tempfile = "3" diff --git a/services/flysim/crates/fly-session/README.md b/services/flysim/crates/fly-session/README.md index 6821bd6..bc98c78 100644 --- a/services/flysim/crates/fly-session/README.md +++ b/services/flysim/crates/fly-session/README.md @@ -3,10 +3,12 @@ The lockstep session coordinator, its phase machine and a synthetic composition over [`flybus`](../flybus). -This crate is the SESSION-01 slice of the session-framework implementation guide: the -sequential transaction of `step-v1`, driven over the Flybus router, with small fake workers -standing in for a brain and an emulator. It contains no public controller API, no implicit -best-effort retry, no real emulator and no real brain. +This crate is the SESSION-01 and SESSION-02 slices of the session-framework implementation +guide: the transaction of `step-v1`, driven over the Flybus router, with small fake workers +standing in for a brain and an emulator, run either in the coordinator's process, on dedicated +threads, or as one agent process per fly and one environment process under a launcher. It +contains no public controller API, no implicit best-effort retry, no real emulator and no real +brain. The domain scalars, method payloads, their validation, the canonical digests and the trace format all come from [`fly-session-types`](../fly-session-types), the CONTRACT-01 crate. This @@ -38,7 +40,51 @@ Ready(k) ─ Prepare all agents concurrently ─────────── | `task` | The task and executor traits, the deterministic counter task, the identity executor | | `rpc` | Domain calls: `req-` serials, incarnation pinning, the retry rule | | `coordinator` | The transaction, the trace, the failure rules and the publication boundary | -| `harness` | The runnable composition: router, two agents, one arena, one coordinator | +| `launcher` | The supervisor: thread budget, identities, start, health check, reap | +| `metrics` | Latency percentiles and the machine's core and memory counters | +| `measure` | The execution-mode comparison of the guide's section 5 | +| `cli` | The binary's subcommands: `agent`, `environment`, `measure` | +| `harness` | The runnable composition: router, the flies, one arena, one coordinator | + +## Execution modes and the launcher + +A participant runs in one of three places, and the same composition code starts it in any of +them. The separate-process mode is the SESSION-02 subject; the other two are what it is +compared against. + +| Mode | Where each participant runs | Transport | +| --- | --- | --- | +| `InProcess` | A task on the coordinator's runtime | in-memory or Unix socket | +| `Thread` | Its own OS thread, with its own runtime | Unix socket | +| `Process` | Its own process: one per fly, one for the world | Unix socket | + +The launcher is the configured supervisor. It owns four things: + +- **The thread budget.** A total allocation, one slice of it reserved for the coordinator and + its router, and one allocation per participant. A request the total cannot cover is refused + as `BUSY` before anything starts. `Agent.Initialize` carries exactly the allocation the + launcher handed out, and an agent refuses an Initialize asking for more than its own, which + is what `workers-v1` means by "within launcher allocation". +- **Identity.** The bus client id, the service name, the worker id and an agent's port binding + are launcher configuration. The launcher says `Worker.Hello` with the identity it configured + and refuses anything that answers as another worker, role or incarnation -- before the + coordinator has pinned a registration. The registration the coordinator pins is the one that + hello returned, never one that was assumed. +- **Health.** `Worker.Status` on the supervisor's own monotonic clock, with the `ipc-v1` + section 6 prototype budgets: probe at two seconds, fail at ten, a separate budget for boot. + A status answer never waits for a mutation, so a busy participant is still a healthy one. +- **Reaping.** `Worker.Shutdown` is the request and the operating system is the guarantee. A + participant that does not stop inside the budget is terminated, and the supervisor reports + which of the two happened. A launcher that is dropped takes its children with it. + +A separate-process participant is a subcommand of this crate's one binary, which is what +`implementation.md` section 2 allows instead of separate worker crates: + +```sh +fly-session agent --socket S --store-root D --client-id C --service N --threads T ... +fly-session environment --socket S --store-root D --client-id C --service N --threads T ... +fly-session measure --steps 300 --agents 1,2,4 +``` ## What it implements @@ -62,6 +108,14 @@ Ready(k) ─ Prepare all agents concurrently ─────────── - **The failure rules.** A partial commit fails the epoch; an uncertain Advance is resolved against its original domain request id and never becomes a second batch; a worker incarnation change invalidates the epoch. +- **A failure stops the epoch rather than neutralising a player.** Every failure carries the + participant it is attributed to, and failing fences the session: the committed boundary + stops moving, the artifact handles are dropped, and no further transition or publication is + allowed. Lifting the fence is a coherent group restore, which is STATE-01's. +- **A bounded diagnosed outcome.** A caller-side deadline on every domain call, on the + coordinator's own clock, so a participant that dies or stops answering produces a typed + failure naming it rather than a hang. An expired deadline is `unknown`, never `none`: a + caller-side timeout is not evidence that nothing was mutated. - **Domain deduplication over bus calls.** Same key, request and body replays its cached reply with fresh delivery ownership over retained artifacts; a changed body is `CONFLICT`; a duplicate of a running operation is `IN_PROGRESS` for that bus call while the original @@ -125,17 +179,41 @@ harness.shutdown().await; - **No state methods.** `State.Capture`, `State.StageRestore` and `State.ActivateRestore` are STATE-01. The phase machine has their edges (`Capturing`, `Restoring`) and the workers do not advertise them as implemented methods. -- **One process.** SESSION-01 runs every participant in one process over the same router. - SESSION-02 is the per-fly process split. - **No audience input.** The admitted pre-step stimulation list exists and is always empty. - **Pacing is coarse.** The pacing deadline rounds one step to whole nanoseconds for sleeping only; simulation time stays rational and that rounding never re-enters the accumulator. +## Measurements + +`fly-session measure` runs the same composition in each mode at one, two and four agents and +reports the thread allocation, the RPC and critical-path percentiles, the memory peaks and the +router's owner, collection and queue counters. **These are local synthetic timings on one +machine and no host capacity claim follows from any of them**; they exist so the three modes +can be compared with each other. Pacing is off for the run, so the samples are work rather +than sleep, and the run report carries the full table. + +What the numbers said on a four-core development box, at 300 transitions per row: + +- A process boundary costs little at the median and shows up in the tail. Two agents: the + critical path was about 7.8 ms p50 in-process, 8.6 ms on threads and 12.7 ms across + processes, while p99 went 12.4 / 12.7 / 26.0 ms. The medians are within a small multiple of + each other; the tails are where a scheduler with more runnable threads than cores appears. +- Four agents needs six threads, which that box does not have, and every mode's tail widens + together. That is the budget being honest, not a property of the process split. +- Memory is the clearest difference: one coordinator at about 14 MiB peak RSS plus roughly + 5.6 MiB per participant process, against a single 11 MiB process for the threaded variant. +- Ownership and queues stayed bounded in every mode and at every agent count: at most 15 live + owners, 11 artifact roots and one queue entry per agent, with the store holding two sealed + frames and 128 bytes at rest. Of 311 frames produced, 309 were collected -- the current and + previous boundary are the two that are still owned. + ## Tests ```text -cargo test -p fly-session # unit + both integration suites +cargo test -p fly-session # unit + all three integration suites cargo run -p fly-session --example session # the runnable synthetic session +cargo build -p fly-session --bin fly-session # the worker binary the launcher starts +cargo run -p fly-session --example processes # the same session in all three modes ``` Every integration test runs over both transports, through the same router code: all but one @@ -150,6 +228,13 @@ because it compares their behaviour traces against each other. transition that just ended; a terminal episode pausing at its own boundary; `Worker.Status` during a session; and sequential, concurrent and reversed dispatch producing one behaviour trace. +- `tests/processes.rs`: the SESSION-02 acceptance bullets, each generated once per execution + mode -- a delayed one-agent result holding the world, a worker or helper death with a + bounded diagnosed outcome, an uncertain Advance that creates no second batch, a partial + Commit that permits no next-step play, supervision and identity, and the launcher thread + allocation -- plus the sequential/reversed/parallel trace comparison across all three modes + and the two process-mode section 4 rows: a router restart during a world advance, and an old + worker's reply after a restart. - `tests/failures.rs`: a duplicate Prepare after a lost reply; a duplicate Commit; the same batch with altered controls; a lost Advance result; a cached artifact consumed by its first caller; one Commit failing after another succeeded; a replaced registration; a reply from diff --git a/services/flysim/crates/fly-session/examples/processes.rs b/services/flysim/crates/fly-session/examples/processes.rs new file mode 100644 index 0000000..54b7721 --- /dev/null +++ b/services/flysim/crates/fly-session/examples/processes.rs @@ -0,0 +1,75 @@ +//! The same synthetic session in all three execution modes, printing the one behaviour trace +//! they agree on. +//! +//! ```sh +//! cargo run -p fly-session --example processes +//! ``` +//! +//! The separate-process run starts one agent process per fly and one environment process +//! through this crate's own binary, so it needs that binary built: +//! +//! ```sh +//! cargo build -p fly-session --bin fly-session +//! ``` + +use fly_session::harness::{ExecutionMode, HarnessConfig, SessionHarness, Via}; +use fly_session::launcher::default_worker_program; + +#[tokio::main(flavor = "multi_thread", worker_threads = 4)] +async fn main() -> Result<(), Box> { + let program = default_worker_program(); + println!("worker program: {}", program.display()); + let mut agreed: Option> = None; + + for mode in ExecutionMode::all() { + let dir = tempfile::tempdir()?; + let config = HarnessConfig { mode, ..HarnessConfig::default() }; + println!( + "\n=== {} : {} threads for {} participants plus the coordinator", + mode.label(), + config.budget()?.total(), + config.agents.len() + 1 + ); + let mut harness = SessionHarness::start(Via::Unix, dir.path(), config).await?; + harness.coordinator.bootstrap().await?; + let reports = harness.coordinator.run(3).await?; + for report in &reports { + println!(" committed boundary {}", report.boundary); + } + for (worker_id, status) in harness.launcher.health_check_all().await { + match status { + Ok(status) => println!( + " {worker_id}: {:?}, progress {}", + status.state, status.progress_counter + ), + Err(e) => println!(" {worker_id}: unhealthy: {e}"), + } + } + let behaviour = harness.coordinator.trace.behavior(); + match &agreed { + None => { + println!(" behaviour trace, {} transitions:", behaviour.len()); + for line in &behaviour { + println!(" {line}"); + } + agreed = Some(behaviour); + } + Some(first) => { + assert_eq!( + &behaviour, first, + "{} produced a different behaviour trace", + mode.label() + ); + println!(" behaviour trace: identical to the first run"); + } + } + let reaped = harness.launcher.reap_all("example").await; + for (worker_id, outcome) in reaped { + println!(" reaped {worker_id}: {outcome:?}"); + } + harness.shutdown().await; + drop(dir); + } + println!("\nall three execution modes produced one behaviour trace"); + Ok(()) +} diff --git a/services/flysim/crates/fly-session/src/agent.rs b/services/flysim/crates/fly-session/src/agent.rs index 68a621b..7ccdfe9 100644 --- a/services/flysim/crates/fly-session/src/agent.rs +++ b/services/flysim/crates/fly-session/src/agent.rs @@ -203,6 +203,9 @@ pub struct AgentConfig { pub incarnation_id: Id, pub tick_duration: RationalNs, pub warmup_ticks: u64, + /// The thread allocation the launcher started this worker within. `workers-v1` requires + /// `Agent.Initialize`'s `workerThreads` to lie inside it. + pub worker_threads: usize, pub faults: AgentFaults, } @@ -369,6 +372,18 @@ impl FakeAgentWorker { if params.worker_threads == 0 { return Err(DomainError::invalid("workerThreads must be >= 1")); } + // `workers-v1`: workerThreads is "within launcher allocation". This worker was started + // with that allocation, so a request for more than it is a capacity refusal made + // before the model is constructed, not a silent reduction to what is available. + if params.worker_threads > self.config.worker_threads as u64 { + return Err(DomainError::before( + ErrorCode::Busy, + format!( + "Agent.Initialize asks for {} worker threads; the launcher allocated {}", + params.worker_threads, self.config.worker_threads + ), + )); + } params.initial_decision_context.validate().map_err(DomainError::invalid)?; let available = FakeAgentWorker::available_actions(¶ms.initial_decision_context)?; // Everything is validated before the model is constructed. diff --git a/services/flysim/crates/fly-session/src/bin/fly-session.rs b/services/flysim/crates/fly-session/src/bin/fly-session.rs new file mode 100644 index 0000000..f8c2c44 --- /dev/null +++ b/services/flysim/crates/fly-session/src/bin/fly-session.rs @@ -0,0 +1,8 @@ +//! This crate's one binary. Every role it can take is a subcommand of it. +//! +//! `implementation.md` section 2: "Worker executables can be subcommands of one binary +//! initially; process boundaries do not require separate repos." + +fn main() -> std::process::ExitCode { + fly_session::cli::main() +} diff --git a/services/flysim/crates/fly-session/src/cli.rs b/services/flysim/crates/fly-session/src/cli.rs new file mode 100644 index 0000000..f13f44f --- /dev/null +++ b/services/flysim/crates/fly-session/src/cli.rs @@ -0,0 +1,251 @@ +//! The subcommands of this crate's one binary. +//! +//! `implementation.md` section 2 allows worker executables to be subcommands of one binary +//! rather than separate crates, and that is what these are: `agent` and `environment` are the +//! two worker roles a launcher starts as separate processes, and `measure` runs the execution +//! modes against each other. +//! +//! ```text +//! fly-session agent --socket S --store-root D --client-id C --service N --threads T ... +//! fly-session environment --socket S --store-root D --client-id C --service N --threads T ... +//! fly-session measure [--steps N] [--agents 1,2,4] [--modes in-process,thread,process] +//! ``` +//! +//! A worker process is told exactly which participant it is. It proves that identity in +//! `Worker.Hello`, so a process started under another one is refused by its own supervisor +//! before the coordinator has pinned anything. + +use std::collections::BTreeMap; +use std::path::PathBuf; +use std::process::ExitCode; + +use crate::agent::AgentFaults; +use crate::environment::EnvironmentFaults; +use crate::launcher::{AgentLaunch, EnvironmentLaunch, ExecutionMode, Started, serve_one}; +use crate::types::*; + +const USAGE: &str = "\ +fly-session [options] + + agent serve one agent worker on a launcher-created endpoint + environment serve the environment worker on a launcher-created endpoint + measure compare the execution modes and print the measurement table + +Worker options (agent and environment): + --socket PATH the launcher's endpoint for this participant + --store-root PATH the router's artifact store root + --client-id ID the configured bus client identity + --service NAME the one service name this worker registers + --threads N the launcher's thread allocation for this worker + --session ID the session this worker belongs to + --incarnation ID this worker's domain incarnation + + agent: --agent ID --port ID --tick-numerator N --tick-denominator N + --warmup-ticks N [--prepare-delay-ms N] [--commit-delay-ms N] + [--fail-commit-at-step N] + environment: --worker ID --ports p1,p2 --step-numerator N --step-denominator N + [--advance-delay-ms N] [--omit-view-at-boundary N] + +Measure options: + --steps N transitions per run (default 200) + --agents 1,2,4 agent counts to compare (default 1,2,4) + --modes LIST in-process, thread, process (default all three) + --worker-threads N within-agent worker threads (default 1) +"; + +/// The binary's entry point. +pub fn main() -> ExitCode { + let mut args = std::env::args_os().skip(1); + let Some(command) = args.next() else { + eprint!("{USAGE}"); + return ExitCode::from(2); + }; + 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)), + "--help" | "-h" | "help" => { + print!("{USAGE}"); + return ExitCode::SUCCESS; + } + other => Err(format!("unknown command {other:?}\n\n{USAGE}")), + }; + match result { + Ok(()) => ExitCode::SUCCESS, + Err(e) => { + eprintln!("fly-session {command}: {e}"); + ExitCode::FAILURE + } + } +} + +/// `--flag value` options. The launcher builds this argv, so the grammar stays small. +#[derive(Debug, Default)] +struct Options(BTreeMap); + +impl Options { + fn parse(args: &[String]) -> 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:?}")); + }; + let value = iter + .next() + .ok_or_else(|| format!("option --{name} needs a value"))?; + if out.insert(name.to_owned(), value.clone()).is_some() { + return Err(format!("option --{name} was given twice")); + } + } + Ok(Options(out)) + } + + fn required(&self, name: &str) -> Result<&str, String> { + self.0 + .get(name) + .map(String::as_str) + .ok_or_else(|| format!("option --{name} is required")) + } + + fn optional(&self, name: &str) -> Option<&str> { + self.0.get(name).map(String::as_str) + } + + fn id(&self, name: &str) -> Result { + parse_id(self.required(name)?).map_err(|e| format!("--{name}: {e}")) + } + + fn u64(&self, name: &str, default: u64) -> Result { + match self.0.get(name) { + None => Ok(default), + Some(value) => value.parse().map_err(|_| format!("--{name}: {value:?} is not a number")), + } + } + + fn opt_u64(&self, name: &str) -> Result, String> { + match self.0.get(name) { + None => Ok(None), + Some(value) => value + .parse() + .map(Some) + .map_err(|_| format!("--{name}: {value:?} is not a number")), + } + } + + fn usize(&self, name: &str, default: usize) -> Result { + Ok(self.u64(name, default as u64)? as usize) + } + + fn path(&self, name: &str) -> Result { + Ok(PathBuf::from(self.required(name)?)) + } + + fn rational(&self, numerator: &str, denominator: &str) -> Result { + let n = self.u64(numerator, 0)?; + let d = self.u64(denominator, 1)?; + RationalNs::new(n, d).map_err(|e| format!("--{numerator}/--{denominator}: {}", e.0)) + } +} + +/// 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)?; + 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 what = match role { + "agent" => Started::Agent(AgentLaunch { + session_id, + agent_id: options.id("agent")?, + port_id: options.id("port")?, + incarnation_id, + tick_duration: options.rational("tick-numerator", "tick-denominator")?, + warmup_ticks: options.u64("warmup-ticks", 0)?, + worker_threads: threads, + 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)?, + }, + client_id: client_id.clone(), + service: service.clone(), + }), + _ => Started::Environment(EnvironmentLaunch { + session_id, + worker_id: options.id("worker")?, + incarnation_id, + step_duration: options.rational("step-numerator", "step-denominator")?, + ports: parse_ports(options.required("ports")?)?, + worker_threads: threads, + faults: EnvironmentFaults { + advance_delay_ms: options.u64("advance-delay-ms", 0)?, + omit_view_at_boundary: options.opt_u64("omit-view-at-boundary")?, + }, + client_id: client_id.clone(), + service: service.clone(), + }), + }; + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(threads) + .enable_all() + .build() + .map_err(|e| format!("runtime: {e}"))?; + runtime.block_on(async move { + let handle = serve_one(&socket, &client_id, &service, &store_root, &what, threads).await?; + // The worker serves until its supervisor's Worker.Shutdown, which it answers before + // it stops. Exiting is then one event, not a race between a reply and a signal. + handle.join().await; + Ok(()) + }) +} + +fn parse_ports(value: &str) -> Result, String> { + value + .split(',') + .filter(|part| !part.is_empty()) + .map(|part| parse_id(part).map_err(|e| format!("--ports: {e}"))) + .collect() +} + +/// Runs the execution-mode comparison and prints its table. +fn measure(options: &Options) -> Result<(), String> { + let mut config = crate::measure::MeasureConfig { + steps: options.u64("steps", 200)?, + worker_threads: options.usize("worker-threads", 1)?, + ..crate::measure::MeasureConfig::default() + }; + if let Some(list) = options.optional("agents") { + config.agent_counts = list + .split(',') + .filter(|p| !p.is_empty()) + .map(|p| p.parse::().map_err(|_| format!("--agents: {p:?}"))) + .collect::, String>>()?; + } + if let Some(list) = options.optional("modes") { + config.modes = list + .split(',') + .filter(|p| !p.is_empty()) + .map(|p| match p { + "in-process" => Ok(ExecutionMode::InProcess), + "thread" => Ok(ExecutionMode::Thread), + "process" => Ok(ExecutionMode::Process), + other => Err(format!("--modes: {other:?}")), + }) + .collect::, String>>()?; + } + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .map_err(|e| format!("runtime: {e}"))?; + let rows = runtime.block_on(crate::measure::run(&config))?; + print!("{}", crate::measure::table(&rows)); + Ok(()) +} diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs index 3f31c87..22f4221 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/services/flysim/crates/fly-session/src/coordinator.rs @@ -10,10 +10,12 @@ //! domain request id, and anything that cannot be resolved fails the epoch. use std::collections::{BTreeMap, BTreeSet}; +use std::time::{Duration, Instant}; use serde_json::{Map, Value, json}; use crate::clock::Pacing; +use crate::metrics::Metrics; use crate::phase::{Phase, PhaseMachine}; use crate::rpc::{self, DomainReply, Serials, WorkerRef}; use crate::task::{ActionExecutor, Task}; @@ -84,11 +86,20 @@ pub struct SessionFailure { pub error: DomainError, pub phase: String, pub detail: String, + /// The participant the failure is attributed to, where one is. + /// + /// `step-v1` section 7 stops the epoch rather than neutralising a player, so a diagnosed + /// outcome has to say which participant it was: the coordinator names it here instead of + /// leaving a caller to read it out of a message. + pub participant: Option, } impl std::fmt::Display for SessionFailure { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{} at {}: {}", self.detail, self.phase, self.error) + match &self.participant { + Some(who) => write!(f, "{} at {} ({who}): {}", self.detail, self.phase, self.error), + None => write!(f, "{} at {}: {}", self.detail, self.phase, self.error), + } } } @@ -96,6 +107,24 @@ impl std::error::Error for SessionFailure {} type Outcome = Result; +/// How long the coordinator waits for a participant before it calls the call uncertain. +/// +/// `ipc-v1` section 6 measures these on the caller's own monotonic clock and gives the +/// prototype values: ten seconds without progress is a failure, with a separate budget for a +/// long boot. They are failure-detection values, not a gameplay latency goal. Without them a +/// dead participant is a hang rather than a diagnosed outcome. +#[derive(Clone, Copy, Debug)] +pub struct Deadlines { + pub call: Duration, + pub boot: Duration, +} + +impl Default for Deadlines { + fn default() -> Deadlines { + Deadlines { call: Duration::from_secs(10), boot: Duration::from_secs(30) } + } +} + /// The bus addresses this session publishes on. Chosen by the composition, not the router. #[derive(Clone, Debug)] pub struct Topics { @@ -121,6 +150,9 @@ pub struct AgentSlot { pub port_id: Id, pub profile: AssetRef, pub seed: i32, + /// The thread allocation the launcher gave this agent, which is what `Agent.Initialize` + /// asks for. It is within the launcher allocation by construction. + pub worker_threads: u64, pub tick_duration: RationalNs, pub warmup_ticks: u64, pub committed_step: u64, @@ -144,6 +176,7 @@ impl AgentSlot { port_id, profile, seed, + worker_threads: 1, tick_duration: RationalNs::ZERO, warmup_ticks: 0, committed_step: 0, @@ -198,6 +231,16 @@ pub struct Coordinator { pub injection_log: Vec, /// How many times an exact duplicate met IN_PROGRESS while resolving an uncertain call. pub in_progress_replies: u64, + /// The caller-side failure-detection budgets of `ipc-v1` section 6. + pub deadlines: Deadlines, + /// Per-method and critical-path latency samples. Local synthetic timings, never a + /// capacity claim. + pub metrics: Metrics, + /// The participant the next failure is attributed to, set around each call to one. + blame: Option, + /// Set when the epoch failed: every old handle, route and reply is invalid from here on + /// and only a coherent restore may lift it. + fenced: bool, started: std::time::Instant, last_advance_request: Option, last_commit_requests: Vec, @@ -246,6 +289,10 @@ impl Coordinator { injections: Injections::default(), injection_log: Vec::new(), in_progress_replies: 0, + deadlines: Deadlines::default(), + metrics: Metrics::default(), + blame: None, + fenced: false, started: std::time::Instant::now(), last_advance_request: None, last_commit_requests: Vec::new(), @@ -312,6 +359,16 @@ impl Coordinator { self.pause.load(std::sync::atomic::Ordering::SeqCst) } + /// Stops pacing to the wall clock, so transitions follow one another as fast as the + /// participants answer. + /// + /// Only one pacing authority is ever active; this removes the coordinator's. It is for a + /// measurement run, where a 60 Hz sleep would be most of every sample and none of it the + /// thing being compared. A session that presents to anyone keeps its pacing. + pub fn disable_pacing(&mut self) { + self.pacing = None; + } + /// Leaves a normal pause at its committed boundary. pub fn resume(&mut self) -> Outcome<()> { let Phase::Paused(k) = self.phases.phase() else { @@ -339,12 +396,37 @@ impl Coordinator { } /// Fails the epoch and records the transition to Failed. + /// + /// The epoch is fenced at the same moment: the session's routes, pinned registrations and + /// artifact handles are no longer valid, and nothing but a coherent restore into a new + /// epoch may lift that. fn fail_now(&mut self, error: DomainError, detail: &str) -> SessionFailure { let phase = self.phases.phase().label(); + let participant = self.blame.take(); let (from, to) = self.phases.fail(); self.trace.phase(from, to); - self.audit.push(format!("fail:{detail}")); - SessionFailure { error, phase, detail: detail.to_owned() } + self.fenced = true; + self.views.clear(); + self.pending_views.clear(); + match &participant { + Some(who) => self.audit.push(format!("fail:{detail}:{who}")), + None => self.audit.push(format!("fail:{detail}")), + } + SessionFailure { error, phase, detail: detail.to_owned(), participant } + } + + /// Names the participant the next failure belongs to. + fn blame(&mut self, who: Option) { + self.blame = who; + } + + /// True once the epoch has failed. Old handles and routes are invalid; the session takes + /// no further step and publishes nothing. + /// + /// Lifting the fence is STATE-01's: a group restore into a fresh epoch from a coherent + /// checkpoint. This slice only establishes it. + pub fn is_fenced(&self) -> bool { + self.fenced } fn agent(&self, agent_id: &Id) -> Option<&AgentSlot> { @@ -571,7 +653,9 @@ impl Coordinator { seed: self.agents[index].seed, initial_input: self.sensory_input(&observation, 0), initial_decision_context: self.agents[index].context.clone(), - worker_threads: 1, + // The launcher's allocation for this agent. `workers-v1` requires it to lie + // within that allocation, and the worker refuses anything larger. + worker_threads: self.agents[index].worker_threads, }; let attachments = self.view_attachments(); let scope = self.scope(0); @@ -707,6 +791,13 @@ struct Job { } /// Issues one domain call with owned arguments, so it can run in its own task. +/// +/// The deadline is the caller's, on the caller's monotonic clock. A participant that has died +/// mid-call is usually reported by the bus itself, because its connection took its +/// registration with it; this bound is what makes the remaining cases -- a live process that +/// stopped answering -- a diagnosed outcome rather than a hang. An expired deadline is +/// deliberately `unknown`: `ipc-v1` section 6 forbids reading a caller-side timeout as proof +/// that nothing was mutated. #[allow(clippy::too_many_arguments)] async fn call_owned( bus: flybus::Client, @@ -717,10 +808,33 @@ async fn call_owned( attachments: Vec<(String, flybus::Artifact)>, request_id: DomainRequestId, want: Vec, + deadline: Duration, ) -> Result { let refs: Vec<(&str, &flybus::Artifact)> = attachments.iter().map(|(n, a)| (n.as_str(), a)).collect(); - rpc::call(&bus, &worker, method, scope, params, &refs, request_id, &want).await + let call = rpc::call(&bus, &worker, method, scope, params, &refs, request_id, &want); + match tokio::time::timeout(deadline, call).await { + Ok(result) => result, + Err(_) => Err(DomainError::new( + ErrorCode::BackendFailure, + format!( + "{method}: {} did not answer within {:?}", + worker.worker_id, deadline + ), + MutationCertainty::Unknown, + )), + } +} + +/// One per-agent call's result, in the shape the phase loops read it. +struct JobResult { + agent_id: Id, + reply: Result, + scope: Option, + worker: WorkerRef, + method: &'static str, + /// How long the call took, for the section 5 percentiles. + elapsed: Duration, } impl Coordinator { @@ -740,6 +854,14 @@ impl Coordinator { .iter() .map(|(n, a)| ((*n).to_owned(), (*a).clone())) .collect(); + // Whatever goes wrong from here until the reply is checked belongs to this worker. + self.blame(Some(worker.worker_id.clone())); + let deadline = if method.ends_with("Initialize") || method == "Worker.Hello" { + self.deadlines.boot + } else { + self.deadlines.call + }; + let started = Instant::now(); let reply = call_owned( self.bus.clone(), worker.clone(), @@ -749,15 +871,21 @@ impl Coordinator { owned, request_id, want.to_vec(), + deadline, ) .await; + self.metrics.record(method, started.elapsed()); let reply = match reply { Ok(reply) => reply, Err(e) => return Err(self.fail_now(e, method)), }; self.check_reply(worker, &reply, &scope, method)?; match reply.result() { - Ok(_) => Ok(reply), + Ok(reply_value) => { + let _ = reply_value; + self.blame(None); + Ok(reply) + } Err(e) => Err(self.fail_now(e, method)), } } @@ -773,6 +901,7 @@ impl Coordinator { scope: &Option, method: &'static str, ) -> Outcome<()> { + self.blame(Some(worker.worker_id.clone())); let (worker_id, incarnation, echoed) = match &reply.outcome { SessionRpcOutcome::Success(s) => { (&s.worker_id, &s.incarnation_id, &s.scope) @@ -833,6 +962,8 @@ impl Coordinator { // The original may still be running, which answers IN_PROGRESS for this bus call and // starts no second mutation. Waiting and asking again is the resolution, not a retry // of the operation. + self.blame(Some(worker.worker_id.clone())); + let deadline = self.deadlines.call; for attempt in 0..200u32 { let reply = call_owned( self.bus.clone(), @@ -843,6 +974,7 @@ impl Coordinator { attachments.clone(), request_id.clone(), want.to_vec(), + deadline, ) .await; let reply = match reply { @@ -851,7 +983,10 @@ impl Coordinator { }; self.check_reply(worker, &reply, &scope, method)?; match reply.result() { - Ok(_) => return Ok(reply), + Ok(_) => { + self.blame(None); + return Ok(reply); + } Err(e) if e.code == ErrorCode::InProgress => { let _ = attempt; self.in_progress_replies += 1; @@ -884,6 +1019,7 @@ impl Coordinator { expected: Option<&Value>, what: &str, ) { + let deadline = self.deadlines.call; let reply = call_owned( self.bus.clone(), worker.clone(), @@ -893,6 +1029,7 @@ impl Coordinator { attachments, request_id, Vec::new(), + deadline, ) .await; let outcome = match reply { @@ -943,6 +1080,7 @@ impl Coordinator { Vec::new(), request_id, Vec::new(), + self.deadlines.call, ) .await?; reply.result().cloned() @@ -964,11 +1102,8 @@ impl Coordinator { } /// Runs one set of per-agent jobs in the configured dispatch order. - async fn run_jobs( - &mut self, - jobs: Vec, - order: DispatchOrder, - ) -> Vec<(Id, Result, Option, WorkerRef, &'static str)> { + async fn run_jobs(&mut self, jobs: Vec, order: DispatchOrder) -> Vec { + let deadline = self.deadlines.call; let mut out = Vec::new(); match order { DispatchOrder::Sequential | DispatchOrder::Reversed => { @@ -977,6 +1112,7 @@ impl Coordinator { jobs.reverse(); } for job in jobs { + let started = Instant::now(); let reply = call_owned( self.bus.clone(), job.worker.clone(), @@ -986,9 +1122,17 @@ impl Coordinator { job.attachments, job.request_id, Vec::new(), + deadline, ) .await; - out.push((job.agent_id, reply, job.scope, job.worker, job.method)); + out.push(JobResult { + agent_id: job.agent_id, + reply, + scope: job.scope, + worker: job.worker, + method: job.method, + elapsed: started.elapsed(), + }); } } DispatchOrder::Concurrent => { @@ -1000,6 +1144,7 @@ impl Coordinator { let scope = job.scope.clone(); let method = job.method; tasks.push(tokio::spawn(async move { + let started = Instant::now(); let reply = call_owned( bus, job.worker, @@ -1009,9 +1154,17 @@ impl Coordinator { job.attachments, job.request_id, Vec::new(), + deadline, ) .await; - (agent_id, reply, scope, worker, method) + JobResult { + agent_id, + reply, + scope, + worker, + method, + elapsed: started.elapsed(), + } })); } for task in tasks { @@ -1024,7 +1177,10 @@ impl Coordinator { } // Completion order never affects anything downstream, so the results are put back in // sorted agent-id order here and nowhere else. - out.sort_by(|a, b| a.0.cmp(&b.0)); + out.sort_by(|a, b| a.agent_id.cmp(&b.agent_id)); + for result in &out { + self.metrics.record(result.method, result.elapsed); + } out } } @@ -1032,6 +1188,20 @@ impl Coordinator { impl Coordinator { /// One complete transition `k -> k+1`. pub async fn step(&mut self) -> Outcome { + if self.fenced { + // A failed epoch's routes, registrations and handles are invalid. There is no + // partial continuation: only a coherent restore into a new epoch resumes play. + return Err(SessionFailure { + error: DomainError::before( + ErrorCode::InvalidPhase, + "the epoch is fenced; only a coherent restore resumes play", + ), + phase: self.phases.phase().label(), + detail: "fenced".to_owned(), + participant: None, + }); + } + let mut step_started = Instant::now(); let Phase::Ready(k) = self.phases.phase() else { return Err(self.fail_now( DomainError::before( @@ -1062,6 +1232,9 @@ impl Coordinator { // Wall time is only pacing. Being late omits the sleep and is reported; it never // skips a world step or a neural tick. pacing.wait().await; + // The critical path is the transaction, not the sleep in front of it: the + // deadline the coordinator was waiting for is pacing, and pacing is not work. + step_started = Instant::now(); } // ---- Phase A: prepare all agents concurrently @@ -1172,6 +1345,8 @@ impl Coordinator { self.pause.store(false, std::sync::atomic::Ordering::SeqCst); self.audit.push(format!("pause:{}", k + 1)); } + // The critical path: one whole transition, pacing sleep included. + self.metrics.record("step", step_started.elapsed()); Ok(StepReport { boundary: k + 1, paused, terminal }) } @@ -1239,7 +1414,8 @@ impl Coordinator { } let results = self.run_jobs(jobs, self.dispatch).await; let mut prepared = Vec::new(); - for (agent_id, reply, scope, worker, method) in results { + for JobResult { agent_id, reply, scope, worker, method, elapsed: _ } in results { + self.blame(Some(agent_id.clone())); let reply = match reply { Ok(reply) => reply, // Some agents are already Prepared. Dispatch stops and the epoch fails; a @@ -1271,6 +1447,7 @@ impl Coordinator { } self.audit.push(format!("prepared:{agent_id}@{k}")); self.stats.prepares += 1; + self.blame(None); prepared.push((agent_id, decision)); } prepared.sort_by(|a, b| a.0.cmp(&b.0)); @@ -1422,6 +1599,9 @@ impl Coordinator { self.last_advance_request = Some(request_id.clone()); let want = vec!["view.arena".to_owned()]; self.audit.push(format!("advance:{k}")); + self.blame(Some(worker.worker_id.clone())); + let advance_deadline = self.deadlines.call; + let advance_started = Instant::now(); let injected = self.injections.at_step == k; let reply = if injected && self.injections.lose_advance_result { @@ -1501,8 +1681,10 @@ impl Coordinator { Vec::new(), request_id.clone(), want.clone(), + advance_deadline, ) .await; + self.metrics.record("Environment.Advance", advance_started.elapsed()); let reply = match reply { Ok(reply) => reply, Err(e) => return Err(self.fail_now(e, "advance")), @@ -1518,6 +1700,7 @@ impl Coordinator { Ok(result) => result, Err(e) => return Err(self.fail_now(e, "advance")), }; + self.blame(None); self.pending_views = reply.artifacts; if injected && self.injections.altered_advance_controls { @@ -1773,13 +1956,18 @@ impl Coordinator { let results = self.run_jobs(jobs, self.dispatch).await; let mut commits = Vec::new(); let mut first_failure = None; - for (agent_id, reply, scope, worker, method) in results { + let mut blamed: Option = None; + for JobResult { agent_id, reply, scope, worker, method, elapsed: _ } in results { + self.blame(Some(agent_id.clone())); match reply { Ok(reply) => { self.check_reply(&worker, &reply, &scope, method)?; match reply.result() { Ok(_) => {} Err(e) => { + if first_failure.is_none() { + blamed = Some(agent_id.clone()); + } first_failure = Some(first_failure.unwrap_or(e)); continue; } @@ -1812,17 +2000,23 @@ impl Coordinator { } self.audit.push(format!("committed:{agent_id}@{k}")); self.stats.commits += 1; + self.blame(None); commits.push((agent_id, result)); } Err(e) => { + if first_failure.is_none() { + blamed = Some(agent_id.clone()); + } first_failure = Some(first_failure.unwrap_or(e)); } } } if let Some(error) = first_failure { // One Commit failed after others succeeded. There is no partial-match - // continuation: the epoch is failed and the group recovers together. + // continuation: the epoch is failed and the group recovers together, and the + // failure names the agent whose Commit failed. let _ = bodies; + self.blame(blamed); return Err(self.fail_now(error, "commit")); } if commits.len() != self.agents.len() { diff --git a/services/flysim/crates/fly-session/src/harness.rs b/services/flysim/crates/fly-session/src/harness.rs index 429ed59..b4a635e 100644 --- a/services/flysim/crates/fly-session/src/harness.rs +++ b/services/flysim/crates/fly-session/src/harness.rs @@ -1,36 +1,37 @@ -//! The runnable synthetic composition: one router, two fake agents, one counter arena and one -//! coordinator, over either transport. +//! The runnable synthetic composition: one router, the configured flies, one counter arena and +//! one coordinator, in whichever execution mode the composition asks for. //! -//! All participants use router semantics even when colocated, so the in-memory and -//! Unix-socket runs exercise the same code. The caller owns the store root directory, which -//! keeps this module free of a temporary-directory dependency. +//! All participants use router semantics even when colocated, so every mode and both +//! transports exercise the same code. The caller owns the store root directory, which keeps +//! this module free of a temporary-directory dependency. +//! +//! The three execution modes are the SESSION-02 comparison: +//! +//! | Mode | Where each participant runs | Transport | +//! | --- | --- | --- | +//! | [`ExecutionMode::InProcess`] | A task on the coordinator's runtime | either | +//! | [`ExecutionMode::Thread`] | Its own OS thread and runtime | Unix socket | +//! | [`ExecutionMode::Process`] | Its own process, one per fly plus one world | Unix socket | use std::collections::BTreeMap; -use std::path::{Path, PathBuf}; +use std::path::Path; use std::sync::Mutex; -use std::sync::atomic::{AtomicU64, Ordering}; -use flybus::{ - Client, ClientConfig, Grants, Pattern, Policy, Router, RouterConfig, ServiceConfig, Transport, - UnixListenerHandle, -}; +use flybus::{Client, Grants, Pattern, Policy, Router, RouterConfig}; -use crate::agent::{AgentConfig, AgentFaults, FakeAgentWorker, synthetic_profile}; +use crate::agent::{AgentFaults, synthetic_profile}; use crate::coordinator::{AgentSlot, Coordinator}; -use crate::environment::{CounterEnvironment, EnvironmentConfig, EnvironmentFaults}; -use crate::rpc::WorkerRef; +use crate::environment::EnvironmentFaults; +use crate::launcher::{ + AgentLaunch, EnvironmentLaunch, Launcher, ReapOutcome, SUPERVISOR_CLIENT, ThreadBudget, +}; use crate::task::{ActionExecutor, CounterTask, IdentityExecutor, Terminal}; // `crate::types` is this crate's facade over the shared `fly-session-types` crate; the // glob keeps the contract's own names in sight instead of restating them. use crate::types::*; -use crate::worker::{StatusCell, WorkerHandle, serve}; +use crate::worker::StatusCell; -/// Which transport the session runs over. Both must produce the same behaviour. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum Via { - Memory, - Unix, -} +pub use crate::launcher::{ExecutionMode, Via}; /// One agent in the composition. #[derive(Clone, Debug)] @@ -41,6 +42,22 @@ pub struct AgentSpec { /// derivation algorithm is specified before the real agent slice. pub seed: i32, pub faults: AgentFaults, + /// The threads this agent asks the launcher for. `Agent.Initialize` carries exactly what + /// the launcher allocated, which `workers-v1` requires it to lie within. + pub worker_threads: usize, +} + +impl AgentSpec { + /// One agent on one thread, with no injected fault. + pub fn new(agent_id: &str, port_id: &str, seed: i32) -> AgentSpec { + AgentSpec { + agent_id: id(agent_id), + port_id: id(port_id), + seed, + faults: AgentFaults::default(), + worker_threads: 1, + } + } } /// The composition the harness builds. @@ -56,6 +73,15 @@ pub struct HarnessConfig { pub warmup_ticks: u64, pub terminal: Terminal, pub environment_faults: EnvironmentFaults, + /// Where each participant runs. + pub mode: ExecutionMode, + /// The total thread allocation the launcher may hand out. `None` sizes it from the + /// composition and the machine, which is what an ordinary run wants; a test that means to + /// exhaust the budget names a number. + pub thread_budget: Option, + /// The threads reserved for the coordinator, its router and its store. + pub coordinator_threads: usize, + pub environment_threads: usize, } impl Default for HarnessConfig { @@ -65,31 +91,45 @@ impl Default for HarnessConfig { epoch: id("e1"), episode_id: id("ep1"), agents: vec![ - AgentSpec { - agent_id: id("fly-a"), - port_id: id("p1"), - seed: 7, - faults: AgentFaults::default(), - }, - AgentSpec { - agent_id: id("fly-b"), - port_id: id("p2"), - seed: 11, - faults: AgentFaults::default(), - }, + AgentSpec { seed: 7, ..AgentSpec::new("fly-a", "p1", 7) }, + AgentSpec { seed: 11, ..AgentSpec::new("fly-b", "p2", 11) }, ], step_hz: 60, tick_ms: 1, warmup_ticks: 10, terminal: Terminal::Never, environment_faults: EnvironmentFaults::default(), + mode: ExecutionMode::InProcess, + thread_budget: None, + coordinator_threads: 1, + environment_threads: 1, } } } +impl HarnessConfig { + /// The threads this composition needs at a minimum: the coordinator, the world and every + /// agent's own allocation. + pub fn required_threads(&self) -> usize { + self.coordinator_threads + + self.environment_threads + + self.agents.iter().map(|a| a.worker_threads).sum::() + } + + /// The budget the launcher runs under: what was configured, or a budget that covers both + /// this composition and this machine's physical cores. + pub fn budget(&self) -> Result { + let total = self + .thread_budget + .unwrap_or_else(|| crate::metrics::physical_cores().max(self.required_threads())); + ThreadBudget::new(total, self.coordinator_threads) + } +} + const ENV_SERVICE: &str = "env.arena"; const ENV_CLIENT: &str = "environment"; const ENV_WORKER: &str = "arena"; +const COORDINATOR_CLIENT: &str = "coordinator"; fn agent_service(agent_id: &Id) -> String { format!("agent.{agent_id}") @@ -105,37 +145,6 @@ fn grants(f: impl FnOnce(&mut Grants)) -> Grants { g } -/// Makes a connection for one launcher-bound participant, over the chosen transport. -struct Connector { - router: Router, - via: Via, - store_root: PathBuf, - sockets: PathBuf, - next_socket: AtomicU64, - listeners: Mutex>, -} - -impl Connector { - async fn client(&self, id: &str) -> Result { - let transport = match self.via { - Via::Memory => self.router.connect_in_memory_as(id), - Via::Unix => { - let n = self.next_socket.fetch_add(1, Ordering::Relaxed); - let path = self.sockets.join(format!("{id}-{n}.sock")); - let listener = self.router.listen_unix_as(&path, id).await.map_err(|e| { - flybus::BusError::new(flybus::ErrorCode::RouterLost, format!("listen: {e}")) - })?; - let transport = Transport::unix(&path).await.map_err(|e| { - flybus::BusError::new(flybus::ErrorCode::RouterLost, format!("connect: {e}")) - })?; - self.listeners.lock().expect("not poisoned").push(listener); - transport - } - }; - Client::connect(transport, ClientConfig::new(id, &self.store_root)).await - } -} - /// What a restarted worker looks like from the outside: a new registration and a new /// incarnation, both different from the ones the coordinator pinned. #[derive(Clone, Debug)] @@ -148,16 +157,17 @@ pub struct Restarted { /// A running synthetic session. pub struct SessionHarness { pub coordinator: Coordinator, - pub environment: WorkerHandle, - pub agents: BTreeMap, pub config: HarnessConfig, pub via: Via, - connector: Connector, + pub mode: ExecutionMode, + /// The supervisor. It owns every participant's lifetime and thread allocation. + pub launcher: Launcher, observers: Mutex>, } impl SessionHarness { - /// Builds the router, the workers and the coordinator. Nothing has stepped yet. + /// Builds the router, launches the workers and builds the coordinator. Nothing has + /// stepped yet. pub async fn start( via: Via, root: &Path, @@ -167,16 +177,29 @@ impl SessionHarness { let sockets = root.join("sockets"); std::fs::create_dir_all(&sockets).expect("the caller owns a writable directory"); + // The launcher's policy: who may connect, and what each may do. Naming a target is not + // authority to use it, so the supervisor calls but never registers or publishes, and a + // worker registers exactly one service and calls nothing. let mut policy = Policy::closed() .client( - "coordinator", + COORDINATOR_CLIENT, grants(|g| { g.call = vec![Pattern::prefix("agent."), Pattern::prefix("env.")]; g.publish = vec![Pattern::prefix("session.")]; g.manage_topics = vec![Pattern::prefix("session.")]; }), ) + .client( + SUPERVISOR_CLIENT, + grants(|g| { + g.call = vec![Pattern::prefix("agent."), Pattern::prefix("env.")]; + }), + ) .client(ENV_CLIENT, grants(|g| g.register = vec![Pattern::exact(ENV_SERVICE)])) + .client( + &format!("{ENV_CLIENT}-r2"), + grants(|g| g.register = vec![Pattern::exact(ENV_SERVICE)]), + ) .client("observer", grants(|g| g.subscribe = vec![Pattern::prefix("session.")])); for spec in &config.agents { let service = agent_service(&spec.agent_id); @@ -196,66 +219,68 @@ impl SessionHarness { let router = Router::new(router_config).map_err(|e| { flybus::BusError::new(flybus::ErrorCode::StoreFailure, format!("router: {e}")) })?; - let connector = Connector { - router, - via, - store_root, - sockets, - next_socket: AtomicU64::new(0), - listeners: Mutex::new(Vec::new()), - }; + + let budget = config.budget().map_err(refusal)?; + let mut launcher = + Launcher::start(router, config.mode, via, &store_root, &sockets, budget).await?; let step_duration = hz(config.step_hz).expect("a positive cadence"); let tick_duration = millis(config.tick_ms).expect("a positive tick"); // The environment first: it owns the world and the descriptor. - let env_client = connector.client(ENV_CLIENT).await?; - let env_service = env_client.register(ENV_SERVICE, ServiceConfig::default()).await?; - let env_incarnation = env_service.incarnation().to_owned(); - let environment = serve( - env_client, - env_service, - CounterEnvironment::new(EnvironmentConfig { + let environment = launcher + .launch_environment(EnvironmentLaunch { session_id: config.session_id.clone(), worker_id: id(ENV_WORKER), incarnation_id: id("arena-inc-1"), step_duration, ports: config.agents.iter().map(|a| a.port_id.clone()).collect(), + worker_threads: config.environment_threads, faults: config.environment_faults.clone(), - }), - ); + client_id: ENV_CLIENT.to_owned(), + service: ENV_SERVICE.to_owned(), + }) + .await + .map_err(refusal)?; + let environment_ref = launcher + .worker(&environment.worker_id) + .expect("just launched") + .worker_ref(); let mut slots = Vec::new(); - let mut agents = BTreeMap::new(); for spec in &config.agents { - let service_name = agent_service(&spec.agent_id); - let client = connector.client(&agent_client(&spec.agent_id)).await?; - let service = client.register(&service_name, ServiceConfig::default()).await?; - let incarnation = service.incarnation().to_owned(); - let handle = serve( - client, - service, - FakeAgentWorker::new(AgentConfig { + let identity = launcher + .launch_agent(AgentLaunch { session_id: config.session_id.clone(), agent_id: spec.agent_id.clone(), + port_id: spec.port_id.clone(), incarnation_id: parse_id(&format!("{}-inc-1", spec.agent_id)) .expect("an agent id plus a suffix is an Id"), tick_duration, warmup_ticks: config.warmup_ticks, + worker_threads: spec.worker_threads, faults: spec.faults.clone(), - }), - ); - slots.push(AgentSlot::new( - WorkerRef::new(&service_name, &incarnation, &spec.agent_id), + client_id: agent_client(&spec.agent_id), + service: agent_service(&spec.agent_id), + }) + .await + .map_err(refusal)?; + let worker_ref = launcher + .worker(&spec.agent_id) + .expect("just launched") + .worker_ref(); + let mut slot = AgentSlot::new( + worker_ref, spec.agent_id.clone(), spec.port_id.clone(), synthetic_profile(&spec.agent_id, &tick_duration, config.warmup_ticks), spec.seed, - )); - agents.insert(spec.agent_id.clone(), handle); + ); + slot.worker_threads = identity.worker_threads as u64; + slots.push(slot); } - let coordinator_client = connector.client("coordinator").await?; + let coordinator_client = launcher.connect(COORDINATOR_CLIENT).await?; let executors: BTreeMap> = config .agents .iter() @@ -268,7 +293,7 @@ impl SessionHarness { config.session_id.clone(), config.epoch.clone(), config.episode_id.clone(), - WorkerRef::new(ENV_SERVICE, &env_incarnation, &id(ENV_WORKER)), + environment_ref, slots, Box::new(CounterTask::new(&config.epoch, config.terminal)), executors, @@ -276,27 +301,42 @@ impl SessionHarness { Ok(SessionHarness { coordinator, - environment, - agents, config, via, - connector, + mode: launcher.mode(), + launcher, observers: Mutex::new(Vec::new()), }) } pub fn router(&self) -> &Router { - &self.connector.router + self.launcher.router() + } + + /// The coordinator and its supervisor, borrowed apart. + /// + /// A supervisor acts while a transition is in flight -- that is what a supervisor is for + /// -- so the two have to be reachable at the same time. + pub fn parts(&mut self) -> (&mut Coordinator, &mut Launcher) { + (&mut self.coordinator, &mut self.launcher) + } + + /// The router's own counters: owners, roots, queued messages and store bytes. + pub fn router_stats(&self) -> flybus::RouterStats { + self.launcher.router().stats() } /// A client for `id`, connected the same way every participant is. + /// + /// An unconfigured client id is refused by the launcher's policy before it can route, so + /// this is not a way around the composition. pub async fn client(&self, id: &str) -> Result { - self.connector.client(id).await + self.launcher.connect(id).await } /// An extra subscriber, for a test that watches the published boundaries. pub async fn observer(&self) -> Result { - let client = self.connector.client("observer").await?; + let client = self.launcher.connect("observer").await?; self.observers.lock().expect("not poisoned").push(client.clone()); Ok(client) } @@ -306,22 +346,6 @@ impl SessionHarness { /// The coordinator still pins the old registration, so its next call to that agent fails /// rather than silently reaching another brain. pub async fn restart_agent(&mut self, agent_id: &Id) -> Result { - let tick_duration = millis(self.config.tick_ms).expect("a positive tick"); - if let Some(old) = self.agents.remove(agent_id) { - old.stop().await; - } - let service_name = agent_service(agent_id); - let client = self.connector.client(&format!("{}-r2", agent_client(agent_id))).await?; - let service = loop { - match client.register(&service_name, ServiceConfig::default()).await { - Ok(service) => break service, - Err(e) if e.code == flybus::ErrorCode::Conflict => { - // The old registration is released when its connection finishes closing. - tokio::time::sleep(std::time::Duration::from_millis(5)).await; - } - Err(e) => return Err(e), - } - }; let spec = self .config .agents @@ -329,53 +353,87 @@ impl SessionHarness { .find(|spec| spec.agent_id == *agent_id) .expect("a configured agent") .clone(); + self.launcher.kill(agent_id).await; + let tick_duration = millis(self.config.tick_ms).expect("a positive tick"); let incarnation_id = parse_id(&format!("{agent_id}-inc-2")).expect("an agent id plus a suffix is an Id"); - let restarted = Restarted { - service: service_name, - service_incarnation: service.incarnation().to_owned(), - incarnation_id: incarnation_id.clone(), - }; - let handle = serve( - client, - service, - FakeAgentWorker::new(AgentConfig { + self.launcher + .launch_agent(AgentLaunch { session_id: self.config.session_id.clone(), agent_id: agent_id.clone(), - incarnation_id, + port_id: spec.port_id.clone(), + incarnation_id: incarnation_id.clone(), tick_duration, warmup_ticks: self.config.warmup_ticks, - faults: spec.faults, - }), - ); - self.agents.insert(agent_id.clone(), handle); - Ok(restarted) + worker_threads: spec.worker_threads, + faults: spec.faults.clone(), + client_id: format!("{}-r2", agent_client(agent_id)), + service: agent_service(agent_id), + }) + .await + .map_err(refusal)?; + let worker = self.launcher.worker(agent_id).expect("just launched"); + Ok(Restarted { + service: worker.identity.service.clone(), + service_incarnation: worker.service_incarnation.clone(), + incarnation_id, + }) + } + + /// Ends one participant without asking it, as a crash would. + pub async fn kill(&mut self, worker_id: &Id) -> ReapOutcome { + self.launcher.kill(worker_id).await + } + + /// The worker id the environment answers to. + pub fn environment_id(&self) -> Id { + id(ENV_WORKER) } /// The agent worker's progress counter, which is its fake model's mutation count. + /// + /// A participant in another process keeps its counter there; use + /// [`SessionHarness::progress_of`], which reads it over the bus in every mode. pub fn agent_mutations(&self, agent_id: &Id) -> u64 { - self.agents.get(agent_id).map(WorkerHandle::progress_counter).unwrap_or_default() + self.launcher + .worker(agent_id) + .and_then(crate::launcher::LaunchedWorker::progress_counter) + .unwrap_or_default() } pub fn environment_mutations(&self) -> u64 { - self.environment.progress_counter() + self.agent_mutations(&id(ENV_WORKER)) } + /// One participant's progress counter, read over the bus. Works in every execution mode. + pub async fn progress_of(&mut self, worker_id: &Id) -> Result { + Ok(self.launcher.health_check(worker_id).await?.progress_counter) + } + + /// The local status cell of a participant in this process, or `None` for one with a + /// process of its own. pub fn agent_status(&self, agent_id: &Id) -> Option { - self.agents.get(agent_id).map(|handle| handle.status.clone()) + self.launcher.worker(agent_id).and_then(crate::launcher::LaunchedWorker::status) } - /// Stops every worker and closes the router. + /// Reaps every participant and closes the router. pub async fn shutdown(self) { - let SessionHarness { coordinator, environment, agents, connector, observers, .. } = self; + let SessionHarness { coordinator, mut launcher, observers, .. } = self; drop(coordinator); - environment.stop().await; - for (_, handle) in agents { - handle.stop().await; - } + launcher.reap_all("shutdown").await; for observer in observers.into_inner().expect("not poisoned") { observer.close().await; } - connector.router.shutdown(); + launcher.router().shutdown(); } } + +/// A launcher refusal, as a bus error: the harness's one error type stays the bus's. +fn refusal(e: DomainError) -> flybus::BusError { + let code = match e.code { + ErrorCode::Busy => flybus::ErrorCode::QuotaExceeded, + ErrorCode::IdentityMismatch => flybus::ErrorCode::NotAuthorized, + _ => flybus::ErrorCode::RouterLost, + }; + flybus::BusError::new(code, e.to_string()) +} diff --git a/services/flysim/crates/fly-session/src/launcher.rs b/services/flysim/crates/fly-session/src/launcher.rs new file mode 100644 index 0000000..e1bb474 --- /dev/null +++ b/services/flysim/crates/fly-session/src/launcher.rs @@ -0,0 +1,1325 @@ +//! The launcher and supervisor: it starts participants, gives them their identities, checks +//! that they are the participants the composition configured, and reaps them. +//! +//! SESSION-02's subject is that one agent process per fly and one environment process under +//! the coordinator behave exactly as the in-process composition does. The same launcher also +//! starts the two comparison variants -- every participant as a task on the coordinator's +//! runtime, and every participant on its own dedicated thread -- so the three can be compared +//! without writing a second composition. +//! +//! ```text +//! Launcher ── thread budget ──> one allocation per participant +//! ── identity ──> client id, service name, worker id, agent/port binding +//! ── Worker.Hello ──> the registration the coordinator then pins +//! ── Worker.Status ──> health, bounded by the supervisor's own clock +//! ── Worker.Shutdown ──> reaped, and terminated if it does not stop +//! ``` +//! +//! The launcher is the configured supervisor: it holds `Worker.Shutdown` authority and the +//! bus grants that go with it. A worker has no authority over it. + +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +use serde_json::{Map, Value}; + +use flybus::{Client, ClientConfig, Router, ServiceConfig, Transport, UnixListenerHandle}; + +use crate::agent::{AgentConfig, AgentFaults, FakeAgentWorker}; +use crate::environment::{CounterEnvironment, EnvironmentConfig, EnvironmentFaults}; +use crate::rpc::WorkerRef; +use crate::types::*; +use crate::worker::{StatusCell, WorkerHandle, serve}; + +/// Which transport a participant's connection runs over. Both must produce the same +/// behaviour, which is a test. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Via { + Memory, + Unix, +} + +/// Where a participant runs. +/// +/// A separate process is the SESSION-02 subject; the other two are the comparison variants +/// the slice is measured against. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum ExecutionMode { + /// Every participant is a task on the coordinator's own runtime. This is SESSION-01. + #[default] + InProcess, + /// Every participant owns a dedicated OS thread and its own runtime, in this process. + Thread, + /// One agent process per fly and one environment process, over Unix-domain sockets. + Process, +} + +impl ExecutionMode { + pub fn label(&self) -> &'static str { + match self { + ExecutionMode::InProcess => "in-process", + ExecutionMode::Thread => "thread", + ExecutionMode::Process => "process", + } + } + + /// A separate process reaches the router only over a socket; the other two may use either. + pub fn transport(&self, configured: Via) -> Via { + match self { + ExecutionMode::InProcess => configured, + ExecutionMode::Thread | ExecutionMode::Process => Via::Unix, + } + } + + pub fn all() -> [ExecutionMode; 3] { + [ExecutionMode::InProcess, ExecutionMode::Thread, ExecutionMode::Process] + } +} + +// ------------------------------------------------------------------------------------------- +// The thread budget + +/// The total thread allocation the launcher may hand out, and what it has handed out. +/// +/// `workers-v1` requires `Agent.Initialize`'s `workerThreads` to lie within the launcher +/// allocation. This is that allocation: the launcher refuses to start a participant whose +/// request would take the composition over its configured total, and an agent endpoint +/// refuses an `Agent.Initialize` asking for more threads than its launcher gave it. +#[derive(Clone, Debug)] +pub struct ThreadBudget { + total: usize, + coordinator: usize, + allocated: BTreeMap, +} + +impl ThreadBudget { + /// A budget of `total` threads, `coordinator` of them reserved for the coordinator, its + /// router and its store. + pub fn new(total: usize, coordinator: usize) -> Result { + if total == 0 { + return Err(DomainError::invalid("a thread budget is at least one thread")); + } + if coordinator > total { + return Err(DomainError::invalid( + "the coordinator's reservation exceeds the total thread budget", + )); + } + Ok(ThreadBudget { total, coordinator, allocated: BTreeMap::new() }) + } + + /// The default budget: one thread per physical core, one of them the coordinator's. + pub fn for_this_machine() -> ThreadBudget { + let cores = crate::metrics::physical_cores().max(2); + ThreadBudget::new(cores, 1).expect("two or more cores make a valid budget") + } + + pub fn total(&self) -> usize { + self.total + } + + pub fn coordinator(&self) -> usize { + self.coordinator + } + + pub fn used(&self) -> usize { + self.coordinator + self.allocated.values().sum::() + } + + pub fn remaining(&self) -> usize { + self.total.saturating_sub(self.used()) + } + + pub fn allocation(&self, who: &Id) -> Option { + self.allocated.get(who).copied() + } + + /// Reserves `want` threads for `who`. A request the total cannot cover is refused before + /// anything is started: a capacity refusal, not a runtime fault. + pub fn allocate(&mut self, who: &Id, want: usize) -> Result { + if want == 0 { + return Err(DomainError::invalid("workerThreads must be >= 1")); + } + if self.allocated.contains_key(who) { + return Err(DomainError::before( + ErrorCode::Conflict, + format!("{who} already holds a thread allocation"), + )); + } + if want > self.remaining() { + return Err(DomainError::before( + ErrorCode::Busy, + format!( + "{who} asked for {want} threads; {} of {} remain in the launcher allocation", + self.remaining(), + self.total + ), + )); + } + self.allocated.insert(who.clone(), want); + Ok(want) + } + + pub fn release(&mut self, who: &Id) { + self.allocated.remove(who); + } + + /// Every allocation, in agent-id order, for the report. + pub fn allocations(&self) -> Vec<(Id, usize)> { + self.allocated.iter().map(|(k, v)| (k.clone(), *v)).collect() + } +} + +// ------------------------------------------------------------------------------------------- +// Identities and policies + +/// Everything the launcher configures about one participant before it exists. +/// +/// The bus client id, the service name and the worker id are launcher configuration. The +/// worker proves it is that participant in `Worker.Hello`; a process started under any other +/// identity is refused there rather than adopted into the session. +#[derive(Clone, Debug)] +pub struct WorkerIdentity { + pub session_id: Id, + pub client_id: String, + pub service: String, + pub worker_id: Id, + pub incarnation_id: Id, + pub role: Role, + /// The port an agent is bound to. The environment owns the ports, not one of them. + pub port_id: Option, + /// The thread allocation this participant runs within. + pub worker_threads: usize, +} + +/// How long the supervisor waits before it calls a participant unhealthy. +/// +/// The `ipc-v1` section 6 prototype values: probe after two seconds without a reply, fail +/// after ten without progress, with a separate budget for a participant that is still +/// starting. These are failure-detection values, not a latency goal. +#[derive(Clone, Copy, Debug)] +pub struct HealthPolicy { + pub probe: Duration, + pub fail: Duration, + pub boot: Duration, +} + +impl Default for HealthPolicy { + fn default() -> HealthPolicy { + HealthPolicy { + probe: Duration::from_secs(2), + fail: Duration::from_secs(10), + boot: Duration::from_secs(30), + } + } +} + +/// What one agent participant is started with. +#[derive(Clone, Debug)] +pub struct AgentLaunch { + pub session_id: Id, + pub agent_id: Id, + pub port_id: Id, + pub incarnation_id: Id, + pub tick_duration: RationalNs, + pub warmup_ticks: u64, + /// What the launcher asks the budget for. + pub worker_threads: usize, + pub faults: AgentFaults, + /// The configured client id. A replacement worker connects under its own. + pub client_id: String, + pub service: String, +} + +/// What the environment participant is started with. +#[derive(Clone, Debug)] +pub struct EnvironmentLaunch { + pub session_id: Id, + pub worker_id: Id, + pub incarnation_id: Id, + pub step_duration: RationalNs, + pub ports: Vec, + pub worker_threads: usize, + pub faults: EnvironmentFaults, + pub client_id: String, + pub service: String, +} + +// ------------------------------------------------------------------------------------------- +// A launched participant + +/// A participant on its own thread, with its own runtime. +struct ThreadWorker { + stop: Option>, + join: Option>, +} + +impl ThreadWorker { + /// Ends the thread and waits for its runtime to finish. + fn stop(&mut self) { + drop(self.stop.take()); + if let Some(join) = self.join.take() { + let _ = join.join(); + } + } + + fn finished(&self) -> bool { + self.join.as_ref().map(|j| j.is_finished()).unwrap_or(true) + } +} + +enum Body { + Task(Option), + Thread(ThreadWorker), + Process(Option), +} + +/// One started participant: its configured identity, the registration a caller pins, and +/// whatever the launcher needs to reap it. +pub struct LaunchedWorker { + pub identity: WorkerIdentity, + /// The bus `serviceIncarnation` a caller pins. Not a process id. + pub service_incarnation: String, + /// The domain `incarnationId` `Worker.Hello` reported. + pub domain_incarnation: Id, + /// The operating-system process, when this participant has one of its own. + pub pid: Option, + /// The highest peak resident set the launcher has read for that process, in KiB. + pub peak_rss_kib: u64, + body: Body, + status: Option, + /// The per-participant socket endpoint, kept alive for the connection's lifetime. + _listener: Option, +} + +impl LaunchedWorker { + /// The reference a caller pins: service, registration, worker id and negotiated + /// incarnation. + pub fn worker_ref(&self) -> WorkerRef { + let mut r = WorkerRef::new( + &self.identity.service, + &self.service_incarnation, + &self.identity.worker_id, + ); + r.domain_incarnation = Some(self.domain_incarnation.clone()); + r + } + + /// The local status cell of a participant in this process. A separate process answers + /// `Worker.Status` over the bus instead, which every mode also supports. + pub fn status(&self) -> Option { + self.status.clone() + } + + /// The progress counter of a participant in this process, or `None` for a separate one. + pub fn progress_counter(&self) -> Option { + self.status.as_ref().map(StatusCell::progress_counter) + } + + /// True while the participant is still running, as far as the operating system knows. + pub fn alive(&mut self) -> bool { + match &mut self.body { + Body::Task(handle) => handle.is_some(), + Body::Thread(thread) => !thread.finished(), + Body::Process(Some(child)) => matches!(child.try_wait(), Ok(None)), + Body::Process(None) => false, + } + } + + fn refresh_rss(&mut self) { + if let Some(pid) = self.pid + && let Some(kib) = crate::metrics::peak_rss_kib_of(pid) + { + self.peak_rss_kib = self.peak_rss_kib.max(kib); + } + } +} + +/// How a participant ended. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ReapOutcome { + /// It answered `Worker.Shutdown` and stopped on its own. + Stopped, + /// It did not stop within the supervisor's budget and was terminated. + Terminated, + /// It was already gone when the launcher reached it. + AlreadyGone, +} + +// ------------------------------------------------------------------------------------------- +// Endpoints: how a participant reaches the router + +struct Endpoints { + router: Router, + via: Via, + store_root: PathBuf, + sockets: PathBuf, + next_socket: AtomicU64, + listeners: Mutex>, +} + +impl Endpoints { + fn socket_path(&self, client_id: &str) -> PathBuf { + let n = self.next_socket.fetch_add(1, Ordering::Relaxed); + self.sockets.join(format!("{client_id}-{n}.sock")) + } + + /// A connection this process owns, over the configured transport. + async fn connect(&self, client_id: &str) -> Result { + let transport = match self.via { + Via::Memory => self.router.connect_in_memory_as(client_id), + Via::Unix => { + let path = self.socket_path(client_id); + let listener = self.listen(&path, client_id).await?; + let transport = Transport::unix(&path).await.map_err(|e| { + flybus::BusError::new(flybus::ErrorCode::RouterLost, format!("connect: {e}")) + })?; + self.listeners.lock().expect("not poisoned").push(listener); + transport + } + }; + Client::connect(transport, ClientConfig::new(client_id, &self.store_root)).await + } + + /// An endpoint for someone else to connect to: a separate process, or a thread with its + /// own runtime. + async fn endpoint_for( + &self, + client_id: &str, + ) -> Result<(PathBuf, UnixListenerHandle), flybus::BusError> { + let path = self.socket_path(client_id); + let listener = self.listen(&path, client_id).await?; + Ok((path, listener)) + } + + async fn listen( + &self, + path: &Path, + client_id: &str, + ) -> Result { + self.router.listen_unix_as(path, client_id).await.map_err(|e| { + flybus::BusError::new(flybus::ErrorCode::RouterLost, format!("listen: {e}")) + }) + } +} + +// ------------------------------------------------------------------------------------------- +// The launcher + +/// Starts, identifies, health-checks and reaps the session's participants. +pub struct Launcher { + endpoints: Endpoints, + mode: ExecutionMode, + /// The worker program, for [`ExecutionMode::Process`]. + program: PathBuf, + budget: ThreadBudget, + health: HealthPolicy, + /// The supervisor's own bus connection. It calls `Worker.Hello`, `Worker.Status` and + /// `Worker.Shutdown`, and nothing else. + supervisor: Client, + workers: BTreeMap, + serial: u64, +} + +/// The bus client id the supervisor connects under. +pub const SUPERVISOR_CLIENT: &str = "launcher"; + +/// The environment variable that names the worker program, for a checkout whose binary is not +/// beside the running executable. +pub const WORKER_PROGRAM_ENV: &str = "FLY_SESSION_WORKER"; + +/// Where the worker program is: by configuration, or beside the running executable. +/// +/// The worker is a subcommand of this crate's one binary, so a build that produced the tests +/// produced it too, one directory above them. +pub fn default_worker_program() -> PathBuf { + if let Some(configured) = std::env::var_os(WORKER_PROGRAM_ENV) { + return PathBuf::from(configured); + } + let name = "fly-session"; + if let Ok(exe) = std::env::current_exe() { + let here = exe.parent().map(Path::to_path_buf); + let above = exe.parent().and_then(Path::parent).map(Path::to_path_buf); + for dir in [here, above].into_iter().flatten() { + let candidate = dir.join(name); + if candidate.is_file() { + return candidate; + } + } + } + PathBuf::from(name) +} + +impl Launcher { + /// Connects the supervisor and prepares the launcher. Nothing is started yet. + pub async fn start( + router: Router, + mode: ExecutionMode, + via: Via, + store_root: impl Into, + sockets: impl Into, + budget: ThreadBudget, + ) -> Result { + let sockets: PathBuf = sockets.into(); + std::fs::create_dir_all(&sockets).map_err(|e| { + flybus::BusError::new(flybus::ErrorCode::StoreFailure, format!("sockets: {e}")) + })?; + let endpoints = Endpoints { + router, + via: mode.transport(via), + store_root: store_root.into(), + sockets, + next_socket: AtomicU64::new(0), + listeners: Mutex::new(Vec::new()), + }; + let supervisor = endpoints.connect(SUPERVISOR_CLIENT).await?; + Ok(Launcher { + endpoints, + mode, + program: default_worker_program(), + budget, + health: HealthPolicy::default(), + supervisor, + workers: BTreeMap::new(), + serial: 0, + }) + } + + pub fn mode(&self) -> ExecutionMode { + self.mode + } + + pub fn budget(&self) -> &ThreadBudget { + &self.budget + } + + pub fn health_policy(&self) -> HealthPolicy { + self.health + } + + pub fn set_health_policy(&mut self, health: HealthPolicy) { + self.health = health; + } + + pub fn set_program(&mut self, program: impl Into) { + self.program = program.into(); + } + + pub fn program(&self) -> &Path { + &self.program + } + + pub fn router(&self) -> &Router { + &self.endpoints.router + } + + pub fn supervisor(&self) -> &Client { + &self.supervisor + } + + pub fn worker(&self, worker_id: &Id) -> Option<&LaunchedWorker> { + self.workers.get(worker_id) + } + + pub fn worker_mut(&mut self, worker_id: &Id) -> Option<&mut LaunchedWorker> { + self.workers.get_mut(worker_id) + } + + pub fn worker_ids(&self) -> Vec { + self.workers.keys().cloned().collect() + } + + /// An ordinary connection for a participant this process drives, such as the coordinator. + pub async fn connect(&self, client_id: &str) -> Result { + self.endpoints.connect(client_id).await + } + + fn next_serial(&mut self) -> u64 { + self.serial += 1; + self.serial + } + + // --------------------------------------------------------------------------------------- + // Starting participants + + /// Starts one agent, in the launcher's configured mode, and identifies it. + pub async fn launch_agent(&mut self, spec: AgentLaunch) -> Result { + let threads = self.budget.allocate(&spec.agent_id, spec.worker_threads)?; + let identity = WorkerIdentity { + session_id: spec.session_id.clone(), + client_id: spec.client_id.clone(), + service: spec.service.clone(), + worker_id: spec.agent_id.clone(), + incarnation_id: spec.incarnation_id.clone(), + role: Role::Agent, + port_id: Some(spec.port_id.clone()), + worker_threads: threads, + }; + let started = self.start_participant(&identity, Started::Agent(spec.clone()), threads).await; + match started { + Ok(worker) => { + let identity = worker.identity.clone(); + self.workers.insert(spec.agent_id.clone(), worker); + Ok(identity) + } + Err(e) => { + self.budget.release(&spec.agent_id); + Err(e) + } + } + } + + /// Starts the environment, in the launcher's configured mode, and identifies it. + pub async fn launch_environment( + &mut self, + spec: EnvironmentLaunch, + ) -> Result { + let threads = self.budget.allocate(&spec.worker_id, spec.worker_threads)?; + let identity = WorkerIdentity { + session_id: spec.session_id.clone(), + client_id: spec.client_id.clone(), + service: spec.service.clone(), + worker_id: spec.worker_id.clone(), + incarnation_id: spec.incarnation_id.clone(), + role: Role::Environment, + port_id: None, + worker_threads: threads, + }; + let started = self + .start_participant(&identity, Started::Environment(spec.clone()), threads) + .await; + match started { + Ok(worker) => { + let identity = worker.identity.clone(); + self.workers.insert(spec.worker_id.clone(), worker); + Ok(identity) + } + Err(e) => { + self.budget.release(&spec.worker_id); + Err(e) + } + } + } + + async fn start_participant( + &mut self, + identity: &WorkerIdentity, + what: Started, + threads: usize, + ) -> Result { + let (body, status, listener) = match self.mode { + ExecutionMode::InProcess => { + let (handle, status) = self.serve_here(identity, &what).await?; + (Body::Task(Some(handle)), Some(status), None) + } + ExecutionMode::Thread => { + let (thread, listener) = self.serve_on_a_thread(identity, &what, threads).await?; + (Body::Thread(thread), None, Some(listener)) + } + ExecutionMode::Process => { + let (child, listener) = self.spawn_process(identity, &what, threads).await?; + (Body::Process(Some(child)), None, Some(listener)) + } + }; + let pid = match &body { + Body::Process(Some(child)) => Some(child.id()), + _ => None, + }; + // The registration is discovered, not assumed: the launcher says hello with the + // identity it configured, and the reply is what the coordinator later pins. + let identified = self.identify(identity).await; + let (service_incarnation, domain_incarnation) = match identified { + Ok(pair) => pair, + Err(e) => { + let mut dying = LaunchedWorker { + identity: identity.clone(), + service_incarnation: String::new(), + domain_incarnation: identity.incarnation_id.clone(), + pid, + peak_rss_kib: 0, + body, + status, + _listener: listener, + }; + terminate(&mut dying); + return Err(e); + } + }; + let mut worker = LaunchedWorker { + identity: identity.clone(), + service_incarnation, + domain_incarnation, + pid, + peak_rss_kib: 0, + body, + status, + _listener: listener, + }; + worker.refresh_rss(); + Ok(worker) + } + + async fn serve_here( + &self, + identity: &WorkerIdentity, + what: &Started, + ) -> Result<(WorkerHandle, StatusCell), DomainError> { + let client = self + .endpoints + .connect(&identity.client_id) + .await + .map_err(|e| launch_error(&identity.worker_id, &e))?; + let service = register_with_retry(&client, &identity.service, self.health.boot) + .await + .map_err(|e| launch_error(&identity.worker_id, &e))?; + Ok(match what { + Started::Agent(spec) => { + let endpoint = FakeAgentWorker::new(agent_config(spec, identity.worker_threads)); + let status = endpoint.status(); + (serve(client, service, endpoint), status) + } + Started::Environment(spec) => { + let endpoint = CounterEnvironment::new(environment_config(spec)); + let status = endpoint.status(); + (serve(client, service, endpoint), status) + } + }) + } + + async fn serve_on_a_thread( + &self, + identity: &WorkerIdentity, + what: &Started, + threads: usize, + ) -> Result<(ThreadWorker, UnixListenerHandle), DomainError> { + let (path, listener) = self + .endpoints + .endpoint_for(&identity.client_id) + .await + .map_err(|e| launch_error(&identity.worker_id, &e))?; + let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>(); + let (ready_tx, ready_rx) = std::sync::mpsc::channel::>(); + let client_id = identity.client_id.clone(); + let service_name = identity.service.clone(); + let store_root = self.endpoints.store_root.clone(); + let what = what.clone(); + let worker_threads = identity.worker_threads; + let join = std::thread::Builder::new() + .name(format!("fly-session-{}", identity.worker_id)) + .spawn(move || { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(threads) + .enable_all() + .build(); + let runtime = match runtime { + Ok(runtime) => runtime, + Err(e) => { + let _ = ready_tx.send(Err(format!("runtime: {e}"))); + return; + } + }; + runtime.block_on(async move { + let served = serve_one( + &path, + &client_id, + &service_name, + &store_root, + &what, + worker_threads, + ) + .await; + let handle = match served { + Ok(handle) => handle, + Err(e) => { + let _ = ready_tx.send(Err(e)); + return; + } + }; + let _ = ready_tx.send(Ok(())); + // The thread lives until the launcher drops its stop sender, whether the + // worker answered Shutdown or not. + let _ = stop_rx.await; + handle.stop().await; + }); + }) + .map_err(|e| { + DomainError::new( + ErrorCode::Internal, + format!("{}: thread: {e}", identity.worker_id), + MutationCertainty::None, + ) + })?; + match ready_rx.recv_timeout(self.health.boot) { + Ok(Ok(())) => Ok((ThreadWorker { stop: Some(stop_tx), join: Some(join) }, listener)), + Ok(Err(e)) => { + drop(stop_tx); + let _ = join.join(); + Err(DomainError::new( + ErrorCode::BackendFailure, + format!("{}: {e}", identity.worker_id), + MutationCertainty::None, + )) + } + Err(_) => { + drop(stop_tx); + Err(DomainError::new( + ErrorCode::BackendFailure, + format!("{} did not register within its boot budget", identity.worker_id), + MutationCertainty::None, + )) + } + } + } + + async fn spawn_process( + &self, + identity: &WorkerIdentity, + what: &Started, + threads: usize, + ) -> Result<(std::process::Child, UnixListenerHandle), DomainError> { + let (path, listener) = self + .endpoints + .endpoint_for(&identity.client_id) + .await + .map_err(|e| launch_error(&identity.worker_id, &e))?; + let mut command = std::process::Command::new(&self.program); + command + .arg(what.subcommand()) + .arg("--socket") + .arg(&path) + .arg("--store-root") + .arg(&self.endpoints.store_root) + .arg("--client-id") + .arg(&identity.client_id) + .arg("--service") + .arg(&identity.service) + .arg("--threads") + .arg(threads.to_string()); + for (flag, value) in what.arguments() { + command.arg(flag).arg(value); + } + command.stdin(std::process::Stdio::null()); + let child = command.spawn().map_err(|e| { + DomainError::new( + ErrorCode::BackendFailure, + format!( + "{}: starting {}: {e}", + identity.worker_id, + self.program.display() + ), + MutationCertainty::None, + ) + })?; + Ok((child, listener)) + } + + // --------------------------------------------------------------------------------------- + // Identity and health + + /// Says hello as the supervisor and returns the registration and domain incarnation. + /// + /// The `expectedWorkerId` and role are the launcher's own configuration, so a participant + /// that is not the one the composition configured is refused here, before the coordinator + /// has pinned anything. + async fn identify(&mut self, identity: &WorkerIdentity) -> Result<(String, Id), DomainError> { + let params = HelloParams { + session_id: identity.session_id.clone(), + expected_worker_id: identity.worker_id.clone(), + role: identity.role, + supported_majors: vec![1], + }; + let request_id = DomainRequestId::from_serial(self.next_serial()); + let deadline = Instant::now() + self.health.boot; + loop { + let payload = object( + SessionRpcRequest { + request_id: request_id.clone(), + scope: None, + params: Value::Object(object(params.to_json())), + } + .to_json(), + ); + // The registration is not pinned yet: this call is how the launcher learns it. + let pending = self + .supervisor + .call(&identity.service, None, "Worker.Hello", payload, &[]) + .await; + match pending { + Ok(mut pending) => { + let service_incarnation = pending.service_incarnation().to_owned(); + let result = tokio::time::timeout(self.health.fail, pending.result()).await; + let result = match result { + Ok(Ok(result)) => result, + // The registration this call reached was the predecessor's, which is + // still letting go: a handover, not the new worker's answer. Nothing + // is adopted from it -- the loop asks again for the live one. + Ok(Err(e)) if handover(&e) && Instant::now() < deadline => { + tokio::time::sleep(Duration::from_millis(5)).await; + continue; + } + Ok(Err(e)) => return Err(launch_error(&identity.worker_id, &e)), + Err(_) => { + return Err(DomainError::new( + ErrorCode::BackendFailure, + format!("{} did not answer Worker.Hello", identity.worker_id), + MutationCertainty::Unknown, + )); + } + }; + let outcome = + SessionRpcOutcome::from_json(&Value::Object(result.outcome().clone())) + .map_err(|e| { + DomainError::invalid(format!( + "{}: unreadable Worker.Hello outcome: {e}", + identity.worker_id + )) + })?; + let value = outcome_result(&outcome)?; + let hello: HelloResult = HelloResult::from_json(value).map_err(|e| { + DomainError::invalid(format!( + "{}: unreadable HelloResult: {e}", + identity.worker_id + )) + })?; + if hello.worker_id != identity.worker_id + || hello.role != identity.role + || hello.incarnation_id != identity.incarnation_id + { + return Err(DomainError::before( + ErrorCode::IdentityMismatch, + format!( + "{} answered as another worker, role or incarnation", + identity.worker_id + ), + )); + } + return Ok((service_incarnation, hello.incarnation_id)); + } + Err(e) if handover(&e) && Instant::now() < deadline => { + // Still starting: it has not registered its service yet, or the + // registration it replaces has not finished closing. + tokio::time::sleep(Duration::from_millis(5)).await; + } + Err(e) => return Err(launch_error(&identity.worker_id, &e)), + } + } + } + + /// Asks one participant for its status, bounded by the supervisor's own clock. + /// + /// The answer never waits for a numerical operation, so this is health and not progress: + /// a worker in the middle of a mutation still answers. + pub async fn health_check(&mut self, worker_id: &Id) -> Result { + let Some(worker) = self.workers.get(worker_id) else { + return Err(DomainError::before( + ErrorCode::IdentityMismatch, + format!("{worker_id} is not a launched participant"), + )); + }; + let service = worker.identity.service.clone(); + let incarnation = worker.service_incarnation.clone(); + let request_id = DomainRequestId::from_serial(self.next_serial()); + let payload = object( + SessionRpcRequest { + request_id, + scope: None, + params: Value::Object(Map::new()), + } + .to_json(), + ); + let call = self + .supervisor + .call(&service, Some(&incarnation), "Worker.Status", payload, &[]); + let result = match tokio::time::timeout(self.health.fail, call).await { + Ok(Ok(mut pending)) => { + match tokio::time::timeout(self.health.fail, pending.result()).await { + Ok(Ok(result)) => result, + Ok(Err(e)) => return Err(launch_error(worker_id, &e)), + Err(_) => return Err(unresponsive(worker_id)), + } + } + Ok(Err(e)) => return Err(launch_error(worker_id, &e)), + Err(_) => return Err(unresponsive(worker_id)), + }; + let outcome = SessionRpcOutcome::from_json(&Value::Object(result.outcome().clone())) + .map_err(|e| DomainError::invalid(format!("{worker_id}: {e}")))?; + let value = outcome_result(&outcome)?; + let status = StatusResult::from_json(value) + .map_err(|e| DomainError::invalid(format!("{worker_id}: {e}")))?; + if let Some(worker) = self.workers.get_mut(worker_id) { + worker.refresh_rss(); + } + Ok(status) + } + + /// Health-checks every participant, and reports each one's answer or its failure. + pub async fn health_check_all(&mut self) -> Vec<(Id, Result)> { + let mut out = Vec::new(); + for worker_id in self.worker_ids() { + let status = self.health_check(&worker_id).await; + out.push((worker_id, status)); + } + out + } + + // --------------------------------------------------------------------------------------- + // Ending participants + + /// Asks one participant to stop, then makes sure it has. + /// + /// `Worker.Shutdown` is the supervisor's request; the operating system is its guarantee. + /// A participant that does not stop within the budget is terminated, which is reported as + /// such rather than as a clean stop. + pub async fn reap(&mut self, worker_id: &Id, reason: &str) -> ReapOutcome { + let Some(worker) = self.workers.get(worker_id) else { + return ReapOutcome::AlreadyGone; + }; + let service = worker.identity.service.clone(); + let incarnation = worker.service_incarnation.clone(); + let request_id = DomainRequestId::from_serial(self.next_serial()); + let params = ShutdownParams { reason: parse_id(reason).unwrap_or_else(|_| id("stop")) }; + let payload = object( + SessionRpcRequest { + request_id, + scope: None, + params: Value::Object(object(params.to_json())), + } + .to_json(), + ); + let asked = tokio::time::timeout( + self.health.probe, + self.supervisor.call_and_wait( + &service, + Some(&incarnation), + "Worker.Shutdown", + payload, + &[], + ), + ) + .await; + let answered = matches!(asked, Ok(Ok(_))); + let mut worker = self.workers.remove(worker_id).expect("just looked it up"); + self.budget.release(worker_id); + worker.refresh_rss(); + match &mut worker.body { + Body::Task(handle) => { + match handle.take() { + Some(handle) => { + handle.stop().await; + if answered { ReapOutcome::Stopped } else { ReapOutcome::Terminated } + } + None => ReapOutcome::AlreadyGone, + } + } + Body::Thread(thread) => { + thread.stop(); + if answered { ReapOutcome::Stopped } else { ReapOutcome::Terminated } + } + Body::Process(child) => match child.take() { + Some(mut child) => wait_or_terminate(&mut child, self.health.probe, answered), + None => ReapOutcome::AlreadyGone, + }, + } + } + + /// Reaps every participant. Used at the end of a session and on every failure path. + pub async fn reap_all(&mut self, reason: &str) -> Vec<(Id, ReapOutcome)> { + let mut out = Vec::new(); + for worker_id in self.worker_ids() { + let outcome = self.reap(&worker_id, reason).await; + out.push((worker_id, outcome)); + } + out + } + + /// Ends one participant without asking it, as a crash would. + /// + /// This is the deliberate injection behind the worker-death row: the process is killed, + /// the thread's runtime is dropped, or the serving task is aborted, and in every case the + /// registration and every owner the connection held go with it. + pub async fn kill(&mut self, worker_id: &Id) -> ReapOutcome { + let Some(mut worker) = self.workers.remove(worker_id) else { + return ReapOutcome::AlreadyGone; + }; + self.budget.release(worker_id); + worker.refresh_rss(); + match &mut worker.body { + Body::Task(handle) => match handle.take() { + Some(handle) => { + handle.stop().await; + ReapOutcome::Terminated + } + None => ReapOutcome::AlreadyGone, + }, + Body::Thread(thread) => { + thread.stop(); + ReapOutcome::Terminated + } + Body::Process(child) => match child.take() { + Some(mut child) => { + let _ = child.kill(); + let _ = child.wait(); + ReapOutcome::Terminated + } + None => ReapOutcome::AlreadyGone, + }, + } + } + + /// The peak resident set of every participant with a process of its own, in KiB, plus the + /// coordinator's own. + pub fn peak_rss_kib(&mut self) -> BTreeMap { + let mut out = BTreeMap::new(); + out.insert( + "coordinator".to_owned(), + crate::metrics::peak_rss_kib().unwrap_or_default(), + ); + for (worker_id, worker) in &mut self.workers { + worker.refresh_rss(); + if worker.pid.is_some() { + out.insert(worker_id.clone(), worker.peak_rss_kib); + } + } + out + } +} + +impl Drop for Launcher { + /// A launcher that goes away takes its participants with it. Leaving a child process + /// behind would be a leak the supervisor is exactly responsible for not producing. + fn drop(&mut self) { + for worker in self.workers.values_mut() { + terminate(worker); + } + } +} + +fn terminate(worker: &mut LaunchedWorker) { + match &mut worker.body { + Body::Task(handle) => { + if let Some(handle) = handle.take() { + handle.abort(); + } + } + Body::Thread(thread) => thread.stop(), + Body::Process(child) => { + if let Some(mut child) = child.take() { + let _ = child.kill(); + let _ = child.wait(); + } + } + } +} + +fn wait_or_terminate( + child: &mut std::process::Child, + budget: Duration, + answered: bool, +) -> ReapOutcome { + let deadline = Instant::now() + budget; + loop { + match child.try_wait() { + Ok(Some(_)) => { + return if answered { ReapOutcome::Stopped } else { ReapOutcome::Terminated }; + } + Ok(None) => { + if Instant::now() >= deadline { + let _ = child.kill(); + let _ = child.wait(); + return ReapOutcome::Terminated; + } + std::thread::sleep(Duration::from_millis(2)); + } + Err(_) => return ReapOutcome::AlreadyGone, + } + } +} + +/// True for a refusal that means "the service this call reached is not the live one yet". +/// +/// A replacement participant registers only once its predecessor's connection has finished +/// closing, so until then a call to that name either finds no route or reaches the route the +/// predecessor is still holding. Neither is an answer, and neither is adopted. +fn handover(e: &flybus::BusError) -> bool { + matches!( + e.code, + flybus::ErrorCode::NoService + | flybus::ErrorCode::CallGone + | flybus::ErrorCode::TargetChanged + ) +} + +fn unresponsive(worker_id: &Id) -> DomainError { + DomainError::new( + ErrorCode::BackendFailure, + format!("{worker_id} did not answer within the supervisor's budget"), + MutationCertainty::Unknown, + ) +} + +fn launch_error(worker_id: &Id, e: &flybus::BusError) -> DomainError { + let mutation = match e.dispatch { + flybus::Dispatch::NotDispatched => MutationCertainty::None, + flybus::Dispatch::Dispatched | flybus::Dispatch::Unknown => MutationCertainty::Unknown, + }; + let code = match e.code { + flybus::ErrorCode::NoService | flybus::ErrorCode::TargetChanged => { + ErrorCode::IdentityMismatch + } + flybus::ErrorCode::NotAuthorized => ErrorCode::IdentityMismatch, + flybus::ErrorCode::Backpressure | flybus::ErrorCode::QuotaExceeded => ErrorCode::Busy, + _ => ErrorCode::BackendFailure, + }; + DomainError::new( + code, + format!("{worker_id}: bus {:?}: {}", e.code, e.message), + mutation, + ) +} + +// ------------------------------------------------------------------------------------------- +// What a participant is + +/// The endpoint a launched participant serves. +#[derive(Clone, Debug)] +pub(crate) enum Started { + Agent(AgentLaunch), + Environment(EnvironmentLaunch), +} + +impl Started { + pub(crate) fn subcommand(&self) -> &'static str { + match self { + Started::Agent(_) => "agent", + Started::Environment(_) => "environment", + } + } + + /// The arguments a separate process needs to be exactly this participant. + fn arguments(&self) -> Vec<(String, String)> { + 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(), + ), + ]; + if let Some(step) = spec.faults.fail_commit_at_step { + args.push(("--fail-commit-at-step".to_owned(), step.to_string())); + } + 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(), + ), + ]; + if let Some(boundary) = spec.faults.omit_view_at_boundary { + args.push(("--omit-view-at-boundary".to_owned(), boundary.to_string())); + } + args + } + } + } +} + +pub(crate) fn agent_config(spec: &AgentLaunch, worker_threads: usize) -> AgentConfig { + AgentConfig { + session_id: spec.session_id.clone(), + agent_id: spec.agent_id.clone(), + incarnation_id: spec.incarnation_id.clone(), + tick_duration: spec.tick_duration, + warmup_ticks: spec.warmup_ticks, + worker_threads, + faults: spec.faults.clone(), + } +} + +pub(crate) fn environment_config(spec: &EnvironmentLaunch) -> EnvironmentConfig { + EnvironmentConfig { + session_id: spec.session_id.clone(), + worker_id: spec.worker_id.clone(), + incarnation_id: spec.incarnation_id.clone(), + step_duration: spec.step_duration, + ports: spec.ports.clone(), + faults: spec.faults.clone(), + } +} + +/// Connects, registers and serves one participant. Used by a thread with its own runtime and, +/// through the worker subcommand, by a separate process. +pub(crate) async fn serve_one( + socket: &Path, + client_id: &str, + service_name: &str, + store_root: &Path, + what: &Started, + worker_threads: usize, +) -> Result { + let client = Client::connect_unix(socket, ClientConfig::new(client_id, store_root)) + .await + .map_err(|e| format!("connect: {}", e.message))?; + let service = register_with_retry(&client, service_name, Duration::from_secs(30)) + .await + .map_err(|e| format!("register: {}", e.message))?; + Ok(match what { + Started::Agent(spec) => serve( + client, + service, + FakeAgentWorker::new(agent_config(spec, worker_threads)), + ), + Started::Environment(spec) => { + serve(client, service, CounterEnvironment::new(environment_config(spec))) + } + }) +} + +/// Registers one exclusive service name, waiting out a predecessor that is still letting go. +/// +/// Registration is exclusive, so a replacement worker meets `CONFLICT` until the connection it +/// replaces has finished closing and the router has released its routes. That is a handover, +/// not a refusal, so it is waited out; every other refusal is returned as it stands. +pub(crate) async fn register_with_retry( + client: &Client, + name: &str, + budget: Duration, +) -> Result { + let deadline = Instant::now() + budget; + loop { + match client.register(name, ServiceConfig::default()).await { + Ok(service) => return Ok(service), + Err(e) if e.code == flybus::ErrorCode::Conflict && Instant::now() < deadline => { + tokio::time::sleep(Duration::from_millis(5)).await; + } + Err(e) => return Err(e), + } + } +} diff --git a/services/flysim/crates/fly-session/src/lib.rs b/services/flysim/crates/fly-session/src/lib.rs index 7594b52..93aea5e 100644 --- a/services/flysim/crates/fly-session/src/lib.rs +++ b/services/flysim/crates/fly-session/src/lib.rs @@ -22,11 +22,15 @@ //! [`step-v1`]: https://example.invalid/step-v1 pub mod agent; +pub mod cli; pub mod clock; pub mod coordinator; pub mod dedup; pub mod environment; pub mod harness; +pub mod launcher; +pub mod measure; +pub mod metrics; pub mod phase; pub mod rpc; pub mod task; @@ -37,5 +41,8 @@ pub mod worker; pub mod types; pub use fly_session_types; -pub use coordinator::{Coordinator, DispatchOrder, Injections, SessionFailure, StepReport}; +pub use coordinator::{ + Coordinator, Deadlines, DispatchOrder, Injections, SessionFailure, StepReport, +}; +pub use launcher::{ExecutionMode, Launcher, ReapOutcome, ThreadBudget, Via}; pub use phase::{Phase, PhaseMachine}; diff --git a/services/flysim/crates/fly-session/src/measure.rs b/services/flysim/crates/fly-session/src/measure.rs new file mode 100644 index 0000000..18d32b7 --- /dev/null +++ b/services/flysim/crates/fly-session/src/measure.rs @@ -0,0 +1,331 @@ +//! The execution-mode comparison the implementation guide's section 5 asks for. +//! +//! It runs the same synthetic session in each execution mode, at one, two and four agents, +//! and reports the thread allocation, the RPC and critical-path percentiles, the memory peaks +//! and the router's owner, collection and queue counters. +//! +//! **These are local synthetic timings on one machine, and no host capacity claim follows +//! from any of them.** They exist so the three modes can be compared with each other: the +//! question this slice has to answer is what a process boundary costs, not how fast anything +//! is. Pacing is switched off for the run, so a transition follows the one before it as fast +//! as the participants answer and the samples are work rather than sleep. + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + +use crate::coordinator::DispatchOrder; +use crate::harness::{AgentSpec, HarnessConfig, SessionHarness, Via}; +use crate::launcher::ExecutionMode; +use crate::metrics::{Percentiles, physical_cores}; + +/// What to compare. +#[derive(Clone, Debug)] +pub struct MeasureConfig { + /// Transitions per run, after the warm-up ones. + pub steps: u64, + /// Transitions run before sampling starts, so first-call costs are not in the samples. + pub warmup_steps: u64, + pub agent_counts: Vec, + pub modes: Vec, + /// The within-agent worker count each agent asks its launcher for. + pub worker_threads: usize, +} + +impl Default for MeasureConfig { + fn default() -> MeasureConfig { + MeasureConfig { + steps: 200, + warmup_steps: 10, + agent_counts: vec![1, 2, 4], + modes: ExecutionMode::all().to_vec(), + worker_threads: 1, + } + } +} + +/// The highest value each router counter reached while the transitions ran. +#[derive(Debug, Default)] +struct Peaks { + owners: AtomicU64, + roots: AtomicU64, + queued: AtomicU64, + sealed: AtomicU64, + store_bytes: AtomicU64, +} + +impl Peaks { + fn observe(&self, stats: &flybus::RouterStats) { + raise(&self.owners, stats.owners as u64); + raise(&self.roots, stats.artifact_roots); + raise(&self.queued, stats.queued as u64); + raise(&self.sealed, stats.sealed_artifacts as u64); + raise(&self.store_bytes, stats.store_bytes); + } + + fn read(&self) -> (usize, u64, usize, usize, u64) { + ( + self.owners.load(Ordering::Relaxed) as usize, + self.roots.load(Ordering::Relaxed), + self.queued.load(Ordering::Relaxed) as usize, + self.sealed.load(Ordering::Relaxed) as usize, + self.store_bytes.load(Ordering::Relaxed), + ) + } +} + +fn raise(slot: &AtomicU64, value: u64) { + slot.fetch_max(value, Ordering::Relaxed); +} + +/// One measured composition. +#[derive(Clone, Debug)] +pub struct Row { + pub mode: ExecutionMode, + pub agents: usize, + pub worker_threads: usize, + pub physical_cores: usize, + pub budget_total: usize, + pub budget_used: usize, + pub steps: u64, + /// `Agent.Prepare`, over every agent and every transition. + pub prepare: Percentiles, + pub commit: Percentiles, + pub advance: Percentiles, + /// `Worker.Status`: an RPC with no domain work behind it, so it is the router and + /// transport floor rather than a measure of the worker. + pub status: Percentiles, + /// One whole transition, pacing excluded: the critical path. + pub step: Percentiles, + pub coordinator_peak_rss_kib: u64, + /// The sum of the peak resident sets of the participants with processes of their own. + pub participants_peak_rss_kib: u64, + pub owners_max: usize, + pub owners_final: usize, + pub artifact_roots_max: u64, + pub queued_max: usize, + pub sealed_max: usize, + pub sealed_final: usize, + pub store_bytes_max: u64, + pub store_bytes_final: u64, + /// One sealed frame per boundary, boundary zero included. + pub frames_produced: u64, +} + +impl Row { + /// Frames the store collected: produced, minus the ones still owned at the end. + pub fn collected(&self) -> u64 { + self.frames_produced.saturating_sub(self.sealed_final as u64) + } +} + +/// Runs the comparison. Every row is one composition in one mode. +pub async fn run(config: &MeasureConfig) -> Result, String> { + let mut rows = Vec::new(); + for mode in &config.modes { + for agents in &config.agent_counts { + rows.push(one(config, *mode, *agents).await?); + } + } + Ok(rows) +} + +async fn one(config: &MeasureConfig, mode: ExecutionMode, agents: usize) -> Result { + let dir = std::env::temp_dir().join(format!( + "fly-session-measure-{}-{agents}-{}", + mode.label(), + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?; + let result = measure_in(config, mode, agents, &dir).await; + let _ = std::fs::remove_dir_all(&dir); + result +} + +async fn measure_in( + config: &MeasureConfig, + mode: ExecutionMode, + agents: usize, + dir: &std::path::Path, +) -> Result { + let specs: Vec = (0..agents) + .map(|i| AgentSpec { + worker_threads: config.worker_threads, + ..AgentSpec::new(&format!("fly-{}", (b'a' + i as u8) as char), &format!("p{}", i + 1), 7 + i as i32) + }) + .collect(); + let harness_config = HarnessConfig { + agents: specs, + mode, + ..HarnessConfig::default() + }; + let budget = harness_config.budget().map_err(|e| e.to_string())?; + let (budget_total, budget_used_floor) = (budget.total(), harness_config.required_threads()); + let mut harness = SessionHarness::start(Via::Unix, dir, harness_config) + .await + .map_err(|e| format!("{}: start: {}", mode.label(), e.message))?; + harness.coordinator.dispatch = DispatchOrder::Concurrent; + harness.coordinator.disable_pacing(); + harness + .coordinator + .bootstrap() + .await + .map_err(|e| format!("{}: bootstrap: {e}", mode.label()))?; + // The warm-up transitions pay the first-call costs; their samples are then discarded. + harness + .coordinator + .run(config.warmup_steps) + .await + .map_err(|e| format!("{}: warm-up: {e}", mode.label()))?; + harness.coordinator.metrics.clear(); + + // The router's counters are sampled *while* transitions run, not between them: a queue + // that is empty at every committed boundary says nothing about whether it stayed bounded + // during the transaction, which is the thing section 4 asks about. + let peaks = Arc::new(Peaks::default()); + let sampling = Arc::new(AtomicBool::new(true)); + let sampler = tokio::spawn({ + let router = harness.router().clone(); + let peaks = peaks.clone(); + let sampling = sampling.clone(); + async move { + while sampling.load(Ordering::Relaxed) { + peaks.observe(&router.stats()); + tokio::time::sleep(std::time::Duration::from_micros(200)).await; + } + peaks.observe(&router.stats()); + } + }); + + let environment = harness.environment_id(); + let mut status = crate::metrics::Metrics::default(); + for _ in 0..config.steps { + harness + .coordinator + .step() + .await + .map_err(|e| format!("{}: step: {e}", mode.label()))?; + // One status call per transition: an RPC the worker answers from a cell rather than + // from its endpoint, so it is the router-and-transport floor the domain calls sit on. + let started = std::time::Instant::now(); + let _ = harness.launcher.health_check(&environment).await; + status.record("Worker.Status", started.elapsed()); + } + sampling.store(false, Ordering::Relaxed); + let _ = sampler.await; + let (owners_max, roots_max, queued_max, sealed_max, store_bytes_max) = peaks.read(); + + let metrics = &harness.coordinator.metrics; + let zero = Percentiles::default(); + let row = Row { + mode, + agents, + worker_threads: config.worker_threads, + physical_cores: physical_cores(), + budget_total, + budget_used: budget_used_floor, + steps: config.steps, + prepare: metrics.percentiles("Agent.Prepare").unwrap_or(zero), + commit: metrics.percentiles("Agent.Commit").unwrap_or(zero), + advance: metrics.percentiles("Environment.Advance").unwrap_or(zero), + status: status.percentiles("Worker.Status").unwrap_or(zero), + step: metrics.percentiles("step").unwrap_or(zero), + coordinator_peak_rss_kib: crate::metrics::peak_rss_kib().unwrap_or_default(), + participants_peak_rss_kib: harness + .launcher + .peak_rss_kib() + .iter() + .filter(|(who, _)| who.as_str() != "coordinator") + .map(|(_, kib)| *kib) + .sum(), + owners_max, + owners_final: harness.router_stats().owners, + artifact_roots_max: roots_max, + queued_max, + sealed_max, + sealed_final: harness.router_stats().sealed_artifacts, + store_bytes_max, + store_bytes_final: harness.router_stats().store_bytes, + frames_produced: config.steps + config.warmup_steps + 1, + }; + harness.shutdown().await; + Ok(row) +} + +/// The measurement table, as Markdown. +pub fn table(rows: &[Row]) -> String { + let mut out = String::new(); + out.push_str( + "Local synthetic timings on one machine. Not a capacity claim, and not a latency goal.\n\n", + ); + if let Some(first) = rows.first() { + out.push_str(&format!( + "Physical cores: {}. Coordinator reservation: 1 thread.\n\n", + first.physical_cores + )); + } + out.push_str( + "| mode | agents | threads/agent | budget used/total | Prepare p50/p95/p99 us | \ +Commit p50/p95/p99 us | Advance p50/p95/p99 us | Status p50/p99 us | step p50/p95/p99 us |\n", + ); + out.push_str("| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n"); + for r in rows { + out.push_str(&format!( + "| {} | {} | {} | {}/{} | {:.0}/{:.0}/{:.0} | {:.0}/{:.0}/{:.0} | \ +{:.0}/{:.0}/{:.0} | {:.0}/{:.0} | {:.0}/{:.0}/{:.0} |\n", + r.mode.label(), + r.agents, + r.worker_threads, + r.budget_used, + r.budget_total, + r.prepare.p50_us(), + r.prepare.p95_us(), + r.prepare.p99_us(), + r.commit.p50_us(), + r.commit.p95_us(), + r.commit.p99_us(), + r.advance.p50_us(), + r.advance.p95_us(), + r.advance.p99_us(), + r.status.p50_us(), + r.status.p99_us(), + r.step.p50_us(), + r.step.p95_us(), + r.step.p99_us(), + )); + } + out.push('\n'); + out.push_str( + "| mode | agents | coordinator peak RSS KiB | participant peak RSS KiB | owners max/final | \ +roots max | queued max | sealed max/final | store bytes max/final | frames produced/collected |\n", + ); + out.push_str("| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n"); + for r in rows { + out.push_str(&format!( + "| {} | {} | {} | {} | {}/{} | {} | {} | {}/{} | {}/{} | {}/{} |\n", + r.mode.label(), + r.agents, + r.coordinator_peak_rss_kib, + r.participants_peak_rss_kib, + r.owners_max, + r.owners_final, + r.artifact_roots_max, + r.queued_max, + r.sealed_max, + r.sealed_final, + r.store_bytes_max, + r.store_bytes_final, + r.frames_produced, + r.collected(), + )); + } + out +} + +/// The rows keyed by mode and agent count, for a caller that wants one of them. +pub fn by_composition(rows: &[Row]) -> BTreeMap<(String, usize), Row> { + rows.iter() + .map(|r| ((r.mode.label().to_owned(), r.agents), r.clone())) + .collect() +} diff --git a/services/flysim/crates/fly-session/src/metrics.rs b/services/flysim/crates/fly-session/src/metrics.rs new file mode 100644 index 0000000..fd5d1a3 --- /dev/null +++ b/services/flysim/crates/fly-session/src/metrics.rs @@ -0,0 +1,176 @@ +//! Latency and resource samples, for the measurements the implementation guide's section 5 +//! asks every slice to report. +//! +//! These are local synthetic timings on one machine. No host capacity claim follows from any +//! number this module produces, and nothing here is a gameplay latency goal: the percentiles +//! exist so that the three execution modes can be compared against each other. + +use std::collections::BTreeMap; + +/// One sample set's order statistics, by nearest rank. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct Percentiles { + pub count: usize, + pub p50_ns: u64, + pub p95_ns: u64, + pub p99_ns: u64, + pub max_ns: u64, +} + +impl Percentiles { + fn of(sorted: &[u64]) -> Percentiles { + let rank = |p: f64| -> u64 { + if sorted.is_empty() { + return 0; + } + let n = sorted.len() as f64; + let index = (p * n).ceil() as usize; + sorted[index.clamp(1, sorted.len()) - 1] + }; + Percentiles { + count: sorted.len(), + p50_ns: rank(0.50), + p95_ns: rank(0.95), + p99_ns: rank(0.99), + max_ns: sorted.last().copied().unwrap_or_default(), + } + } + + pub fn p50_us(&self) -> f64 { + self.p50_ns as f64 / 1000.0 + } + + pub fn p95_us(&self) -> f64 { + self.p95_ns as f64 / 1000.0 + } + + pub fn p99_us(&self) -> f64 { + self.p99_ns as f64 / 1000.0 + } +} + +/// Named duration samples. One name is one measured path: a domain method, or the +/// coordinator's whole critical path for a transition. +#[derive(Clone, Debug, Default)] +pub struct Metrics { + samples: BTreeMap>, +} + +impl Metrics { + pub fn record(&mut self, what: &str, elapsed: std::time::Duration) { + let ns = u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX); + self.samples.entry(what.to_owned()).or_default().push(ns); + } + + pub fn percentiles(&self, what: &str) -> Option { + let mut values = self.samples.get(what)?.clone(); + values.sort_unstable(); + Some(Percentiles::of(&values)) + } + + pub fn names(&self) -> Vec { + self.samples.keys().cloned().collect() + } + + pub fn count(&self, what: &str) -> usize { + self.samples.get(what).map(Vec::len).unwrap_or_default() + } + + pub fn clear(&mut self) { + self.samples.clear(); + } +} + +/// This process's peak resident set, in KiB, from its own status file. +pub fn peak_rss_kib() -> Option { + peak_rss_of("/proc/self/status") +} + +/// One child process's peak resident set, in KiB. `None` once the child is gone. +pub fn peak_rss_kib_of(pid: u32) -> Option { + peak_rss_of(&format!("/proc/{pid}/status")) +} + +fn peak_rss_of(path: &str) -> Option { + let text = std::fs::read_to_string(path).ok()?; + for line in text.lines() { + if let Some(rest) = line.strip_prefix("VmHWM:") { + return rest.split_whitespace().next()?.parse().ok(); + } + } + None +} + +/// How many physical cores this machine has, counted as distinct (package, core) pairs. +/// +/// Falls back to the logical count, which is what a thread budget has to use when the +/// topology cannot be read. +pub fn physical_cores() -> usize { + if let Ok(text) = std::fs::read_to_string("/proc/cpuinfo") { + let mut pairs: std::collections::BTreeSet<(String, String)> = + std::collections::BTreeSet::new(); + let (mut package, mut core) = (None, None); + for line in text.lines() { + if line.trim().is_empty() { + if let (Some(p), Some(c)) = (package.take(), core.take()) { + pairs.insert((p, c)); + } + continue; + } + if let Some((key, value)) = line.split_once(':') { + match key.trim() { + "physical id" => package = Some(value.trim().to_owned()), + "core id" => core = Some(value.trim().to_owned()), + _ => {} + } + } + } + if let (Some(p), Some(c)) = (package, core) { + pairs.insert((p, c)); + } + if !pairs.is_empty() { + return pairs.len(); + } + } + logical_cores() +} + +/// How many hardware threads this machine reports. +pub fn logical_cores() -> usize { + std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn percentiles_use_nearest_rank() { + let mut m = Metrics::default(); + for ns in 1..=100u64 { + m.record("x", std::time::Duration::from_nanos(ns)); + } + let p = m.percentiles("x").expect("recorded"); + assert_eq!(p.count, 100); + assert_eq!(p.p50_ns, 50); + assert_eq!(p.p95_ns, 95); + assert_eq!(p.p99_ns, 99); + assert_eq!(p.max_ns, 100); + assert!(m.percentiles("y").is_none()); + } + + #[test] + fn a_single_sample_is_every_percentile() { + let mut m = Metrics::default(); + m.record("x", std::time::Duration::from_nanos(7)); + let p = m.percentiles("x").expect("recorded"); + assert_eq!((p.p50_ns, p.p95_ns, p.p99_ns, p.max_ns), (7, 7, 7, 7)); + } + + #[test] + fn the_machine_reports_at_least_one_core() { + assert!(physical_cores() >= 1); + assert!(logical_cores() >= 1); + assert!(peak_rss_kib().unwrap_or(1) > 0); + } +} diff --git a/services/flysim/crates/fly-session/src/worker.rs b/services/flysim/crates/fly-session/src/worker.rs index 7f0c92e..aeb6ed2 100644 --- a/services/flysim/crates/fly-session/src/worker.rs +++ b/services/flysim/crates/fly-session/src/worker.rs @@ -224,6 +224,22 @@ impl WorkerHandle { let _ = self.task.await; self.client.close().await; } + + /// Stops serving without waiting. The connection closes when the last handle to it is + /// dropped, which this does. For a supervisor's `Drop`, where there is no runtime to wait + /// on. + pub fn abort(self) { + self.task.abort(); + } + + /// Waits until the worker stops serving, which `Worker.Shutdown` makes it do. + /// + /// A worker process awaits this and then exits, so the supervisor's `Worker.Shutdown` and + /// the process's exit are the same event rather than two racing ones. + pub async fn join(self) { + let _ = self.task.await; + self.client.close().await; + } } /// Registers `service_name` and serves `endpoint` on it until the service ends or Shutdown. diff --git a/services/flysim/crates/fly-session/tests/common/mod.rs b/services/flysim/crates/fly-session/tests/common/mod.rs index c95912c..92e1866 100644 --- a/services/flysim/crates/fly-session/tests/common/mod.rs +++ b/services/flysim/crates/fly-session/tests/common/mod.rs @@ -4,7 +4,7 @@ use std::time::Duration; -use fly_session::harness::{HarnessConfig, SessionHarness, Via}; +use fly_session::harness::{ExecutionMode, HarnessConfig, SessionHarness, Via}; use fly_session::types::*; pub const WAIT: Duration = Duration::from_secs(20); @@ -94,3 +94,55 @@ pub async fn within(what: &str, f: impl std::future::Future) -> T Err(_) => panic!("{what}: timed out"), } } + +/// Generates one test per execution mode from an `async fn name(mode: ExecutionMode)`. +/// +/// The separate-process mode is the SESSION-02 subject; the other two are the variants it is +/// compared against, and a row that holds in one must hold in all three. +#[macro_export] +macro_rules! all_modes { + ($($name:ident),* $(,)?) => { + mod in_process { + $( + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn $name() { + super::$name($crate::common::mode_in_process()).await + } + )* + } + mod dedicated_thread { + $( + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn $name() { + super::$name($crate::common::mode_thread()).await + } + )* + } + mod separate_process { + $( + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn $name() { + super::$name($crate::common::mode_process()).await + } + )* + } + }; +} + +pub fn mode_in_process() -> ExecutionMode { + ExecutionMode::InProcess +} + +pub fn mode_thread() -> ExecutionMode { + ExecutionMode::Thread +} + +pub fn mode_process() -> ExecutionMode { + ExecutionMode::Process +} + +/// A fixture in one execution mode. The transport is the mode's own: a separate process +/// reaches the router only over a socket. +pub async fn mode_fixture(mode: ExecutionMode, config: HarnessConfig) -> Fixture { + fixture(Via::Unix, HarnessConfig { mode, ..config }).await +} diff --git a/services/flysim/crates/fly-session/tests/processes.rs b/services/flysim/crates/fly-session/tests/processes.rs new file mode 100644 index 0000000..248a060 --- /dev/null +++ b/services/flysim/crates/fly-session/tests/processes.rs @@ -0,0 +1,600 @@ +//! SESSION-02 acceptance: one agent process per fly and one environment process under the +//! coordinator, compared with the in-process and dedicated-thread variants. +//! +//! Every acceptance bullet is one named test here, generated once per execution mode, so a +//! rule that holds in one process holds across a process boundary too. The two process-mode +//! failure rows of section 4 that SESSION-01 could not reach in one process -- a router +//! restart during a world advance, and an old worker's reply after a restart -- are at the +//! end and run in the separate-process mode. + +mod common; + +use std::collections::BTreeMap; +use std::time::{Duration, Instant}; + +use common::{at, count, fly_a, fly_b, mode_fixture, within}; +use fly_session::agent::AgentFaults; +use fly_session::coordinator::{DispatchOrder, Injections}; +use fly_session::environment::EnvironmentFaults; +use fly_session::harness::{ExecutionMode, HarnessConfig, Via}; +use fly_session::launcher::{ReapOutcome, ThreadBudget}; +use fly_session::phase::Phase; +use fly_session::types::*; + +all_modes!( + a_delayed_one_agent_result_holds_the_world, + a_worker_death_has_a_bounded_diagnosed_outcome, + a_helper_death_has_a_bounded_diagnosed_outcome, + an_uncertain_advance_never_creates_a_second_batch, + a_partial_commit_never_permits_next_step_play, + every_participant_answers_its_supervisor, + worker_threads_lie_within_the_launcher_allocation, +); + +const STEPS: u64 = 4; + +fn two_agents(mode: ExecutionMode) -> HarnessConfig { + HarnessConfig { mode, ..HarnessConfig::default() } +} + +// ------------------------------------------------------------------------------------------- +// Acceptance: sequential, reversed and parallel completion produce equivalent traces + +/// `step-v1` section 8, across the process boundary: sequential, concurrent and reversed +/// dispatch, in all three execution modes, produce one behaviour trace. +/// +/// This reuses the wave-1 comparator -- the behaviour half of the section 8 trace, with +/// request ids, bus correlation and wall time excluded -- so "a process behaves like a task" +/// is the same assertion that "a reordered dispatch behaves like an ordered one" was. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn sequential_reversed_and_parallel_completion_agree() { + let mut behaviours: BTreeMap> = BTreeMap::new(); + for mode in ExecutionMode::all() { + for order in [ + DispatchOrder::Sequential, + DispatchOrder::Concurrent, + DispatchOrder::Reversed, + ] { + let mut config = two_agents(mode); + // Deliberately unequal completion times, so a concurrent run really does finish + // out of dispatch order whichever side of a process boundary the agents are on. + config.agents[0].faults = + AgentFaults { prepare_delay_ms: 12, ..AgentFaults::default() }; + config.agents[1].faults = AgentFaults { commit_delay_ms: 9, ..AgentFaults::default() }; + let mut f = mode_fixture(mode, config).await; + f.harness.coordinator.dispatch = order; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + within("run", f.harness.coordinator.run(STEPS)).await.unwrap(); + let behaviour = f.harness.coordinator.trace.behavior(); + assert_eq!(behaviour.len() as u64, STEPS); + behaviours.insert(format!("{}/{order:?}", mode.label()), behaviour); + f.shutdown().await; + } + } + let mut iter = behaviours.iter(); + let (first_name, first) = iter.next().expect("at least one run"); + for (name, behaviour) in iter { + assert_eq!( + behaviour, first, + "{name} produced a different behaviour trace from {first_name}" + ); + } +} + +// ------------------------------------------------------------------------------------------- +// Acceptance: a delayed one-agent result holds the world + +/// One agent takes far longer than the other to prepare. No `Environment.Advance` is sent +/// until every agent is Prepared, and the world is still at its old boundary while the +/// coordinator waits. +async fn a_delayed_one_agent_result_holds_the_world(mode: ExecutionMode) { + let mut config = two_agents(mode); + config.agents[1].faults = AgentFaults { prepare_delay_ms: 400, ..AgentFaults::default() }; + let mut f = mode_fixture(mode, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let environment = f.harness.environment_id(); + let before = within("progress", f.harness.progress_of(&environment)).await.unwrap(); + + let (coordinator, launcher) = f.harness.parts(); + // The supervisor watches the world while the transition is in flight. That is what a + // supervisor is for, and `Worker.Status` answers without waiting for a mutation. + let (stepped, held) = tokio::join!( + async { within("step", coordinator.step()).await }, + async { + tokio::time::sleep(Duration::from_millis(120)).await; + within("status", launcher.health_check(&environment)).await + } + ); + let report = stepped.expect("the transition completes once the slow agent answers"); + assert_eq!(report.boundary, 1); + let held = held.expect("the environment answers its supervisor during the wait"); + assert_eq!( + held.progress_counter, before, + "the world may not advance while one agent is still preparing" + ); + assert_eq!( + held.state, + WorkerState::Ready, + "the environment is at a committed boundary, not advancing" + ); + + // And the ordering the audit records says the same thing from the coordinator's side. + let audit = f.harness.coordinator.audit.clone(); + let advance = at(&audit, "advance:0"); + for agent in [fly_a(), fly_b()] { + assert!( + at(&audit, &format!("prepared:{agent}@0")) < advance, + "{agent} must be Prepared before the world advances: {audit:?}" + ); + } + assert_eq!(f.harness.coordinator.stats().advances, 1); + f.shutdown().await; +} + +// ------------------------------------------------------------------------------------------- +// Acceptance: worker or helper death has a bounded diagnosed outcome + +/// One agent dies in the middle of its Prepare. The epoch fails with a typed cause naming +/// that agent, within the caller's own budget, and nothing continues on the remainder. +async fn a_worker_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) { + let mut config = two_agents(mode); + config.agents[1].faults = AgentFaults { prepare_delay_ms: 5_000, ..AgentFaults::default() }; + let mut f = mode_fixture(mode, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let started = Instant::now(); + + let (coordinator, launcher) = f.harness.parts(); + let (stepped, reaped) = tokio::join!( + async { within("step", coordinator.step()).await }, + async { + tokio::time::sleep(Duration::from_millis(80)).await; + launcher.kill(&fly_b()).await + } + ); + assert_eq!(reaped, ReapOutcome::Terminated); + let failure = stepped.expect_err("a dead participant is a failed epoch, not a slow one"); + assert!( + started.elapsed() < Duration::from_secs(20), + "the outcome must be bounded, not a hang" + ); + assert_eq!( + failure.participant.as_deref(), + Some(fly_b().as_str()), + "the failure names the participant: {failure}" + ); + assert_ne!( + failure.error.mutation, + MutationCertainty::None, + "a participant that died mid-call leaves an uncertain mutation, never a clean none" + ); + assert_eq!(f.harness.coordinator.phase(), Phase::Failed); + assert!(f.harness.coordinator.is_fenced()); + // No partial continuation: no world step, no publication, and no next transition. + assert_eq!(f.harness.coordinator.stats().advances, 0); + assert_eq!(count(&f.harness.coordinator.audit, "publish:1"), 0); + let again = f.harness.coordinator.step().await.expect_err("a fenced epoch takes no step"); + assert_eq!(again.error.code, ErrorCode::InvalidPhase); + f.shutdown().await; +} + +/// The environment helper dies in the middle of the world advance. Same rule: a typed cause +/// naming it, bounded, and no half-transition afterwards. +async fn a_helper_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) { + let config = HarnessConfig { + environment_faults: EnvironmentFaults { + advance_delay_ms: 5_000, + ..EnvironmentFaults::default() + }, + ..two_agents(mode) + }; + let mut f = mode_fixture(mode, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let environment = f.harness.environment_id(); + let started = Instant::now(); + + let (coordinator, launcher) = f.harness.parts(); + let (stepped, reaped) = tokio::join!( + async { within("step", coordinator.step()).await }, + async { + tokio::time::sleep(Duration::from_millis(200)).await; + launcher.kill(&environment).await + } + ); + assert_eq!(reaped, ReapOutcome::Terminated); + let failure = stepped.expect_err("a dead world is a failed epoch"); + assert!(started.elapsed() < Duration::from_secs(20), "bounded, not a hang"); + assert_eq!( + failure.participant.as_deref(), + Some(environment.as_str()), + "the failure names the participant: {failure}" + ); + assert_ne!(failure.error.mutation, MutationCertainty::None); + assert_eq!(f.harness.coordinator.phase(), Phase::Failed); + assert!(f.harness.coordinator.is_fenced()); + assert_eq!(f.harness.coordinator.stats().advances, 0); + // The agents prepared and are not asked to prepare again or to commit anything. + assert_eq!(f.harness.coordinator.stats().commits, 0); + assert_eq!(count(&f.harness.coordinator.audit, "publish:1"), 0); + f.shutdown().await; +} + +// ------------------------------------------------------------------------------------------- +// Acceptance: an uncertain Advance never creates a second batch + +/// The Advance result is lost after the world already stepped. The coordinator resolves the +/// same operation against its original domain request id; the world advances once per +/// transition and the batch is never re-sent as a new one. +async fn an_uncertain_advance_never_creates_a_second_batch(mode: ExecutionMode) { + let clean = { + let mut f = mode_fixture(mode, two_agents(mode)).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + within("run", f.harness.coordinator.run(STEPS)).await.unwrap(); + let environment = f.harness.environment_id(); + let world = within("progress", f.harness.progress_of(&environment)).await.unwrap(); + let out = (f.harness.coordinator.trace.behavior(), world); + f.shutdown().await; + out + }; + + let mut f = mode_fixture(mode, two_agents(mode)).await; + f.harness.coordinator.injections = Injections { + at_step: 2, + lose_advance_result: true, + ..Injections::default() + }; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + within("run", f.harness.coordinator.run(STEPS)).await.unwrap(); + let environment = f.harness.environment_id(); + let world = within("progress", f.harness.progress_of(&environment)).await.unwrap(); + + assert_eq!(f.harness.coordinator.stats().advances, STEPS, "one advance per transition"); + assert_eq!( + world, clean.1, + "the world moved exactly as often as it did without the loss" + ); + assert_eq!( + f.harness.coordinator.trace.behavior(), + clean.0, + "an uncertain Advance changes no behaviour, so it created no second batch" + ); + // Every transition has exactly one batch, and every batch id is its own. + let batches: Vec = f + .harness + .coordinator + .trace + .transitions + .iter() + .map(|t| t.behaviour.batch_id.clone()) + .collect(); + let unique: std::collections::BTreeSet = batches.iter().cloned().collect(); + assert_eq!(unique.len(), batches.len(), "one batch id per transition: {batches:?}"); + let injections = f.harness.coordinator.injection_log.clone(); + assert!( + injections.iter().any(|o| o.what == "lost-advance-result" && o.identical), + "the loss must happen after dispatch, so the outcome really is uncertain: {injections:?}" + ); + f.shutdown().await; +} + +// ------------------------------------------------------------------------------------------- +// Acceptance: a partial Commit never permits next-step play + +/// One agent's Commit fails after the other's succeeded. The epoch fails naming that agent, +/// the boundary does not move, nothing is published and there is no next transition. +async fn a_partial_commit_never_permits_next_step_play(mode: ExecutionMode) { + let mut config = two_agents(mode); + config.agents[1].faults = + AgentFaults { fail_commit_at_step: Some(1), ..AgentFaults::default() }; + let mut f = mode_fixture(mode, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + within("step", f.harness.coordinator.step()).await.unwrap(); + let environment = f.harness.environment_id(); + + let failure = within("step", f.harness.coordinator.step()) + .await + .expect_err("one failed Commit fails the epoch"); + assert_eq!( + failure.participant.as_deref(), + Some(fly_b().as_str()), + "the failure names the agent whose Commit failed: {failure}" + ); + assert_eq!(f.harness.coordinator.phase(), Phase::Failed); + assert!(f.harness.coordinator.is_fenced()); + + // The world moved once inside the failing transition -- the Advance is what the Commit + // follows -- and it moves no further. There is no next-step play on a partial commit. + let world_before = within("progress", f.harness.progress_of(&environment)).await.unwrap(); + let again = f.harness.coordinator.step().await.expect_err("no play after a partial commit"); + assert_eq!(again.error.code, ErrorCode::InvalidPhase); + let world_after = within("progress", f.harness.progress_of(&environment)).await.unwrap(); + assert_eq!(world_after, world_before, "no next world step follows a partial commit"); + let status = within("status", f.harness.launcher.health_check(&environment)).await.unwrap(); + assert_eq!( + status.current_scope.unwrap().step, + 2, + "the world stays at the boundary the failed transition reached" + ); + let audit = f.harness.coordinator.audit.clone(); + assert_eq!(count(&audit, "publish:2"), 0); + assert_eq!(f.harness.coordinator.committed_boundary(), None); + f.shutdown().await; +} + +// ------------------------------------------------------------------------------------------- +// Supervision: identity, health and reaping + +/// Every participant answers the supervisor with the identity the launcher configured, and +/// stops when it is asked to. +async fn every_participant_answers_its_supervisor(mode: ExecutionMode) { + let mut f = mode_fixture(mode, two_agents(mode)).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + within("run", f.harness.coordinator.run(2)).await.unwrap(); + + let environment = f.harness.environment_id(); + for who in [fly_a(), fly_b(), environment.clone()] { + let worker = f.harness.launcher.worker(&who).expect("a launched participant"); + assert_eq!(worker.identity.worker_id, who); + assert_eq!(worker.domain_incarnation, worker.identity.incarnation_id); + assert!(!worker.service_incarnation.is_empty()); + let status = within("health", f.harness.launcher.health_check(&who)).await.unwrap(); + assert_eq!(status.state, WorkerState::Ready, "{who} is healthy at a boundary"); + } + // The agents carry their configured port identities; the environment owns the ports. + assert_eq!( + f.harness.launcher.worker(&fly_a()).unwrap().identity.port_id.as_deref(), + Some("p1") + ); + assert_eq!( + f.harness.launcher.worker(&fly_b()).unwrap().identity.port_id.as_deref(), + Some("p2") + ); + assert!(f.harness.launcher.worker(&environment).unwrap().identity.port_id.is_none()); + + // A worker that is not the one the caller expects refuses to negotiate at all. + let worker = f.harness.coordinator.agent_ref(&fly_a()).cloned().unwrap(); + let wrong = serde_json::json!({ + "sessionId": "demo", + "expectedWorkerId": "fly-z", + "role": "agent", + "supportedMajors": [1], + }); + let err = within( + "hello", + f.harness.coordinator.probe_raw(&worker, "Worker.Hello", None, wrong), + ) + .await + .expect_err("a worker is not whoever a caller says it is"); + assert_eq!(err.code, ErrorCode::IdentityMismatch); + + // Asking a participant to stop stops it, and the supervisor says which kind of stop it was. + let outcome = f.harness.launcher.reap(&fly_a(), "test").await; + assert_eq!(outcome, ReapOutcome::Stopped, "a live participant answers Worker.Shutdown"); + assert_eq!(f.harness.launcher.reap(&fly_a(), "test").await, ReapOutcome::AlreadyGone); + f.shutdown().await; +} + +/// `workers-v1`: `Agent.Initialize`'s `workerThreads` lies within the launcher allocation. +/// +/// The budget refuses an allocation it cannot cover before anything is started, and an agent +/// refuses an `Agent.Initialize` asking for more threads than its launcher gave it. +async fn worker_threads_lie_within_the_launcher_allocation(mode: ExecutionMode) { + // The budget itself: a total, a coordinator reservation, and a refusal that names both. + let mut budget = ThreadBudget::new(4, 1).unwrap(); + assert_eq!(budget.remaining(), 3); + assert_eq!(budget.allocate(&id("arena"), 1).unwrap(), 1); + assert_eq!(budget.allocate(&id("fly-a"), 2).unwrap(), 2); + let refused = budget.allocate(&id("fly-b"), 1).expect_err("the budget is spent"); + assert_eq!(refused.code, ErrorCode::Busy); + budget.release(&id("fly-a")); + assert_eq!(budget.allocate(&id("fly-b"), 1).unwrap(), 1); + assert_eq!(budget.allocate(&id("fly-b"), 1).expect_err("already held").code, ErrorCode::Conflict); + + // A composition the configured budget cannot cover never starts. + let config = HarnessConfig { + mode, + thread_budget: Some(2), + ..HarnessConfig::default() + }; + let dir = tempfile::tempdir().expect("a temporary directory"); + let refused = fly_session::harness::SessionHarness::start(Via::Unix, dir.path(), config).await; + let refused = refused.err().expect("two threads cannot hold a coordinator, a world and two flies"); + assert_eq!(refused.code, flybus::ErrorCode::QuotaExceeded, "{}", refused.message); + drop(dir); + + // And the worker's own check: it was launched with one thread, so an Initialize asking + // for eight is refused before the model is constructed. + let mut f = mode_fixture(mode, two_agents(mode)).await; + let worker = f.harness.coordinator.agent_ref(&fly_a()).cloned().unwrap(); + let profile = fly_session::agent::synthetic_profile( + &fly_a(), + &millis(1).unwrap(), + f.harness.config.warmup_ticks, + ); + let params = serde_json::json!({ + "agentId": "fly-a", + "profile": profile.to_json(), + "seed": 7, + "initialInput": {"boundary": "0", "views": [], "structured": null}, + "initialDecisionContext": { + "schema": fly_session::task::context_schema().to_json(), + "value": {}, + }, + "workerThreads": 8, + }); + let err = within( + "initialize", + f.harness.coordinator.probe_raw( + &worker, + "Agent.Initialize", + Some(scope_at("demo", "e1", 0)), + params, + ), + ) + .await + .expect_err("eight threads are not within a one-thread allocation"); + assert_eq!(err.code, ErrorCode::Busy); + assert_eq!(err.mutation, MutationCertainty::None, "nothing was constructed"); + // The allocation the coordinator actually sends is the one the launcher handed out. + assert_eq!(f.harness.launcher.worker(&fly_a()).unwrap().identity.worker_threads, 1); + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + f.shutdown().await; +} + +// ------------------------------------------------------------------------------------------- +// Section 4 rows SESSION-01 could not reach in one process + +/// Row: "Router restarts during a world advance | Old handles/routes invalid; epoch fails and +/// restores coherently." +/// +/// The restore half is STATE-01's. What SESSION-02 establishes is the half before it: the +/// epoch fails with a typed cause naming the participant the coordinator was talking to, the +/// session is fenced, every artifact handle of that store incarnation is gone, and no +/// boundary, publication or further transition follows. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_router_restart_during_a_world_advance_fences_the_epoch() { + let mode = ExecutionMode::Process; + let config = HarnessConfig { + environment_faults: EnvironmentFaults { + advance_delay_ms: 3_000, + ..EnvironmentFaults::default() + }, + ..two_agents(mode) + }; + let mut f = mode_fixture(mode, config).await; + // The router is gone in a moment, so the supervisor must not spend its full budget + // asking a participant that can no longer be reached. + f.harness.launcher.set_health_policy(fly_session::launcher::HealthPolicy { + probe: Duration::from_millis(200), + fail: Duration::from_millis(500), + boot: Duration::from_secs(30), + }); + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let boundary_before = f.harness.coordinator.observation().unwrap().boundary; + let router = f.harness.router().clone(); + let started = Instant::now(); + + let (coordinator, _launcher) = f.harness.parts(); + let (stepped, ()) = tokio::join!( + async { within("step", coordinator.step()).await }, + async { + // Mid-advance: the world has been asked to move and has not answered yet. + tokio::time::sleep(Duration::from_millis(250)).await; + router.shutdown(); + } + ); + let failure = stepped.expect_err("a lost router fails the epoch"); + assert!(started.elapsed() < Duration::from_secs(20), "bounded, not a hang"); + assert_eq!( + failure.participant.as_deref(), + Some(f.harness.environment_id().as_str()), + "the failure names the participant the coordinator was waiting for: {failure}" + ); + assert_ne!( + failure.error.mutation, + MutationCertainty::None, + "the world may have stepped; a lost router is never proof that it did not" + ); + assert_eq!(f.harness.coordinator.phase(), Phase::Failed); + assert!( + f.harness.coordinator.is_fenced(), + "old handles and routes are invalid from here on" + ); + assert_eq!(f.harness.coordinator.stats().advances, 0, "no boundary was committed"); + assert_eq!(count(&f.harness.coordinator.audit, "publish:1"), 0); + assert_eq!( + f.harness.coordinator.observation().unwrap().boundary, + boundary_before, + "the committed observation is still the one from before the advance" + ); + // Nothing reconnects into the active epoch: a new call on the old route is refused. + let again = f.harness.coordinator.step().await.expect_err("a fenced epoch takes no step"); + assert_eq!(again.error.code, ErrorCode::InvalidPhase); + f.shutdown().await; +} + +/// Row: "Old worker replies after restore | Stale epoch/incarnation rejected", with real +/// processes. +/// +/// A restarted agent is a new process, a new registration and a new domain incarnation. The +/// coordinator pinned the old registration, so its next call fails rather than reaching the +/// replacement; and the replacement, followed deliberately, refuses an operation from the +/// epoch the old process belonged to. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_old_worker_reply_after_a_restart_is_rejected_on_stale_epoch_or_incarnation() { + let mode = ExecutionMode::Process; + let mut f = mode_fixture(mode, two_agents(mode)).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + within("step", f.harness.coordinator.step()).await.unwrap(); + + let old = f.harness.coordinator.agent_ref(&fly_b()).cloned().unwrap(); + let old_pid = f.harness.launcher.worker(&fly_b()).unwrap().pid; + assert!(old_pid.is_some(), "a separate-process agent has a process of its own"); + let restarted = f.harness.restart_agent(&fly_b()).await.unwrap(); + let new_pid = f.harness.launcher.worker(&fly_b()).unwrap().pid; + assert_ne!(old_pid, new_pid, "a restart is a new process"); + assert_ne!( + restarted.service_incarnation, old.bus_incarnation, + "a replacement registration is a new incarnation" + ); + + // Following the new registration while still pinning the old worker's negotiated + // incarnation is rejected: this is the shape an old worker's reply would arrive in. + let stale = fly_session::rpc::WorkerRef { + service: restarted.service.clone(), + bus_incarnation: restarted.service_incarnation.clone(), + worker_id: fly_b(), + domain_incarnation: old.domain_incarnation.clone(), + }; + assert_ne!(old.domain_incarnation, Some(restarted.incarnation_id.clone())); + let err = within("status", f.harness.coordinator.status(&stale)) + .await + .expect_err("the replacement is not the incarnation this epoch negotiated"); + assert_eq!(err.error.code, ErrorCode::IdentityMismatch); + assert_eq!(err.participant.as_deref(), Some(fly_b().as_str())); + assert_eq!(f.harness.coordinator.phase(), Phase::Failed); + assert!(f.harness.coordinator.is_fenced()); + assert_eq!(f.harness.coordinator.stats().advances, 1, "no world step under a lost pin"); + f.shutdown().await; +} + +/// The other half of the same row: the replacement process is live and refuses an operation +/// naming the epoch the old process belonged to, rather than applying it to a fresh brain. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_restarted_worker_refuses_an_operation_from_the_old_epoch() { + let mode = ExecutionMode::Process; + let mut f = mode_fixture(mode, two_agents(mode)).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + within("step", f.harness.coordinator.step()).await.unwrap(); + let restarted = f.harness.restart_agent(&fly_b()).await.unwrap(); + + let replacement = fly_session::rpc::WorkerRef::new( + &restarted.service, + &restarted.service_incarnation, + &fly_b(), + ); + let params = serde_json::json!({ + "agentId": "fly-b", + "profileDigest": digest_of_bytes(b"whatever"), + "interval": {"numerator": "16666667", "denominator": "1"}, + "decisionContextDigest": digest_of_bytes(b"whatever"), + "preStepStimulations": [], + }); + let err = within( + "stale epoch", + f.harness.coordinator.probe_raw( + &replacement, + "Agent.Prepare", + Some(scope_at("demo", "e1", 1)), + params, + ), + ) + .await + .expect_err("an uninitialized replacement has no epoch to prepare in"); + assert!( + matches!(err.code, ErrorCode::StaleEpoch | ErrorCode::InvalidPhase), + "a replacement refuses the old epoch's work: {err}" + ); + assert_eq!(err.mutation, MutationCertainty::None, "nothing was applied to a fresh brain"); + assert_eq!(f.harness.coordinator.stats().advances, 1); + f.shutdown().await; +} diff --git a/services/flysim/crates/fly-session/tests/session.rs b/services/flysim/crates/fly-session/tests/session.rs index feb5d7f..bc7c7bf 100644 --- a/services/flysim/crates/fly-session/tests/session.rs +++ b/services/flysim/crates/fly-session/tests/session.rs @@ -379,12 +379,7 @@ async fn sequential_concurrent_and_reversed_orders_agree() { /// specific. async fn a_single_agent_composition_runs_the_same_transaction(via: Via) { let config = HarnessConfig { - agents: vec![AgentSpec { - agent_id: id("fly-a"), - port_id: id("p1"), - seed: 7, - faults: AgentFaults::default(), - }], + agents: vec![AgentSpec::new("fly-a", "p1", 7)], ..HarnessConfig::default() }; let mut f: Fixture = fixture(via, config).await;