alex/flybrain
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge feat/sf-session-02: per-fly processes, a launcher with thread budgets, and the fault behaviour

Browse source
This commit is contained in:
acamilo 2026-09-22 16:00:57 +00:00
commit ba545d9225
27 changed files with 4334 additions and 276 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
docs/design/session-framework/ipc-v1.md
View file

@ -94,10 +94,14 @@ interface HelloResult {
workerId: Id; incarnationId: Id; role: "agent" | "environment" | "coordinator";
buildDigest: Digest; contractDigest: Digest;
capabilities: Id[];
limits: { maxAgents: number; maxPorts: number };
limits: { maxAgents: number; maxPorts: number; workerThreads: number };
}
```
`limits.workerThreads` is the thread allocation the worker's launcher started it within; it
is an integer >=1 and its rule belongs to [worker interfaces](workers-v1.md) section 2, whose
2026-09-22 amendment added it.
The bus supplies caller identity; do not accept a forged caller in params. Bind a worker's
session authority to the expected coordinator identity/incarnation during negotiation and
initialization. Wrong worker/role, no common major or missing required capability refuses

10
docs/design/session-framework/workers-v1.md
View file

@ -99,6 +99,16 @@ interface AgentInitializeResult {
}
```
**Amendment, 2026-09-22 (SESSION-02).** `HelloResult.limits` gains `workerThreads`, an
integer >=1 reporting the allocation the launcher started that worker within, because
"within launcher allocation" above had no wire-level proof: the launcher passes the number to
the worker out of band, and a coordinator that is not also its own launcher had no contract
path to it. Hello is where a worker already proves its identity and reports its limits, so the
allocation belongs there. A caller asking for more than the worker reports is refused with
`BUSY` before the model is constructed, which this section already required; the amendment
only makes the number visible to whoever must respect it. It changes `contractDigest`, which
[session RPC](ipc-v1.md) section 4 already provides for.
The profile fixes warm-up/calibration behavior and supported schema versions. Validate inputs
and required roles before model construction. Install the initial sensory input, warm the
brain with learning disabled, calibrate the fixed readout and establish Ready(0). Do not

12
packages/session-types/src/workers.ts
View file

@ -43,6 +43,8 @@ export const MAX_ACKNOWLEDGE = 16;
export const MAX_ENGINE_FRAME_LEN = 64;
/** Not stated by a document; this crate's choice, published in the schema set. */
export const MAX_CAPABILITIES = 32;
/** The largest `workerThreads` a launcher may allocate to one worker (workers-v1 2). */
export const MAX_WORKER_THREADS = 4096;
export const MAX_SUPPORTED_MAJORS = 8;
export const MAX_MESSAGE_CODE_POINTS = 512;
@ -305,7 +307,7 @@ export function readAgentInitializeParams(value: unknown): AgentInitializeParams
seed: reader.int('seed', -2_147_483_648, 2_147_483_647),
initialInput: readSensoryInput(reader.value('initialInput')),
initialDecisionContext: readTypedValue(reader.value('initialDecisionContext')),
workerThreads: reader.int('workerThreads', 1, 4_096),
workerThreads: reader.int('workerThreads', 1, MAX_WORKER_THREADS),
};
reader.finish();
return params;
@ -776,7 +778,12 @@ export interface HelloResult {
buildDigest: Digest;
contractDigest: Digest;
capabilities: Id[];
limits: { maxAgents: number; maxPorts: number };
/**
* `workerThreads` is the thread allocation this worker was launched within, added by the
* 2026-09-22 amendment to workers-v1 section 2: the bound "within launcher allocation" had
* no wire on which a caller could learn the allocation.
*/
limits: { maxAgents: number; maxPorts: number; workerThreads: number };
}
export interface StatusResult {
@ -860,6 +867,7 @@ export function readHelloResult(value: unknown): HelloResult {
const limits = {
maxAgents: limitsReader.int('maxAgents', 1, MAX_AGENTS),
maxPorts: limitsReader.int('maxPorts', 1, MAX_PORTS),
workerThreads: limitsReader.int('workerThreads', 1, MAX_WORKER_THREADS),
};
limitsReader.finish();
reader.finish();

6
services/flysim/crates/fly-session-types/fixtures/contract-digest.json
View file

@ -1,9 +1,9 @@
{
"description": "contractDigest is the SHA-256 of the canonical schema set in schema-set.json.",
"contractDigest": "7932aef30c4d2d16e428081affc4e0ad187987f5b361138d553e54fd7f843b50",
"contractDigest": "d8f29a49b5df05ad8f75f7f5790a3f8cde9c5ad23a685137474c649c3c9da36d",
"schemaSetVersion": 1,
"schemaSetBytes": 26685,
"schemaSetBytes": 26814,
"types": 53,
"enums": 11,
"limits": 25
"limits": 26
}

76
services/flysim/crates/fly-session-types/fixtures/invalid.json
View file

@ -2802,7 +2802,8 @@
],
"limits": {
"maxAgents": 4,
"maxPorts": 4
"maxPorts": 4,
"workerThreads": 1
}
},
"reason": "an agent must advertise agent-step-v1"
@ -2823,7 +2824,8 @@
],
"limits": {
"maxAgents": 5,
"maxPorts": 4
"maxPorts": 4,
"workerThreads": 1
}
},
"reason": "the first composition allows four agents"
@ -2844,11 +2846,77 @@
],
"limits": {
"maxAgents": 4,
"maxPorts": 4
"maxPorts": 4,
"workerThreads": 1
}
},
"reason": "v1 selects major 1"
},
{
"name": "hello result without its launcher allocation",
"type": "HelloResult",
"value": {
"selectedMajor": 1,
"selectedMinor": 0,
"workerId": "fly-a",
"incarnationId": "inc-1",
"role": "agent",
"buildDigest": "44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66",
"contractDigest": "cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf",
"capabilities": [
"agent-step-v1"
],
"limits": {
"maxAgents": 4,
"maxPorts": 4
}
},
"reason": "limits.workerThreads is required by the 2026-09-22 workers-v1 amendment"
},
{
"name": "hello result promising more threads than a launcher may allocate",
"type": "HelloResult",
"value": {
"selectedMajor": 1,
"selectedMinor": 0,
"workerId": "fly-a",
"incarnationId": "inc-1",
"role": "agent",
"buildDigest": "44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66",
"contractDigest": "cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf",
"capabilities": [
"agent-step-v1"
],
"limits": {
"maxAgents": 4,
"maxPorts": 4,
"workerThreads": 4097
}
},
"reason": "limits.workerThreads is at most maxWorkerThreads"
},
{
"name": "hello result reporting no threads at all",
"type": "HelloResult",
"value": {
"selectedMajor": 1,
"selectedMinor": 0,
"workerId": "fly-a",
"incarnationId": "inc-1",
"role": "agent",
"buildDigest": "44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66",
"contractDigest": "cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf",
"capabilities": [
"agent-step-v1"
],
"limits": {
"maxAgents": 4,
"maxPorts": 4,
"workerThreads": 0
}
},
"reason": "limits.workerThreads is at least one"
},
{
"name": "hello params with no supported majors",
"type": "HelloParams",
@ -3897,4 +3965,4 @@
"reason": "a commit acknowledges the transition's next boundary"
}
]
}
}

2
services/flysim/crates/fly-session-types/fixtures/schema-set.json
View file

File diff suppressed because one or more lines are too long

14
services/flysim/crates/fly-session-types/fixtures/valid.json
View file

@ -1758,12 +1758,13 @@
],
"limits": {
"maxAgents": 4,
"maxPorts": 4
"maxPorts": 4,
"workerThreads": 1
}
},
"note": "",
"canonical": "{\"buildDigest\":\"44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66\",\"capabilities\":[\"agent-step-v1\",\"checkpoint-v1\",\"pixel-observation-v1\"],\"contractDigest\":\"cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf\",\"incarnationId\":\"inc-1\",\"limits\":{\"maxAgents\":4,\"maxPorts\":4},\"role\":\"agent\",\"selectedMajor\":1,\"selectedMinor\":0,\"workerId\":\"fly-a\"}",
"digest": "9f8cbc9dfe7a7532c2e41b53c4ce001973ca95703a81aad30ee2b1c8b640bc13"
"canonical": "{\"buildDigest\":\"44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66\",\"capabilities\":[\"agent-step-v1\",\"checkpoint-v1\",\"pixel-observation-v1\"],\"contractDigest\":\"cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf\",\"incarnationId\":\"inc-1\",\"limits\":{\"maxAgents\":4,\"maxPorts\":4,\"workerThreads\":1},\"role\":\"agent\",\"selectedMajor\":1,\"selectedMinor\":0,\"workerId\":\"fly-a\"}",
"digest": "262ea112e24dcdbdc1a5050b6dd6d031eace6a6a4cdca814913ac89fc9e39666"
},
{
"name": "hello result for an environment",
@ -1782,12 +1783,13 @@
],
"limits": {
"maxAgents": 1,
"maxPorts": 1
"maxPorts": 1,
"workerThreads": 1
}
},
"note": "",
"canonical": "{\"buildDigest\":\"44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66\",\"capabilities\":[\"world-step-v1\",\"checkpoint-v1\"],\"contractDigest\":\"cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf\",\"incarnationId\":\"inc-2\",\"limits\":{\"maxAgents\":1,\"maxPorts\":1},\"role\":\"environment\",\"selectedMajor\":1,\"selectedMinor\":0,\"workerId\":\"world\"}",
"digest": "0dc33a6cbf6bf66c4d28dc41b17b2eb7918b55c65469f4161e009f662f11bca2"
"canonical": "{\"buildDigest\":\"44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66\",\"capabilities\":[\"world-step-v1\",\"checkpoint-v1\"],\"contractDigest\":\"cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf\",\"incarnationId\":\"inc-2\",\"limits\":{\"maxAgents\":1,\"maxPorts\":1,\"workerThreads\":1},\"role\":\"environment\",\"selectedMajor\":1,\"selectedMinor\":0,\"workerId\":\"world\"}",
"digest": "06ced6fff0d810005ab2598cc9fb72a138303daef0787232871c9afa43b0eac0"
},
{
"name": "status result before initialization",

9
services/flysim/crates/fly-session-types/src/schema.rs
View file

@ -239,6 +239,11 @@ pub const LIMITS: &[LimitSchema] = &[
value: crate::workers::MAX_CAPABILITIES as u64,
source: "crate",
},
LimitSchema {
name: "maxWorkerThreads",
value: crate::workers::MAX_WORKER_THREADS,
source: "workers-v1 2",
},
LimitSchema {
name: "maxSupportedMajors",
value: crate::workers::MAX_SUPPORTED_MAJORS as u64,
@ -651,8 +656,8 @@ pub const SCHEMAS: &[TypeSchema] = &[
),
req(
"limits",
"{maxAgents:int,maxPorts:int}",
"1..=4 agents and 1..=4 ports",
"{maxAgents:int,maxPorts:int,workerThreads:int}",
"1..=4 agents, 1..=4 ports, and the launcher allocation this worker runs in",
),
],
},

19
services/flysim/crates/fly-session-types/src/workers.rs
View file

@ -33,6 +33,9 @@ pub const MAX_ACKNOWLEDGE: usize = 16;
pub const MAX_ENGINE_FRAME_LEN: usize = 64;
/// Negotiated capability ids. Not a stated bound; recorded in the schema set.
pub const MAX_CAPABILITIES: usize = 32;
/// The largest `workerThreads` a launcher may allocate to one worker (`workers-v1` 2).
pub const MAX_WORKER_THREADS: u64 = 4_096;
/// Supported majors in Worker.Hello. Not a stated bound; recorded in the schema set.
pub const MAX_SUPPORTED_MAJORS: usize = 8;
/// Domain error messages are <=512 code points (ipc-v1 section 7).
@ -651,7 +654,7 @@ impl DomainType for AgentInitializeParams {
let seed = i32_field(&mut f, "seed")?;
let initial_input = SensoryInput::from_json(f.value("initialInput")?)?;
let initial_decision_context = TypedValue::from_json(f.value("initialDecisionContext")?)?;
let worker_threads = f.int("workerThreads", 1, 4_096)?;
let worker_threads = f.int("workerThreads", 1, MAX_WORKER_THREADS)?;
f.finish()?;
let p = AgentInitializeParams {
agent_id,
@ -1958,6 +1961,13 @@ pub struct HelloResult {
pub capabilities: Vec<String>,
pub max_agents: u64,
pub max_ports: u64,
/// The thread allocation this worker was launched within.
///
/// `workers-v1` bounds `Agent.Initialize`'s `workerThreads` by "within launcher
/// allocation" and, before the 2026-09-22 amendment, named no wire on which a caller could
/// learn it. This is that wire: the worker reports what its launcher gave it, and a caller
/// that meant to ask for more finds out here rather than after the model exists.
pub worker_threads: u64,
}
impl HelloResult {
@ -1990,13 +2000,14 @@ impl DomainType for HelloResult {
let build_digest = f.string("buildDigest")?.to_owned();
let contract_digest = f.string("contractDigest")?.to_owned();
let capabilities = id_list(&mut f, "capabilities", 0, MAX_CAPABILITIES)?;
let (max_agents, max_ports) = {
let (max_agents, max_ports, worker_threads) = {
let v = f.value("limits")?;
let mut l = Fields::new(v, "HelloResult.limits")?;
let max_agents = l.int("maxAgents", 1, MAX_AGENTS as u64)?;
let max_ports = l.int("maxPorts", 1, MAX_PORTS as u64)?;
let worker_threads = l.int("workerThreads", 1, MAX_WORKER_THREADS)?;
l.finish()?;
(max_agents, max_ports)
(max_agents, max_ports, worker_threads)
};
f.finish()?;
let r = HelloResult {
@ -2008,6 +2019,7 @@ impl DomainType for HelloResult {
capabilities,
max_agents,
max_ports,
worker_threads,
};
r.validate()?;
Ok(r)
@ -2031,6 +2043,7 @@ impl DomainType for HelloResult {
obj(vec![
("maxAgents", Value::from(self.max_agents)),
("maxPorts", Value::from(self.max_ports)),
("workerThreads", Value::from(self.worker_threads)),
]),
),
])

11
services/flysim/crates/fly-session/Cargo.toml
View file

@ -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"

149
services/flysim/crates/fly-session/README.md
View file

@ -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,54 @@ Ready(k) ─ Prepare all agents concurrently ───────────
| `task` | The task and executor traits, the deterministic counter task, the identity executor |
| `rpc` | Domain calls: `req-<U64>` 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". The allocation is on the wire,
not only in the launcher's own record: `HelloResult.limits.workerThreads` reports it, under
the dated 2026-09-22 amendment to `workers-v1` section 2 that this slice added, so a
coordinator that is not also its own launcher can read the bound it has to respect.
- **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, incarnation or thread allocation
-- 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 +111,31 @@ 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.
- **The `ipc-v1` section 6 procedure, on the path that reaches it.** A call that goes two
seconds without a terminal reply is *uncertain*, not failed. The coordinator then queries
the same operation -- a fresh bus call carrying the original domain request id and body,
pinned to the same incarnation, with its retained attachments -- absorbing `IN_PROGRESS`
while the original is still running. Only when that ends without a definite answer, or the
incarnation is gone, or the retained result expired, is the epoch failed. A merely slow
participant therefore finishes its step, and `step-v1` section 7's "query/retransmit same
request to same incarnation; never new batch" is the same code path for a slow Advance.
The procedure has two explicit bounds, and they do not mean the same thing. **`resolve`, 8
seconds, is the working limit**: two to notice plus eight to resolve is section 6's ten
seconds without progress. **`resolve_attempts`, 8192, is a guard**, not the limit -- the
procedure pauses 2 ms between attempts, so the guard is over sixteen seconds of pauses
alone, twice the budget, and an attempt whose call expires costs a whole probe on top. At
these values the budget is always what fires. Which one did is recorded in
`Coordinator::last_resolution` and named in the failure's own message, so an exhausted
resolution never has to be explained by arithmetic.
- **A bounded diagnosed outcome.** Those budgets are the coordinator's own, on its 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,23 +199,66 @@ 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.
Every row runs in a child process of its own. A peak-memory figure is a high-water mark that
never falls, so rows sharing one process would each report where that process had already
been: the column would sort itself by row position rather than by mode, and the mode ranking
would reverse when the rows were reordered. One child per row is what makes the number belong
to the row.
What the numbers said on a four-core development box, at 300 transitions per row:
- A process boundary costs about a fifth of the critical path at the median. Two agents: 10.0
ms p50 in-process, 12.2 ms on threads, 12.1 ms across processes, with p99 at 21.4 / 19.4 /
21.4 ms. The dedicated-thread and separate-process variants are within noise of each other,
so what is being paid for is leaving the coordinator's runtime, not crossing a socket.
- `Worker.Status` -- an RPC answered from a cell with no domain work behind it -- is the
router and transport floor: 0.60 / 0.89 / 1.17 ms p50 for the three modes at two agents.
- Four agents needs six threads, which that box does not have, and every mode's tail widens
together. That is the budget being honest about oversubscription, not a property of the
process split.
- Memory is where the split really shows, but not where the first version of this note said.
The coordinator's own peak is roughly the same in all three modes and is *lowest* in process
mode -- 8.9 / 9.2 / 7.9 MiB at one agent -- because the workers are no longer inside it.
What the split costs is the children: about 5.7 MiB per participant process, so the whole
composition is roughly 9 MiB on threads against 39 MiB across processes at four agents.
- Ownership, collection and queues stayed bounded in every mode and at every agent count: at
most 15 live owners, 11 artifact roots and one queued entry per agent, with the store at
rest holding two sealed frames and 128 bytes. Of 311 frames observed, 309 were collected --
the two still owned are the current and previous boundary. The frame count is taken from the
behaviour trace's observation boundaries rather than calculated from the step count, so a
backend sealing two frames per boundary would show up instead of being hidden.
## 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
are generated twice by `both_transports!`, and
`sequential_concurrent_and_reversed_orders_agree` walks both transports inside one test
because it compares their behaviour traces against each other.
The three integration suites do not all run over both transports, and cannot:
- `tests/session.rs` and `tests/failures.rs` are in-process compositions and run over both,
through the same router code. All but one test in them is generated twice by
`both_transports!`; `sequential_concurrent_and_reversed_orders_agree` walks both transports
inside one test, because it compares their behaviour traces against each other.
- `tests/processes.rs` runs over the Unix socket only, in all three execution modes. A
participant in a process of its own has no in-memory transport to reach the router by, so
the mode is the axis that suite varies and the transport is fixed.
- `tests/session.rs`: one world advance per complete batch; every agent Prepared before the
advance; one task evaluation per transition; every agent committed before the next Prepare or
@ -150,6 +267,14 @@ 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 slow participant resolved rather than failed, 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

75
services/flysim/crates/fly-session/examples/processes.rs Normal file
View file

@ -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<dyn std::error::Error>> {
let program = default_worker_program();
println!("worker program: {}", program.display());
let mut agreed: Option<Vec<String>> = 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(&fly_session::types::id("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(())
}

19
services/flysim/crates/fly-session/src/agent.rs
View file

@ -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(&params.initial_decision_context)?;
// Everything is validated before the model is constructed.
@ -635,6 +650,10 @@ impl WorkerEndpoint for FakeAgentWorker {
self.status.clone()
}
fn worker_threads(&self) -> u64 {
self.config.worker_threads as u64
}
fn methods(&self) -> Vec<&'static str> {
vec!["Agent.Initialize", "Agent.Prepare", "Agent.Commit"]
}

8
services/flysim/crates/fly-session/src/bin/fly-session.rs Normal file
View file

@ -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()
}

282
services/flysim/crates/fly-session/src/cli.rs Normal file
View file

@ -0,0 +1,282 @@
//! 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 <command> [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
measure-row measure one row and print it as JSON (one child per row)
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)
--warmup-steps N transitions run before sampling starts (default 10)
--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)
measure-row options: --mode NAME --agents N, plus the measure options above. Each row runs in
a process of its own, so its memory peak is its own rather than the peak of the rows before
it.
";
/// 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<String> = args.map(|a| a.to_string_lossy().into_owned()).collect();
let result = match command.as_str() {
"agent" | "environment" => Options::parse(&rest).and_then(|o| serve(&command, &o)),
"measure" => Options::parse(&rest).and_then(|o| measure(&o)),
"measure-row" => Options::parse(&rest).and_then(|o| measure_row(&o)),
"--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<String, String>);
impl Options {
fn parse(args: &[String]) -> Result<Options, String> {
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<Id, String> {
parse_id(self.required(name)?).map_err(|e| format!("--{name}: {e}"))
}
fn u64(&self, name: &str, default: u64) -> Result<u64, String> {
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<Option<u64>, 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<usize, String> {
Ok(self.u64(name, default as u64)? as usize)
}
fn path(&self, name: &str) -> Result<PathBuf, String> {
Ok(PathBuf::from(self.required(name)?))
}
fn rational(&self, numerator: &str, denominator: &str) -> Result<RationalNs, String> {
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<Vec<Id>, String> {
value
.split(',')
.filter(|part| !part.is_empty())
.map(|part| parse_id(part).map_err(|e| format!("--ports: {e}")))
.collect()
}
fn measure_config(options: &Options) -> Result<crate::measure::MeasureConfig, String> {
let mut config = crate::measure::MeasureConfig {
steps: options.u64("steps", 200)?,
warmup_steps: options.u64("warmup-steps", 10)?,
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::<usize>().map_err(|_| format!("--agents: {p:?}")))
.collect::<Result<Vec<usize>, String>>()?;
}
if let Some(list) = options.optional("modes") {
config.modes = list
.split(',')
.filter(|p| !p.is_empty())
.map(parse_mode)
.collect::<Result<Vec<ExecutionMode>, String>>()?;
}
Ok(config)
}
fn parse_mode(name: &str) -> Result<ExecutionMode, String> {
match name {
"in-process" => Ok(ExecutionMode::InProcess),
"thread" => Ok(ExecutionMode::Thread),
"process" => Ok(ExecutionMode::Process),
other => Err(format!("unknown mode {other:?}")),
}
}
/// Runs the execution-mode comparison and prints its table.
///
/// One child per row: a peak-memory figure is only that row's if nothing else ran in the
/// process that produced it.
fn measure(options: &Options) -> Result<(), String> {
let config = measure_config(options)?;
let program = std::env::current_exe().map_err(|e| format!("current exe: {e}"))?;
let rows = crate::measure::run(&config, &program)?;
print!("{}", crate::measure::table(&rows));
Ok(())
}
/// Measures exactly one row and prints it as one JSON object. The parent's child.
fn measure_row(options: &Options) -> Result<(), String> {
let config = measure_config(options)?;
let mode = parse_mode(options.required("mode")?)?;
let agents = options.usize("agents", 2)?;
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.map_err(|e| format!("runtime: {e}"))?;
let row = runtime.block_on(crate::measure::one(&config, mode, agents))?;
println!("{}", row.to_json());
Ok(())
}

588
services/flysim/crates/fly-session/src/coordinator.rs
View file

@ -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<Id>,
}
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,69 @@ impl std::error::Error for SessionFailure {}
type Outcome<T> = Result<T, SessionFailure>;
/// The caller-side failure-detection budgets of `ipc-v1` section 6, on the coordinator's own
/// monotonic clock.
///
/// Section 6 is a two-stage procedure, and these are its two stages. `probe` is how long a
/// call may go without a terminal reply before it is *uncertain*; an uncertain call is not a
/// failed one, so the coordinator then runs the section 6 resolution -- a fresh bus call
/// carrying the original domain request id and body, pinned to the same incarnation -- for at
/// most `resolve` and at most `resolve_attempts` tries. Only when that ends without a definite
/// answer, or the incarnation is gone, or the retained result expired, is the epoch failed.
///
/// The prototype values follow section 6: probe at two seconds without a reply, give up at
/// ten seconds without progress -- two to notice plus eight to resolve -- with a separate
/// budget for a long boot. They are failure-detection values, not a gameplay latency goal.
///
/// **Which bound ends a resolution.** `resolve` is the one that does, at these values.
/// Between attempts the procedure sleeps [`RESOLVE_PAUSE`], so the fastest the attempt count
/// can be spent is `resolve_attempts * RESOLVE_PAUSE`; the default 8192 attempts is over
/// sixteen seconds of pauses alone, twice the eight-second budget, and an attempt whose call
/// expires costs a whole `probe` on top. `resolve_attempts` is therefore a second, coarser
/// stop for a pathological loop that costs nothing per turn, not the working limit. Both are
/// explicit because section 6 forbids filling this gap with an implicit best-effort policy,
/// and [`ResolutionEnd`] says which of them fired.
#[derive(Clone, Copy, Debug)]
pub struct Deadlines {
/// Without a terminal reply for this long, the call is uncertain.
pub probe: Duration,
/// The resolution's own budget, measured from its first attempt. The working limit.
pub resolve: Duration,
/// How many times the resolution may re-ask, as a guard rather than the working limit.
/// Never a retry of the operation: every attempt carries the original request id and body.
pub resolve_attempts: u32,
/// A separate, larger budget for `Worker.Hello` and the `Initialize` methods.
pub boot: Duration,
}
/// How long the resolution waits between attempts.
pub const RESOLVE_PAUSE: Duration = Duration::from_millis(2);
/// What ended a resolution, so a caller can tell an exhausted budget from an exhausted
/// attempt count rather than reading one number out of a message.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ResolutionEnd {
/// A matching terminal result arrived.
Answered,
/// The `resolve` budget ran out. At the default values this is the one that fires.
BudgetExpired,
/// The `resolve_attempts` guard ran out first, which needs a pause short enough or an
/// attempt count low enough for it to be reached before the budget.
AttemptsExhausted,
}
impl Default for Deadlines {
fn default() -> Deadlines {
Deadlines {
probe: Duration::from_secs(2),
resolve: Duration::from_secs(8),
// Over sixteen seconds of pauses: the budget above is what terminates.
resolve_attempts: 8192,
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 +195,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 +221,7 @@ impl AgentSlot {
port_id,
profile,
seed,
worker_threads: 1,
tick_duration: RationalNs::ZERO,
warmup_ticks: 0,
committed_step: 0,
@ -198,6 +276,20 @@ pub struct Coordinator {
pub injection_log: Vec<InjectionOutcome>,
/// How many times an exact duplicate met IN_PROGRESS while resolving an uncertain call.
pub in_progress_replies: u64,
/// How many uncertain calls ran the `ipc-v1` section 6 resolution.
pub resolutions: u64,
/// How the last resolution ended, so a test or a supervisor can tell which bound fired.
pub last_resolution: Option<ResolutionEnd>,
/// 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<Id>,
/// 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<DomainRequestId>,
last_commit_requests: Vec<TraceRequest>,
@ -246,6 +338,12 @@ impl Coordinator {
injections: Injections::default(),
injection_log: Vec::new(),
in_progress_replies: 0,
resolutions: 0,
last_resolution: None,
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 +410,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 +447,46 @@ 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<Id>) {
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
}
/// How many artifact handles this session still owns: the committed boundary's views plus
/// any the world has produced but not yet committed.
///
/// A fence drops both sets, which is what "old handles are invalid" means on this side of
/// the router. It is exposed so that can be asserted rather than described.
pub fn live_view_handles(&self) -> usize {
self.views.len() + self.pending_views.len()
}
fn agent(&self, agent_id: &Id) -> Option<&AgentSlot> {
@ -359,6 +501,20 @@ impl Coordinator {
///
/// Nothing here advances the environment or produces a gameplay reward.
pub async fn bootstrap(&mut self) -> Outcome<()> {
if self.fenced {
// Defence in depth: the phase machine refuses `Failed -> Ready(0)` anyway, but a
// fenced session should not be sending Hello and Initialize to live workers on the
// way to finding that out.
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,
});
}
self.hello_environment().await?;
for index in 0..self.agents.len() {
self.hello_agent(index).await?;
@ -571,7 +727,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 +865,29 @@ struct Job {
}
/// Issues one domain call with owned arguments, so it can run in its own task.
///
/// What one bus call came back as.
///
/// The expiry is its own variant rather than an error, because `ipc-v1` section 6 treats the
/// two differently: a refusal is an answer and can fail the epoch, while an expired deadline
/// is only an *uncertain* call and owes the resolution procedure first. Collapsing them into
/// one error is how a merely slow participant loses an epoch.
enum CallOutcome {
// Boxed: a `DomainReply` carries its artifact handles, and the other two variants are a
// unit and one error. Without the box every caller's `Result` is sized for the reply.
Answered(Box<DomainReply>),
/// The caller's deadline expired with no terminal reply.
Expired,
/// The bus or the reply itself refused. This is an answer, even when it is a bad one.
Refused(DomainError),
}
/// 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.
#[allow(clippy::too_many_arguments)]
async fn call_owned(
bus: flybus::Client,
@ -717,10 +898,55 @@ async fn call_owned(
attachments: Vec<(String, flybus::Artifact)>,
request_id: DomainRequestId,
want: Vec<String>,
) -> Result<DomainReply, DomainError> {
deadline: Duration,
) -> CallOutcome {
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(Ok(reply)) => CallOutcome::Answered(Box::new(reply)),
Ok(Err(e)) => CallOutcome::Refused(e),
Err(_) => CallOutcome::Expired,
}
}
/// The error an unresolved uncertain call finally becomes.
///
/// Deliberately `unknown`: `ipc-v1` section 6 forbids reading a caller-side timeout as proof
/// that nothing was mutated.
fn unresolved(method: &str, worker: &WorkerRef, end: ResolutionEnd, bound: &str) -> DomainError {
let why = match end {
ResolutionEnd::BudgetExpired => "its resolution budget",
ResolutionEnd::AttemptsExhausted => "its resolution attempt guard",
ResolutionEnd::Answered => "its resolution",
};
DomainError::new(
ErrorCode::BackendFailure,
format!(
"{method}: {} never resolved; {why} of {bound} ran out and the operation's \
outcome is unknown",
worker.worker_id
),
MutationCertainty::Unknown,
)
}
/// One per-agent call's result, in the shape the phase loops read it.
///
/// The request is carried back beside the outcome so an uncertain call can be resolved
/// against its *original* domain request id and body. Recomputing it would be a second
/// operation, which `step-v1` section 7 forbids.
struct JobResult {
agent_id: Id,
outcome: CallOutcome,
scope: Option<Scope>,
worker: WorkerRef,
method: &'static str,
request_id: DomainRequestId,
params: Map<String, Value>,
attachments: Vec<(String, flybus::Artifact)>,
/// How long the call took, for the section 5 percentiles.
elapsed: Duration,
}
impl Coordinator {
@ -740,24 +966,44 @@ impl Coordinator {
.iter()
.map(|(n, a)| ((*n).to_owned(), (*a).clone()))
.collect();
let reply = call_owned(
// 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.probe
};
let started = Instant::now();
let outcome = call_owned(
self.bus.clone(),
worker.clone(),
method,
scope.clone(),
params,
owned,
request_id,
params.clone(),
owned.clone(),
request_id.clone(),
want.to_vec(),
deadline,
)
.await;
let reply = match reply {
Ok(reply) => reply,
Err(e) => return Err(self.fail_now(e, method)),
self.metrics.record(method, started.elapsed());
let reply = match outcome {
CallOutcome::Answered(reply) => *reply,
// Uncertain, not failed: section 6 owes this call its resolution first.
CallOutcome::Expired => {
return self
.resolve(worker, method, scope.clone(), params, owned, request_id, want)
.await;
}
CallOutcome::Refused(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 +1019,7 @@ impl Coordinator {
scope: &Option<Scope>,
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)
@ -816,9 +1063,22 @@ impl Coordinator {
Ok(())
}
/// Resolves an uncertain operation: a fresh bus call with the original domain request id
/// and body, pinned to the same service incarnation. It never issues a new request id and
/// never recomputes a decision.
/// The `ipc-v1` section 6 resolution of an uncertain call.
///
/// Step 2 of that section: while the same bus, service and worker incarnation still exist,
/// issue a fresh bus call carrying the **original** domain request id and body with its
/// retained input attachments. The worker's own deduplication answers it from the record
/// of the first attempt, so this queries rather than repeats: it never issues a new request
/// id, never recomputes a decision, and never becomes a second batch.
///
/// Step 3: only a matching terminal result resolves it. `IN_PROGRESS` means the original is
/// still running and no second mutation was started, so the procedure waits and asks again.
/// An expiry inside the procedure is likewise not an answer.
///
/// Step 4: the epoch fails when the routes or ownership were lost, the incarnation changed,
/// the retained result expired, or the procedure's own bounded budget ran out. Those bounds
/// are [`Deadlines`] and are explicit, because section 6 refuses an implicit best-effort
/// policy here.
#[allow(clippy::too_many_arguments)]
async fn resolve(
&mut self,
@ -833,8 +1093,23 @@ 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.
for attempt in 0..200u32 {
let reply = call_owned(
self.blame(Some(worker.worker_id.clone()));
let probe = self.deadlines.probe;
let budget = self.deadlines.resolve;
let attempts = self.deadlines.resolve_attempts;
let started = Instant::now();
self.resolutions += 1;
self.last_resolution = None;
self.audit.push(format!("resolve:{}:{method}", worker.worker_id));
// The budget is the working limit and the attempt count is a guard; whichever runs
// out is recorded, so "it gave up" is never an unexplained number.
let mut end = ResolutionEnd::AttemptsExhausted;
for _ in 0..attempts {
if started.elapsed() >= budget {
end = ResolutionEnd::BudgetExpired;
break;
}
let outcome = call_owned(
self.bus.clone(),
worker.clone(),
method,
@ -843,31 +1118,42 @@ impl Coordinator {
attachments.clone(),
request_id.clone(),
want.to_vec(),
probe,
)
.await;
let reply = match reply {
Ok(reply) => reply,
Err(e) => return Err(self.fail_now(e, method)),
let reply = match outcome {
CallOutcome::Answered(reply) => *reply,
// Still no answer. The original may simply be slow; asking again is the
// procedure, and the request id it carries is unchanged.
CallOutcome::Expired => {
tokio::time::sleep(RESOLVE_PAUSE).await;
continue;
}
// Step 4: routes or ownership lost, or the incarnation is gone.
CallOutcome::Refused(e) => return Err(self.fail_now(e, method)),
};
self.check_reply(worker, &reply, &scope, method)?;
match reply.result() {
Ok(_) => return Ok(reply),
Err(e) if e.code == ErrorCode::InProgress => {
let _ = attempt;
self.in_progress_replies += 1;
tokio::time::sleep(std::time::Duration::from_millis(2)).await;
Ok(_) => {
self.blame(None);
self.last_resolution = Some(ResolutionEnd::Answered);
return Ok(reply);
}
Err(e) if e.code == ErrorCode::InProgress => {
self.in_progress_replies += 1;
tokio::time::sleep(RESOLVE_PAUSE).await;
}
// A terminal refusal, including RESULT_EXPIRED: definite, so the epoch fails.
Err(e) => return Err(self.fail_now(e, method)),
}
}
Err(self.fail_now(
DomainError::new(
ErrorCode::BackendFailure,
"the uncertain operation never resolved",
MutationCertainty::Unknown,
),
method,
))
self.last_resolution = Some(end);
let bound = match end {
ResolutionEnd::AttemptsExhausted => format!("{attempts} attempts"),
_ => format!("{budget:?}"),
};
let error = unresolved(method, worker, end, &bound);
Err(self.fail_now(error, method))
}
/// Sends the same domain request again on a fresh bus call and reports what came back,
@ -884,6 +1170,7 @@ impl Coordinator {
expected: Option<&Value>,
what: &str,
) {
let deadline = self.deadlines.probe;
let reply = call_owned(
self.bus.clone(),
worker.clone(),
@ -893,10 +1180,11 @@ impl Coordinator {
attachments,
request_id,
Vec::new(),
deadline,
)
.await;
let outcome = match reply {
Ok(reply) => match reply.result() {
CallOutcome::Answered(reply) => match reply.result() {
Ok(result) => InjectionOutcome {
what: what.to_owned(),
code: None,
@ -908,11 +1196,18 @@ impl Coordinator {
identical: false,
},
},
Err(e) => InjectionOutcome {
CallOutcome::Refused(e) => InjectionOutcome {
what: what.to_owned(),
code: Some(e.code),
identical: false,
},
// A probe is a diagnostic, not a phase of the transaction: it reports what it saw
// and never resolves anything on the session's behalf.
CallOutcome::Expired => InjectionOutcome {
what: what.to_owned(),
code: Some(ErrorCode::BackendFailure),
identical: false,
},
};
self.injection_log.push(outcome);
}
@ -934,7 +1229,7 @@ impl Coordinator {
_ => Map::new(),
};
let request_id = self.serials.next(&worker.service);
let reply = call_owned(
let outcome = call_owned(
self.bus.clone(),
worker.clone(),
method,
@ -943,9 +1238,19 @@ impl Coordinator {
Vec::new(),
request_id,
Vec::new(),
self.deadlines.probe,
)
.await?;
reply.result().cloned()
.await;
match outcome {
CallOutcome::Answered(reply) => reply.result().cloned(),
CallOutcome::Refused(e) => Err(e),
CallOutcome::Expired => Err(unresolved(
method,
worker,
ResolutionEnd::BudgetExpired,
&format!("{:?}", self.deadlines.probe),
)),
}
}
/// The sensory input one agent is permitted to consume at `boundary`.
@ -964,11 +1269,10 @@ impl Coordinator {
}
/// Runs one set of per-agent jobs in the configured dispatch order.
async fn run_jobs(
&mut self,
jobs: Vec<Job>,
order: DispatchOrder,
) -> Vec<(Id, Result<DomainReply, DomainError>, Option<Scope>, WorkerRef, &'static str)> {
/// A job's deadline is the section 6 probe, not the failure point: an expiry here starts
/// the resolution, which the phase loops run one at a time with the session in hand.
async fn run_jobs(&mut self, jobs: Vec<Job>, order: DispatchOrder) -> Vec<JobResult> {
let deadline = self.deadlines.probe;
let mut out = Vec::new();
match order {
DispatchOrder::Sequential | DispatchOrder::Reversed => {
@ -977,18 +1281,30 @@ impl Coordinator {
jobs.reverse();
}
for job in jobs {
let reply = call_owned(
let started = Instant::now();
let outcome = call_owned(
self.bus.clone(),
job.worker.clone(),
job.method,
job.scope.clone(),
job.params,
job.attachments,
job.request_id,
job.params.clone(),
job.attachments.clone(),
job.request_id.clone(),
Vec::new(),
deadline,
)
.await;
out.push((job.agent_id, reply, job.scope, job.worker, job.method));
out.push(JobResult {
agent_id: job.agent_id,
outcome,
scope: job.scope,
worker: job.worker,
method: job.method,
request_id: job.request_id,
params: job.params,
attachments: job.attachments,
elapsed: started.elapsed(),
});
}
}
DispatchOrder::Concurrent => {
@ -1000,18 +1316,30 @@ impl Coordinator {
let scope = job.scope.clone();
let method = job.method;
tasks.push(tokio::spawn(async move {
let reply = call_owned(
let started = Instant::now();
let outcome = call_owned(
bus,
job.worker,
job.method,
job.scope,
job.params,
job.attachments,
job.request_id,
job.params.clone(),
job.attachments.clone(),
job.request_id.clone(),
Vec::new(),
deadline,
)
.await;
(agent_id, reply, scope, worker, method)
JobResult {
agent_id,
outcome,
scope,
worker,
method,
request_id: job.request_id,
params: job.params,
attachments: job.attachments,
elapsed: started.elapsed(),
}
}));
}
for task in tasks {
@ -1024,7 +1352,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 +1363,20 @@ impl Coordinator {
impl Coordinator {
/// One complete transition `k -> k+1`.
pub async fn step(&mut self) -> Outcome<StepReport> {
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 +1407,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 +1520,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,12 +1589,31 @@ 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 {
let reply = match reply {
Ok(reply) => reply,
for job in results {
let JobResult {
agent_id,
outcome,
scope,
worker,
method,
request_id,
params,
attachments,
elapsed: _,
} = job;
self.blame(Some(agent_id.clone()));
let reply = match outcome {
CallOutcome::Answered(reply) => *reply,
// Uncertain: the agent may be slow rather than gone. Query the same operation
// against the same incarnation before the epoch is failed. A Prepare that
// already ran answers from its own record, so no tick is repeated.
CallOutcome::Expired => {
self.resolve(&worker, method, scope.clone(), params, attachments, request_id, &[])
.await?
}
// Some agents are already Prepared. Dispatch stops and the epoch fails; a
// prepared agent is never asked to prepare again.
Err(e) => return Err(self.fail_now(e, method)),
CallOutcome::Refused(e) => return Err(self.fail_now(e, method)),
};
self.check_reply(&worker, &reply, &scope, method)?;
let decision: PreparedDecision = match reply.parse() {
@ -1271,6 +1640,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 +1792,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.probe;
let advance_started = Instant::now();
let injected = self.injections.at_step == k;
let reply = if injected && self.injections.lose_advance_result {
@ -1492,7 +1865,7 @@ impl Coordinator {
)
.await?
} else {
let reply = call_owned(
let outcome = call_owned(
self.bus.clone(),
worker.clone(),
"Environment.Advance",
@ -1501,16 +1874,36 @@ impl Coordinator {
Vec::new(),
request_id.clone(),
want.clone(),
advance_deadline,
)
.await;
let reply = match reply {
Ok(reply) => reply,
Err(e) => return Err(self.fail_now(e, "advance")),
};
self.check_reply(&worker, &reply, &Some(scope.clone()), "advance")?;
match reply.result() {
Ok(_) => reply,
Err(e) => return Err(self.fail_now(e, "advance")),
self.metrics.record("Environment.Advance", advance_started.elapsed());
match outcome {
CallOutcome::Answered(reply) => {
let reply = *reply;
self.check_reply(&worker, &reply, &Some(scope.clone()), "advance")?;
match reply.result() {
Ok(_) => reply,
Err(e) => return Err(self.fail_now(e, "advance")),
}
}
// `step-v1` section 7 is imperative for this row: "Advance acknowledgment
// lost | Query/retransmit same request to same incarnation; never new batch."
// The resolution carries the original batch id and controls, so the world is
// asked about the operation it already has rather than given another one.
CallOutcome::Expired => {
self.resolve(
&worker,
"Environment.Advance",
Some(scope.clone()),
params.clone(),
Vec::new(),
request_id.clone(),
&want,
)
.await?
}
CallOutcome::Refused(e) => return Err(self.fail_now(e, "advance")),
}
};
@ -1518,6 +1911,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 +2167,52 @@ 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 {
match reply {
Ok(reply) => {
let mut blamed: Option<Id> = None;
for job in results {
let JobResult {
agent_id,
outcome,
scope,
worker,
method,
request_id,
params,
attachments: job_attachments,
elapsed: _,
} = job;
self.blame(Some(agent_id.clone()));
// Uncertain: resolve the same Commit against the same incarnation first. A Commit
// that already ran replays its cached reply, so no reward is applied twice.
let outcome = match outcome {
CallOutcome::Expired => {
match self
.resolve(
&worker,
method,
scope.clone(),
params,
job_attachments,
request_id,
&[],
)
.await
{
Ok(reply) => CallOutcome::Answered(Box::new(reply)),
Err(failure) => return Err(failure),
}
}
other => other,
};
match outcome {
CallOutcome::Answered(reply) => {
let reply = *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 +2245,24 @@ impl Coordinator {
}
self.audit.push(format!("committed:{agent_id}@{k}"));
self.stats.commits += 1;
self.blame(None);
commits.push((agent_id, result));
}
Err(e) => {
CallOutcome::Refused(e) => {
if first_failure.is_none() {
blamed = Some(agent_id.clone());
}
first_failure = Some(first_failure.unwrap_or(e));
}
CallOutcome::Expired => unreachable!("an expiry was resolved just above"),
}
}
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() {

6
services/flysim/crates/fly-session/src/environment.rs
View file

@ -36,6 +36,8 @@ pub struct EnvironmentConfig {
/// The world's fixed reduced step duration. 60 Hz is `1/60` s.
pub step_duration: RationalNs,
pub ports: Vec<Id>,
/// The thread allocation the launcher started this worker within.
pub worker_threads: usize,
pub faults: EnvironmentFaults,
}
@ -425,6 +427,10 @@ impl WorkerEndpoint for CounterEnvironment {
self.status.clone()
}
fn worker_threads(&self) -> u64 {
self.config.worker_threads as u64
}
fn methods(&self) -> Vec<&'static str> {
vec!["Environment.Initialize", "Environment.Advance"]
}

365
services/flysim/crates/fly-session/src/harness.rs
View file

@ -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<usize>,
/// 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::<usize>()
}
/// 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<ThreadBudget, DomainError> {
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<Vec<UnixListenerHandle>>,
}
impl Connector {
async fn client(&self, id: &str) -> Result<Client, flybus::BusError> {
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<Id, WorkerHandle>,
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<Vec<Client>>,
}
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<Id, Box<dyn ActionExecutor>> = 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<Client, flybus::BusError> {
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<Client, flybus::BusError> {
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<Restarted, flybus::BusError> {
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,88 @@ 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,
})
}
/// The agent worker's progress counter, which is its fake model's mutation count.
pub fn agent_mutations(&self, agent_id: &Id) -> u64 {
self.agents.get(agent_id).map(WorkerHandle::progress_counter).unwrap_or_default()
/// 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
}
pub fn environment_mutations(&self) -> u64 {
self.environment.progress_counter()
/// 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, when
/// this process is where that counter lives.
///
/// `None` means "not observable from here", not "nothing happened": a participant with a
/// process of its own keeps its counter there. [`SessionHarness::progress_of`] reads it
/// over the bus and works in every mode.
pub fn agent_mutations(&self, agent_id: &Id) -> Option<u64> {
self.launcher
.worker(agent_id)
.and_then(crate::launcher::LaunchedWorker::progress_counter)
}
pub fn environment_mutations(&self) -> Option<u64> {
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<u64, DomainError> {
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<StatusCell> {
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(&id("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())
}

1344
services/flysim/crates/fly-session/src/launcher.rs Normal file
View file

File diff suppressed because it is too large Load diff

9
services/flysim/crates/fly-session/src/lib.rs
View file

@ -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, ResolutionEnd, SessionFailure, StepReport,
};
pub use launcher::{ExecutionMode, Launcher, ReapOutcome, ThreadBudget, Via};
pub use phase::{Phase, PhaseMachine};

488
services/flysim/crates/fly-session/src/measure.rs Normal file
View file

@ -0,0 +1,488 @@
//! 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.
//!
//! **Every row runs in a process of its own.** The coordinator's memory figure is a peak --
//! `VmHWM` never falls -- so several rows sharing one process would each report where that
//! process had already been rather than what its own mode costs, and the column would order
//! itself by row position instead of by mode. The parent spawns one `measure-row` child per
//! row and reads its result back, so each figure belongs to the row that produced it.
use std::collections::BTreeMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use serde_json::{Value, json};
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<usize>,
pub modes: Vec<ExecutionMode>,
/// 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,
/// Frames this run actually observed, counted from the behaviour trace: every sensory
/// view of every transition, plus the one the environment sealed for boundary zero.
///
/// Counted rather than calculated, so a backend that sealed two frames per boundary would
/// show up here instead of being hidden by arithmetic.
pub frames_observed: u64,
}
impl Row {
/// Frames the store collected: observed, minus the ones still owned at the end.
pub fn collected(&self) -> u64 {
self.frames_observed.saturating_sub(self.sealed_final as u64)
}
/// The row as one JSON object, for the child that measured it to hand back.
pub fn to_json(&self) -> Value {
let p = |x: &Percentiles| {
json!({"count": x.count, "p50": x.p50_ns, "p95": x.p95_ns, "p99": x.p99_ns,
"max": x.max_ns})
};
json!({
"mode": self.mode.label(),
"agents": self.agents,
"workerThreads": self.worker_threads,
"physicalCores": self.physical_cores,
"budgetTotal": self.budget_total,
"budgetUsed": self.budget_used,
"steps": self.steps,
"prepare": p(&self.prepare),
"commit": p(&self.commit),
"advance": p(&self.advance),
"status": p(&self.status),
"step": p(&self.step),
"coordinatorPeakRssKib": self.coordinator_peak_rss_kib,
"participantsPeakRssKib": self.participants_peak_rss_kib,
"ownersMax": self.owners_max,
"ownersFinal": self.owners_final,
"artifactRootsMax": self.artifact_roots_max,
"queuedMax": self.queued_max,
"sealedMax": self.sealed_max,
"sealedFinal": self.sealed_final,
"storeBytesMax": self.store_bytes_max,
"storeBytesFinal": self.store_bytes_final,
"framesObserved": self.frames_observed,
})
}
/// Reads back what a `measure-row` child printed.
pub fn from_json(value: &Value) -> Result<Row, String> {
let u = |name: &str| -> Result<u64, String> {
value
.get(name)
.and_then(Value::as_u64)
.ok_or_else(|| format!("measure row: {name} is missing or not a number"))
};
let p = |name: &str| -> Result<Percentiles, String> {
let v = value
.get(name)
.ok_or_else(|| format!("measure row: {name} is missing"))?;
let f = |k: &str| v.get(k).and_then(Value::as_u64).unwrap_or_default();
Ok(Percentiles {
count: f("count") as usize,
p50_ns: f("p50"),
p95_ns: f("p95"),
p99_ns: f("p99"),
max_ns: f("max"),
})
};
let mode = match value.get("mode").and_then(Value::as_str) {
Some("in-process") => ExecutionMode::InProcess,
Some("thread") => ExecutionMode::Thread,
Some("process") => ExecutionMode::Process,
other => return Err(format!("measure row: unknown mode {other:?}")),
};
Ok(Row {
mode,
agents: u("agents")? as usize,
worker_threads: u("workerThreads")? as usize,
physical_cores: u("physicalCores")? as usize,
budget_total: u("budgetTotal")? as usize,
budget_used: u("budgetUsed")? as usize,
steps: u("steps")?,
prepare: p("prepare")?,
commit: p("commit")?,
advance: p("advance")?,
status: p("status")?,
step: p("step")?,
coordinator_peak_rss_kib: u("coordinatorPeakRssKib")?,
participants_peak_rss_kib: u("participantsPeakRssKib")?,
owners_max: u("ownersMax")? as usize,
owners_final: u("ownersFinal")? as usize,
artifact_roots_max: u("artifactRootsMax")?,
queued_max: u("queuedMax")? as usize,
sealed_max: u("sealedMax")? as usize,
sealed_final: u("sealedFinal")? as usize,
store_bytes_max: u("storeBytesMax")?,
store_bytes_final: u("storeBytesFinal")?,
frames_observed: u("framesObserved")?,
})
}
}
/// Runs the comparison, one child process per row.
///
/// `program` is this crate's binary; each row is measured by a `measure-row` invocation of it
/// so that the row's memory peak is its own and not the accumulated peak of the rows before
/// it. The order of the rows therefore cannot change any of their numbers.
pub fn run(config: &MeasureConfig, program: &std::path::Path) -> Result<Vec<Row>, String> {
let mut rows = Vec::new();
for mode in &config.modes {
for agents in &config.agent_counts {
rows.push(row_in_a_child(config, program, *mode, *agents)?);
}
}
Ok(rows)
}
fn row_in_a_child(
config: &MeasureConfig,
program: &std::path::Path,
mode: ExecutionMode,
agents: usize,
) -> Result<Row, String> {
let output = std::process::Command::new(program)
.arg("measure-row")
.arg("--mode")
.arg(mode.label())
.arg("--agents")
.arg(agents.to_string())
.arg("--steps")
.arg(config.steps.to_string())
.arg("--warmup-steps")
.arg(config.warmup_steps.to_string())
.arg("--worker-threads")
.arg(config.worker_threads.to_string())
.stdin(std::process::Stdio::null())
.output()
.map_err(|e| format!("measure-row {}: {e}", mode.label()))?;
if !output.status.success() {
return Err(format!(
"measure-row {} {agents}: exited {}: {}",
mode.label(),
output.status,
String::from_utf8_lossy(&output.stderr).trim()
));
}
let text = String::from_utf8_lossy(&output.stdout);
let line = text
.lines()
.rev()
.find(|l| l.trim_start().starts_with('{'))
.ok_or_else(|| format!("measure-row {}: no row on stdout", mode.label()))?;
let value: Value = serde_json::from_str(line)
.map_err(|e| format!("measure-row {}: unreadable row: {e}", mode.label()))?;
Row::from_json(&value)
}
/// Measures exactly one row, in this process. The `measure-row` subcommand's body.
pub async fn one(
config: &MeasureConfig,
mode: ExecutionMode,
agents: usize,
) -> Result<Row, String> {
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<Row, String> {
let specs: Vec<AgentSpec> = (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,
// Counted from the behaviour trace: every sensory view of every transition this run
// recorded, plus the frame the environment sealed for boundary zero.
frames_observed: 1 + harness
.coordinator
.trace
.transitions
.iter()
.map(|t| t.behaviour.observation_boundaries.len() as u64)
.sum::<u64>(),
};
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. Every row was measured in \
a process of its own, so no figure depends on the order of the rows.\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 observed/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_observed,
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()
}

176
services/flysim/crates/fly-session/src/metrics.rs Normal file
View file

@ -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<String, Vec<u64>>,
}
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<Percentiles> {
let mut values = self.samples.get(what)?.clone();
values.sort_unstable();
Some(Percentiles::of(&values))
}
pub fn names(&self) -> Vec<String> {
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<u64> {
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<u64> {
peak_rss_of(&format!("/proc/{pid}/status"))
}
fn peak_rss_of(path: &str) -> Option<u64> {
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);
}
}

29
services/flysim/crates/fly-session/src/worker.rs
View file

@ -188,6 +188,12 @@ pub trait WorkerEndpoint: Send + 'static {
fn capabilities(&self) -> Vec<Id>;
fn status_cell(&self) -> StatusCell;
/// The thread allocation this worker's launcher started it within.
///
/// `Worker.Hello` reports it, so a caller bounded by `workers-v1`'s "within launcher
/// allocation" can read the allocation instead of being told it out of band.
fn worker_threads(&self) -> u64;
/// The domain methods this endpoint implements, beyond the common `Worker.*` set.
/// Anything else returns UNSUPPORTED without entering the endpoint.
fn methods(&self) -> Vec<&'static str>;
@ -224,6 +230,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.
@ -259,7 +281,7 @@ async fn run<E: WorkerEndpoint>(
) {
// Identity and capabilities are fixed for the endpoint's lifetime, so the shell reads them
// once and never takes the endpoint mutex to answer Hello or Status.
let (worker_id, incarnation_id, session_id, role, capabilities, status, methods) = {
let (worker_id, incarnation_id, session_id, role, capabilities, status, methods, threads) = {
let e = endpoint.lock().await;
(
e.worker_id(),
@ -269,6 +291,7 @@ async fn run<E: WorkerEndpoint>(
e.capabilities(),
e.status_cell(),
e.methods(),
e.worker_threads(),
)
};
let mut running: Vec<tokio::task::JoinHandle<()>> = Vec::new();
@ -305,6 +328,7 @@ async fn run<E: WorkerEndpoint>(
&incarnation_id,
role,
&capabilities,
threads,
);
let _ = responder.reply(outcome.to_outcome(), &[]).await;
continue;
@ -606,6 +630,7 @@ fn failure(
failure_outcome(request_id, worker_id, incarnation_id, scope, error)
}
#[allow(clippy::too_many_arguments)]
fn hello(
request: &SessionRpcRequest,
session_id: &Id,
@ -613,6 +638,7 @@ fn hello(
incarnation_id: &Id,
role: Role,
capabilities: &[Id],
worker_threads: u64,
) -> SessionRpcOutcome {
let params: HelloParams = match HelloParams::from_json(&request.params) {
Ok(params) => params,
@ -668,6 +694,7 @@ fn hello(
capabilities: capabilities.to_vec(),
max_agents: MAX_AGENTS as u64,
max_ports: MAX_PORTS as u64,
worker_threads,
};
success(
request,

54
services/flysim/crates/fly-session/tests/common/mod.rs
View file

@ -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<T>(what: &str, f: impl std::future::Future<Output = T>) -> 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
}

13
services/flysim/crates/fly-session/tests/failures.rs
View file

@ -33,6 +33,15 @@ both_transports!(
const STEPS: u64 = 4;
const INJECT_AT: u64 = 2;
/// The mutation counter of a participant running in this process. These suites are all
/// in-process compositions, so it is always there; SESSION-02's are not, and read it over the
/// bus instead.
fn local_mutations(f: &Fixture, agent_id: &Id) -> u64 {
f.harness
.agent_mutations(agent_id)
.expect("an in-process participant keeps its counter in this process")
}
/// What a run of the standard composition produced.
struct Run {
behaviour: Vec<String>,
@ -51,8 +60,8 @@ async fn run_with(via: Via, injections: Injections) -> Run {
let run = Run {
behaviour: f.harness.coordinator.trace.behavior(),
mutations: vec![
(fly_a(), f.harness.agent_mutations(&fly_a())),
(fly_b(), f.harness.agent_mutations(&fly_b())),
(fly_a(), local_mutations(&f, &fly_a())),
(fly_b(), local_mutations(&f, &fly_b())),
],
counter: f
.harness

823
services/flysim/crates/fly-session/tests/processes.rs Normal file
View file

@ -0,0 +1,823 @@
//! 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::ResolutionEnd;
use fly_session::phase::Phase;
use fly_session::types::*;
all_modes!(
a_slow_participant_is_resolved_rather_than_failed,
a_resolution_says_which_of_its_two_bounds_ended_it,
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<String, Vec<String>> = 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}"
);
}
}
// -------------------------------------------------------------------------------------------
// ipc-v1 section 6: an uncertain call is resolved, not failed
/// A participant that is merely slow -- slower than the caller's probe, faster than the
/// resolution's budget -- finishes its step. The epoch is not lost, and the resolution adds no
/// second operation.
///
/// This is the `ipc-v1` section 6 procedure on the path that actually reaches it: the probe
/// expires, the coordinator queries the same request id against the same incarnation, the
/// worker answers `IN_PROGRESS` while its original is still running and then replays its
/// cached reply. `step-v1` section 7's Advance row is the same rule, so the world is slow here
/// too and its batch is never re-sent as a new one.
async fn a_slow_participant_is_resolved_rather_than_failed(mode: ExecutionMode) {
// A clean run of the same composition, to compare against.
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(2)).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 config = two_agents(mode);
config.agents[1].faults = AgentFaults { prepare_delay_ms: 500, ..AgentFaults::default() };
config.environment_faults =
EnvironmentFaults { advance_delay_ms: 500, ..EnvironmentFaults::default() };
let mut f = mode_fixture(mode, config).await;
// A probe well inside both delays, and a resolution budget well outside them: the point is
// a call that expires and an operation that is nevertheless fine.
f.harness.coordinator.deadlines = fly_session::Deadlines {
probe: Duration::from_millis(120),
resolve: Duration::from_secs(20),
resolve_attempts: 4096,
boot: Duration::from_secs(30),
};
within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
let reports = within("run", f.harness.coordinator.run(2))
.await
.expect("a slow participant is resolved, not failed");
assert_eq!(reports.len(), 2);
assert_eq!(f.harness.coordinator.phase(), Phase::Ready(2));
assert!(!f.harness.coordinator.is_fenced(), "a slow answer is not a lost epoch");
assert!(
f.harness.coordinator.resolutions >= 2,
"both the slow Prepare and the slow Advance must have run the resolution, not {}",
f.harness.coordinator.resolutions
);
assert!(
f.harness.coordinator.in_progress_replies > 0,
"the resolution must have met the original still running"
);
assert_eq!(
f.harness.coordinator.last_resolution,
Some(ResolutionEnd::Answered),
"the resolution ended by being answered, not by running out of anything"
);
// No second operation anywhere: one advance per transition, one batch id per transition,
// and the same behaviour as the run that never timed out.
assert_eq!(f.harness.coordinator.stats().advances, 2);
let environment = f.harness.environment_id();
let world = within("progress", f.harness.progress_of(&environment)).await.unwrap();
assert_eq!(world, clean.1, "the world moved exactly as often as in the clean run");
assert_eq!(
f.harness.coordinator.trace.behavior(),
clean.0,
"resolving an uncertain call changes no behaviour"
);
let batches: std::collections::BTreeSet<Id> = f
.harness
.coordinator
.trace
.transitions
.iter()
.map(|t| t.behaviour.batch_id.clone())
.collect();
assert_eq!(batches.len(), 2, "one batch id per transition, never a second batch");
// And the agents took exactly the ticks the clean run took: a resolution is a query.
for transition in &f.harness.coordinator.trace.transitions {
for agent in &transition.behaviour.agents {
assert!(agent.ticks_advanced == 16 || agent.ticks_advanced == 17);
}
}
f.shutdown().await;
}
/// The resolution has two bounds, and which one ended it is never left to be guessed.
///
/// `resolve` is the working limit at the default values -- the attempt guard is over sixteen
/// seconds of pauses against an eight-second budget -- so an unresponsive participant runs the
/// budget out. Setting the guard low instead ends the same resolution the other way, and the
/// failure says so both in `last_resolution` and in its own message.
async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode) {
// The budget is what ends it at ordinary settings: a generous attempt guard, a short
// budget, and a participant far slower than either.
let mut config = two_agents(mode);
config.agents[1].faults = AgentFaults { prepare_delay_ms: 30_000, ..AgentFaults::default() };
let mut f = mode_fixture(mode, config).await;
f.harness.coordinator.deadlines = fly_session::Deadlines {
probe: Duration::from_millis(50),
resolve: Duration::from_millis(300),
resolve_attempts: 8192,
boot: Duration::from_secs(30),
};
within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
let started = Instant::now();
let failure = within("step", f.harness.coordinator.step())
.await
.expect_err("a participant that never answers exhausts the resolution");
assert_eq!(f.harness.coordinator.last_resolution, Some(ResolutionEnd::BudgetExpired));
assert!(
failure.error.message.contains("resolution budget"),
"the message names the bound that fired: {failure}"
);
assert_eq!(failure.participant.as_deref(), Some(fly_b().as_str()));
assert_eq!(failure.error.mutation, MutationCertainty::Unknown);
assert!(
started.elapsed() < Duration::from_secs(20),
"the budget, not the 30-second participant, is what ended it"
);
assert!(f.harness.coordinator.is_fenced());
f.shutdown().await;
// The guard is what ends it when it is set below the budget: three attempts against a
// budget the participant could never reach anyway.
let mut config = two_agents(mode);
config.agents[1].faults = AgentFaults { prepare_delay_ms: 30_000, ..AgentFaults::default() };
let mut f = mode_fixture(mode, config).await;
f.harness.coordinator.deadlines = fly_session::Deadlines {
probe: Duration::from_millis(50),
resolve: Duration::from_secs(600),
resolve_attempts: 3,
boot: Duration::from_secs(30),
};
within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
let failure = within("step", f.harness.coordinator.step())
.await
.expect_err("three attempts are not enough to resolve a silent participant");
assert_eq!(f.harness.coordinator.last_resolution, Some(ResolutionEnd::AttemptsExhausted));
assert!(
failure.error.message.contains("attempt guard") && failure.error.message.contains("3 attempts"),
"the message names the bound that fired and its size: {failure}"
);
assert_eq!(failure.participant.as_deref(), Some(fly_b().as_str()));
f.shutdown().await;
}
// -------------------------------------------------------------------------------------------
// 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<Id> = f
.harness
.coordinator
.trace
.transitions
.iter()
.map(|t| t.behaviour.batch_id.clone())
.collect();
let unique: std::collections::BTreeSet<Id> = 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");
}
// Every participant reports the allocation its launcher gave it, which is the wire the
// 2026-09-22 `workers-v1` amendment added. The launcher refused anything else at start,
// so a caller reads it here rather than being told it out of band.
for who in [fly_a(), fly_b(), environment.clone()] {
let worker = f.harness.coordinator.agent_ref(&who).cloned().unwrap_or_else(|| {
f.harness.coordinator.environment_ref().clone()
});
let params = serde_json::json!({
"sessionId": "demo",
"expectedWorkerId": who.as_str(),
"role": if who == environment { "environment" } else { "agent" },
"supportedMajors": [1],
});
let result = within(
"hello",
f.harness.coordinator.probe_raw(&worker, "Worker.Hello", None, params),
)
.await
.expect("a worker answers its own identity");
let reported = result["limits"]["workerThreads"].as_u64();
assert_eq!(
reported,
Some(f.harness.launcher.worker(&who).unwrap().identity.worker_threads as u64),
"{who} must report the allocation its launcher gave it"
);
}
// 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(), &id("test")).await;
assert_eq!(outcome, ReapOutcome::Stopped, "a live participant answers Worker.Shutdown");
assert_eq!(
f.harness.launcher.reap(&fly_a(), &id("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;
assert!(
f.harness.coordinator.live_view_handles() > 0,
"boundary 0's view is owned before the router goes away"
);
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.live_view_handles(),
0,
"the fence drops every artifact handle of the old store incarnation"
);
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, in two parts, because the two refusals are different
/// refusals and each deserves its own exact code.
///
/// A restarted worker is a *fresh* process: it has no epoch at all, so the old epoch's work is
/// refused on phase, not on timeline. The stale-epoch half of the row needs a worker that has
/// an epoch and has left it, which in process mode is the agent that did not restart.
#[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();
// Part one: a live agent process, initialized under epoch e1, meets an operation from
// another epoch. This is the row's stale-epoch half, with a real child process.
let live = f.harness.coordinator.agent_ref(&fly_a()).cloned().unwrap();
let err = within(
"stale epoch",
f.harness.coordinator.probe_raw(
&live,
"Agent.Prepare",
Some(scope_at("demo", "e0", 1)),
prepare_params("fly-a"),
),
)
.await
.expect_err("an old epoch cannot mutate a worker that belongs to this one");
assert_eq!(err.code, ErrorCode::StaleEpoch);
assert_eq!(err.mutation, MutationCertainty::None, "refused before any mutation");
// Part two: the replacement process. It is a fresh worker with no epoch at all, so the
// same request is refused on phase rather than on timeline -- and, either way, nothing
// from the old epoch is applied to a fresh brain.
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 err = within(
"uninitialized replacement",
f.harness.coordinator.probe_raw(
&replacement,
"Agent.Prepare",
Some(scope_at("demo", "e1", 1)),
prepare_params("fly-b"),
),
)
.await
.expect_err("an uninitialized replacement has no epoch to prepare in");
assert_eq!(
err.code,
ErrorCode::InvalidPhase,
"a fresh process has no epoch to be stale about: {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;
}
/// A well-formed `Agent.Prepare` body, for a probe whose subject is the scope rather than the
/// payload.
fn prepare_params(agent_id: &str) -> serde_json::Value {
serde_json::json!({
"agentId": agent_id,
"profileDigest": digest_of_bytes(b"whatever"),
"interval": {"numerator": "16666667", "denominator": "1"},
"decisionContextDigest": digest_of_bytes(b"whatever"),
"preStepStimulations": [],
})
}

17
services/flysim/crates/fly-session/tests/session.rs
View file

@ -35,7 +35,7 @@ const STEPS: u64 = 3;
async fn one_world_advance_per_complete_batch(via: Via) {
let mut f = default_fixture(via).await;
within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
let before = f.harness.environment_mutations();
let before = f.harness.environment_mutations().expect("an in-process arena");
let reports = within("run", f.harness.coordinator.run(STEPS)).await.unwrap();
assert_eq!(reports.len() as u64, STEPS);
assert_eq!(f.harness.coordinator.stats().advances, STEPS);
@ -46,7 +46,10 @@ async fn one_world_advance_per_complete_batch(via: Via) {
"the world is at exactly one boundary per batch"
);
// The environment's progress counter moves once per advance and not otherwise.
assert_eq!(f.harness.environment_mutations() - before, STEPS);
assert_eq!(
f.harness.environment_mutations().expect("an in-process arena") - before,
STEPS
);
assert_eq!(count(&f.harness.coordinator.audit, "advance:0"), 1);
assert_eq!(f.harness.coordinator.trace.transitions.len() as u64, STEPS);
f.shutdown().await;
@ -227,7 +230,8 @@ async fn bootstrap_cannot_advance_the_world_or_produce_a_reward(via: Via) {
// Warm-up did run, with learning disabled, so the models did mutate.
for agent in [fly_a(), fly_b()] {
assert!(
f.harness.agent_mutations(&agent) >= f.harness.config.warmup_ticks,
f.harness.agent_mutations(&agent).expect("an in-process agent")
>= f.harness.config.warmup_ticks,
"warm-up ticks are real mutations"
);
}
@ -379,12 +383,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;