# Fusee documentation # Get started with Fusee Fusee is a library for workflows that need to survive a restart. Write a function in your language, wrap work in named steps, and let Fusee record their results. When the process restarts, the workflow can reuse a recorded result and continue from there. ## Your first workflow [Install the SDK and CLI](installation.md), then follow the quickstart for your language. You'll build a workflow that greets someone, waits for approval, and resumes after you kill the process. ```ts import { workflow } from "@bitsinfra/fusee"; export const greet = workflow("greet", async (ctx, name: string) => { return await ctx.step("compose", () => `hello ${name}`); }); ``` [Start the TypeScript quickstart →](tutorial-typescript.md) ```python import fusee @fusee.workflow("greet") async def greet(ctx: fusee.Ctx, name: str) -> str: return await ctx.step("compose", lambda: f"hello {name}") ``` [Start the Python quickstart →](tutorial-python.md) ```rust use fusee::{Ctx, Error, WorkflowRef}; const GREET: WorkflowRef = WorkflowRef::new("greet"); async fn greet(ctx: Ctx, name: String) -> fusee::Result { ctx.step("compose", || async { Ok::<_, Error>(format!("hello {name}")) }).await } ``` [Start the Rust quickstart →](tutorial-rust.md) Go and Java embed the same engine. Start with the [Go SDK](../bindings/go/README.md) or the [Java SDK](../bindings/java/README.md) for build instructions and a checkout workflow. ## Where to go next - **[Core concepts](concepts.md)** — how workflows, runs, steps, and the journal fit together. - **[Examples](../examples/README.md)** — run a payment saga, an agent loop, a scheduled job, a human approval, or a cart in any of the five languages. - **[API reference](api/index.md)** — look up a method in your SDK or compare an operation across languages. - **[Deployment options](guides/running.md)** — choose between embedding the engine, running a server, and replicating the journal. # Installation Fusee runs on Linux and macOS. Choose the SDK for your application. The TypeScript and Python packages include the CLI used in the quickstarts. ## TypeScript Use Node.js 22.18 or later to run the TypeScript quickstart. In an existing project, install the SDK from npm: ```sh npm install @bitsinfra/fusee ``` Import it with `import { Engine, workflow } from "@bitsinfra/fusee"`. The package also provides the `fusee` CLI. To try the 1.0.0 starter in a new directory, run: ```sh mkdir hello && cd hello npm exec --yes --package=@bitsinfra/fusee@1.0.0 -- fusee init --lang ts npm install 'fusee@npm:@bitsinfra/fusee@1.0.0' npx --no-install fusee dev ``` The published 1.0.0 CLI generates files that import `fusee`. The npm alias above installs the public SDK under that name for the starter. Continue at [Run it](tutorial-typescript.md#2-run-it). ## Python Use Python 3.9 or later. Install the SDK and CLI from PyPI in a virtual environment: ```sh mkdir hello && cd hello python3 -m venv .venv . .venv/bin/activate python -m pip install fusee==1.0.0 fusee init --lang python fusee dev ``` Keep the environment active when running CLI commands. Continue at [Run it](tutorial-python.md#2-run-it). ## Rust Use Rust 1.88 or later. Add the SDK to an existing Cargo project with `cargo add fusee`. To start the tutorial in a new directory, install the CLI from crates.io: ```sh cargo install --locked fusee-cli --version 1.0.0 mkdir hello && cd hello fusee init --lang rust fusee dev -- cargo run ``` Continue at [Run it](tutorial-rust.md#2-run-it). ## Go Use Go 1.23 or later with cgo enabled, a C compiler, and the Rust toolchain. From the Fusee checkout: ```sh cargo build --release -p fusee-c (cd examples/go && go run ./payment_saga) ``` The [Go SDK guide](../bindings/go/README.md) covers linking the engine into your application. ## Java Use JDK 17 or later, Maven, a C compiler, and the Rust toolchain. From the Fusee checkout: ```sh cargo build --release -p fusee-c (cd bindings/java && mvn install) java -cp bindings/java/target/classes examples/java/PaymentSaga.java ``` The [Java SDK guide](../bindings/java/README.md) describes the Maven artifact and how it loads the native library. # TypeScript quickstart Ten minutes. You will start a workflow, kill the process in the middle of it, and watch it carry on from where it was. Before you begin, follow the [TypeScript installation instructions](installation.md#typescript). If you already started the sample there, continue at [Run it](#2-run-it). ## 1. A project ```sh mkdir hello && cd hello npm exec --yes --package=@bitsinfra/fusee@1.0.0 -- fusee init --lang ts npm install 'fusee@npm:@bitsinfra/fusee@1.0.0' ``` `fusee init` writes three files: - `workflows.ts`: the workflow, a plain value apart from any engine. - `app.ts`: opens an engine, starts one run, and keeps working. - `workflows.test.ts`: a test with the SDK’s test engine. It also writes `AGENTS.md` (and a `CLAUDE.md` that imports it): the rules a coding assistant needs to change the workflows safely. The 1.0.0 starter imports the SDK through the `fusee` alias installed above. Run the commands below from the project directory. The workflow composes a greeting in a step, pauses five seconds, then waits for an approval signal: ```ts export const approve = signal("approve"); export const greet = workflow("greet", async (ctx, name: string) => { const greeting = await ctx.step("compose", () => `hello ${name}`); await ctx.sleep("pause", "5s"); const ok = await ctx.waitFor(approve, { timeout: "1h" }); return ok ? greeting : `${greeting} (not approved)`; }); ``` A step's result is recorded once it runs. When the code runs again after a crash, `compose` returns its recorded result instead of running again; the sleep and the wait pick up where they were. ## 2. Run it ```sh npx --no-install fusee dev ``` `fusee dev` runs `node app.ts` under the dev profile and prints: ``` fusee: dev profile (/path/to/hello/.fusee/dev): every run's history is kept, flushes use fsync (a process crash loses nothing; power loss may). Not for production. fusee: UI and HTTP API at http://127.0.0.1:7601 started greet-1; approve it with: fusee runs signal greet-1 approve true ``` Open http://127.0.0.1:7601: the run `greet-1` is there, with its `compose` step done and its pause running. `fusee dev` restarts the app when a file changes or the app exits. ## 3. Kill it Within the five seconds of the pause, in another terminal: ```sh pkill -9 -f app.ts ``` That is a crash: no shutdown, no flush. `fusee dev` starts the app again, and the run continues from its journal. The UI shows `compose` once; it did not run a second time. ## 4. Approve it ```sh npx --no-install fusee runs signal greet-1 approve true ``` The app prints `hello world`. `fusee runs ls` lists the run as `ok`; `fusee runs show greet-1` prints its timeline. ## 5. Test it ```sh node --test workflows.test.ts ``` The test opens a `TestEngine` (in memory, under the dev profile), starts the run, moves the virtual clock past the five-second pause (`t.advance("5s")`), signals it, and checks the result. See [testing your workflows](testing-workflows.md). ## Next - [Errors and retries](errors.md): every error code and what to do. - [Features](features.md): steps, signals, queries, updates, children, compensation, schedules, objects, streams, limits. - [Operations](operations.md) and [deploying](../deploy/README.md): a server, replicated groups, monitoring. # Python quickstart Ten minutes. You will start a workflow, kill the process in the middle of it, and watch it carry on from where it was. Before you begin, follow the [Python installation instructions](installation.md#python). If you already started the sample there, continue at [Run it](#2-run-it). ## 1. A project ```sh mkdir hello && cd hello fusee init --lang python ``` `fusee init` writes three files: - `workflows.py`: the workflow, a plain value apart from any engine. - `app.py`: opens an engine, starts one run, and keeps working. - `test_workflows.py`: a test with `fusee.testing`. It also writes `AGENTS.md` (and a `CLAUDE.md` that imports it): the rules a coding assistant needs to change the workflows safely. The workflow composes a greeting in a step, pauses five seconds, then waits for an approval signal: ```python @fusee.workflow("greet") async def greet(ctx: fusee.Ctx, name: str) -> str: greeting = await ctx.step("compose", lambda: f"hello {name}") await ctx.sleep("pause", "5s") ok = await ctx.wait_for(approve, timeout="1h") return greeting if ok else f"{greeting} (not approved)" ``` A step's result is recorded once it runs. When the code runs again after a crash, `compose` returns its recorded result instead of running again; the sleep and the wait pick up where they were. ## 2. Run it ```sh fusee dev ``` `fusee dev` runs `app.py` under the dev profile and prints: ``` fusee: dev profile (/path/to/hello/.fusee/dev): every run's history is kept, each run is replayed against the code when it parks or ends. Not for production. fusee: UI and HTTP API at http://127.0.0.1:7601 started greet-1; approve it with: fusee runs signal greet-1 approve true ``` Open http://127.0.0.1:7601: the run `greet-1` is there, with its `compose` step done and its pause running. `fusee dev` restarts the app when a file changes or the app exits. ## 3. Kill it Within the five seconds of the pause, in another terminal: ```sh pkill -9 -f app.py ``` That is a crash: no shutdown, no flush. `fusee dev` starts the app again, and the run continues from its journal. The UI shows `compose` once; it did not run a second time. ## 4. Approve it ```sh fusee runs signal greet-1 approve true ``` The app prints `hello world`. `fusee runs ls` lists the run as `ok`; `fusee runs show greet-1` prints its timeline. ## 5. Test it ```sh python test_workflows.py ``` The test opens a `TestEngine` (in memory, under the dev profile), starts the run, signals it, and checks the result, the steps it recorded, and that the code replays against its journal (`t.check()`). See [testing your workflows](testing-workflows.md). ## Next - [Errors and retries](errors.md): every error code and what to do. - [Features](features.md): steps, signals, queries, updates, children, compensation, schedules, objects, streams, limits. - [Operations](operations.md) and [deploying](../deploy/README.md): a server, replicated groups, monitoring. # Rust quickstart Ten minutes. You will start a workflow, kill the process in the middle of it, and watch it carry on from where it was. Before you begin, follow the [Rust installation instructions](installation.md#rust). If you already started the sample there, continue at [Run it](#2-run-it). ## 1. A project ```sh mkdir hello && cd hello fusee init --lang rust ``` `fusee init` writes `Cargo.toml` and `src/main.rs`, and `AGENTS.md` (which a `CLAUDE.md` imports) with the rules a coding assistant needs. The workflow is a reference (`WorkflowRef`) and a function: ```rust pub const GREET: WorkflowRef = WorkflowRef::new("greet"); pub const APPROVE: Signal = Signal::new("approve"); engine.register(GREET, |ctx: Ctx, name: String| async move { let greeting: String = ctx .step("compose", || async { Ok::<_, Error>(format!("hello {name}")) }) .await?; ctx.sleep("pause", Duration::from_secs(5)).await?; let ok = ctx.wait_for(APPROVE).timeout(Duration::from_secs(3600)).await?; Ok(if ok == Some(true) { greeting } else { format!("{greeting} (not approved)") }) }); ``` A step's result is recorded once it runs. When the code runs again after a crash, `compose` returns its recorded result instead of running again; the sleep and the wait pick up where they were. ## 2. Run it ```sh fusee dev -- cargo run ``` Under `fusee dev` the engine runs with the dev profile (`FUSEE_PROFILE=dev`): it keeps every run's history, replays each run against the code when it waits or ends, and serves the UI: ``` fusee: dev profile (/path/to/hello/.fusee/dev): every run's history is kept, each run is replayed against the code when it waits or ends. Not for production. fusee: UI and HTTP API at http://127.0.0.1:7601 started greet-1; approve it: fusee runs signal greet-1 approve true ``` Open http://127.0.0.1:7601 to see the run. ## 3. Kill it Within the five seconds of the pause, in another terminal: ```sh pkill -9 fusee-app ``` `fusee dev` starts the app again; the run continues from its journal, and `compose` does not run twice. ## 4. Approve it ```sh fusee runs signal greet-1 approve true ``` The app prints `hello world`. ## 5. Test it Move the workflow's closure into a function, `async fn greet(ctx: Ctx, name: String) -> fusee::Result`, so a test can register it too, and add the test kit to `Cargo.toml`: ```toml [dev-dependencies] fusee = { path = "../fusee/crates/fusee", features = ["testing"] } ``` ```rust #[tokio::test] async fn greets_after_approval() -> fusee::Result<()> { let t = fusee::testing::TestEngine::open().await?; t.register(GREET, greet); let run = t.engine().start(GREET, "g-1", &"test".to_string()).await?; run.signal(APPROVE, &true).await?; // The five-second pause is skipped: virtual time. assert_eq!(t.result::("g-1").await?, "hello test"); t.check().await // strict replay } ``` See [testing your workflows](testing-workflows.md). ## Next - [Errors and retries](errors.md), [features](features.md), [operations](operations.md), [deploying](../deploy/README.md). # Concepts **Workflow.** Ordinary async code in Rust, TypeScript or Python, registered under a type name and version. A run is identified by a workflow ID you choose. Starting an ID that is running fails unless the start says to use the existing run (or replace it); starting an ID whose last run finished starts a new run unless the start says otherwise. See [ID policies](features.md#workflow-id-and-child-policies). **Journal.** Every operation a workflow makes through `ctx` is a journal entry: `step`, `sleep`, `wait_for`, `child`, `signal`, `now`, `random`, `uuid`, `await_completion`, `set_attributes`, an update handler's result. Accepted updates, signals and cancellation are journal entries too, and so are heartbeats and [stream](features.md#streams) chunks, which take no position. Operations are numbered in the order the code calls them. Their results are recorded in the order they are handed to the code. After a crash the code runs again from the top, and the journal hands back the recorded results in the same order, so the code takes the same path. Code between `ctx` calls must be deterministic. **Step.** `ctx.step(name, fn)` runs `fn` once every earlier record of the run is durable, records its result, and returns it immediately. On replay the recorded result is returned and `fn` does not run. If the process died after `fn` ran but before its result was durable, `fn` runs again: steps are at least once, and `ctx.key()` gives a stable idempotency key that makes them exactly once at services that accept such keys. A step marked read-only runs `fn` at once instead, without waiting for the records before it; see [read-only steps](features.md#read-only-steps). **Shards.** A workflow ID hashes to one of the engine's shards. Each shard is single-threaded and owns one log, its in-memory state and its timers. Messages between shards (child starts and results, signals between workflows) go through a durable outbox and are acknowledged by the receiver. **Running ahead of the disk.** Appending a record makes it visible at once. A flusher makes records durable in batches; exactly one flush is in flight per shard, and the next starts as soon as the last completes, so batches grow with load and an idle shard flushes a single record immediately. Three kinds of things wait for durability, and nothing else does: 1. a step's function, for the records before it (unless the step is [read-only](features.md#read-only-steps)); 2. an output: an acknowledgment, a result, a message to another shard, or a stream chunk handed to a reader; 3. nothing else. Code keeps running. **Objects.** An [object type](features.md#objects-keyed-state) is a set of handlers over per-key JSON state. Each key is a long-lived run (`/`) and each call is an update of it: exclusive calls run one at a time per key, shared calls run at once and only read. An exclusive handler's state changes are committed with its result, in its `UpdateDone` record. The engine continues a key's run as new with a snapshot of its state when its journal grows or before it parks, so a key's history stays short. **Parking.** A run that is only waiting on timers, signals or children is written to the run store and dropped from memory after `park_after` (default 1 s), or sooner, least recently used first, while loaded runs take more than `memory_budget` (default 128 MiB across shards). A parked run costs a 32-byte table entry plus its share of two hash indexes, under 48 bytes; parking again appends only the records added since its last frame. It is loaded back when a timer fires or a signal arrives. **Checkpoints.** Each shard checkpoints periodically. A checkpoint plus the run store plus the log after it reconstructs the shard, so older log segments are deleted. **Determinism errors.** If the code calls a different operation than its journal recorded at that position, the run stops with an error that names both, for example "the journal recorded step 'charge' here but the code called step 'refund'". The run stays stopped until code that matches is deployed; it is not marked failed. **Versions.** Register new code as a new version. Runs finish on the version they started with; new starts use the newest. When hot retention expires, a shard removes the run from lookups. It rewrites a run store once obsolete frames occupy at least 8 MiB and half its bytes. Two checkpoints keep their required store generations and log tails. To keep full histories beyond hot retention, enable Rust `Config::archive_history(true)`, TypeScript `archiveHistory: true`, or Python `archive_history=True`. The checkpoint worker writes checksummed immutable segments under the shard’s `archive/` directory. `fusee export` can read them after the live run expires. Archive files keep the journal’s encoded payloads and grow with retained history. The default keeps archival off. ## Checkpoints and cold state A checkpoint records the shard’s durable log position, the run-store generation and its exact byte prefix, and references for active runs and outgoing messages. An unchanged active journal can reuse its existing stored prefix. Recovery loads the newest readable retained checkpoint and replays later records. Segment cleanup removes a contiguous prefix safe for the older surviving checkpoint. Finishing or expiring an ID leaves enough information in the store to keep an old value from reappearing after recovery. For attribute searches, each shard builds a secondary index from the checkpointed prefix. Records newer than that prefix are checked separately. When the index is unavailable the query scans the stored state. With history archival enabled, terminal histories go to immutable checksummed files. Retrying an existing archive write verifies its contents before treating it as complete. # Examples Five small applications, each in Python, TypeScript, Rust, Go and Java. They print the same results in every language, and `scripts/check_examples.sh` runs all twenty-five in CI. | | Python | TypeScript | Rust | Go | Java | | --- | --- | --- | --- | --- | --- | | Payment saga: reserve, charge, wait for shipping; compensations undo on failure | [py](python/payment_saga.py) | [ts](typescript/payment-saga.mjs) | [rs](../crates/fusee/examples/payment_saga.rs) | [go](go/payment_saga/main.go) | [java](java/PaymentSaga.java) | | Agent loop: model and tool calls, each a step | [py](python/agent_loop.py) | [ts](typescript/agent-loop.mjs) | [rs](../crates/fusee/examples/agent_loop.rs) | [go](go/agent_loop/main.go) | [java](java/AgentLoop.java) | | Scheduled job: a schedule starts a report each second (a cron expression in a real job) | [py](python/scheduled_job.py) | [ts](typescript/scheduled-job.mjs) | [rs](../crates/fusee/examples/scheduled_job.rs) | [go](go/scheduled_job/main.go) | [java](java/ScheduledJob.java) | | Human approval: ask, wait up to three days for a decision, act on it | [py](python/human_approval.py) | [ts](typescript/human-approval.mjs) | [rs](../crates/fusee/examples/human_approval.rs) | [go](go/human_approval/main.go) | [java](java/HumanApproval.java) | | Cart object: one object per user, calls that take turns, a repeated call that runs once | [py](python/cart_object.py) | [ts](typescript/cart-object.mjs) | [rs](../crates/fusee/examples/cart_object.rs) | [go](go/cart_object/main.go) | [java](java/CartObject.java) | Each keeps its data in `./data/` under the directory it runs from, and its workflow IDs attach to an earlier run (`use-existing`), so running one again, or killing it midway and running it again, continues where it was. Run one: ```sh PYTHONPATH=bindings/python/python python3 examples/python/payment_saga.py cargo run --release -p fusee --example payment_saga (cd examples/go && go run ./payment_saga) java -cp bindings/java/target/classes examples/java/PaymentSaga.java ``` The TypeScript examples import `fusee`: run them from a project where it is installed (`npm install @bitsinfra/fusee`), as `scripts/check_examples.sh` arranges. `invoked_worker.py` and `invoked-worker.mjs` show a worker the server starts on demand ([server mode](../docs/server-mode.md)). # Errors and retries A step's function talks to the world, and the world fails. This guide is about deciding, for each failure, whether to try again, give up, or undo. The reference is [errors](../errors.md); the codes and the failure envelope are there. ## What is retried by default A step that throws is retried: three attempts, about 100 ms and then 200 ms apart (each delay drawn between half of it and all of it, so many runs that fail together do not retry together). The delay taken is recorded, so a replay after a crash waits for the same time. ```ts const charge = await ctx.step("charge", (s) => payments.charge(order, { idempotencyKey: s.key }), { retry: { attempts: 5, initial: "200ms", max: "1m", maxElapsed: "10m" }, }); ``` ```python charge = await ctx.step("charge", lambda s: payments.charge(order, idempotency_key=s.key), retry=fusee.Retry(attempts=5, initial="200ms", max="1m", max_elapsed="10m")) ``` Pass the step's key to the service as its idempotency key. A step's function can run more than once (a retry, or a crash between the call and its recording), and the key is how the service tells a repeat from a new request. `fusee.testing`'s `crash_everywhere` finds steps that forget it ([testing your workflows](../testing-workflows.md)). ## Failures that must not be retried A declined card stays declined. Mark it final and the step fails at once: ```ts class CardDeclined extends FatalError {} throw new CardDeclined("card declined", { code: "card_declined", details: { last4 } }); ``` ```python class CardDeclined(fusee.Fatal): code = "card_declined" raise CardDeclined("card declined") ``` In Rust, return `Failure::fatal("card declined").kind("CardDeclined").code("card_declined")`; in Go, `fusee.Fatal(err)`; in Java, throw a `FatalException`. Or keep the error as it is and say which types the policy does not retry: `nonRetryable: [CardDeclined]` (TypeScript), `non_retryable=(CardDeclined,)` (Python), `NonRetryable: []string{"CardDeclined"}` (Go), `.nonRetryable("CardDeclined")` (Java). `retryIf(failure, attempt)` decides case by case. ## What the caller sees The failure travels as an envelope: its message, type, code and details reach the caller in any language. A step that failed for good raises `StepFailed` in the workflow, with the step's failure as its cause: ```ts try { await ctx.step("charge", chargeCard); } catch (e) { if (e instanceof StepError && e.cause.code === "card_declined") return { status: "declined" }; throw e; } ``` ```python async def pay(ctx, order): try: await ctx.step("charge", charge_card) except fusee.StepError as e: if e.cause.code == "card_declined": return {"status": "declined"} raise ``` A workflow that throws fails with `WorkflowFailed`, its failure as the cause; a child's failure reaches its parent as `ChildFailed`. Each wraps the one below, so the original type and code survive any number of levels, across languages: a Go workflow's `CardDeclined` is the TypeScript caller's `e.cause.cause.type`. ## Timeouts `timeout` bounds one attempt; `heartbeatTimeout` fails an attempt that has not reported progress (`s.heartbeat(progress)`) for that long, so a long step can be cut short when it hangs rather than when it ends. Both are retried under the policy. A workflow's `timeout` bounds the whole run and ends it `TimedOut`. ## Pausing instead of failing Some failures outlast any retry policy but not a fix: a bug, an expired credential, a service that is down for the afternoon. Pause the run when a step's retries run out, instead of failing it, and resume it once the cause is fixed. The run continues as if the step had succeeded the first time; its parent and callers keep waiting meanwhile. ```ts await ctx.step("charge", (s) => payments.charge(order, { idempotencyKey: s.key }), { retry: { attempts: 5, onExhausted: "pause" }, }); // Once payments are back: await engine.run("order-1").resume(); ``` ```python await ctx.step("charge", lambda s: payments.charge(order, idempotency_key=s.key), retry=fusee.Retry(attempts=5, on_exhausted="pause")) # Once payments are back: await engine.run("order-1").resume() ``` `onExhausted: "pause"` on the engine makes it every step's default. Final failures still fail at once. Paused runs show as `paused` in listings and on the UI's Needs attention screen, and `fusee runs bulk resume --status paused` resumes them all; see [pausing when retries run out](../features.md#pausing-when-retries-run-out). ## When retrying is not enough Retries cover a failure that goes away. For one that does not, undo what the run already did: see [sagas](sagas.mdx). For a step that must not run twice at all, make the service idempotent with the key; Fusee cannot make the outside world exactly-once by itself, only make the attempts and their order exact. # Sagas: undoing what a run did A checkout reserves stock, charges a card and books shipping. If the charge fails after the reservation, the reservation must be released. A saga is a workflow that registers, for each step that changes the world, the step that undoes it. ```ts const checkout = workflow("checkout", async (ctx, order) => { const hold = await ctx.step("reserve", (s) => inventory.reserve(order.sku, s.key)); ctx.compensate("release", () => inventory.release(hold)); const charge = await ctx.step("charge", (s) => payments.charge(order.cents, s.key)); ctx.compensate("refund", () => payments.refund(charge)); await ctx.step("ship", (s) => shipping.book(order, s.key)); return charge; }); ``` ```python @fusee.workflow("checkout") async def checkout(ctx, order): hold = await ctx.step("reserve", lambda s: inventory.reserve(order["sku"], s.key)) ctx.compensate("release", lambda: inventory.release(hold)) charge = await ctx.step("charge", lambda s: payments.charge(order["cents"], s.key)) ctx.compensate("refund", lambda: payments.refund(charge)) await ctx.step("ship", lambda s: shipping.book(order, s.key)) return charge ``` If the workflow fails or is cancelled, the registered compensations run, newest first, each as a step: `refund`, then `release`. Then the run ends with its failure. A compensation is a step like any other, so it is recorded, retried under its policy and, after a crash, not run again once done. One that fails after its retries is logged and the others still run. ## Rules that keep a saga correct - **Register the undo after the step succeeds.** A step that failed did nothing to undo (if it could have half-done something, make its undo safe to run on nothing, and register it before). - **Give every step and every undo the step's key** as the service's idempotency key. A crash can make either run twice. - **Undo is not rollback.** The charge happened; the refund is a second event the customer may see. Order steps so the ones hardest to undo come last. - **Cancel is a request.** `run.cancel(reason)` reaches the code as `Cancelled` at its next operation, and compensations run. `terminate` ends the run at once: nothing runs after it, not even compensation. ## Without `compensate` The same thing by hand, when the undo depends on how far the run got: ```ts try { await ctx.step("ship", (s) => shipping.book(order, s.key)); } catch (e) { await ctx.step("refund", () => payments.refund(charge)); throw e; } ``` The [payment saga example](../../examples/README.md) runs in five languages. Testing a saga: `crash_everywhere` crashes the run once at each step boundary and requires the same result and the same outside effects, so a missing idempotency key shows up as a test failure ([testing your workflows](../testing-workflows.md)). # Waiting for people A refund over a limit needs a manager. An expense needs finance. A contract waits for a signature. The run asks, waits (minutes or weeks, at no cost while it waits), and continues when the answer comes. ## A signal and a timeout ```ts const decision = signal("decision"); const expense = workflow("expense", async (ctx, claim) => { await ctx.step("ask", () => mail.send(finance, approvalLink(ctx.workflowId))); const answer = await ctx.waitFor(decision, { timeout: "3d" }); if (answer === undefined) return "expired"; if (!answer.approved) return `rejected by ${answer.by}`; return ctx.step("pay", (s) => payments.reimburse(claim, s.key)); }); // The approval page's handler, any process, any time later: await engine.run("expense-7").signal(decision, { approved: true, by: "dana" }); ``` ```python decision = fusee.signal("decision", dict) @fusee.workflow("expense") async def expense(ctx, claim): await ctx.step("ask", lambda: mail.send(finance, approval_link(ctx.workflow_id))) answer = await ctx.wait_for(decision, timeout="3d") if answer is None: return "expired" if not answer["approved"]: return f"rejected by {answer['by']}" return await ctx.step("pay", lambda s: payments.reimburse(claim, s.key)) ``` ```rust const DECISION: Signal = Signal::new("decision"); let answer = ctx .wait_for(DECISION) .timeout(Duration::from_secs(3 * 86_400)) .await?; let Some(answer) = answer else { return Ok("expired: nobody answered in three days".into()); }; ``` See the [complete approval workflow](../../crates/fusee/examples/human_approval.rs) for the request, the decision type, and the caller that sends the signal. While it waits, the run is parked: nothing of it is in memory, and a restart or a deploy changes nothing. A signal that arrives before the code reaches `waitFor` is kept and taken when it gets there. `fusee runs signal expense-7 decision '{"approved": true, "by": "dana"}'` and the UI's Signal button send the same signal, and both record who sent it ([audit](../features.md#audit)). ## A link that completes a step When the answer comes from a system that only knows how to call a URL (an e-signature service, a payment page), wait on a completion instead: ```ts const signed = await ctx.awaitCompletion("signature", (cb) => esign.send(contract, { webhook: cb.url })); ``` `cb.url` completes that step when something `POST`s the value to it (`…/v1/fail/` fails it). The URL is unguessable, names the run, step and attempt, and stops working once used. ## Showing where a run is A query answers from the run's state without changing it: ```ts ctx.onQuery("status", () => (answered ? "decided" : "waiting for finance")); const status = await engine.run("expense-7").query("status"); ``` Reminders are a sleep in a loop with the wait: `ctx.sleep("remind-1", "1d")` raced against the signal with `ctx.race`. Escalation is a second wait with a new approver. Each is a few lines of ordinary code, recorded step by step. The [human approval example](../../examples/README.md) runs in five languages. # Agents An agent is a loop: ask a model, run the tool it asks for, give it the result, repeat. Each model call and each tool call is a step, so a crash in the middle of a long session resumes at the step it was on, and the model is not asked again for answers it already gave. ```ts const agent = workflow("agent", async (ctx, question) => { const history = [{ role: "user", content: question }]; for (let turn = 0; turn < 20; turn++) { const reply = await ctx.step(`model-${turn}`, () => model.complete(history)); if (!reply.toolCall) return reply.text; const result = await ctx.step(`tool-${turn}-${reply.toolCall.name}`, (s) => tools[reply.toolCall.name](reply.toolCall.args, { idempotencyKey: s.key }), ); history.push(reply.message, { role: "tool", content: result }); } throw new FatalError("no answer after 20 turns"); }); ``` Step names carry the turn, so the journal reads as the conversation did. Tool calls that change the world take the step's key, like any step. An agent already written with the Vercel AI SDK, Pydantic AI or the OpenAI Agents SDK becomes durable without rewriting its loop: see [agent patterns](agent-patterns.md), which also covers sessions, tool approvals, interrupting and the other shapes agents take. ## Streaming tokens A user watching an answer wants tokens as they come. `modelCall` makes the model call a step whose tokens are a stream, readable from anywhere while the step runs: ```ts import { modelCall } from "@bitsinfra/fusee/ai"; const text = await modelCall(ctx, `model-${turn}`, async function* (resume) { // `resume`: the tokens an earlier attempt already streamed. Ask the model // to continue from there (an assistant prefill) instead of starting over. yield* model.stream(history, { prefill: resume }); }); // A web handler, in any process: follow the stream as it grows. for await (const token of engine.run(id).stream(`model-${turn}`)) res.write(token); ``` ```python from fusee.ai import model_call text = await model_call(ctx, f"model-{turn}", lambda resume: model.stream(history, prefill=resume)) ``` Readers see only tokens that are durable, so nothing they show is taken back after a crash; a reader that reconnects with its offset continues exactly where it was. A retried call keeps what was streamed and appends after it. The [streams reference](../features.md#streams) has the rules. ## Long sessions - **Context windows.** A run's journal grows by two records per turn. After a few hundred turns, summarize the history and continue as new with the summary: `return ctx.continueAsNew({ summary })`. - **Human in the loop.** A tool that needs approval is a wait: see [waiting for people](human-approval.mdx). - **Rate limits.** Put model calls under a limit, `limit: "openai"` on the step, sized with `engine.limit("openai", { perSecond: 50 })`; steps over it wait their turn instead of failing. - **Cost and time caps.** A workflow `timeout`, or a step counter in the loop as above. The [agent loop example](../../examples/README.md) runs in five languages. # Agent patterns Each pattern below is built from what [agents](agents.mdx) already uses: steps, streams, children, waits and objects. None needs anything else, and each keeps the guarantee of a step: a crash resumes at the call that was running, and nothing that finished runs again. ## The Vercel AI SDK An agent written with the AI SDK becomes durable without changing its loop: `durableModel` makes each model call a step and `durableTools` makes each tool call one. ```ts import { generateText, stepCountIs } from "ai"; import { openai } from "@ai-sdk/openai"; import { workflow } from "@bitsinfra/fusee"; import { durableModel, durableTools } from "@bitsinfra/fusee/ai/vercel"; const support = workflow("support", async (ctx, question: string) => { const { text } = await generateText({ model: durableModel(ctx, openai("gpt-5")), tools: durableTools(ctx, { lookupOrder, refund }), maxRetries: 0, // the steps retry, and their retries are recorded stopWhen: stepCountIs(10), prompt: question, }); return text; }); ``` - Model calls are steps named `model`, tool calls `tool:`. Steps are matched by position, so a determinism error names the call that moved. - A tool's `execute` gets the AI SDK's `toolCallId`: pass it to services that take an idempotency key, as a step passes `s.key`. - `streamText` works too: each call's text is a stream (`model#1`, `model#2`, …) readers follow while it runs (see [streaming to a chat UI](#streaming-to-a-chat-ui)). - Set the AI SDK's own retries to zero. The step's retry policy applies instead, `onExhausted: "pause"` included: an agent whose provider is down waits for it instead of losing the conversation. ## Pydantic AI `FuseeDurability` makes a Pydantic AI agent's model requests and tool calls steps; `run` runs the agent as part of the workflow. ```python import fusee from pydantic_ai import Agent from fusee.ai.pydantic import FuseeDurability, run agent = Agent( "openai:gpt-5", name="support", tools=[lookup_order, refund], capabilities=[FuseeDurability(model_options={"retry": fusee.Retry(on_exhausted="pause")})], ) @fusee.workflow("support") async def support(ctx, question: str) -> str: result = await run(ctx, lambda: agent.run(question)) return result.output ``` - Steps are named after the agent and its toolsets (`support__model.request`, `support__function_toolset__.call_tool:refund`): name the agent and give each toolset an `id`. - A tool's own step options are its metadata's `"fusee"` entry; `False` calls it without a step. - Pydantic AI's waits between its own retries are durable sleeps. - Outside `run` the agent is an ordinary agent, so the same agent serves tests and scripts. - Install with `pip install 'fusee[pydantic-ai]'`. ## The OpenAI Agents SDK `run` is `Runner.run` with the agent made durable: model calls are steps named `model`, function tool calls `tool:`, and the agents it hands off to are made durable too. ```python import fusee from agents import Agent, function_tool from agents.tool_context import ToolContext from fusee.ai.openai import run @function_tool def refund(ctx: ToolContext, order: str, amount: int) -> str: return payments.refund(order, amount, idempotency_key=ctx.tool_call_id) agent = Agent(name="support", instructions="Help with orders.", tools=[refund]) @fusee.workflow("support") async def support(ctx, question: str) -> str: result = await run(ctx, agent, question, model_options={"retry": fusee.Retry(on_exhausted="pause")}) return result.final_output ``` - `model_options` and `tool_options` are the steps' options (retries, timeout, limit). The OpenAI client's own retries happen inside one attempt; set them to zero to leave retrying to the step. - `durable_agent(agent)` is the durable copy on its own, for code that calls `Runner` itself; outside a workflow it is an ordinary agent. - Hosted tools (web search, file search) run inside the model call, so they are part of its step. - Install with `pip install 'fusee[openai-agents]'`. ## Sessions: one conversation per key An object keeps a conversation's history under its key, and its exclusive handler runs one turn at a time, in the order the messages came. ```ts import { modelCall } from "@bitsinfra/fusee/ai"; const chats = engine.object("chat", { say: async (ctx, text: string) => { const history = ctx.state.get("history") ?? []; history.push({ role: "user", content: text }); const reply = await modelCall(ctx, "reply", (resume) => model.stream(history, { prefill: resume })); history.push({ role: "assistant", content: reply }); ctx.state.set("history", history); return reply; }, history: shared((ctx) => ctx.state.get("history") ?? []), }); await chats.key("user-7").call("say", "Where is my order?"); ``` ```python async def say(ctx, text): history = ctx.state.get("history") or [] history.append({"role": "user", "content": text}) reply = await model_call(ctx, "reply", lambda resume: model.stream(history, prefill=resume)) history.append({"role": "assistant", "content": reply}) ctx.state.set("history", history) return reply chats = fusee.object("chat", {"say": say}) ``` A support engineer can fix a session's history from outside with [`fusee objects edit chat user-7`](../features.md#editing-state-from-outside). ## Approval before a tool runs A tool that changes something a person must agree to waits for them. The wait holds no process: the run parks until the answer comes. ```ts const tools = durableTools(ctx, { refund }, { approve: { tools: ["refund"], ask: (cb, call) => slack.post(`Refund ${call.input.amount}? ${cb.url}`) }, }); ``` Completing the URL with `true` runs the tool; `{ "approved": false, "reason": "..." }` gives the model the refusal as the tool's result. Without the adapter, the same is `ctx.awaitCompletion` before the step, or `ctx.waitFor(approval, { timeout: "1d" })` on a signal (see [waiting for people](human-approval.mdx)). ## Interrupt and regenerate Run each turn as its own child so it can be stopped. Cancelling the run aborts the model call's request (its step's `s.signal`), and the next turn starts from the history as it was. ```ts const turn = workflow("turn", async (ctx, history: Message[]) => ctx.step("model", (s) => model.complete(history, { signal: s.signal })), ); // The user pressed "stop", then "regenerate": await engine.run("chat-7/turn-3").cancel("interrupted"); await engine.start(turn, "chat-7/turn-3b", history); ``` ## Parallel tool calls Calls that do not depend on each other run at once. Each is its own step, so a crash keeps the ones that finished. ```ts const [weather, flights] = await ctx.all([ ctx.step("tool:weather", () => weatherApi.get(city)), ctx.step("tool:flights", () => flightsApi.search(city, date)), ]); ``` ```python weather, flights = await ctx.all([ ctx.step("tool:weather", lambda: weather_api.get(city)), ctx.step("tool:flights", lambda: flights_api.search(city, date)), ]) ``` ## Racing models Ask two models and take the first answer; the slower call is cancelled. ```ts const answer = await ctx.race([ ctx.step("model:fast", (s) => fast.complete(prompt, { signal: s.signal })), ctx.step("model:large", (s) => large.complete(prompt, { signal: s.signal })), ]); ``` ## A workflow as a tool A tool can be a whole workflow, with its own steps, retries and journal: the agent waits for it as a child. ```ts const research = workflow("research", async (ctx, topic: string) => { /* many steps */ }); const tools = { research: { description: "Researches a topic", execute: ({ topic }, { toolCallId }) => ctx.child(research, `${ctx.workflowId}/${toolCallId}`, topic) }, }; ``` The child's ID comes from the call, so a replay attaches to the same child instead of starting another. ## Orchestrator and workers A planner splits the work, children do the parts, and a final step joins them. ```ts const plan = await ctx.step("plan", () => planner.split(task)); const parts = await ctx.all(plan.parts.map((p, i) => ctx.child(writer, `${ctx.workflowId}/part-${i}`, p))); return ctx.step("join", () => editor.merge(parts)); ``` ## A sequential pipeline Each stage is its own model call, so a failure in a later stage does not repeat the earlier ones. ```python outline = await ctx.step("outline", lambda: model.complete(f"Outline: {topic}")) draft = await ctx.step("draft", lambda: model.complete(f"Write from: {outline}")) final = await ctx.step("edit", lambda: model.complete(f"Edit: {draft}")) ``` ## Evaluate and retry A judge scores each answer; a low score asks again, up to a bound the journal enforces. ```ts for (let round = 1; round <= 3; round++) { const answer = await ctx.step("model", () => model.complete(prompt)); const score = await ctx.step("judge", () => judge.score(prompt, answer)); if (score >= 0.8) return answer; prompt = `${prompt}\nA reviewer scored the last answer ${score}. Improve it.`; } throw new FatalError("no answer passed review in 3 rounds"); ``` ## Streaming to a chat UI A model call's tokens are a stream of the run. A web handler in any process follows it; a reader that reconnects passes its offset and misses nothing. ```ts // In the workflow: const reply = await modelCall(ctx, "reply", (resume) => model.stream(history, { prefill: resume })); // In the web server, with server-sent events: app.get("/chat/:id/stream", async (req, res) => { res.setHeader("Content-Type", "text/event-stream"); for await (const token of engine.run(req.params.id).stream("reply", { from: Number(req.query.from ?? 0) })) { res.write(`data: ${JSON.stringify(token)}\n\n`); } res.end(); }); ``` The HTTP API serves the same stream without your server: `GET /v1/runs/{id}/streams/reply` with `Accept: text/event-stream`. ## Notify when ready A long task tells someone when it is done: the last step calls a webhook with the step's key, so a crash after the call does not notify twice. ```ts const report = await ctx.child(research, `${ctx.workflowId}/research`, topic); await ctx.step("notify", (s) => fetch(webhookUrl, { method: "POST", headers: { "Idempotency-Key": s.key }, body: JSON.stringify({ id: ctx.workflowId, report }), })); ``` A caller that would rather wait asks for the result instead: `await engine.run(id).result()`, or `GET /v1/runs/{id}/result`. ## Undo on failure An agent that books, buys or sends registers how to undo each action. If the run fails or is cancelled, the compensations run, newest first. ```ts const booking = await ctx.step("tool:book", (s) => hotels.book(room, { idempotencyKey: s.key })); ctx.compensate("tool:cancel-booking", () => hotels.cancel(booking.id)); ``` ## Long conversations A run's journal grows by a few records per turn. After a few hundred turns, summarize and continue as a new run with the summary. ```python @fusee.workflow("chat") async def chat(ctx, state): history = await run_turns(ctx, state) if ctx.journal_size() > 1000: summary = await ctx.step("summarize", lambda: model.summarize(history)) return ctx.continue_as_new({"summary": summary}) ``` ## Rate and cost limits Model calls take a permit of a limit, so many runs share a provider's quota and wait their turn instead of failing. A workflow timeout bounds what a runaway agent can spend. ```ts engine.limit("openai", { perSecond: 50, concurrent: 20 }); await ctx.step("model", () => model.complete(history), { limit: "openai", fairnessKey: tenantId }); await engine.start(agent, id, question, { timeout: "15m" }); ``` ## When the provider is down A model call that keeps failing can pause its run instead of failing it: the conversation waits, with everything it has, until someone resumes it. ```ts await ctx.step("model", () => model.complete(history), { retry: { attempts: 5, max: "30s", onExhausted: "pause" }, }); // Once the provider is back: // fusee runs bulk resume --type support --status paused ``` See [pausing when retries run out](../features.md#pausing-when-retries-run-out). # Writing Fusee code with an AI assistant Claude, Cursor, Codex and the others write Fusee code well once they know the handful of rules that make a workflow replay. This page is those rules, each with the version a model tends to write and the one that works. Point your assistant here, or at the whole documentation as one file: [`llms-full.txt`](https://fusee.bitsinfra.com/llms-full.txt) (the index is [`llms.txt`](https://fusee.bitsinfra.com/llms.txt)), or at one page as Markdown by adding `.md` to its address (`https://fusee.bitsinfra.com/guides/sagas.md`). A project made with `fusee init` has the same rules in its `AGENTS.md`, which a `CLAUDE.md` imports. ## Workflow code is replayed After a crash, the workflow function runs again from the top against its journal: recorded steps return their results without running. So the code between `ctx` calls must make the same decisions every time. **Await only `ctx` operations in workflow code.** Anything else (a fetch, a database query, a sleep with `setTimeout` or `asyncio.sleep`) goes inside a step. ```ts // Wrong: the fetch runs again on every replay, and its answer is not recorded. const user = await fetch(`/users/${id}`).then((r) => r.json()); // Right: const user = await ctx.step("load-user", () => fetch(`/users/${id}`).then((r) => r.json())); ``` **Read time, randomness and IDs through the context.** They are recorded, so a replay sees the values the first run saw. ```python # Wrong: a replay gets another time and another ID. started = time.time() order_id = str(uuid.uuid4()) # Right: started = ctx.now() order_id = ctx.uuid() ``` **Keep step names stable and in the same order.** Steps are matched to the journal by position and name. Renaming or reordering steps of a workflow that has runs in flight is a determinism error: register the change as a new version, or guard it with `ctx.patched("id")` so older runs take the old path. ## A step's function may run more than once A step runs again if the process died after its function ran but before its result was durable. Pass the step's key to every service that accepts an idempotency key, and the effect happens once. ```ts // Wrong: a retry after a crash charges twice. await ctx.step("charge", () => stripe.charges.create({ amount, customer })); // Right: await ctx.step("charge", (s) => stripe.charges.create({ amount, customer }, { idempotencyKey: s.key })); ``` ```python await ctx.step("charge", lambda s: stripe.Charge.create(amount=amount, customer=customer, idempotency_key=s.key)) ``` ## Failures - Errors thrown in a step are retried (three attempts by default). A failure that retrying cannot fix is a `FatalError` (`fusee.Fatal` in Python, `Failure::fatal` in Rust): a declined card, a validation error. - `retry: { attempts, initial, max, maxElapsed }` sets the policy; `onExhausted: "pause"` makes a run wait for `fusee runs resume` instead of failing when the retries run out. - To undo earlier steps when a later one fails, register compensations with `ctx.compensate(name, fn)`; they run newest first. ## Values Step results, inputs and signals are stored as JSON. A `Date` comes back as a string, a `Map` or a class instance as a plain object, unless the engine opens with the `rich` serializer (`@bitsinfra/fusee/rich`). In Python, annotate inputs and outputs with dataclasses or pydantic models to get them back typed. ## Testing Use the test kit, not a real engine with sleeps: virtual time skips `ctx.sleep`, and `crashEverywhere` (`crash_everywhere`) runs a workflow once per step boundary with a crash there, finding steps that forget their idempotency key. See [testing your workflows](../testing-workflows.md). ## Checklist for a reviewer - Every `await` in workflow code is a `ctx` call. - No `Date.now()`, `Math.random()`, `time.time()`, `uuid4()` or environment reads in workflow code outside a step. - Every step that changes something outside passes `s.key` on. - Step names are fixed strings, or built only from the input and earlier step results. - Final failures are `FatalError`; everything else may be retried. - A change to the steps of a workflow with runs in flight is a new version or a patch. # Schedules and timers Three tools, from the smallest: - **A sleep** inside a run: `await ctx.sleep("cool-off", "10m")`. Durable: a crash or a deploy during the ten minutes changes nothing. - **A delayed start**: `engine.start(report, "report-7", input, { delay: "1h" })` (or `at: date`). The run exists under its ID from now; `result()` waits through the delay. - **A schedule**: a run on every cron time or interval, until you delete it. ## A schedule ```ts const s = await engine.schedule( "nightly-report", { cron: "0 2 * * *", timezone: "Europe/London", overlap: "skip" }, report, { recipients: ["ops@example.com"] }, ); await s.pause(); await s.resume(); await s.trigger(); await s.backfill(new Date("2026-11-01"), new Date("2026-11-07")); await s.delete(); ``` ```python s = await engine.schedule("nightly-report", {"cron": "0 2 * * *", "timezone": "Europe/London", "overlap": "skip"}, report, {"recipients": ["ops@example.com"]}) ``` ```rust let spec = ScheduleSpec { cron: Some("0 2 * * *".into()), timezone: "Europe/London".into(), overlap: Overlap::Skip, workflow: "report".into(), input: serde_json::json!({ "recipients": ["ops@example.com"] }), ..Default::default() }; let schedule = engine.schedule("nightly-report", spec).await?; ``` `fusee schedules ls`, `fusee schedules pause nightly-report` and the UI's Schedules screen do the same from outside. ## What it guarantees - **Each time runs once.** A run's ID is the schedule's name and its time (`nightly-report@2026-11-09T02:00:00Z`), and a start attaches to a run that has the ID. Two processes, a crash during the start, a backfill over a time that already ran: still one run per time. - **Missed times catch up.** After downtime, the times missed within `catchupWindow` (default a year) start, oldest first. Set it to an hour for a job where only the latest matters. - **Overlap is a choice.** `skip` (the default) skips a time while the previous run still runs; `buffer-one` keeps one waiting; `allow-all` starts every time; `cancel-other` cancels the running one first. - **Time zones.** A cron in `Europe/London` fires at 02:00 local time across daylight-saving changes; a time that does not exist that night is skipped, one that happens twice runs once. - **Jitter** spreads starts of many schedules over a window (recorded, so the same on replay). A schedule is itself a workflow (`fusee.schedule/`) running the same native code in every SDK, so it has no scheduler process to deploy and keeps working through restarts and failovers. ## Choosing | You want | Use | | --- | --- | | a pause in a process | `ctx.sleep` | | one run later | a delayed start | | a run every hour, or at 02:00 | a schedule | | the latest of a burst of events, after a quiet period | `engine.debounce` | | events gathered into batches | `engine.batch` | The [scheduled job example](../../examples/README.md) runs in five languages. # Testing your workflows The test kit gives a test an engine in memory under the [dev profile](operations.md): every run's history is kept, each run can be replayed against your code (strict replay), and the result of a stuck run fails at once instead of waiting. ## Rust: `fusee::testing` ```toml [dev-dependencies] fusee = { version = "1.0.0", features = ["testing"] } ``` The feature exists only for tests: a build without it carries none of the kit's hooks, so the step path is the same as in production. ```rust use fusee::testing::TestEngine; use std::time::Duration; #[tokio::test] async fn renews_after_a_day() -> fusee::Result<()> { let t = TestEngine::open().await?; t.register(RENEW, renew); // Replace a step's function by workflow type and step name. t.mock_step(RENEW, "charge", |_attempt| Ok(serde_json::json!("ch_1"))); t.engine().start(RENEW, "r-1", &"pro".to_string()).await?; // The workflow sleeps a day: while every run waits on a timer, result() // moves the virtual clock to it, so this takes milliseconds. let out: String = t.result("r-1").await?; assert_eq!(out, "pro:42:ch_1"); assert_eq!(t.steps("r-1").await?, ["quote", "charge"]); t.check().await?; // strict replay of every run Ok(()) } ``` | | | | --- | --- | | `TestEngine::open()`, `with_shards(n)` | a `memory` engine, dev profile, nothing served | | `register(wf, f)` | as `Engine::register`, and again after each crash | | `mock_step(wf, step, f)` | `f(attempt)` returns the step's result, or `Err(Failure)` to fail that attempt (retried under the step's policy unless final) | | `advance(d)`, `now_ms()` | move the virtual clock; timers that fall due fire | | `result(id)` | the result; while every run waits on a timer (and no step function waits on outside work), time skips to it | | `crash()` | drop the engine's memory and recover every run from its journal: the code replays, recorded steps do not run again | | `steps(id)`, `journal(id)`, `snapshot(id)` | what the run recorded; `snapshot` is stable JSON for snapshot tests | | `check()` | replays every run against the registered code; an error names each run and step that differ | ### Finding steps that are not idempotent A step's function can run twice: when the process dies after the function ran and before its result was recorded, the step runs again after recovery. `crash_everywhere` runs a workflow once per step boundary with such a crash at each, and requires the same result and the same outside effects as a run without one: ```rust use fusee::testing::{crash_everywhere, Case}; #[tokio::test] async fn checkout_survives_a_crash_anywhere() { crash_everywhere(|t| async move { let payments = Arc::new(FakePayments::default()); // fresh per run t.register(CHECKOUT, checkout_with(payments.clone())); Ok(Case { workflow: "checkout".into(), id: "order-1".into(), input: serde_json::json!({"cents": 500}), effects: Box::new(move || payments.snapshot()), }) }) .await .unwrap(); } ``` A step that charges a card without passing `ctx.key()` to the payment service as an idempotency key fails with a message naming the step and both sets of effects. ## Python: `fusee.testing` ```python from fusee.testing import TestEngine async def test_renewal(): t = await TestEngine.open(workflows=[renew]) t.mock_step(renew, "charge", lambda attempt: {"id": "ch_1"}) await t.start(renew, "r-1", "pro") # The workflow sleeps a day: result() skips time while every run waits. assert await t.result("r-1") == "pro:ch_1" assert await t.steps("r-1") == ["quote", "charge"] await t.crash() # drop memory, recover from the journals await t.check() # strict replay of every run await t.close() ``` `mock_step` takes a function of the attempt number; raising fails that attempt. `advance("1d")` moves the clock; `result()` moves it to the next timer while no step function is running. `crash_everywhere(build)` takes a function that returns a `Case(workflows, workflow, id, input, effects)` with fresh fakes and raises `AssertionError` naming the step whose rerun changed the result or the effects. ## TypeScript: `@bitsinfra/fusee/testing` ```ts import { TestEngine, crashEverywhere } from "@bitsinfra/fusee/testing"; const t = await TestEngine.open({ workflows: [renew] }); t.mockStep(renew, "charge", (attempt) => ({ id: "ch_1" })); await t.start(renew, "r-1", "pro"); assert.equal(await t.result("r-1"), "pro:ch_1"); // a day's sleep, skipped await t.crash(); await t.check(); await t.close(); await crashEverywhere(() => { const payments = new FakePayments(); // fresh per run return { workflows: [checkoutWith(payments)], workflow: "checkout", id: "o-1", input: order, effects: () => payments.snapshot() }; }); ``` The virtual clock of the Python and TypeScript kits is the process's: one test engine at a time. # Versioning and deploys A run that started last month replays its journal against today's code. Changing workflow code is safe when the new code makes the same journal operations, in the same order, for the runs that already passed that point. This guide is how to change code anyway. ## What changes are safe - Anything inside a step's function: the step runs again only if it never finished, and its recorded result is used otherwise. - New code after the point every running run has passed. - Queries, logging, and anything that reads without making journal operations. Renaming a step, removing one, adding one before others, or changing the order of steps is not safe for running runs. Fusee notices: the run stops (`stuck`, with a `Determinism` error naming the step that differs), keeps its state, and waits for code that matches. Nothing is lost; the [errors page](../errors.md#determinism) lists the ways out. ## Three ways to change code **A patch** keeps both paths in one function: ```ts const fee = ctx.patched("refund-rounding") ? Math.round(x) : Math.floor(x); ``` `patched` is true for new runs and for running ones the first time their code reaches it after the deploy, false for runs that passed it before. Once no running run predates the patch, replace it with `ctx.deprecatePatch("refund-rounding")` and keep the new branch; later, delete the call. **A version** keeps two functions: ```ts const checkoutV1 = workflow("checkout", { version: 1 }, oldCode); const checkoutV2 = workflow("checkout", { version: 2 }, newCode); ``` New runs start under the newest registered version; a run keeps the version it started under. `fusee versions` lists live runs per type and version: when v1 has none, delete its code. **A reset** moves one run onto the new code: `fusee runs reset order-7 --before charge` starts it again from its journal cut just before `charge`, with the signals it received after. ## Before the deploy Replay a sample of real journals against the new code in CI: ```sh fusee export --sample 1000 ./data -o journals/ fusee replay-check journals/ --module ./dist/workflows.js ``` `replay-check` runs the new code against each exported journal without running any step, and names the first difference for each run that would get stuck. Under the dev profile every run is replayed against the code as it parks or ends, so a change that breaks running runs fails in development, at the line that changed. ## During the deploy Old and new processes can run side by side: each run is held by one process at a time, and a run waiting for code that only the new version has (`waiting_for_code`) waits until a process with that code takes its shard. Roll forward; to roll back, keep the patch branches until no run needs them. # API at a glance Generated from [`api/spec.json`](../api/spec.json) by `api/gen.py`; do not edit. Every operation has one name, one argument order and one option set, written in each language's case. `api/check.py` checks the SDKs against it. ## Definitions | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **workflow** | name, fn | version, input, output, timeout, concurrency | `WorkflowRef::new` | `workflow` | `workflow` | `Define` | `workflow` | | **signal** | name | schema | `Signal::new` | `signal` | `signal` | `Signal` | `signal` | | **query** | name | schema | `Query::new` | `query` | `query` | `Query` | `query` | | **update** | name | schema | `Update::new` | `update` | `update` | `Update` | `update` | | **object** | name, handlers | version | `Engine::object_type(name).exclusive(..)` | `object` | `object` | `Object` | `object` | | **shared** | handler | | `ObjectType::shared` | `shared` | `shared` | `Shared` | `shared` | - **workflow**: A workflow type: a plain value, apart from any engine. Pass it to Engine.open({ workflows }), engine.start, ctx.child. The options form is workflow(name, options, fn). - **signal**: A signal's name and payload type. A plain string works wherever a definition does. - **query**: A query's name, argument and result types. - **update**: An update's name, argument and result types. - **object**: A keyed object type. Handlers are exclusive unless wrapped in shared(). - **shared**: Marks an object handler as shared: it reads state and runs beside other shared calls. ## Engine and Client (one interface) | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **start** | workflow, id, input | idPolicy, timeout, version, delay, at, attributes | `start` | `start` | `start` | `Start` | `start` | | **run** | id | | `run` | `run` | `run` | `Run` | `run` | | **signalWithStart** | workflow, id, input, signal, value | idPolicy, timeout, version | `signal_with_start` | `signalWithStart` | `signal_with_start` | `SignalWithStart` | `signalWithStart` | | **updateWithStart** | workflow, id, input, update, arg | idPolicy, timeout, version, updateId | `update_with_start` | `updateWithStart` | `update_with_start` | `UpdateWithStart` | `updateWithStart` | | **object** | type, key | | `object` | `object` | `object` | `Object` | `object` | | **schedule** | name, spec, workflow, input | | `schedule` | `schedule` | `schedule` | `Schedule` | `schedule` | | **schedules** | | | `Engine::list_schedules` | `schedules` | `schedules` | `Schedules` | `schedules` | | **list** | | type, status, version, startedAfter, startedBefore, attributes, limit, pageSize | `Engine::list_where` | `list` | `list` | `List` | `list` | | **complete** | token, value | | `Engine::complete` | `complete` | `complete` | `Complete` | `complete` | | **fail** | token, failure | | `fail` | `fail` | `fail` | `Fail` | `fail` | | **debounce** | workflow, key, input | window, maxWait, dedupWindow, eventId | `debounce` | `debounce` | `debounce` | `Debounce` | `debounce` | | **batch** | workflow, key, event | maxEvents, maxWait, dedupWindow, eventId | `batch` | `batch` | `batch` | `Batch` | `batch` | | **close** | | | `close` | `close` | `close` | `Close` | `close` | - **run**: A handle; makes no call. - **schedule**: Creates or updates the schedule. spec: { cron, every, timezone, overlap, catchupWindow, jitter, paused }. - **schedules**: schedules.get(name), schedules.list(). ## Engine only | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **open** | | dir, url, durability, workflows, objects, shards, workers, module, profile, serializer, codec, keepHistory, recordStacks, onExhausted, retention, telemetry, logger | `open` | `open` | `open` | `Open` | `open` | | **limit** | name, spec | | `limit` | `limit` | `limit` | `Limit` | `limit` | | **limits** | | | `limits` | `limits` | `limits` | `Limits` | `limits` | | **serve** | | port, host, token, ui | `Config::http / http::serve` | `serve` | `serve` | `Serve` | `serve` | | **on** | event, handler | | `log::set_hook` | `on` | `on` | `On` | `on` | | **metrics** | | | `metrics` | `metrics` | `metrics` | `Metrics` | `metrics` | - **open**: No arguments: FUSEE_DIR, else FUSEE_URL (a client), else ./fusee-data. - **limit**: spec: { concurrent, perSecond }. - **serve**: Serves the HTTP API (and the UI) from its own thread. - **on**: Events: stuck, waitingForCode. ## A run | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **signal** | signal, value | | `signal` | `signal` | `signal` | `Signal` | `signal` | | **query** | query, arg | | `query` | `query` | `query` | `Query` | `query` | | **update** | update, arg | id | `update` | `update` | `update` | `Update` | `update` | | **result** | | | `result` | `result` | `result` | `Result` | `result` | | **status** | | | `status` | `status` | `status` | `Status` | `status` | | **describe** | | | `describe` | `describe` | `describe` | `Describe` | `describe` | | **cancel** | | reason | `cancel` | `cancel` | `cancel` | `Cancel` | `cancel` | | **terminate** | | reason | `terminate` | `terminate` | `terminate` | `Terminate` | `terminate` | | **reset** | beforeStep | reapplySignals, version | `reset` | `reset` | `reset` | `Reset` | `reset` | | **retry** | | | `retry` | `retry` | `retry` | `Retry` | `retry` | | **resume** | | | `resume` | `resume` | `resume` | `Resume` | `resume` | | **stream** | name | from, wait | `stream` | `stream` | `stream` | `Stream` | `stream` | - **describe**: The summary and the journal as a typed union of records. - **retry**: Resets a failed run to before its failed step. - **resume**: Runs a paused step again with a fresh set of attempts. ## An object's key | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **call** | handler, arg | id | `call` | `call` | `call` | `Call` | `call` | | **send** | handler, arg | id | `send` | `send` | `send` | `Send` | `send` | | **state** | | | `state` | `state` | `state` | `State` | `state` | | **describe** | | | `describe` | `describe` | `describe` | `Describe` | `describe` | | **delete** | | | `delete` | `delete` | `delete` | `Delete` | `delete` | ## A schedule | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **pause** | | | `pause` | `pause` | `pause` | `Pause` | `pause` | | **resume** | | | `resume` | `resume` | `resume` | `Resume` | `resume` | | **trigger** | | | `trigger` | `trigger` | `trigger` | `Trigger` | `trigger` | | **backfill** | from, to | | `backfill` | `backfill` | `backfill` | `Backfill` | `backfill` | | **describe** | | | `describe` | `describe` | `describe` | `Describe` | `describe` | | **delete** | | | `delete` | `delete` | `delete` | `Delete` | `delete` | ## Workflow context | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **step** | name, fn | retry, timeout, heartbeatTimeout, readOnly, limit, priority, fairnessKey, fairnessWeight, output, thread | `step` | `step` | `step` | `Step` | `step` | | **sleep** | name, duration | | `sleep` | `sleep` | `sleep` | `Sleep` | `sleep` | | **sleepUntil** | name, at | | `Ctx::sleep_until` | `sleepUntil` | `sleep_until` | `SleepUntil` | `sleepUntil` | | **waitFor** | signal | timeout | `wait_for` | `waitFor` | `wait_for` | `WaitFor` | `waitFor` | | **child** | workflow, id, input | parentClose, timeout | `child` | `child` | `child` | `Child` | `child` | | **startChild** | workflow, id, input | parentClose, timeout | `start_child` | `startChild` | `start_child` | `StartChild` | `startChild` | | **signalWorkflow** | id, signal, value | | `signal_workflow` | `signalWorkflow` | `signal_workflow` | `SignalWorkflow` | `signalWorkflow` | | **cancelWorkflow** | id | reason | `cancel_workflow` | `cancelWorkflow` | `cancel_workflow` | `CancelWorkflow` | `cancelWorkflow` | | **object** | type, key | | `Ctx::object` | `object` | `object` | `Object` | `object` | | **onQuery** | query, handler | | `on_query` | `onQuery` | `on_query` | `OnQuery` | `onQuery` | | **onUpdate** | update, handler | validate | `on_update` | `onUpdate` | `on_update` | `OnUpdate` | `onUpdate` | | **awaitCompletion** | name, send | retry, timeout | `await_completion` | `awaitCompletion` | `await_completion` | `AwaitCompletion` | `awaitCompletion` | | **compensate** | name, fn | retry, timeout | `compensate` | `compensate` | `compensate` | `Compensate` | `compensate` | | **continueAsNew** | input | version | `continue_as_new` | `continueAsNew` | `continue_as_new` | `ContinueAsNew` | `continueAsNew` | | **patched** | id | | `patched` | `patched` | `patched` | `Patched` | `patched` | | **deprecatePatch** | id | | `deprecate_patch` | `deprecatePatch` | `deprecate_patch` | `DeprecatePatch` | `deprecatePatch` | | **setAttributes** | attributes | | `set_attributes` | `setAttributes` | `set_attributes` | `SetAttributes` | `setAttributes` | | **stream** | name | | `stream` | `stream` | `stream` | `Stream` | `stream` | | **now** | | | `Ctx::now_ms` | `now` | `now` | `Now` | `now` | | **random** | | | `random` | `random` | `random` | `Random` | `random` | | **uuid** | | | `uuid` | `uuid` | `uuid` | `UUID` | `uuid` | | **all** | operations | | `futures::join! (Rust's own)` | `all` | `all` | `All` | `all` | | **race** | operations | | `futures::select! (Rust's own)` | `race` | `race` | `Race` | `race` | - **step**: fn may take the step context s. - **awaitCompletion**: send(cb) receives { token, url }. - **compensate**: Registers a step that runs, newest first, if the workflow fails or is cancelled. ## Step context | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **heartbeat** | progress | | `ctx.heartbeat` | `heartbeat` | `heartbeat` | `Heartbeat` | `heartbeat` | | **progress** | | | `ctx.progress` | `progress` | `progress` | `Progress` | `progress` | | **stream** | name | | `ctx.stream` | `stream` | `stream` | `Stream` | `stream` | ## Errors The same codes in every SDK; each has a page in [errors.md](errors.md). - [`NotFound`](errors.md#notfound) - [`AlreadyExists`](errors.md#alreadyexists) - [`UnknownWorkflow`](errors.md#unknownworkflow) - [`StepFailed`](errors.md#stepfailed) - [`WorkflowFailed`](errors.md#workflowfailed) - [`ChildFailed`](errors.md#childfailed) - [`Cancelled`](errors.md#cancelled) - [`TimedOut`](errors.md#timedout) - [`Terminated`](errors.md#terminated) - [`Determinism`](errors.md#determinism) - [`Rejected`](errors.md#rejected) - [`Codec`](errors.md#codec) - [`Connection`](errors.md#connection) - [`Io`](errors.md#io) - [`Busy`](errors.md#busy) - [`Closed`](errors.md#closed) # API at a glance Generated from [`api/spec.json`](../api/spec.json) by `api/gen.py`; do not edit. Every operation has one name, one argument order and one option set, written in each language's case. `api/check.py` checks the SDKs against it. ## Definitions | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **workflow** | name, fn | version, input, output, timeout, concurrency | `WorkflowRef::new` | `workflow` | `workflow` | `Define` | `workflow` | | **signal** | name | schema | `Signal::new` | `signal` | `signal` | `Signal` | `signal` | | **query** | name | schema | `Query::new` | `query` | `query` | `Query` | `query` | | **update** | name | schema | `Update::new` | `update` | `update` | `Update` | `update` | | **object** | name, handlers | version | `Engine::object_type(name).exclusive(..)` | `object` | `object` | `Object` | `object` | | **shared** | handler | | `ObjectType::shared` | `shared` | `shared` | `Shared` | `shared` | - **workflow**: A workflow type: a plain value, apart from any engine. Pass it to Engine.open({ workflows }), engine.start, ctx.child. The options form is workflow(name, options, fn). - **signal**: A signal's name and payload type. A plain string works wherever a definition does. - **query**: A query's name, argument and result types. - **update**: An update's name, argument and result types. - **object**: A keyed object type. Handlers are exclusive unless wrapped in shared(). - **shared**: Marks an object handler as shared: it reads state and runs beside other shared calls. ## Engine and Client (one interface) | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **start** | workflow, id, input | idPolicy, timeout, version, delay, at, attributes | `start` | `start` | `start` | `Start` | `start` | | **run** | id | | `run` | `run` | `run` | `Run` | `run` | | **signalWithStart** | workflow, id, input, signal, value | idPolicy, timeout, version | `signal_with_start` | `signalWithStart` | `signal_with_start` | `SignalWithStart` | `signalWithStart` | | **updateWithStart** | workflow, id, input, update, arg | idPolicy, timeout, version, updateId | `update_with_start` | `updateWithStart` | `update_with_start` | `UpdateWithStart` | `updateWithStart` | | **object** | type, key | | `object` | `object` | `object` | `Object` | `object` | | **schedule** | name, spec, workflow, input | | `schedule` | `schedule` | `schedule` | `Schedule` | `schedule` | | **schedules** | | | `Engine::list_schedules` | `schedules` | `schedules` | `Schedules` | `schedules` | | **list** | | type, status, version, startedAfter, startedBefore, attributes, limit, pageSize | `Engine::list_where` | `list` | `list` | `List` | `list` | | **complete** | token, value | | `Engine::complete` | `complete` | `complete` | `Complete` | `complete` | | **fail** | token, failure | | `fail` | `fail` | `fail` | `Fail` | `fail` | | **debounce** | workflow, key, input | window, maxWait, dedupWindow, eventId | `debounce` | `debounce` | `debounce` | `Debounce` | `debounce` | | **batch** | workflow, key, event | maxEvents, maxWait, dedupWindow, eventId | `batch` | `batch` | `batch` | `Batch` | `batch` | | **close** | | | `close` | `close` | `close` | `Close` | `close` | - **run**: A handle; makes no call. - **schedule**: Creates or updates the schedule. spec: { cron, every, timezone, overlap, catchupWindow, jitter, paused }. - **schedules**: schedules.get(name), schedules.list(). ## Engine only | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **open** | | dir, url, durability, workflows, objects, shards, workers, module, profile, serializer, codec, keepHistory, recordStacks, onExhausted, retention, telemetry, logger | `open` | `open` | `open` | `Open` | `open` | | **limit** | name, spec | | `limit` | `limit` | `limit` | `Limit` | `limit` | | **limits** | | | `limits` | `limits` | `limits` | `Limits` | `limits` | | **serve** | | port, host, token, ui | `Config::http / http::serve` | `serve` | `serve` | `Serve` | `serve` | | **on** | event, handler | | `log::set_hook` | `on` | `on` | `On` | `on` | | **metrics** | | | `metrics` | `metrics` | `metrics` | `Metrics` | `metrics` | - **open**: No arguments: FUSEE_DIR, else FUSEE_URL (a client), else ./fusee-data. - **limit**: spec: { concurrent, perSecond }. - **serve**: Serves the HTTP API (and the UI) from its own thread. - **on**: Events: stuck, waitingForCode. ## A run | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **signal** | signal, value | | `signal` | `signal` | `signal` | `Signal` | `signal` | | **query** | query, arg | | `query` | `query` | `query` | `Query` | `query` | | **update** | update, arg | id | `update` | `update` | `update` | `Update` | `update` | | **result** | | | `result` | `result` | `result` | `Result` | `result` | | **status** | | | `status` | `status` | `status` | `Status` | `status` | | **describe** | | | `describe` | `describe` | `describe` | `Describe` | `describe` | | **cancel** | | reason | `cancel` | `cancel` | `cancel` | `Cancel` | `cancel` | | **terminate** | | reason | `terminate` | `terminate` | `terminate` | `Terminate` | `terminate` | | **reset** | beforeStep | reapplySignals, version | `reset` | `reset` | `reset` | `Reset` | `reset` | | **retry** | | | `retry` | `retry` | `retry` | `Retry` | `retry` | | **resume** | | | `resume` | `resume` | `resume` | `Resume` | `resume` | | **stream** | name | from, wait | `stream` | `stream` | `stream` | `Stream` | `stream` | - **describe**: The summary and the journal as a typed union of records. - **retry**: Resets a failed run to before its failed step. - **resume**: Runs a paused step again with a fresh set of attempts. ## An object's key | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **call** | handler, arg | id | `call` | `call` | `call` | `Call` | `call` | | **send** | handler, arg | id | `send` | `send` | `send` | `Send` | `send` | | **state** | | | `state` | `state` | `state` | `State` | `state` | | **describe** | | | `describe` | `describe` | `describe` | `Describe` | `describe` | | **delete** | | | `delete` | `delete` | `delete` | `Delete` | `delete` | ## A schedule | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **pause** | | | `pause` | `pause` | `pause` | `Pause` | `pause` | | **resume** | | | `resume` | `resume` | `resume` | `Resume` | `resume` | | **trigger** | | | `trigger` | `trigger` | `trigger` | `Trigger` | `trigger` | | **backfill** | from, to | | `backfill` | `backfill` | `backfill` | `Backfill` | `backfill` | | **describe** | | | `describe` | `describe` | `describe` | `Describe` | `describe` | | **delete** | | | `delete` | `delete` | `delete` | `Delete` | `delete` | ## Workflow context | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **step** | name, fn | retry, timeout, heartbeatTimeout, readOnly, limit, priority, fairnessKey, fairnessWeight, output, thread | `step` | `step` | `step` | `Step` | `step` | | **sleep** | name, duration | | `sleep` | `sleep` | `sleep` | `Sleep` | `sleep` | | **sleepUntil** | name, at | | `Ctx::sleep_until` | `sleepUntil` | `sleep_until` | `SleepUntil` | `sleepUntil` | | **waitFor** | signal | timeout | `wait_for` | `waitFor` | `wait_for` | `WaitFor` | `waitFor` | | **child** | workflow, id, input | parentClose, timeout | `child` | `child` | `child` | `Child` | `child` | | **startChild** | workflow, id, input | parentClose, timeout | `start_child` | `startChild` | `start_child` | `StartChild` | `startChild` | | **signalWorkflow** | id, signal, value | | `signal_workflow` | `signalWorkflow` | `signal_workflow` | `SignalWorkflow` | `signalWorkflow` | | **cancelWorkflow** | id | reason | `cancel_workflow` | `cancelWorkflow` | `cancel_workflow` | `CancelWorkflow` | `cancelWorkflow` | | **object** | type, key | | `Ctx::object` | `object` | `object` | `Object` | `object` | | **onQuery** | query, handler | | `on_query` | `onQuery` | `on_query` | `OnQuery` | `onQuery` | | **onUpdate** | update, handler | validate | `on_update` | `onUpdate` | `on_update` | `OnUpdate` | `onUpdate` | | **awaitCompletion** | name, send | retry, timeout | `await_completion` | `awaitCompletion` | `await_completion` | `AwaitCompletion` | `awaitCompletion` | | **compensate** | name, fn | retry, timeout | `compensate` | `compensate` | `compensate` | `Compensate` | `compensate` | | **continueAsNew** | input | version | `continue_as_new` | `continueAsNew` | `continue_as_new` | `ContinueAsNew` | `continueAsNew` | | **patched** | id | | `patched` | `patched` | `patched` | `Patched` | `patched` | | **deprecatePatch** | id | | `deprecate_patch` | `deprecatePatch` | `deprecate_patch` | `DeprecatePatch` | `deprecatePatch` | | **setAttributes** | attributes | | `set_attributes` | `setAttributes` | `set_attributes` | `SetAttributes` | `setAttributes` | | **stream** | name | | `stream` | `stream` | `stream` | `Stream` | `stream` | | **now** | | | `Ctx::now_ms` | `now` | `now` | `Now` | `now` | | **random** | | | `random` | `random` | `random` | `Random` | `random` | | **uuid** | | | `uuid` | `uuid` | `uuid` | `UUID` | `uuid` | | **all** | operations | | `futures::join! (Rust's own)` | `all` | `all` | `All` | `all` | | **race** | operations | | `futures::select! (Rust's own)` | `race` | `race` | `Race` | `race` | - **step**: fn may take the step context s. - **awaitCompletion**: send(cb) receives { token, url }. - **compensate**: Registers a step that runs, newest first, if the workflow fails or is cancelled. ## Step context | Operation | Arguments | Options | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | --- | --- | | **heartbeat** | progress | | `ctx.heartbeat` | `heartbeat` | `heartbeat` | `Heartbeat` | `heartbeat` | | **progress** | | | `ctx.progress` | `progress` | `progress` | `Progress` | `progress` | | **stream** | name | | `ctx.stream` | `stream` | `stream` | `Stream` | `stream` | ## Errors The same codes in every SDK; each has a page in [errors.md](errors.md). - [`NotFound`](errors.md#notfound) - [`AlreadyExists`](errors.md#alreadyexists) - [`UnknownWorkflow`](errors.md#unknownworkflow) - [`StepFailed`](errors.md#stepfailed) - [`WorkflowFailed`](errors.md#workflowfailed) - [`ChildFailed`](errors.md#childfailed) - [`Cancelled`](errors.md#cancelled) - [`TimedOut`](errors.md#timedout) - [`Terminated`](errors.md#terminated) - [`Determinism`](errors.md#determinism) - [`Rejected`](errors.md#rejected) - [`Codec`](errors.md#codec) - [`Connection`](errors.md#connection) - [`Io`](errors.md#io) - [`Busy`](errors.md#busy) - [`Closed`](errors.md#closed) # fusee (Node.js and Bun) Durable execution at the speed of the disk, embedded in your process. ```ts import { Engine } from "@bitsinfra/fusee"; const engine = await Engine.open({ dir: "./data" }); const wf = engine.workflow("greet", async (ctx, name: string) => { const greeting = await ctx.step("compose", () => `hello ${name}`); await ctx.sleep("pause", "1s"); return greeting; }); console.log(await (await wf.start("greet-1", "world")).result()); await engine.close(); ``` Workflow code must be deterministic between `ctx` calls: use `ctx.now()`, `ctx.random()` and `ctx.uuid()`, and await only `ctx` operations (or `ctx.all` / `ctx.race` of them). Step functions may do anything. API: - `ctx`: `step(name, fn, { retry, timeout, heartbeatTimeout, limit, priority, fairnessKey, fairnessWeight, readOnly })`, `heartbeat(progress)`, `progress()`, `awaitCompletion(name, send)`, `sleep(name, duration)`, `waitFor(name, { timeout })`, `child(type, id, input, { parentClose })`, `spawnChild`, `signal(target, name, value)`, `cancelWorkflow(target)`, `onQuery(name, fn)`, `onUpdate(name, { validate, handle })`, `continueAsNew(input)`, `journalSize()`, `patched(id)`, `deprecatePatch(id)`, `setAttributes(obj)`, `now/random/uuid`, `key()`, `all`, `race`. - `run`: `signal`, `result`, `started()`, `update(name, arg, { id })`, `query(name, arg)`, `cancel`, `terminate`, `reset({ beforeStep })`, `resume()` (a run paused by a step out of retries, see below), `inspect`. - Step options: `retry: { attempts, initial, max, maxElapsed, nonRetryable, retryIf, onExhausted }`: `onExhausted: "pause"` pauses the run when the retries run out, until `run.resume()`; `Engine.open({ onExhausted: "pause" })` makes it every step's default ([Pausing](../../docs/features.md#pausing-when-retries-run-out)). - `workflow`: `start(id, input, { idPolicy, timeout })`, `updateWithStart(id, input, name, arg)`, `debounce(key, input, { window, maxWait, dedupWindow, id })`, `batch(key, event, { maxEvents, maxWait, dedupWindow, id })`. - `engine`: `complete(token, value)`, `fail(token, error)`, `schedule(name, { cron | every, timezone, overlap }, workflow, input)`, `list({ where })`, `limit(name, { concurrent, perSecond })`, `limitInUse(name)`, `limitWaiting(name)`, `exportJournal(id)`; `Engine.open({ codec: aesGcm(keys) })`; `new Replayer().workflow(name, fn).replayFile(path)`. - Queues: steps waiting for a limit's permit go by `priority` (higher first), then take turns by `fairnessKey` (`fairnessWeight` permits per turn); the order is not journaled. `debounce` collapses starts with one key inside a window into one run with the latest input; `batch` starts one run with every event once `maxEvents` arrived or `maxWait` passed. Both resolve once the event is durable with the run that will take it (`@`); `run.started()` waits until it exists. See [Debounce](../../docs/features.md#debounce) and [Batching](../../docs/features.md#batching). ```ts engine.limit("search", { concurrent: 4 }); const reindex = engine.workflow("reindex", async (ctx, doc: Doc) => ctx.step("index", () => search.index(doc), { limit: "search", priority: doc.urgent ? 10 : 0, fairnessKey: doc.tenant })); const run = await reindex.debounce(doc.id, doc, { window: "5s", maxWait: "1m", id: eventId }); await run.started(); ``` - Streams: `ctx.stream(name)` returns a writer; `append(value)` (a `Uint8Array` is bytes) never waits for the disk, `close()` completes the stream, and inside a step `recorded()` returns what the step's earlier attempts recorded, so an attempt that runs again continues after it (the step's result completes its stream). `run.stream(name, { from })` and `client.run(id).stream(name, { from })` are async iterables of the durable chunks; `reader.offset` is where to reconnect, `reader.status` how it ended. See [Streams](../../docs/features.md#streams). ```ts const answer = await ctx.step("generate", async () => { const out = ctx.stream("tokens"); let text = out.recorded().join(""); for await (const t of model.stream(prompt, { prefill: text })) { out.append(t); text += t; } return text; }); for await (const t of engine.run(id).stream("tokens", { from: 0 })) process.stdout.write(t); ``` - Objects (keyed state): `engine.object(type, { handler, other: shared(fn) })` registers handlers by name, exclusive (one call at a time per key, in arrival order) unless marked `shared` (they run at once and only read). A handler is workflow code with `ctx.state` (`get`, `set`, `clear`, `clearAll`, `keys`; JSON values, committed when an exclusive handler returns) and `ctx.objectKey`. `engine.objectRef(type, key)` has `call(handler, arg, { id })` (calls with one ID run once), `send(...)` (resolves once accepted), `state()`, `inspect()` and `delete()`, and edits the state from outside: `setState(field, value)`, `patchState(ops)` (JSON Patch), `clearState(field?)`, `editState(edit, { ifVersion })` ([Editing state](../../docs/features.md#editing-state-from-outside)); `engine.objects(type)` lists keys. From workflow code, `ctx.object(type, key).call(...)` and `.send(...)` are journal operations that run exactly once. With `workers`, register objects in the workers' module; calls go to the worker that owns the key. Through a server: `client.objectRef(type, key)`. See [Objects](../../docs/features.md#objects-keyed-state). ```ts engine.object("cart", { add: async (ctx, item) => { const items = ctx.state.get("items") ?? []; items.push(await ctx.step("price", () => pricing.quote(item))); ctx.state.set("items", items); return items.length; }, get: shared((ctx) => ctx.state.get("items") ?? []), }); const n = await engine.objectRef("cart", userId).call("add", item, { id: requestId }); ``` - `Engine.open` options beyond storage: `telemetry: openTelemetry(api)` (a `fusee.workflow` span per run, in the trace of the request that started it, with a `fusee.step` span per attempt below it, and step and workflow metrics; see [traces](../../docs/operations.md#traces)), `inlineFlush: false` (every flush on the log thread; slower for a lone sequential workflow on a fast disk, so off by default), `driver` (`"native"`, the default on Node: the engine's turns run back to back in one event-loop callback, each drained of microtasks as a macrotask would be; `"macrotask"`, always on Bun, or `FUSEE_DRIVER=macrotask`: one event-loop callback per turn), and `workers: n, module` (workflow code on n worker threads, each owning a set of shards and loading `module` to register its workflows; this thread routes calls by workflow ID; see [Across threads](../../docs/features.md#across-threads-typescript)). - The pipelined flush (the default in local mode with the shared log, on hosts with more than two cores): each record goes to the log's threads as soon as a delivery appends it, and the next step's gate spins for it while a core is spare (shards and host threads that worked in the last 2 ms fit in the host's cores), instead of flushing inline at the gate. Faster for a lone chain on Linux (22 against 33 µs a step on an i8g.2xlarge). `FUSEE_SDK_FLUSH=inline` flushes inline at the gate instead, which is faster for a fan-out. - Payloads are JSON, except `Uint8Array` inputs, step results, signals and results, which are recorded as raw bytes and come back as bytes after a restart. See [docs/features.md](../../docs/features.md). On an AWS i8g.2xlarge (instance-store NVMe, a 12 µs flush), a step takes 19.1 µs from one step to the next with eight workflows live ([benchmark](../../docs/benchmark.md#sdks)). Build from source: `cargo build --release -p fusee-node && node scripts/copy-native.mjs && npx tsc`. ## Agents `@bitsinfra/fusee/ai` makes a model call a step whose tokens stream to readers (`modelCall`). `@bitsinfra/fusee/ai/vercel` makes an agent written with the Vercel AI SDK durable without changing its loop: `durableModel(ctx, model)` makes each model call a step (`streamText` too: its text is the stream `model#`), and `durableTools(ctx, tools, { approve })` makes each tool call a step, with tools that wait for a person's approval first. ```ts import { generateText } from "ai"; import { durableModel, durableTools } from "@bitsinfra/fusee/ai/vercel"; const { text } = await generateText({ model: durableModel(ctx, openai("gpt-5")), tools: durableTools(ctx, { search }), maxRetries: 0, prompt, }); ``` See [Agents](../../docs/guides/agents.mdx) and [Agent patterns](../../docs/guides/agent-patterns.md). ## Server and replicated modes A worker can acquire a server lease over TCP or a Unix socket. A separate application uses `Client` to start, signal, query, update or complete work. ```ts const engine = await Engine.open({ durability: "server", address: "127.0.0.1:7600" }); const client = await Client.connect("127.0.0.1:7600"); const run = await client.start("checkout", "order-42", { amount: 42 }); await run.signal("approved", true); const batched = await client.batch("ingest", "tenant-7", event, { maxEvents: 100, maxWait: "2s" }); await batched.started(); client.close(); ``` `client.debounce(type, key, input, opts)` and `client.batch(type, key, event, opts)` take the options of `workflow.debounce` and `workflow.batch`. Register workflows on the worker as in local mode. For replicated operation start a worker process on each of three or five machines, each with its own data directory and member index. Pass `durability: "replicated"`, `replicaId`, `peers` and `dir` to `Engine.open`. Each process serves replication traffic while the call waits for its member to become primary. Close the engine before opening it again after loss of the primary role. Replicas require a trusted network. The protocol has no peer authentication. ## Invoked workers (scale to zero) A server started with `--invoke URL` POSTs to that URL when a shard that no worker holds has work due (a start, a signal, a timer, a child's result). `invocationHandler` answers: each request opens a fresh engine for the invocation's shards, runs them until they are idle or the budget ends, closes it and returns. Nothing stays up while workflows sleep. ```ts import { invocationHandler } from "@bitsinfra/fusee"; const handler = invocationHandler({ address: "fusee.internal:7600?ca=/etc/fusee/ca.pem&token-file=/etc/fusee/token", secret: process.env.FUSEE_INVOKE_SECRET, // the server's --invoke-secret-file register: (engine) => { engine.workflow("order", order); }, }); http.createServer(handler.node).listen(8080); // node:http app.post("/fusee", handler.node); // Express, before any body parser (or after express.raw()) export const lambda = handler.lambda; // AWS Lambda (function URL, API Gateway, ALB) export default { fetch: handler.fetch }; // Bun.serve, Deno, Vercel, Cloudflare ``` `handler.invoke({ headers, body })` serves any other framework and resolves to `{ statusCode, headers, body }`. A request without a valid signature gets 401 before anything connects. Other `OpenOptions` (`codec`, `telemetry`, `retention`...) pass through to each engine. Semantics, the time budget and the server's options are in [server mode](../../docs/server-mode.md#invoked-workers). # fusee (Python) Durable execution at the speed of the disk, embedded in your process (asyncio). Python 3.9 or later; typed (`py.typed`). ```python import asyncio import fusee @fusee.workflow("greet") async def greet(ctx: fusee.Ctx, name: str) -> str: greeting = await ctx.step("compose", lambda: f"hello {name}") await ctx.sleep("pause", "1s") return greeting async def main(): async with await fusee.Engine.open("./data", workflows=[greet]) as engine: run = await engine.start(greet, "greet-1", "world") print(await run.result()) asyncio.run(main()) ``` Kill the process during the sleep and run it again: the run resumes where it was, and `compose` does not run twice. ## Definitions Workflows, signals, queries, updates and object types are plain values, made apart from any engine, so a module can define them and a client can import them for their names and types alone: ```python review = fusee.signal("review", bool) status = fusee.query("status") @fusee.workflow("checkout") async def checkout(ctx: fusee.Ctx, order: Order) -> Receipt: # Order: dataclass, pydantic, msgspec, TypedDict quote = await ctx.step("quote", lambda: pricing.quote(order)) charge = await ctx.step("charge", lambda s: payments.charge(quote, key=s.key), retry=fusee.Retry(attempts=5)) ctx.on_query(status, lambda _: {"phase": "review"}) approved = await ctx.wait_for(review, timeout="1d") # bool | None if not approved: await ctx.step("refund", lambda: payments.refund(charge)) return Receipt(order, charge) engine = await fusee.Engine.open("./data", workflows=[checkout]) run = await engine.start(checkout, "order-123", order) # Run[Receipt] await run.signal(review, True) receipt = await run.result() # a Receipt again ``` - `@fusee.workflow("name")` (or bare `@fusee.workflow`, named after the function), with `version=`, `timeout=`, `input=`, `output=`. The input's and result's annotations are read once and decide what the code receives and what `run.result()` returns: dataclasses, pydantic models, msgspec structs, `datetime`, `UUID`, enums, and lists and dicts of them. A start whose input does not match its type is refused before anything is recorded. - `fusee.signal(name, type)`, `fusee.query(name, arg=, result=)`, `fusee.update(name, arg=, result=)`. A plain string works wherever a definition does, for calls across languages. - `fusee.object("cart", {"add": add, "items": fusee.shared(items)})`, or `cart = fusee.object("cart")` with `@cart.exclusive` and `@cart.shared`. - `Engine.open(workflows=[...], objects=[...])` registers them; `engine.register(...)` adds more later. Registering a type twice is an error that names both places. ## The step context A step function may take one argument, the step context `s`: | | | | --- | --- | | `s.key` | an idempotency key, stable across retries and recovery | | `s.attempt` | 1, 2, ... | | `s.heartbeat(p)`, `s.progress()` | progress, also from before a crash | | `s.stream(name)` | the step's own stream | | `s.log` | a logger tagged with the workflow ID, run and step | It is built only when the function declares the parameter (`lambda s: ...`); `lambda: ...` costs nothing. Cancellation of the workflow, its deadline and the step's own timeouts cancel the step's coroutine. A synchronous step function runs on the event loop and holds every other run while it runs: make it `async`, or pass `thread=True` to run it on a thread (or open the engine with `sync_steps="thread"`). Under the `dev` profile a synchronous step that holds the loop over 50 ms is reported once with both fixes. ## Errors and retries A failed step's error reaches the caller whole: the step's name, and its failure with the exception's type, `code`, `details`, stack and cause. ```python class CardDeclined(fusee.Fatal): # Fatal: never retried code = "card_declined" raise CardDeclined("card declined", details={"last4": "4242"}) try: await run.result() except fusee.StepError as e: e.step, e.cause.type, e.cause.code, e.cause.details # "charge", "CardDeclined", ... ``` - A failure is final when it is a `fusee.Fatal`, when the exception has `retryable = False`, when its type (or a cause's) is in `Retry(non_retryable=(...))`, or when `Retry(retry_if=lambda failure, attempt: ...)` says so. `Retry(attempts=3, initial="100ms", factor=2, max="10s", jitter=True, max_elapsed=None)` is the default policy. - A step result JSON cannot encode is a `CodecError` naming the path of the value (`a set at $.items[1].tags`), and is not retried. - Every error has `code` (the same in every SDK) and `docs`, a link to its page in [docs/errors.md](../../docs/errors.md): `NotFoundError`, `AlreadyExistsError`, `UnknownWorkflowError`, `StepError`, `WorkflowError`, `ChildError`, `CancelledError`, `TimeoutError`, `TerminatedError`, `DeterminismError`, `RejectedError`, `CodecError`, `ConnectionError`, `IoError`, `BusyError`, `ClosedError`. - `Engine.open(record_stacks=False)` keeps stack traces out of the journal. Mistakes fail where they are made: starting a type no code is registered for raises `UnknownWorkflowError` with the registered types and the nearest name; a run whose code no longer matches its journal logs one line with the difference and the three fixes (and `engine.on("stuck", fn)` is called); a run that fails keeps its whole journal for `run.describe()` and `run.retry()`. ## The engine and the client `Engine` (embedded) and `Client` (through a Fusee server) have the same interface, so code that starts, signals and reads runs does not know which it holds: | | | | --- | --- | | `start(wf, id, input, id_policy=, timeout=, version=)` | a `Run`; `id_policy={"running": "use-existing"}` attaches to a running ID | | `run(id, wf=None)` | a handle; `wf` types its result | | `signal_with_start(wf, id, input, signal, value)` | a `Run`; starts it if it is not running, and the signal is recorded before its code first runs | | `update_with_start(wf, id, input, update, arg)` | `(run, result)` | | `object(type, key)` | `call(handler, arg, id=)`, `send`, `state()`, `describe()`, `delete()`; `set_state(field, value)`, `patch_state(ops)`, `clear_state(field=None)`, `edit_state(edit, if_version=)` | | `schedule(name, spec, wf, input)` | `spec`: a cron string, `{"every": "1h"}`, or a `ScheduleSpec`; `schedules.get(name)`, `schedules.list()` | | `list(type=, status=, started_after=, started_before=, attributes=)` | an async iterator of `RunSummary` | | `complete(token, value)`, `fail(token, failure)` | async completions | | `debounce(wf, key, input, window=, max_wait=, event_id=)`, `batch(wf, key, event, max_events=, max_wait=, event_id=)` | queued starts | | `close()` | also `async with` | A `Run` has `signal(sig, value)`, `query(q, arg)`, `update(u, arg, id=)`, `result()`, `status()` (a `RunSummary`), `describe()` (the summary and the journal), `cancel(reason)`, `terminate(reason)`, `reset(before_step)`, `retry()` (from the failed step), `resume()` (a run paused by a step whose retries ran out: `Retry(on_exhausted="pause")`, or `Engine.open(..., on_exhausted="pause")` for every step), `stream(name, from_=)` and `started()`. The engine alone has `limit(name, concurrent=, per_second=)`, `limits()`, `limit_usage(name)`, `on(event, fn)`, `metrics()`, `shard_metrics()`, `serve(addr, token=)` (the HTTP API and the UI) and `export_journal(id)`. For code without an event loop (Django, Flask, scripts), `fusee.sync.Client` has the same methods, blocking: `Client.connect(address)`, then `client.start(...)`, `run.result(timeout=30)`, `run.signal(...)`. `Engine.open()` with no directory reads `FUSEE_DIR`, else `FUSEE_URL` (a worker of a server), else `./fusee-data`; so one program runs embedded on a laptop and against a server in production. ## The `dev` profile `Engine.open(profile="dev")` or `FUSEE_PROFILE=dev` (never inferred) says so on stderr, keeps every run's history, flushes with `fsync` instead of `F_FULLFSYNC` on macOS (a process crash loses nothing; power loss may), reports synchronous steps that hold the event loop, replays each run against the code when it parks or ends and logs any difference (strict replay), makes `result()` of a stuck run raise `DeterminismError` at once instead of waiting, shows `ctx.log` records on stderr when logging is not configured, and serves the UI (`FUSEE_HTTP`, else port 7601, else a free port; `http="off"` serves nothing). ## Testing ```python from fusee.testing import TestEngine t = await TestEngine.open(workflows=[checkout]) t.mock_step(checkout, "charge", lambda attempt: {"id": "ch_1"}) run = await t.engine.start(checkout, "order-1", order) assert await t.steps("order-1") == ["quote", "charge"] await t.check() # strict replay of every run ``` See [docs/testing-workflows.md](../../docs/testing-workflows.md). ## Workflow code Workflow coroutines run under Fusee's deterministic scheduler and may only await `ctx` operations; awaiting anything else is an error that says to put it in a step. `ctx` has `step`, `sleep(name, duration)`, `sleep_until(name, at)`, `wait_for(signal, timeout=)`, `child(wf, id, input, parent_close=, timeout=)`, `start_child`, `signal_workflow(id, sig, value)`, `cancel_workflow(id, reason=)`, `object(type, key)`, `on_query`, `on_update(update, handler, validate=)`, `await_completion(name, send)` (`send(cb)` gets `cb.token` and, when the engine serves its API, `cb.url`), `compensate(name, fn)` (a step that runs, newest first, if the workflow fails or is cancelled), `continue_as_new`, `patched`, `deprecate_patch`, `set_attributes`, `stream`, `now`, `random`, `uuid`, `all`, `race`, `log` and `journal_size`. Durations are milliseconds, `"500ms"`, `"10s"`, `"5m"`, `"2h"`, `"1d"`, or a `timedelta`. Payloads are JSON (no pickle), except `bytes`, recorded raw. Steps waiting for a limit's permit (`limit=`) go by `priority` (higher first), then take turns by `fairness_key` (`fairness_weight` permits per turn). See [docs/features.md](../../docs/features.md) for streams, objects, debounce, batching, schedules, patching, encryption (`Engine.open(codec=fusee.aes_gcm(keys))`), telemetry (`telemetry=fusee.open_telemetry()`) and replay testing (`fusee.Replayer(workflows=[...]).replay_file(path)`). ## Agents `fusee.ai.model_call` makes a model call a step whose tokens stream to readers. Agents written with a framework become durable without changing their loop: each model call and each tool call is a step, so a crash resumes at the call that was running. ```python from fusee.ai.pydantic import FuseeDurability, run # pip install 'fusee[pydantic-ai]' agent = Agent("openai:gpt-5", name="support", tools=[refund], capabilities=[FuseeDurability()]) @fusee.workflow("support") async def support(ctx, question): return (await run(ctx, lambda: agent.run(question))).output ``` `fusee.ai.openai.run(ctx, agent, question)` does the same for the OpenAI Agents SDK (`pip install 'fusee[openai-agents]'`). See [Agents](../../docs/guides/agents.mdx) and [Agent patterns](../../docs/guides/agent-patterns.md). ## Server and replicated modes ```python engine = await fusee.Engine.open(durability="server", address="127.0.0.1:7600", workflows=[checkout]) client = await fusee.Client.connect("127.0.0.1:7600", token_file="/etc/fusee/token", tls={"ca": "/etc/fusee/ca.pem"}) run = await client.start(checkout, "order-42", order) ``` Several Python processes can run one server's workflows, each on its own core: start each worker with `max-shards=N` (and `shm=1` for shared memory on the same host), as in `address="unix:/run/fusee.sock?shm=1&max-shards=2"`. Each takes the free shards it may hold and runs only their workflows; all of them append to the server's one log. A worker's own `engine.start` or `engine.run(id)` works for IDs of its own shards only; use a `Client` for the others. For replication, start a worker on each of three or five machines with `durability="replicated"`, `replica_id`, `peers` and `dir`. ## Invoked workers (scale to zero) A server started with `--invoke URL` POSTs to that URL when a shard that no worker holds has work due. `invocation_handler` answers: each request opens a fresh engine for the invocation's shards, runs them until they are idle or the budget ends, and closes it. ```python app = fusee.invocation_handler( "fusee.internal:7600?ca=/etc/fusee/ca.pem&token-file=/etc/fusee/token", workflows=[order], secret=os.environ["FUSEE_INVOKE_SECRET"], # the server's --invoke-secret-file ) # ASGI: uvicorn module:app; AWS Lambda: app.lambda_handler ``` ## Building `maturin develop` in this directory, or `cargo build --release -p fusee-python` and copy the library to `python/fusee/_native.abi3.so` (on macOS remove the old file first: overwriting a loaded library in place invalidates its signature). # Fusee for Go Durable workflows in Go, on the same engine as the Rust, TypeScript and Python SDKs. Workflow code is plain Go: each `fusee.Step`, sleep, wait or child is a journal operation recorded in the engine's log and replayed after a crash. ```go e, err := fusee.Open(fusee.Options{Dir: "./data"}) defer e.Close() checkout := fusee.Register(e, "checkout", func(c *fusee.Ctx, o Order) (string, error) { quote, err := fusee.Step(c, "quote", func(ctx context.Context) (Quote, error) { return pricing.Quote(ctx, o) }) if err != nil { return "", err } charge, err := fusee.Step(c, "charge", func(ctx context.Context) (Charge, error) { return payments.Charge(ctx, quote, fusee.Key(ctx)) // idempotency key }, fusee.StepOptions{Retry: &fusee.RetryPolicy{Attempts: 5}}) if err != nil { return "", err } if err := c.Sleep("cool-off", 10*time.Minute); err != nil { return "", err } approved, ok, err := fusee.WaitFor[bool](c, "review", 24*time.Hour) if err != nil { return "", err } if ok && approved { return charge.ID, nil } return fusee.Step(c, "refund", func(ctx context.Context) (string, error) { return payments.Refund(ctx, charge) }) }) run, err := checkout.Start(ctx, "order-123", order) err = run.Signal(ctx, "review", true) id, err := run.Result(ctx) ``` ## Building The package uses cgo and links `libfusee_c`, the engine's C interface (`bindings/c`, header `bindings/c/include/fusee.h`). Build it first, from the repository root: ```bash cargo build --release -p fusee-c cd bindings/go && go test ./... ``` By default the package links `target/release/libfusee_c.a` statically, so a Go binary carries the engine and needs no shared library. Build tags choose otherwise: `fusee_dynamic` links `target/release/libfusee_c.dylib` / `.so` (with an rpath to it), `fusee_system` links `-lfusee_c` from the system or `CGO_LDFLAGS`. On macOS the linker may warn that objects were built for a newer macOS than the one being linked; it is harmless (set `MACOSX_DEPLOYMENT_TARGET` when building the library to match). Go 1.23 or newer. ## How it runs The engine is embedded, as in the TypeScript and Python SDKs: Go runs and replays workflow code, and the engine records the journal, decides when a step's inputs are durable, and hands the Go side events and deliveries through a flat interface (numeric run keys, events as rows of numbers, commit positions read from shared memory, payloads as JSON or raw bytes). Every durability mode goes through it: `memory`, `local`, and `server` and `replicated`, where the native core speaks the log service protocol to `fusee server` or to the other replicas itself: one core speaks the protocol for every SDK. An engine has one loop goroutine. Workflow code runs on goroutines the engine schedules one at a time: the goroutine holding the engine hands it to a run's goroutine and waits until that one blocks on a journal operation or returns, through channel handoffs, so whatever one wrote the next sees. Between two deliveries a run's code runs until all its goroutines are blocked, exactly as on replay, which is what makes the journal the same live and replaying. The engine handle moves between OS threads with the goroutine holding it; the C interface allows that (calls must not overlap, and need no particular thread). The loop is not locked to one OS thread: a handoff to a locked thread costs about 25 µs on macOS, against 0.2 µs between ordinary goroutines. **Step functions run on goroutines of their own**, concurrently with everything else on the engine, and receive a `context.Context` that is cancelled when the attempt times out or the engine closes. So a slow step does not hold up other workflows. Two consequences: - A step function must not read or write the workflow's variables (it runs beside the workflow's code): give it what it needs, and return its result. `fusee.Key(ctx)`, `fusee.Heartbeat(ctx, v)`, `fusee.Progress[T](ctx)` and `fusee.Stream(ctx, name)` work inside it. - Parallel steps record in the order they finish. Replay follows the journal, so this is only visible in the journal itself. `StepOptions{Inline: true}` runs the function on the workflow's goroutine while it holds the engine instead: no handoff (a sequential chain of inline no-op steps costs 1.5 µs a step against 3.8 µs), the order it starts in, and access to workflow state, but nothing else on the engine runs until it returns and `Timeout` cannot cut it short. Use it for fast, CPU-only functions. ## Streams A run appends chunks (JSON, or `[]byte` as bytes) to named streams, and readers follow them from an offset ([docs/features.md](../../docs/features.md#streams)): ```go answer, err := fusee.Step(c, "generate", func(ctx context.Context) (string, error) { out, err := fusee.Stream(ctx, "tokens") // the step's stream: its result completes it if err != nil { return "", err } prev, err := fusee.Recorded[string](out) // what an earlier attempt recorded if err != nil { return "", err } text := strings.Join(prev, "") for token := range model.Stream(ctx, prompt, text) { // continue after the prefix if err := out.Append(token); err != nil { return "", err } text += token } return text, nil }) // Anywhere outside workflow code: this process, or a Client through a server. r := e.Run(id).Stream("tokens", lastSeen) for chunk, err := range r.All(ctx) { if err != nil { return err } w.Write(chunk.JSON) // chunk.Bytes for a chunk of bytes; fusee.ChunkAs[T](chunk) } // r.Offset(): where to reconnect; r.Status(): how the stream ended ``` - **Appending never waits.** From a step's goroutine `Append` queues the chunk for the engine's loop and returns; the loop records every queued chunk in its next turn, in order, before the step's result. A failure (say, the stream belongs to another step) is returned by the next `Append` or `Close`. An attempt that timed out cannot append once its retry started. - **Readers see only durable chunks**, so nothing a reader saw is taken back by a crash. `r.Next(ctx)` waits (a long poll of 30 s per request, `r.Wait(d)`) and reports the end once the stream is complete. - **A step that runs again keeps the recorded prefix**: `RecordedCount` and `Recorded[T]` give it what earlier attempts recorded (after a crash, what was durable), and `Closed` is true when an earlier attempt completed the stream (appends are then dropped). - **Workflow code** opens the run's stream with `c.Stream(name)`: its appends are recorded once (a replay appends nothing), and `Close` or the run's end completes it. A step appends to the same stream by opening it with `fusee.Stream(ctx, name)` (a goroutine has no identity, so a writer workflow code opened cannot tell which step calls it; it is for workflow code and inline steps). - In server mode the engine answers the `streamRead` calls clients send through the server, whatever their language. ## Objects An object type is a set of handlers; each key is one object with its own JSON state, and calls to a key's exclusive handlers take turns ([docs/features.md](../../docs/features.md#objects-keyed-state)): ```go cart, err := fusee.RegisterObject(e, "cart", fusee.Handlers{ "add": fusee.Exclusive(func(c *fusee.ObjectCtx, item string) (int, error) { price, err := fusee.Step(c.Ctx, "price", func(ctx context.Context) (int, error) { return pricing.Quote(ctx, item) }) if err != nil { return 0, err } items, _, err := fusee.StateGet[[]string](c.State(), "items") if err != nil { return 0, err } items = append(items, fmt.Sprintf("%s@%d", item, price)) return len(items), c.State().Set("items", items) // committed with the result }), "get": fusee.Shared(func(c *fusee.ObjectCtx, _ any) ([]string, error) { items, _, err := fusee.StateGet[[]string](c.State(), "items") return items, err }), }) // From outside: an ID makes a retried call run once. n, err := fusee.CallRef[int](ctx, cart.Key("user-7"), "add", "tea", fusee.CallOptions{ID: "req-42"}) // From workflow code: a journal operation, run exactly once. n, err = fusee.Call[int](c.Object("cart", "user-7"), "add", "tea") ``` `ref.State(ctx)` reads the committed state, `ref.Inspect(ctx)` adds the running and waiting calls, `e.Objects(ctx, "cart", 0)` lists the keys, and `client.Object("cart", k)` calls a key through a server. ## Deterministic replay in Go Workflow code (the function given to `Register`, functions started with `fusee.Go`, update handlers) must make the same journal operations in the same order on every run: - Wait, branch on time or randomness, and run code concurrently only through the package: `fusee.Step`, `c.Sleep`, `fusee.WaitFor`, `fusee.Child`, `fusee.All`, `fusee.Race`, `fusee.Go`, `c.Now`, `c.Random`, `c.UUID`. - Start no goroutines, and use no channels, `select`, mutexes, timers, `time.Now`, `time.Sleep` or `math/rand` in workflow code. Step functions may do anything. - Map iteration order is random in Go: do not let it decide which journal operations run, or in which order (sort the keys). - A `*fusee.Ctx` and the futures it made belong to one run; a `Ctx` used outside its own goroutine panics. - Do not call `Engine` or `Run` methods from workflow code or inline steps (they wait for the engine, which the caller holds): use the `Ctx` (`c.Signal`, `fusee.Child`, `c.CancelWorkflow`). - Do not recover the panics the package raises, and do not swallow a `*fusee.DeterminismError`: the engine has stopped the run. Code that changed under a recorded run stops it with a `*fusee.DeterminismError` (status `stuck`) naming the step the journal has and the one the code asked for; deploy code that matches, or use `c.Patched(id)` for new branches. ## API | | | | --- | --- | | Engine | `fusee.Open(fusee.Options{Durability, Dir, Address, ReplicaID, Peers, Shards, Retention, ParkAfter, CheckpointInterval, Flush, SegmentBytes, KeepHistory, ArchiveHistory, Codec, InlineFlush, OnExhausted})`, `e.Close()` | | Workflows | `fusee.Define[In, Out](name, fn, fusee.RegisterOptions{Version})` (a definition apart from any engine: `fusee.Options{Workflows: []fusee.Definition{def}}`, `def.Register(e)`, `def.On(e)`), `fusee.Register[In, Out](e, name, fn, fusee.RegisterOptions{Version})`, `fusee.WorkflowOf[In, Out](e, name)`; `wf.Start(ctx, id, in, fusee.StartOptions{Timeout, IfRunning, IfFinished, Version})`, `fusee.UpdateWithStart[R](ctx, wf, id, in, update, arg, opts)`, `wf.Debounce`, `wf.Batch` | | Steps | `fusee.Step[T](c, name, fn, fusee.StepOptions{Retry, Timeout, HeartbeatTimeout, Limit, Priority, FairnessKey, FairnessWeight, ReadOnly, Inline})`, `fusee.StartStep` (a future); `fusee.RetryPolicy{Attempts, Backoff, Initial, Max, NonRetryable, RetryIf, OnExhausted}` (`fusee.Pause` pauses the run when the retries run out, until `Resume`); the step's `context.Context` is its step context: `fusee.Key(ctx)`, `fusee.Heartbeat(ctx, v)`, `fusee.Progress[T](ctx)`, `ctx.Done()` on cancellation, timeout or run end | | Waiting | `c.Sleep(name, d)`, `c.StartSleep`, `fusee.WaitFor[T](c, signal, timeout)` (ok false on timeout), `fusee.StartWait` | | Children and signals | `fusee.Child[O](c, type, id, input, fusee.ChildOptions{ParentClose, Timeout})`, `fusee.StartChild`, `c.SpawnChild`, `c.Signal(target, name, v)`, `c.CancelWorkflow(target, reason)` | | Concurrency | `fusee.Go[T](c, fn)`, `fusee.All(c, futures...)`, `fusee.Race(c, futures...)` (index; losers dropped); `future.Get(c)` | | Queries and updates | `fusee.OnQuery[A, R](c, name, fn)`, `fusee.OnUpdate[A, R](c, name, handle, validate)`; `run.Query`, `run.Update`, typed `fusee.Query[R]`, `fusee.Update[R]` | | Durable values | `c.Now()`, `c.Random()`, `c.UUID()`, `c.JournalSize()` | | Versions and attributes | `c.Patched(id)`, `c.DeprecatePatch(id)`, `c.SetAttributes(map)`, `c.ContinueAsNew(input, fusee.ContinueOptions{Version})` | | Completion tokens | `fusee.AwaitCompletion[T](c, name, send)`; `e.Complete(ctx, token, v)`, `e.Fail(ctx, token, msg)` | | Streams | `c.Stream(name)` (workflow code), `fusee.Stream(ctx, name)` (a step function): `w.Append(v)`, `w.Close()`, `fusee.Recorded[T](w)`, `w.RecordedRaw()`, `w.RecordedCount()`, `w.Closed()`; `run.Stream(name, from)`: `Next(ctx)`, `All(ctx)` (an iterator), `Offset`, `Status`, `End`, `RunID`, `Wait(d)`, `OfRun(run)`; a `fusee.Chunk` has `Offset`, `JSON` or `Bytes`, `Value()`, and `fusee.ChunkAs[T](chunk)` | | Runs | `e.Run(id)`, `fusee.RunOf[Out](e, id)`: `Signal`, `Result`, `Started`, `Update`, `Query`, `Cancel`, `Terminate`, `Reset`, `Resume` (a paused run), `Inspect`, `Stream` | | Engine | `e.Serve(addr, token...)` (the HTTP API and UI, answered by this engine), `e.On(event, fn)` (`"stuck"`, `"paused"`), `e.List(ctx, fusee.ListOptions{Limit, Where})`, `e.Limit(name, fusee.LimitOptions{Concurrent, PerSecond})`, `e.LimitInUse`, `e.LimitWaiting`, `e.Schedule(ctx, name, workflow, input, fusee.ScheduleOptions{...})`, `e.ScheduleHandle(name)` (`Pause`, `Resume`, `Trigger`, `Backfill`, `Describe`, `Delete`), `e.ExportJournal`, `e.ShardMetrics`, `e.Stats` | | Objects | `fusee.RegisterObject(e, type, fusee.Handlers{...})` with `fusee.Exclusive` and `fusee.Shared`; in a handler `c.State()` (`fusee.StateGet[T]`, `Set`, `Clear`, `ClearAll`, `Keys`, `Raw`), `c.ObjectKey()`; `c.Object(type, key)` (`fusee.Call[R]`, `fusee.StartCall[R]`, `Send`); `e.Object(type, key)` and `ty.Key(key)` (`Call`, `fusee.CallRef[R]`, `Send`, `State`, `Inspect`, `Delete`; from outside, `SetState`, `PatchState`, `ClearState` and `EditState` with `IfVersion`), `e.Objects(ctx, type, n)` | | Remote | `fusee.Connect(address)`: `Call`, `Start`, `UpdateWithStart`, `Debounce`, `Batch`, `List`, `Complete`, `Fail`, `Object(type, key)`, `Run(id)` (`Result`, `Started`, `Signal`, `Query`, `Update`, `Cancel`, `Terminate`, `Reset`, `Resume`, `Stream`) | | Invoked workers | `fusee.NewInvocationHandler(fusee.InvocationOptions{Engine, Address, Secret, Register})`: an `http.Handler`, and `Invoke(ctx, signature, body)`; `fusee.VerifySignature` | | Errors | Failures travel as the [envelope](../../docs/errors.md#the-failure-envelope): return a `*fusee.Failure{Message, Type, Code, Details, Retryable, Cause}` (or any error: its message, type name and wrapped errors are recorded); `fusee.Fatal(err)` is never retried; `fusee.FailureOf(err)`. Raised: `*StepError{Step, Detail, Cause}`, `*WorkflowError{Code, ID, Cause}` (a failed run, child or update), `*fusee.Error{Code, Message}`, `*CancelledError` (return `fusee.Cancelled(reason)` to finish cancelled), `*TimeoutError`, `*TerminatedError`, `*DeterminismError`, `*RejectedError`, `fusee.ErrClosed` | Calls that wait take a `context.Context`; cancelling it stops the wait (a start or signal already sent still happens). Payloads are JSON (`encoding/json`); a `[]byte` is recorded as raw bytes, as `bytes` and `Uint8Array` are in Python and TypeScript, so the SDKs read each other's payloads. In server mode (`Durability: "server"`, `Address: "host:port"`) the engine answers calls that clients send through the server, whatever their language. ## Not supported - Worker parts: one engine's workflow code runs on one goroutine at a time (the step functions run in parallel). TypeScript's `workers` option, which splits shards across threads, has no Go counterpart; run several engines on separate directories instead. - A `Replayer` for exported journals, and OpenTelemetry instruments. - Replica metrics while a replicated member waits to lead. ## Tests and conformance ```bash go test -race ./... # server tests start ../../target/release/fusee (FUSEE_CLI) go run ./cmd/conformance > go.json # then conformance/compare.py with the other SDKs go run ./cmd/bench memory # step-to-step latency, calls per step ``` `go test` covers steps, retries, sleeps, signals, children, races, `fusee.Go`, updates, queries, continue-as-new, limits, timeouts, heartbeats, completion tokens, schedules, debounce and batch, a clean close and reopen, a process killed mid-run and reopened, a determinism error after a code change, and server mode against `fusee server` with a client. Stream tests (`stream_test.go`) cover chunk order and completeness while a step streams, reconnecting from an offset, a retry continuing after the recorded prefix, workflow appends recorded once across a reopen and replay, a process killed mid-stream with a resuming reader that sees no chunk twice, an attempt that timed out, and readers through a server (the Go worker answering `streamRead`); `FUSEE_STREAM_COST=1 go test -run StreamCost` prints append cost and reader latency (see [docs/features.md](../../docs/features.md#streams)). The conformance runner prints the 40 scenarios' journals, streams included, and `conformance/remote.py` runs a Go worker and a Go client against the other languages' through a server, a stream read among them. ## Cost per step A sequential chain of no-op steps in memory mode (`go run ./cmd/bench`), against the other SDKs' `bench.mjs` and `bench.py` on the same 18-core Apple M-series laptop, p50 step to step: | | p50 | calls into the engine per step | | --- | --- | --- | | Go | 3.8 µs | 7.0 (2 ticks) | | Go, inline steps | 1.5 µs | 3.1 | | Node | 3.8 µs | 5.0 | | Python | 8.1 µs | 3.1 | Go's default path costs what Node's does (ratio 1.0) and half of Python's (0.47); inline steps cost 0.39 of Node's and 0.19 of Python's. The default path's extra calls come from the step finishing on another goroutine: its result comes back through the loop, which ticks between. # Fusee for Java Durable workflows in Java 17 and later, on the same engine as the Rust, TypeScript, Python and Go SDKs. Workflow code is plain Java: each `ctx.step`, sleep, wait or child is a journal operation recorded in the engine's log and replayed after a crash. ```java try (Engine e = Engine.open(Options.local("./data"))) { Workflow checkout = e.register("checkout", Order.class, String.class, (ctx, o) -> { Quote quote = ctx.step("quote", Quote.class, () -> pricing.quote(o)); Charge charge = ctx.step("charge", Charge.class, () -> payments.charge(quote, ctx.key()), // idempotency key StepOptions.defaults().retry(RetryPolicy.attempts(5))); ctx.sleep("cool-off", Duration.ofMinutes(10)); Optional approved = ctx.waitFor("review", Boolean.class, Duration.ofHours(24)); if (approved.orElse(false)) return charge.id(); return ctx.step("refund", String.class, () -> payments.refund(charge)); }); Run run = checkout.start("order-123", order); run.signal("review", true); String id = run.result(); } ``` ## Building The SDK is a Maven project (`dev.fusee:fusee`, no runtime dependencies) over `libfusee_jni`, a small JNI layer (`native/fusee_jni.c`) statically linked with `libfusee_c`, the engine's C interface (`bindings/c`, header `bindings/c/include/fusee.h`). Build the C interface first, from the repository root, then the SDK: ```bash cargo build --release -p fusee-c cd bindings/java && mvn test ``` `mvn` runs `native/build.sh` before compiling: it compiles the JNI layer with the system C compiler against the JDK's `jni.h` (and checks every C signature against the header `javac -h` generates from `Native.java`), links `../../target/release/libfusee_c.a`, and puts the one shared library at `target/classes/dev/fusee/native//libfusee_jni.`, so it ends up in the jar. `` is `darwin-arm64`, `darwin-x86_64`, `linux-x86_64` or `linux-aarch64`. At run time the loader takes, in order, the file named by the system property `fusee.library`, the library for this platform on the class path (copied to a temporary file when it is inside a jar), or `fusee_jni` from `java.library.path`. A jar for several platforms: build the library on each (`cargo build --release -p fusee-c && native/build.sh out/`, or set `FUSEE_LIB_DIR` and `FUSEE_PLATFORM` for a cross-compiled `libfusee_c.a`), copy each `out/dev/fusee/native/` directory into `target/classes/dev/fusee/native/`, and package with `mvn package -Dfusee.native.skip=true`. CI builds and tests `linux-x86_64`, `linux-aarch64` and `darwin-arm64`. Building needs a JDK (17 or later) and a C compiler; running needs a JRE only. ## JNI, not the Foreign Function & Memory API The FFM API is final in Java 22; in Java 21 it is a preview that needs `--enable-preview` at compile and run time and ties the class files to exactly that release, and Java 17 has only an incubator version with a different API. JNI runs on every LTS release without flags. Measured on the same functions in the same library (Java 21, macOS arm64, 2 M calls, median of 5 rounds; `bench/FfmBench.java`): | | JNI | FFM (21, preview) | | --- | --- | --- | | `fusee_abi_version()` | 4–20 ns | 9–30 ns (12–15 ns with `isTrivial`) | | `fusee_engine_code_state(e, u32, u64, *u32)` | 7–19 ns | 17–42 ns | | `fusee_engine_limit_usage(e, str, *u32, *u64)`, a `byte[]` and a `long[]` | 334–558 ns | 54–117 ns | A call with primitive arguments costs less through JNI than through FFM. What costs is a JNI function the native side calls back into the JVM: each array access (`GetByteArrayRegion`, `SetLongArrayRegion`) costs about 70 ns on macOS arm64, where HotSpot switches the thread's W^X mode on every entry into the VM. So the calls on the step path (`begin_step`, `step_done`, `host`, `quiescent`, `tick`, `finish`) make none: Java writes their byte arguments into a direct `ByteBuffer` the native side registered once, and they leave numbers and bytes in a second one. The engine's commit positions and its owed-tick flag are read from direct buffers over `fusee_io`, without a call. Calls off the step path (start, signal, inspect, list) pass arrays. ## How it runs The engine is embedded, as in the other SDKs: Java runs and replays workflow code, and the engine records the journal, decides when a step's inputs are durable, and hands Java events and deliveries through the C interface. Every durability mode goes through it: `memory`, `local`, and `server` and `replicated`, where the native core speaks the log service protocol to `fusee server` or to the other replicas itself. An engine has one loop thread. Workflow code runs on threads the engine schedules one at a time, Go's model: the thread holding the engine hands it to a run's thread and waits until that one blocks on a journal operation or returns, and between two deliveries a run's code runs until all its threads are blocked, exactly as on replay. On Java 21 and later the loop and workflow code are virtual threads (a handoff is an unpark and a park on the same carrier, and a waiting run costs no OS thread); on Java 17 they are platform threads. Two shortcuts keep a run on its own thread: a workflow thread about to block takes the run's next delivery itself when every other thread of the run is blocked (what the loop would do next), so a chain of inline steps makes no handoff at all; and one with nothing to deliver runs the loop's turns itself until its run has work, so a step's result reaches it with one wake. **Step functions run on threads of their own**, from a cached pool of platform threads shared by every engine, whose most recently idle worker spins for 50 µs before it parks (and the thread waiting for the engine's next event spins for 30 µs), so a step reaches a running thread without a wake. A step function may block; it is interrupted when its attempt times out or the engine closes. For many concurrent steps that block on I/O, pass `Options.stepExecutor(Executors.newVirtualThreadPerTaskExecutor())`. Two consequences of their own threads: - A step function must not read or write workflow state that workflow code changes meanwhile: give it what it needs, and return its result. `ctx.key()`, `ctx.heartbeat(v)`, `ctx.progress(type)` and `ctx.stream(name)` work inside it, on the step's thread. - Parallel steps record in the order they finish. Replay follows the journal, so this is only visible in the journal itself. `StepOptions.defaults().inline()` runs the function on the workflow's thread while it holds the engine instead: no handoff, the order it starts in, and access to workflow state, but nothing else on the engine runs until it returns and a timeout cannot cut it short. Use it for fast, CPU-only functions. The system properties `fusee.threads` (`virtual` or `platform`), `fusee.steps=virtual`, `fusee.spin.us`, `fusee.wait.spin.us` and `fusee.pool.spin.us` change these choices (0 turns a spin off). ## Streams A run appends chunks (JSON, or `byte[]` as bytes) to named streams, and readers follow them from an offset ([docs/features.md](../../docs/features.md#streams)): ```java String answer = ctx.step("generate", String.class, () -> { StreamWriter out = ctx.stream("tokens"); // the step's stream: its result completes it String text = String.join("", out.recorded(String.class)); // what an earlier attempt recorded for (String token : model.stream(prompt, text)) { // continue after the prefix out.append(token); text += token; } return text; }); // Anywhere outside workflow code: this process, or a Client through a server. StreamReader tokens = engine.run(id).stream("tokens", lastSeen); for (Chunk c : tokens) response.write(c.as(String.class)); // c.bytes() for a chunk of bytes // tokens.offset(): where to reconnect; tokens.status(): how the stream ended ``` - **Appending never waits.** From a step's thread `append` queues the chunk for the engine's loop and returns; the loop records every queued chunk in its next turn, in order, before the step's result. A failure (say, the stream belongs to another step) is thrown by the next `append` or `close`. An attempt that timed out cannot append once its retry started. In workflow code and inline steps an append is recorded at once. - **Readers see only durable chunks**, so nothing a reader saw is taken back by a crash. `hasNext()` waits (a long poll of 30 s per request, `waitFor(d)`) and returns false once the stream is complete. - **A step that runs again keeps the recorded prefix**: `recordedCount()` and `recorded(T.class)` give it what earlier attempts recorded (after a crash, what was durable), and `closed()` is true when an earlier attempt completed the stream (appends are then dropped). - **Workflow code** opens the run's stream with `ctx.stream(name)`: its appends are recorded once (a replay appends nothing), and `close()` or the run's end completes it. Step functions may append to it as well: the writer finds the step from the calling thread. - In server mode the engine answers the `streamRead` calls clients send through the server, whatever their language. ## Objects An object type is a set of handlers; each key is one object with its own JSON state, and calls to a key's exclusive handlers take turns ([docs/features.md](../../docs/features.md#objects-keyed-state)): ```java ObjectType cart = engine.registerObject("cart", Map.of( "add", ObjectHandler.exclusive(String.class, (c, item) -> { int price = c.ctx().step("price", Integer.class, () -> pricing.quote(item)); List items = c.state().get("items", new TypeRef>() {}); List next = items == null ? new ArrayList<>() : new ArrayList<>(items); next.add(item + "@" + price); c.state().set("items", next); // committed with the result return next.size(); }), "get", ObjectHandler.shared(Object.class, (c, ignored) -> c.state().raw("items")))); // From outside: an ID makes a retried call run once. int n = cart.key("user-7").call("add", "tea", Integer.class, CallOptions.id("req-42")); // From workflow code: a journal operation, run exactly once. int m = ctx.object("cart", "user-7").call("add", "tea", Integer.class); ``` `ref.state()` reads the committed state, `ref.inspect().object()` adds the running and waiting calls, `engine.objects("cart", 0)` lists the keys, and `client.object("cart", k)` calls a key through a server. ## Deterministic replay in Java Workflow code (the function given to `register`, functions started with `ctx.async`, update handlers) must make the same journal operations in the same order on every run: - Wait, branch on time or randomness, and run code concurrently only through the `Ctx`: `step`, `sleep`, `waitFor`, `child`, `all`, `race`, `async`, `now`, `random`, `uuid`. - Start no threads or executors, and use no `CompletableFuture`, locks, `wait`/`notify`, `Thread.sleep`, timers, `System.currentTimeMillis`, `Instant.now`, `Math.random`, `Random` or `UUID.randomUUID` in workflow code. Step functions may do anything. - `HashMap` and `HashSet` iteration order is not part of a program's meaning: do not let it decide which journal operations run, or in which order (use `LinkedHashMap`, or sort). - A `Ctx` and the `Op`s it made belong to one run. A `Ctx` used outside its run's workflow threads throws `IllegalStateException`. - Do not call `Engine` or `Run` methods from workflow code or inline steps (they wait for the engine, which the caller holds; they throw `IllegalStateException`): use the `Ctx` (`ctx.signal`, `ctx.child`, `ctx.cancelWorkflow`). - Do not catch `Throwable` or `Error` around `Ctx` calls (a run that ends unwinds its threads with an `Error`), and do not swallow a `DeterminismException`: the engine has stopped the run. Code that changed under a recorded run stops it with a `DeterminismException` (status `stuck`) naming the step the journal has and the one the code asked for; deploy code that matches, or use `ctx.patched(id)` for new branches. ## Payloads Payloads are JSON. The built-in codec (`JsonCodec.builtIn()`, no dependencies) handles strings, numbers, booleans, enums (by name), records, classes with public fields (or getters and setters) and a public no-argument constructor, `List`, `Set`, `Map` with string keys, arrays, `Optional`, `UUID` and `java.time` values (as ISO text), and `Json` (a JSON value kept as text). Generic types take a `TypeRef`: `new TypeRef>() {}`. A `byte[]` payload is recorded as raw bytes, as `[]byte`, `bytes` and `Uint8Array` are in Go, Python and TypeScript, so the SDKs read each other's payloads. To use Jackson, implement `JsonCodec` in two lines (the interface's Javadoc has them) and pass it with `Options.json(codec)`. ## API | | | | --- | --- | | Engine | `Engine.open(Options.local(dir) / memory() / server(address) / replicated(id, peers, dir)` with `.shards`, `.retention`, `.parkAfter`, `.checkpointInterval`, `.flush`, `.segmentBytes`, `.keepHistory`, `.archiveHistory`, `.codecKey(id, key)`, `.inlineFlush`, `.onExhausted`, `.json`, `.stepExecutor)`, `e.close()` | | Workflows | `WorkflowDef.workflow(name, In.class, Out.class, (ctx, in) -> ...)` (a definition apart from any engine: `Options.workflows(defs...)`, `def.register(e)`, `def.on(e)`, `.version(n)`), `e.register(name, [version,] In.class, Out.class, (ctx, in) -> ...)` (or `TypeRef`s), `e.workflow(name, In.class, Out.class)`; the engine and a `Client` share `start(type, id, input[, opts])`, `signalWithStart(type, id, input, signal, value[, opts])`, `updateWithStart(type, id, input, update, arg[, opts])`, `debounce`, `batch`, `run(id)`, `list`; `wf.start(id, in, new StartOptions().timeout(d).ifRunning(..).ifFinished(..).version(n))`, `wf.updateWithStart(id, in, update, arg, R.class)` (`.run()`, `.result()`), `wf.debounce(key, in, DebounceOptions.window(d).maxWait(d).id(..))`, `wf.batch(key, event, new BatchOptions().maxEvents(n).maxWait(d).id(..))` | | Steps | `ctx.step(name, T.class, () -> ..., StepOptions.defaults().retry(RetryPolicy.attempts(n).initial(d).max(d).fixed().nonRetryable("CardDeclined").retryIf((failure, attempt) -> ...).onExhausted(Exhausted.PAUSE)).timeout(d).heartbeatTimeout(d).limit(name).priority(p).fairness(key, weight).readOnly().inline())`, `ctx.startStep` (an `Op`); a function that takes one gets its step context: `ctx.step(name, T.class, s -> ...)` with `s.key()`, `s.attempt()`, `s.name()`, `s.workflowId()`, `s.heartbeat(v)`; inside a plain one: `ctx.key()`, `ctx.heartbeat(v)`, `ctx.progress(T.class)` | | Waiting | `ctx.sleep(name, d)`, `ctx.startSleep`, `ctx.waitFor(signal, T.class)`, `ctx.waitFor(signal, T.class, timeout)` (an `Optional`, empty on timeout), `ctx.startWait` | | Children and signals | `ctx.child(type, id, input, O.class, new ChildOptions().parentClose("abandon").timeout(d))`, `ctx.startChild`, `ctx.spawnChild`, `ctx.signal(target, name, v)`, `ctx.cancelWorkflow(target, reason)` | | Concurrency | `ctx.async(() -> ...)`, `ctx.all(ops...)`, `ctx.race(ops...)` (the index; losers dropped); `op.get()`, `op.isDone()`, `op.timedOut()` | | Queries and updates | `ctx.onQuery(name, A.class, a -> ...)`, `ctx.onQuery(name, () -> ...)`, `ctx.onUpdate(name, A.class, a -> ..., a -> { if (bad) throw ...; })`; `run.query(name, arg, R.class)`, `run.update(name, arg, R.class, UpdateOptions.id(..))` | | Durable values | `ctx.now()` (`Instant`), `ctx.random()`, `ctx.uuid()` (`UUID`), `ctx.journalSize()` | | Versions and attributes | `ctx.patched(id)`, `ctx.deprecatePatch(id)`, `ctx.setAttributes(map)`, `return ctx.continueAsNew(input[, version])` | | Completion tokens | `ctx.awaitCompletion(name, T.class, token -> ...)`; `e.complete(token, v)`, `e.fail(token, message)` | | Streams | `ctx.stream(name)` (workflow code, or a step function on its thread): `append(v)`, `close()`, `recorded(T.class)`, `recordedCount()`, `closed()`; `run.stream(name[, from])`, a `StreamReader` (`Iterator` and `Iterable`): `offset()`, `status()`, `end()`, `runId()`, `waitFor(d)`, `ofRun(run)`; a `Chunk` has `offset()`, `json()` or `bytes()`, `value()`, `as(T.class)` | | Runs | `e.run(id[, O.class])`: `signal`, `result([timeout])`, `started([timeout])`, `update`, `query`, `cancel`, `terminate`, `reset(beforeStep[, new ResetOptions().dropSignals().version(n)])`, `resume()` (a run paused when a step's retries ran out), `inspect()` (a `RunDetail`: summary and journal), `stream(name[, from])` | | Engine | `e.on(event, (workflowId, message) -> ...)` (`"stuck"`, `"paused"`), `e.list(ListOptions.where("status", "ok").and("amount", Map.of("gte", 100)).limit(n))`, `e.limit(name, LimitOptions.concurrent(n).withPerSecond(r))`, `e.limitInUse`, `e.limitWaiting`, `e.schedule(name, workflow, input, ScheduleOptions.cron(expr) / every(d)...)`, `e.scheduleHandle(name)` (`pause`, `resume`, `trigger`, `backfill`, `describe`, `delete`), `e.exportJournal(id)`, `e.shardMetrics()`, `e.stats()`, `e.shards()` | | Objects | `engine.registerObject(type, Map.of(name, ObjectHandler.exclusive(Arg.class, fn), ...))` (and `shared`); in a handler `c.state()` (`get`, `set`, `clear`, `clearAll`, `keys`, `raw`), `c.key()`, `c.ctx()`; `ctx.object(type, key)` (`call`, `startCall`, `send`); `engine.object(type, key)` (`call`, `send`, `state`, `inspect`, `delete`, with `CallOptions.id`; from outside, `setState`, `patchState`, `clearState` and `editState(edit, ifVersion)`), `engine.objects(type, n)` | | Remote | `Client.connect(address)`: `call`, `start`, `updateWithStart`, `debounce`, `batch`, `list`, `complete`, `fail`, `object(type, key)`, `run(id)` (`result`, `started`, `signal`, `query`, `update`, `cancel`, `terminate`, `reset`, `resume`, `stream`) | | Invoked workers | `new InvocationHandler(address, secret, options, register)`: `httpHandler()` for the JDK's `HttpServer`, `invoke(signature, body)` for any framework; `InvocationHandler.verifySignature` | | Errors | Failures travel as the [envelope](../../docs/errors.md#the-failure-envelope): throw a `FailureException` (`.code(..)`, `.details(..)`, `.type(..)`, `.retryable(false)`; subclass it to name the type) or a `FatalException` (never retried); any other exception records its message, class name and cause. `Failure` (`message()`, `type()`, `code()`, `details()`, `retryable()`, `cause()`, `isFinal()`, `Failure.of(t)`). Raised: `StepFailedException` (`step()`, `detail()`, `failure()`), `WorkflowFailedException` (`code()` `WorkflowFailed` or `ChildFailed`, `failure()`, `workflowId()`), `FuseeException` (`code()`: `NotFound`, `Closed`, `Invalid`, ...), `CancelledException` (throw one to finish cancelled), `TimedOutException`, `TerminatedException`, `DeterminismException`, `RejectedException` (`reason()`) | Calls that wait block the calling thread; an interrupt ends the wait with a `FuseeException` of code `Interrupted` (a start or signal already sent still happens). In server mode (`Options.server("host:port")`) the engine answers calls that clients send through the server, whatever their language. ## Not supported - Worker parts: one engine's workflow code runs one thread at a time (the step functions run in parallel). TypeScript's `workers` option has no Java counterpart; run several engines on separate directories instead. - A `Replayer` for exported journals, and OpenTelemetry instruments. - Replica metrics while a replicated member waits to lead. - Published jars for Linux and macOS together: CI builds each platform's library; assembling one jar from them is the manual step above. - On Java 17 (platform threads), many concurrent runs cost an OS thread wake per resumed run: about 5,000 steps/s with 1,000 concurrent workflows on the machine below, against 100,000–135,000 on Java 21 (the sequential chain is as fast: 2.7 µs, inline 0.8 µs). Measured on Java 21 with `-Dfusee.threads=platform`, the path Java 17 takes; CI runs the tests on Java 17 itself. ## Tests and conformance ```bash mvn test # server tests start ../../target/release/fusee (-Dfusee.cli=...) java -cp target/classes:target/test-classes dev.fusee.cmd.Conformance > java.json java -cp target/classes:target/test-classes dev.fusee.cmd.Bench memory ``` `mvn test` covers steps, retries, sleeps, signals, children, races, `ctx.async` and `all`, updates and queries, update-with-start, cancel, terminate and start policies, workflow timeouts, continue-as-new, patches, attributes and listing, limits, step timeouts and heartbeats, completion tokens, schedules, debounce and batch, export, raw bytes, payloads larger than the buffers the step path shares with the native side, inline steps and keys, the JSON codec, a clean close and reopen, a process killed with SIGKILL mid-run and reopened, a determinism error after a code change, payload encryption with key rotation (no plaintext in the data directory), and server mode against `fusee server` with a client. `StreamsTest` covers chunk order and completeness while a step streams, reconnecting from an offset, a retry continuing after the recorded prefix, workflow appends (one from a step's thread) recorded once across a reopen and replay, a process killed mid-stream with a resuming reader that sees no chunk twice, an attempt that timed out, inline steps, and readers through a server (the Java worker answering `streamRead`); `FUSEE_STREAM_COST=1 mvn test -Dtest=StreamsTest#cost` prints append cost and reader latency (see [docs/features.md](../../docs/features.md#streams)). The conformance runner prints the 40 scenarios' journals, streams included, for `conformance/compare.py`, and `conformance/remote.py` runs a Java worker and a Java client against the other languages' through a server, a stream read among them. ## Cost per step A sequential chain of no-op steps in memory mode (`dev.fusee.cmd.Bench`), against the other SDKs' benchmarks on the same machine (an 18-core Apple M-series laptop), p50 step to step, median of three interleaved runs, then 1,000 concurrent 20-step workflows (Node: 2,000). Calls per step for Node and Python are from the Go SDK's measurement. | | p50 | inline steps | calls into the engine per step | steps/s, concurrent | | --- | --- | --- | --- | --- | | Java 21 | 2.8 µs | 1.2 µs | 7.7 (1.7 ticks) | 135,000 | | Java 21, one carrier | 2.8 µs | 1.1 µs | 7.6 (1.6 ticks) | 271,000 | | Go | 4.8 µs | 1.1 µs | 7.0 (2.0 ticks) | 436,000 | | Node | 2.2 µs | | 5.0 | 264,000 | | Python | 4.8 µs | | 3.1 | 114,000 | Java's default path costs 0.58 of Go's, 1.3 times Node's and 0.58 of Python's; inline steps cost 1.1 times Go's. Concurrent throughput is 0.31 of Go's, 0.51 of Node's and 1.2 times Python's: with many runs, each resumed run is a virtual-thread switch the engine waits for, and with idle carriers the scheduler wakes one on each switch (2 µs a handoff, against 0.2 µs on one carrier). A JVM that runs mostly workflows can pass `-Djdk.virtualThreadScheduler.parallelism=1` (the "one carrier" row; the engine runs one workflow thread at a time anyway, and step functions run on the pool's platform threads), but that setting applies to every virtual thread in the JVM. With step functions on a virtual thread each (`-Dfusee.steps=virtual`) the chain costs 11–13 µs and the concurrent case 120,000–140,000 steps/s. In local mode with the default flush (`F_FULLFSYNC` on macOS) the chain is disk-bound: 5.1 ms against Go's 4.8 ms, and 1,460 steps/s concurrent against Go's 1,350. # Workflow features Besides steps, sleeps, signals and child workflows, Fusee has the workflow features below. Each one is a record type in the journal plus an SDK call, with the same durability rules as steps, the same simulator checkers and the same cross-SDK conformance scenarios. Names below are TypeScript; Rust and Python use the same names in their own case (`on_update`, `await_completion`, `continue_as_new`), Go in its own (`fusee.OnUpdate`, `fusee.AwaitCompletion`, `ctx.ContinueAsNew`; see [bindings/go](../bindings/go/README.md)), and Java as TypeScript does, with types for results (`ctx.onUpdate(name, A.class, ...)`, `ctx.awaitCompletion(name, T.class, ...)`, `ctx.continueAsNew`; see [bindings/java](../bindings/java/README.md)). Read-only steps, first below, are Fusee's own: an option on a step rather than a record type. ## Read-only steps ```ts const order = await ctx.step("load", () => db.getOrder(id), { readOnly: true }); const price = await ctx.step("quote", () => pricing.quote(order), { readOnly: true }); const charge = await ctx.step("charge", () => payments.charge(price, { idempotencyKey: ctx.key() })); ``` A step's function normally waits until every record before it is durable, so a crash can never take back an input it saw. A read-only step skips that wait and runs at once. Mark steps that change nothing outside the process: lookups, fetches, model calls. The result is recorded and replayed like any step's, and later steps and outputs still wait for it to be durable. Above, `quote` runs as soon as `load` returns, and `charge` waits for one flush that carries both results instead of two. The cost: if the process fails before the records ahead of a read-only step are durable, the step may run again with different inputs, because it read state the crash took back. That is harmless for a read and wrong for a step with effects, which is why it is opt-in per step. The option is not journaled, so adding or removing it never changes a journal. `awaitCompletion` ignores it: its send function always has an effect. In Rust: `StepOptions::default().read_only()`; in Python: `read_only=True`. ## Queries ```ts ctx.onQuery("status", () => state); // in the workflow const s = await run.query("status"); // anywhere ``` The handler runs on the workflow's shard against its in-memory state; a parked run is loaded first, and a run that is still replaying its journal is answered once it has caught up. A query writes nothing. Its answer leaves the process only once the state it read is durable, normally the next flush. A handler must not change state or call `ctx`. ## Updates and update-with-start ```ts ctx.onUpdate("addItem", { validate: (item) => (item.qty > 0 ? undefined : "quantity must be positive"), handle: async (item) => { const price = await ctx.step("price", () => pricing.quote(item)); cart.push({ ...item, price }); return cart.length; }, }); const n = await run.update("addItem", item, { id: "req-42" }); const { run, result } = await checkout.updateWithStart("order-123", order, "addItem", item); ``` - The validator runs first, against the current state. A refusal records nothing; the caller gets a `RejectedError`. - An accepted update is appended with its update ID and delivered to the code at its place in the journal, so its handler's operations replay at the same positions. A retried call with the same ID gets the recorded result instead of running the handler twice. The result is released once its records are durable. - `updateWithStart` starts the workflow if it is not running and puts the update in the start record itself, so the two never come apart; if the workflow is running, the update goes to it. The validator does not run for an update delivered with the start. - Callers waiting on an update whose workflow ends first get an error; the accepted update stays in the journal. ## Step heartbeats ```ts await ctx.step("export", async () => { for (let page = ctx.progress() ?? 0; page < pages; page++) { await exportPage(page); ctx.heartbeat(page + 1); } }, { heartbeatTimeout: "30s" }); ``` Heartbeats are coalesced (only the latest per step is appended, just before the next flush) and never wait for the disk. A step retried after a crash gets its last durable progress from `ctx.progress()`; a retry within the same process gets the latest progress. With `heartbeatTimeout`, an attempt that stops sending heartbeats fails and is retried under its retry policy. ## Streams ```ts // The model call is the step; its tokens are a stream. const answer = await ctx.step("generate", async () => { const out = ctx.stream("tokens"); let text = out.recorded().join(""); // what an earlier attempt recorded for await (const token of model.stream(prompt, { prefill: text })) { out.append(token); text += token; } return text; }); // Anywhere: a web handler, another process through a server, the UI. const tokens = engine.run(id).stream("tokens", { from: lastSeen }); for await (const token of tokens) res.write(token); // tokens.offset: where to reconnect ``` A run appends chunks (JSON, or bytes) to named streams; readers follow a stream from an offset as it grows. The rules: - **Appending never waits.** `append` records a chunk and returns: no flush per chunk. Chunks go to the log with whatever else is flushing, so one write covers every chunk appended since the last. A chunk takes no journal position (as a heartbeat), so adding or removing streaming never shifts a journal, and its offsets are 0, 1, 2, ... in the order of the appends. - **Who writes it, and what completes it.** `ctx.stream(name)` inside a step's function opens a stream of that step: the step's final result completes it (a `StreamEnd` record with the step's status is appended just before the result's). In workflow code it opens a stream of the run, which the code's `close()` or the run's end completes; the run's steps may append to it too, so one stream can carry every step of an agent loop. A step cannot write a stream another step opened, and nothing can append to a complete stream. - **Readers see only durable chunks.** A read returns the chunks from its offset that are durable (committed to a quorum, when replicated), or waits until one is (a long poll, 30 s by default, after which it asks again). It ends once the stream is complete and it has every chunk. A crash can lose chunks that were not durable yet, but never one a reader saw, so nothing a reader shows has to be taken back. The price is one flush between an append and its readers (tens of microseconds on a local disk) against tens of milliseconds between a model's tokens, which is why there is no option to read ahead of the disk. - **Reconnecting.** A stream is addressed by offset: a reader that reconnects with the offset it reached (`reader.offset`) continues exactly there, in this process, in another one through a server, or after the worker failed over. A reader stays on the run it started on when the workflow ID moves on (continue-as-new, reset). - **A step that runs again keeps the recorded prefix.** When a step that streamed runs again (a retry, or after a crash: what was durable), its earlier chunks stay and the new attempt appends after them. `recorded()` hands it those chunks, so it continues instead of starting over: pass them to the model as the start of its answer (an assistant prefill) and stream the rest. The model is not asked again for the recorded tokens, and readers see each token once. Streams have no retraction: an attempt that ignores `recorded()` and starts over appends a second answer after the first one's start. If an earlier attempt completed the stream and the crash took only the step's result, `ctx.stream` reports it closed (`closed`) and appends are dropped. - **Replay appends nothing.** Workflow code runs again from the top after a crash; its appends are matched to the recorded ones, in order, and only those beyond them are appended. A replayed step's function does not run, so it appends nothing either. - **Retention and memory.** Chunks are journal records: a loaded run's chunks count toward `memoryBudget`, go to the run store when the run parks and come back when it loads, and a finished run keeps its streams (in its summary, without `keepHistory` as well) until its retention ends. For readers that follow a stream, each stream keeps its newest durable chunks in memory (64 KiB); a reader behind them reads the run's journal once, and what it read (up to 8 MiB) is kept for its next reads. A read's answer is the same JSON everywhere (see `fusee::StreamBatch`): `{"run", "from", "next", "chunks", "bytes", "end"}`, where `bytes` lists the chunks that are bytes (base64) and `end` is `null` or `{"status", "closed"}`. `status` is `ok` for a stream its writer closed or whose step succeeded, a failed step's `failed` (or `cancelled`), or the run's status when the run ended with the stream open. ### In every SDK | | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | | Open | `ctx.stream(name)?` | `ctx.stream(name)` | `ctx.stream(name)` | `c.Stream(name)` in workflow code, `fusee.Stream(ctx, name)` in a step function | `ctx.stream(name)` | | Append | `w.append(&v)?`, `w.append_bytes(&b)?` | `w.append(v)` (a `Uint8Array` is bytes) | `w.append(v)` (`bytes` are bytes) | `w.Append(v)` (a `[]byte` is bytes) | `w.append(v)` (a `byte[]` is bytes) | | Complete | `w.close()?` | `w.close()` | `w.close()` | `w.Close()` | `w.close()` | | Earlier attempts' chunks | `w.recorded::()?`, `w.recorded_len()`, `w.is_closed()` | `w.recorded()`, `w.recordedCount`, `w.closed` | `w.recorded()`, `w.recorded_count`, `w.closed` | `fusee.Recorded[T](w)`, `w.RecordedCount()`, `w.Closed()` | `w.recorded(T.class)`, `w.recordedCount()`, `w.closed()` | | Read | `run.stream(name).from(n)`, `reader.next().await?` | `for await (const c of run.stream(name, { from }))` | `async for c in run.stream(name, from_=n)` | `for c, err := range run.Stream(name, n).All(ctx)`, `reader.Next(ctx)` | `for (Chunk c : run.stream(name, n))` (an `Iterator`) | | Where it stopped | `reader.offset()`, `reader.end()` | `reader.offset`, `reader.status` | `reader.offset`, `reader.status` | `reader.Offset()`, `reader.Status()`, `reader.End()` | `reader.offset()`, `reader.status()`, `reader.end()` | | From another process | `Client::stream(id, name, from)`, `Client::stream_read` | `client.run(id).stream(name, { from })` | `client.run(id).stream(name, from_=n)` | `client.Run(id).Stream(name, n)` | `client.run(id).stream(name, n)` | Go and Java run step functions on goroutines and threads of their own, so an append there is queued for the engine's loop and returns (one step's appends stay in order, and come before its result); a failure is returned by the next append or close. In Go a step opens its stream from the context its function received (a goroutine has no identity to find the step by); in Java `ctx.stream` inside a step function finds the step from its thread, and a writer workflow code opened appends for the step whose thread calls it. An attempt that timed out cannot append after its retry started. The C interface (`bindings/c`, version 2) has the operations for any other SDK: `FUSEE_OP_STREAM_OPEN`, `_APPEND`, `_CLOSE` and `_RECORDED` and `fusee_engine_stream_read`, and a worker in any language answers the `streamRead` calls clients send through a server. The web UI shows a run's streams as they grow, and `GET /v1/runs//streams/?from=N&wait=MS` reads one (`Accept: text/event-stream` follows it as it grows). A chunk is a journal record, so a reader that reconnects continues at its offset, a step that runs again continues after what it recorded, and a crash never takes back a chunk a reader saw. ## Async step completion ```ts const approved = await ctx.awaitCompletion("approval", (token) => mail.sendApproval(token)); // in another process, any time later: await engine.complete(token, true); // or engine.fail(token, "rejected") ``` The send function runs as a step does (after its gate, retried under the retry policy). Each attempt gets its own token, which names the workflow, the run, the step and the attempt; a completion with an older attempt's token, or after the step finished, is refused. The completion is recorded like a step result; the caller of `complete` is answered once it is durable. While it waits, a Rust run is parked out of memory (TypeScript and Python runs stay loaded). In server mode any machine can complete a token through `fusee_server::Client`. ### Callback URLs When the engine serves its HTTP API (the dev profile, `Config::http`, `engine.serve()`), the send function also gets a URL that completes the step: `POST` the value as JSON to it, or to `…/v1/fail/` with a failure. Behind a proxy, `engine.set_callback_url(base)` says where the API is reached from outside. ```python await ctx.await_completion("approval", lambda cb: mail.send_link(cb.url)) ``` ```rust ctx.await_completion("approval", |cb: fusee::Completion| mail.send_link(cb.url)) ``` ## Compensation ```python hold = await ctx.step("reserve", lambda: inventory.reserve(sku)) ctx.compensate("release", lambda: inventory.release(hold)) await ctx.step("charge", lambda: payments.charge(cents)) # fails: "release" runs ``` A compensation is a step registered after the step it undoes. If the workflow fails or is cancelled, every registered compensation runs as a step under its name, newest first, then the run ends with its failure. A compensation that fails after its retries is logged and the others still run. They are steps like any other: recorded, retried, and replayed, not run again, after a crash. Rust: `ctx.compensate("release", move || release(hold))`. ## Signal with start ```python await engine.signal_with_start(cart, "cart-7", Cart(), add, item) ``` Signals the run `id`, starting it with the input first if it is not running. The start and the signal reach the run's shard back to back, so the signal is recorded before a new run's code first runs; the call returns once both are durable. Rust: `engine.signal_with_start(CART, "cart-7", &cart, ADD, &item)`; over the protocol, `signalWithStart`. ## Cancellation into steps Cancelling, terminating or closing a run cancels its running step functions: in Rust and in Python's async steps at their next `await` (the future is dropped; the coroutine gets `CancelledError`). A Python step on a thread (`thread=True`) cannot be interrupted; it reads `s.cancelled` to stop early. ## Continue-as-new ```ts if (ctx.journalSize() > 1000) return ctx.continueAsNew(state); ``` Ends the run and starts a new run of the same workflow ID with an empty journal, in one record (the old run's end carries the new run's start). Signals and updates that arrive afterwards go to the new run; callers waiting for the result follow it. The old run's records become eligible for compaction. `ctx.journalSize()` counts the run's journal operations and is deterministic. ## Patching ```ts const fee = ctx.patched("refund-rounding") ? Math.round(x) : Math.floor(x); ``` `patched` is true for new runs and for running ones the first time their code reaches this point after the deploy, and false while replaying a journal that passed this point before the patch existed. The decision is recorded as a marker that takes no journal position, so adding a patch never shifts the journal. Once no running workflow predates the patch, replace the call with `ctx.deprecatePatch("refund-rounding")` and keep only the new branch; later the call can be deleted. Pinned versions stay the default; patching is how a workflow sleeping for months picks up a fix. ## Terminate and reset - `run.terminate(reason)` ends a run at once. No more of its code runs, the result of a step still running is discarded, and its children follow their parent close policy. - `run.reset({ beforeStep: "charge" })` starts a new run of the same ID whose journal is the old one cut just before the named step's first record, then runs the code from there. The copied records do not send their signals again; children they started are attached to, not started again. Signals received after the cut are reapplied unless `reapplySignals: false`. The old run is kept and marked `reset` (ended, if it was still running). A finished run can be reset when the engine keeps history (`keepHistory`). - `ctx.key()` includes the run, so steps after the cut get new idempotency keys and really run again. ## Pausing when retries run out A step that fails its last attempt fails its run, and its parent, the callers waiting on `result()` and the calls queued behind an object's handler all get the error. A step can pause the run instead, so the cause is fixed (the fix deployed, if it is in the code) and the run continues as if the step had succeeded the first time: ```ts await ctx.step("charge", (s) => payments.charge(order, { idempotencyKey: s.key }), { retry: { attempts: 5, onExhausted: "pause" }, }); const engine = await Engine.open({ workflows, onExhausted: "pause" }); // every step's default // Later, once payments are back: await engine.run("order-1").resume(); ``` ```python await ctx.step("charge", charge, retry=fusee.Retry(attempts=5, on_exhausted="pause")) engine = await fusee.Engine.open(dir, on_exhausted="pause") await engine.run("order-1").resume() ``` Rust `RetryPolicy::attempts(5).on_exhausted(Exhausted::Pause)` and `Config::on_exhausted`, Go `RetryPolicy{Attempts: 5, OnExhausted: fusee.Pause}` and `Options.OnExhausted`, Java `RetryPolicy.attempts(5).onExhausted(Exhausted.PAUSE)` and `Options.onExhausted`. The default is to fail. - **What pauses.** A failure the policy would retry, on the last attempt or once `maxElapsed` has passed: an exception, a step timeout, a missed heartbeat. A final failure (`FatalError`, `retryable = false`, `nonRetryable`) still fails at once. An async completion's `send` function fails when its retries run out; only steps pause. - **While paused.** The run's status is `paused`. Its parent, callers of `result()` and an object's queued calls keep waiting; shared handlers still run. A paused run parks like a suspended one. Its deadline keeps running: a workflow timeout ends it as `timed_out`. A cancellation reaches the paused step (it ends `Cancelled`, without another attempt, and compensations run); terminate and reset work as on any run. - **Resume.** `run.resume()` in every client, `fusee runs resume `, `fusee runs bulk resume --status paused`, `POST /v1/runs/{id}/resume`, and Resume on the UI's Needs attention screen. The step runs again at once with a fresh set of attempts: `s.attempt` counts from 1 again, and `maxElapsed` from the resume; the journal goes on numbering the step's attempts. A resume records who asked (`by`, see [Audit](#audit)); one for a run with nothing paused is refused. - **Which code.** The resume runs the code the process has then. A fix that leaves the steps before the paused one as they were needs nothing more; one that changes them is a [patch](#patching), as for any deploy. - **Where it shows.** `paused` in listings, `fusee runs ls --status paused`, the Needs attention screen, and the `runs_paused` gauge. Each pause writes one warning line naming the run, the step and the failure; Python, Go and Java also call `engine.on("paused", ...)` handlers. In the journal the last failed attempt is an `Attempt` whose retry time is "until resumed" (`"paused": true` in normalized journals), and a resume is a `Resume` record naming the step and the attempt it resumes after. Neither is written on the successful step path. ## Schedules ```ts const s = await engine.schedule("nightly-report", { cron: "0 2 * * *", timezone: "Europe/London", overlap: "skip" }, "report", input); await s.pause(); await s.resume(); await s.trigger(); await s.backfill(new Date("2026-11-01"), new Date("2026-11-07")); await s.describe(); await s.delete(); ``` A schedule is itself a Fusee workflow (`fusee.schedule/`) built on durable timers, so there is no separate scheduler. Cron has five fields plus aliases (`@daily`, `@hourly`...); `every` sets a fixed interval. Each run's workflow ID is the schedule name and its scheduled time (`nightly-report@2026-11-09T02:00:00Z`) and attaches to an existing run with that ID, so one time never fires twice, even across crashes and backfills. - Overlap: `skip` while the previous run runs, `buffer-one`, `allow-all`, or `cancel-other` (cancel the running one, then start). - `jitter` delays each start by a recorded random amount; after downtime, missed times within `catchupWindow` (default one year) still start. - Runs a schedule starts are abandoned by it: deleting the schedule leaves them running. - The schedule workflow is the same native code in every SDK and continues as new every 50 actions. ## Search attributes ```ts ctx.setAttributes({ customerId, status: "paid", amount }); engine.list({ where: { status: "paid", amount: { gte: 100 }, type: "checkout" } }); ``` Attributes are recorded in the journal (`null` removes a key) and are never encrypted. A filter matches the run's `type`, `status`, `id`, `startedMs` and any attribute, with equality or `eq`, `ne`, `gt`, `gte`, `lt`, `lte`. Each shard answers from memory for loaded runs and from the run store for parked and finished ones; a finished run keeps its merged attributes there even without `keepHistory`. The engine merges the shards' answers. ## Concurrency and rate limits ```ts engine.limit("stripe", { concurrent: 10, perSecond: 100 }); await ctx.step("charge", () => stripe.charge(...), { limit: "stripe" }); ``` A step with a limit takes a permit after its gate and before each attempt, and returns it when the attempt ends. In one process every shard shares one counting semaphore and token bucket. In server mode the server persists each definition and renews held permits on its monotonic clock. The replicated primary records definitions, grants, renewals and releases in the quorum log. A successor waits for prior leases to lapse. The concurrency bound counts valid leases. Work already dispatched to an external service can continue after its lease expires. `engine.limit_waiting(name)` (Rust) counts this process's steps waiting for a permit. ### Priority and fairness ```ts await ctx.step("charge", () => stripe.charge(...), { limit: "stripe", priority: 10, fairnessKey: tenantId, fairnessWeight: 2, }); ``` Steps waiting for a limit's permit are served in this order: 1. **Priority**, higher first (default 0, any 32-bit integer). A lower level waits while a higher one has waiters, so a steady stream of urgent work can hold background work back indefinitely; that is what strict priority means. 2. **Fairness key**, within one level: waiters are grouped by key and the groups take turns (round-robin), a key taking up to its weight in permits per turn (default 1; the latest waiter of a key sets it). A tenant with ten thousand queued steps gets one permit in each round, the same as a tenant with one. A key that runs out of waiters leaves the rotation and rejoins at the back. Steps without a key share one group. 3. **Arrival order** within a key. With neither option every step is in one group at one level, which is the arrival order limits always had. The rate limit applies to whoever is next: when the bucket is empty, the next waiter waits for the refill and the others wait behind it. The order is not journaled, so adding or changing a priority never changes a journal. A waiting step holds no permit and no record; after a crash its code reaches the step again with the same options and asks again, at the same priority and key, behind the waiters that asked first. In server mode the server keeps the queue (every waiting ticket with the priority it was first asked with) and decides the order, so the same rules hold across worker processes; a worker asks again only for its own best waiting ticket, and the server may answer with whichever of that worker's tickets it serves next. Replicated groups record the priority with each request in the quorum log, and checkpoints carry the queue with its rotation. ```python await ctx.step("charge", lambda: stripe.charge(...), limit="stripe", priority=10, fairness_key=tenant_id, fairness_weight=2) ``` In Rust: `StepOptions::default().limit("stripe").priority(10).fairness(&tenant).fairness_weight(2)`. `engine.limitInUse(name)` / `limitWaiting(name)` (TypeScript), `limit_in_use` / `limit_waiting` (Python and Rust) report a limit's held permits and this process's waiting steps. Priority and fairness order the steps waiting for one named limit, shared by every step that calls one external resource, whichever workflow type or shard it runs on; there is no task queue in between because the engine runs the step itself. Fairness keys share one bound between keys; use one limit per key for a per-tenant cap. ## Runs at once ```ts const report = workflow("report", { concurrency: { limit: 2, key: (input) => input.tenant } }, async (ctx, input) => { // At most two reports per tenant run at a time; the rest wait their turn. }); ``` ```python @fusee.workflow("report", concurrency=fusee.Concurrency(2, key=lambda input: input["tenant"])) async def report(ctx, input): ... ``` At most `limit` runs of the type, or of each key, run at a time; the others wait in the order they were started and run as turns free up. In Rust, `engine.limit_runs("report", 2)` and `.concurrency_key(tenant)` on the start; over the HTTP API, `options.concurrencyKey`. A waiting run is a slot run, `fusee.queued/`, until its turn: the UI lists it as queued, and `result(id)` waits through the queue. The type's (or key's) coordinator, `fusee.concurrency/[/]`, keeps the queue and the runs holding a turn in its journal, so a run keeps its turn while it sleeps for a week and across crashes, and each waiting run is admitted once. Nothing of this touches a step: it costs a few records per run, at its start and its end. - Starts from outside a workflow take turns (an engine, a client, the HTTP API); runs a workflow starts do not, including a child, a schedule's run and a delayed start. - Cancelling a run that waits takes it out of the queue. Terminate the run itself, not its slot, once it runs. - The newest start's `limit` applies, so a deploy can change it. - [Limits](#concurrency-and-rate-limits) bound steps, not runs: use them for a rate-limited API, and runs at once for how many runs work at a time. ## Debounce ```ts const run = await reindex.debounce("doc-42", doc, { window: "5s", maxWait: "1m" }); await run.started(); // the run exists once the window closes const out = await run.result(); ``` ```python run = await reindex.debounce("doc-42", doc, window="5s", max_wait="1m") await run.started() out = await run.result() ``` Starts of a workflow type with the same key inside the window collapse into one run. The run begins when the window closes: `window` after the latest start (trailing edge), or `maxWait` after the first start, whichever comes first (without `maxWait` a steady stream of starts delays it indefinitely). The run's input is the latest start's; the latest start's window settings apply. Each key of each workflow type has its own window. **What the caller gets back.** The call resolves once the start is durable, with the handle of the run that will take it: its workflow ID (`@`, chosen when the window opens) is known at once, but the run exists only when the window closes. Until then `status` is empty and `result` fails with `NotFound`; `started()` waits. Every start that joined the window gets the same handle, although only the latest start's input is used. **Across crashes.** Each key has a coordinator, a built-in workflow (`fusee.debounce`, ID `fusee.debounce//`) made from existing records, as schedules are. A start is an update sent with update-with-start: one record starts the coordinator if needed and delivers the start; the update's handler puts it in the open window and returns the receipt; a durable timer closes the window; a child start (abandoned by its parent) begins the run. So: - an acknowledged start is part of exactly one window, across any crash: the handler runs again at its place in the coordinator's journal; - a window starts one run: its ID was recorded when the window opened, and the child start is sent again after a crash with that ID and attaches; - a start that fails was not counted, unless the failure came after the start became durable (the caller's own process crashed, or a network reply was lost). Pass an event ID (`id`) to make such a retry safe: the key answers a repeated ID with the first receipt and adds nothing, for `dedupWindow` (default ten minutes) after that event, and for its latest 1,024 IDs. The coordinator ends once no window is open and no remembered ID is younger than its dedup window, so idle keys cost nothing; it continues as new every few hundred journal operations, carrying the open window and the remembered IDs. A start that reaches a coordinator as it ends is refused before its handler runs and the client sends it to the next one, which is what `debounce` does. A coordinator is an ordinary run: `inspect`, listings and the UI show it, and its `describe` query returns the open window and its due time. Terminating a coordinator drops its open window. `fusee export --sample` leaves coordinators out, as it does schedules. In Rust: `wf.debounce(key, &input, &Debounce::new(window).max_wait(cap))` or `debounce_with_id`. The event ID is `{ id }` in TypeScript and `id=` in Python, and the dedup window `dedupWindow` / `dedup_window`. ## Batching ```ts const run = await ingest.batch("tenant-7", event, { maxEvents: 100, maxWait: "2s" }); ``` ```python run = await ingest.batch("tenant-7", event, max_events=100, max_wait="2s") ``` Events for a workflow type with the same batch key accumulate until the batch holds `maxEvents` events or `maxWait` has passed since its first event; then one run starts with every event, in the order they were accepted, as its input (an array; at most 10,000 events). The call resolves once the event is durable, with the handle of the batch's run, as for debounce. The durability rules are the debounce ones: an acknowledged event is in exactly one batch's input, once; a batch starts one run; a retried event with the same `id` is added once. The coordinator type is `fusee.batch`, ID `fusee.batch//`. A full batch starts its run from the update's handler, so a batch never holds more than `maxEvents`. Events can come from anywhere a start can: callers outside the engine, and workflow code through a step (`ctx.step("enqueue", () => ingest.batch(key, e, { id: ctx.key() }))`), where the step's idempotency key makes the event exactly once despite the step's at-least-once retries. In Rust: `Workflow, O>::batch(key, &event, &Batch::new(max, wait))` or `batch_with_id`. ### In every SDK The coordinators are native code shared by every SDK, so the SDKs add only calls, no workflow code. Each call is update-with-start on the key's coordinator, built by `fusee::queue::request` and sent again when the coordinator it reached was ending ("the workflow ended before the update finished"): | | Rust | TypeScript | Python | Go | Java | | --- | --- | --- | --- | --- | --- | | Debounce | `wf.debounce(key, &input, &Debounce)`, `debounce_with_id` | `wf.debounce(key, input, { window, maxWait, dedupWindow, id })` | `await wf.debounce(key, input, window=, max_wait=, dedup_window=, id=)` | `wf.Debounce(ctx, key, input, fusee.DebounceOptions{Window, MaxWait, DedupWindow, ID})` | `wf.debounce(key, input, DebounceOptions.window(d).maxWait(d).dedupWindow(d).id(id))` | | Batch | `wf.batch(key, &event, &Batch)`, `batch_with_id` | `wf.batch(key, event, { maxEvents, maxWait, dedupWindow, id })` | `await wf.batch(key, event, max_events=, max_wait=, dedup_window=, id=)` | `wf.Batch(ctx, key, event, fusee.BatchOptions{MaxEvents, MaxWait, DedupWindow, ID})` | `wf.batch(key, event, new BatchOptions().maxEvents(n).maxWait(d).dedupWindow(d).id(id))` | | Wait for the run | `RunHandle::started()` | `run.started()` | `await run.started()` | `run.Started(ctx)` | `run.started()` | | From another process | `Client::update_with_start` with `queue::request` | `client.debounce(type, key, input, opts)`, `client.batch(...)`, `remoteRun.started()` | `client.debounce(type, key, input, ...)`, `client.batch(...)`, `remote_run.started()` | `client.Debounce(ctx, type, key, input, opts)`, `client.Batch(...)`, `remoteRun.Started(ctx)` | `client.debounce(type, key, input, opts)`, `client.batch(...)`, `remoteRun.started()` | They work from a `Client` through a server (or a replicated server group) whatever language the worker is written in, and with worker threads in TypeScript (`Engine.open({ workers })`): the main thread routes each event to the worker that owns its coordinator's ID, and each run starts on the worker that owns its own ID. The conformance suite checks that a debounced start and two batches journal identically in every SDK. ## Objects (keyed state) ```ts import { shared } from "@bitsinfra/fusee"; engine.object("cart", { // Exclusive (the default): one call at a time per key, in arrival order. add: async (ctx, item) => { const price = await ctx.step("price", () => pricing.quote(item)); const items = ctx.state.get("items") ?? []; items.push({ ...item, price }); ctx.state.set("items", items); return items.length; }, // Shared: runs at once, next to other calls; reads only. get: shared(async (ctx) => ctx.state.get("items") ?? []), }); const cart = engine.objectRef("cart", "user-7"); const n = await cart.call("add", item, { id: "req-42" }); // the handler's result await cart.send("add", item); // once it is accepted const state = await cart.state(); // { items: [...] } // In a workflow or another object's handler: durable, exactly once. const m = await ctx.object("cart", "user-7").call("add", item); ctx.object("cart", "user-7").send("add", item); ``` An object type is a set of handlers; each key of the type is one object with its own key/value state, built from Fusee's own records: a key is a long-lived run and each call is an update of it, so it needs no new kind of storage and works in every mode (embedded, worker threads, server, replicated) and through every client. **A key is a run.** The first call to a key creates a run with workflow ID `/` and workflow type ``, marked as an object in its `Start` record. It never finishes on its own (terminating it deletes the key: the next call starts with empty state). `inspect`, listings, export and the web UI show it like any run; when it is idle it is parked like a suspended run (a table entry under 48 bytes and a store frame of one record). An ID held by a run that is not an object of that type refuses the call. **Handlers.** - *Exclusive* handlers (the default) run one at a time per key, in the order their calls reach the key's shard, and may read and change state. - *Shared* handlers run as soon as they arrive, concurrently with each other and with a running exclusive handler. They read state and cannot change it. - Both are workflow code: `ctx.step`, `sleep`, `awaitCompletion`, child workflows, calls to other objects and streams all work, with the usual replay rules. Each call is an update of the key's run: it starts at its place in the journal and its operations replay at the same positions. Handlers have no validators. **State.** JSON values under string keys: `get(key)` (absent: `None` in Rust and Python, `undefined` in TypeScript), `set(key, value)` (null clears), `clear(key)`, `clearAll()`, `keys()`. Reads are synchronous: the engine holds the key's state next to its run. - An exclusive handler sees its own writes at once. They take effect for everyone else when it returns: its `UpdateDone` record carries them with its result, so the change and the result are durable together, and released together. A handler that fails (an exception, or a step that fails for good) changes nothing. - A shared handler sees the state as last committed, never a running handler's uncommitted writes. - Reads are not journaled. Commits happen when handlers return, at positions the journal fixes, so a handler replayed after a crash reads exactly what it read the first time. **When a call is durable, and exactly once.** A call is *accepted* when its handler starts: its `Update` record is appended. From then on it runs to completion exactly once across crashes and its result is released once durable. An exclusive call that arrives while another runs waits for the key in memory at the key's shard, in arrival order; a crash drops waiting calls (none of them had started) and their callers send them again: - **From a client** (`call`): pass an idempotency ID. A call sent again with the same ID, whether the first one was waiting, running or done, runs once and every attempt gets the same recorded result. A key remembers the IDs and results of its calls with IDs while they are in its current run, and across compaction (below) its latest 1,024 for an hour after the compaction that kept each. Without an ID a retry is a new call. `send` is a call whose caller does not take the result: it resolves once the call is accepted. - **From a workflow or a handler** (`ctx.object(type, key).call`): the call is a journal operation, recorded as a `Child` record flagged as a call, and goes through the durable outbox like a child start. The outbox sends it again until the key accepts it, the key recognizes it by an ID made from the caller's journal position, and the result comes back to the caller's journal as a `ChildDone` (the key's `Update` record names its caller, so the answer is sent again after a crash on either side). So the handler runs exactly once and its result is recorded once. `ctx.object(...).send` is the same without waiting: it returns at once and the outbox delivers it. **Compaction.** A key's history is bounded. When no handler is running and either the key's run has made 128 journal operations or it is about to be parked, the engine continues the run as new with a snapshot as the new run's input: the committed state and the IDs and results of recent calls, in one record. Calls waiting for the key move to the new run. So loading a parked key, recovering one, or reading its state never replays more than about 128 operations of handler code, and an idle key's journal is one record. Reading a key's state never runs code: it is the snapshot plus the changes recorded since. A new version of the object's code takes over at the next compaction; calls running at a deploy replay against the new code as workflow updates do. **Deadlocks.** An exclusive handler that calls an exclusive handler of its own key, or keys whose exclusive handlers call each other in a cycle, wait forever. Use `send` for such calls, or a shared handler. **Cost.** A call is an update: an `Update` and an `UpdateDone` record (which also carries the state changes) and one flush before the result is released, plus a compaction record every 128 operations or idle period. **Code that is not registered.** A call from workflow code to a type no process has registered waits (the outbox sends it again each second) until one does; register object types right after the engine opens. **Reading from outside.** `state()` returns the committed state, once it is durable (as a query). `inspect()` returns the key's run with its journal, its state and its pending calls (running and waiting). `engine.objects(type)` lists a type's keys. | | Rust | TypeScript | Python | | --- | --- | --- | --- | | Define | `engine.object("cart").exclusive("add", h).shared("get", h).register()` | `engine.object("cart", { add, get: shared(get) })` | `cart = engine.object("cart")`, `@cart.exclusive`, `@cart.shared` | | Handler | `\|ctx: ObjectCtx, arg: A\| async move { ... }` | `async (ctx, arg) => ...` | `async def add(ctx, arg)` | | State | `ctx.state().get::(k)?`, `set(k, &v)?`, `clear(k)?`, `clear_all()?`, `keys()?` | `ctx.state.get(k)`, `set(k, v)`, `clear(k)`, `clearAll()`, `keys()` | `ctx.state.get(k)`, `set(k, v)`, `clear(k)`, `clear_all()`, `keys()` | | The key | `ctx.object_key()` | `ctx.objectKey` | `ctx.object_key` | | Call from outside | `engine.object_ref("cart", k).call_with_id::<_, R>("add", &item, id)` | `engine.objectRef("cart", k).call("add", item, { id })` | `await engine.object_ref("cart", k).call("add", item, id=...)` | | Send from outside | `.send("add", &item)` | `.send("add", item, { id })` | `await ref.send("add", item, id=...)` | | Call from code | `ctx.object("cart", k).call::<_, R>("add", &item).await?` | `await ctx.object("cart", k).call("add", item)` | `await ctx.object("cart", k).call("add", item)` | | Send from code | `ctx.object("cart", k).send("add", &item)?` | `ctx.object("cart", k).send("add", item)` | `ctx.object("cart", k).send("add", item)` | | Read | `.state().await?`, `.inspect().await?`, `engine.objects("cart", n)` | `.state()`, `.inspect()`, `engine.objects("cart")` | `await ref.state()`, `ref.inspect()`, `engine.objects("cart")` | | From another process | `Client::object_call`, `object_send`, `object_state` | `client.objectRef("cart", k)` | `client.object_ref("cart", k)` | | | Go | Java | | --- | --- | --- | | Define | `fusee.RegisterObject(e, "cart", fusee.Handlers{"add": fusee.Exclusive(add), "get": fusee.Shared(get)})` | `e.registerObject("cart", Map.of("add", ObjectHandler.exclusive(Item.class, add), "get", ObjectHandler.shared(Object.class, get)))` | | Handler | `func(c *fusee.ObjectCtx, arg A) (R, error)` (steps: `fusee.Step(c.Ctx, ...)`) | `(ObjectCtx c, A arg) -> ...` (steps: `c.ctx().step(...)`) | | State | `fusee.StateGet[T](c.State(), k)`, `Set(k, v)`, `Clear(k)`, `ClearAll()`, `Keys()`, `Raw(k)` | `c.state().get(k, T.class)`, `set(k, v)`, `clear(k)`, `clearAll()`, `keys()`, `raw(k)` | | The key | `c.ObjectKey()` | `c.key()` | | Call from outside | `fusee.CallRef[R](ctx, e.Object("cart", k), "add", item, fusee.CallOptions{ID: id})` | `e.object("cart", k).call("add", item, R.class, CallOptions.id(id))` | | Send from outside | `ref.Send(ctx, "add", item, fusee.CallOptions{ID: id})` | `ref.send("add", item, CallOptions.id(id))` | | Call from code | `fusee.Call[R](c.Object("cart", k), "add", item)` | `ctx.object("cart", k).call("add", item, R.class)` | | Send from code | `c.Object("cart", k).Send("add", item)` | `ctx.object("cart", k).send("add", item)` | | Read | `ref.State(ctx)`, `ref.Inspect(ctx)`, `e.Objects(ctx, "cart", n)` | `ref.state()`, `ref.inspect()`, `e.objects("cart", n)` | | From another process | `client.Object("cart", k)` | `client.object("cart", k)` | State reads are synchronous and not journaled (the exclusive order and the journal's commit points make them replay-safe). A call waiting for its key is held at the key in memory rather than in a durable queue, so it is durable at its caller (a workflow's outbox, or a client retrying with the same ID) until it starts; a client's `send` resolves once the call has started. The engine keeps the lock and the state, deduplicates calls across continue-as-new and compacts a key's journal. ### Editing state from outside A key with bad state can be repaired without a handler written for it: ```ts const cart = engine.objectRef("cart", "user-7"); await cart.setState("note", "gift wrap"); await cart.patchState([{ op: "remove", path: "/items/0" }]); // JSON Patch, RFC 6902 await cart.clearState("coupon"); // one field; none named: all of it const { version } = cart.inspect().object; await cart.editState({ set: { owner: "ana" } }, { ifVersion: version }); ``` ```bash fusee objects set cart user-7 note '"gift wrap"' fusee objects patch cart user-7 '[{"op": "remove", "path": "/items/0"}]' --dry-run fusee objects clear cart user-7 coupon fusee objects edit cart user-7 # $EDITOR on the state; what changed is applied ``` - **An edit is an exclusive call.** Every object type has the handler `fusee.state.edit`, which the engine runs: it waits for the key like any exclusive call, and its changes commit with its result in the `UpdateDone` record a handler's writes make. It replays, deduplicates by call ID, is sealed by the codec and compacts as a handler's call does, and handlers that run after it see the new state. - **The edit**, applied in this order: `clearAll`, `clear` (fields; absent ones are ignored), `set` (fields and values; `null` removes one), `patch` (a JSON Patch over the state as one object). A patch that does not apply changes nothing and names the operation. The answer is `{version, changed}`. - **Versions.** A key's version is a hash of its state (`version` in `inspect()`'s object detail and in `fusee objects show`). `ifVersion` (`--if-version`, HTTP `If-Match`) refuses an edit if the state changed since that version was read, so an editor never overwrites a change it did not show. - **HTTP.** `PUT /v1/objects/{type}/{key}/state/{field}` (the value), `DELETE /v1/objects/{type}/{key}/state/{field}`, `PATCH /v1/objects/{type}/{key}/state` (a JSON Patch), `POST /v1/objects/{type}/{key}/state` (an edit), `DELETE /v1/objects/{type}/{key}/state` (all of it); a stale `If-Match` answers 412. The protocol call is `editState`. The UI's Objects screen edits a key's state as JSON, shows the fields the edit changes, and saves it with the version it was read at. - **Who asked.** Edits through the HTTP API, a server or the CLI carry the requester (`by`, as in [Audit](#audit)) in the edit, which the key's `Update` record keeps. A read token cannot edit; a namespace token edits only keys whose ID is under its namespace. The same calls in every SDK: Rust `ObjectRef::edit_state`, `set_state`, `patch_state`, `clear_state`; Python `edit_state`, `set_state`, `patch_state`, `clear_state` (`if_version=`); Go `EditState`, `SetState`, `PatchState`, `ClearState` (`StateEdit{IfVersion}`); Java `editState(edit, ifVersion)`, `setState`, `patchState`, `clearState`. ## Payload encryption ```ts const engine = await Engine.open({ dir, codec: aesGcm({ k2: newKey, k1: oldKey }) }); ``` Workflow inputs and results, step results, signals, update arguments and results, child inputs and heartbeat progress are encrypted with AES-256-GCM inside the application before they become records, so the log, checkpoints, the run store, replicas and a Fusee server hold only ciphertext. Record headers (IDs, names, positions), error messages and search attributes stay readable. Each payload carries its key ID: the first key encrypts and every listed key decrypts, so keys rotate by adding a new first key. In Rust: `Config::local(dir).codec(fusee::aes_gcm([("k1", key)]))`; in Python: `Engine.open(dir, codec=aes_gcm({"k1": key}))`. The HTTP API and `fusee runs show` show an encrypted payload as `{"$encoded": ""}`, and the UI as *encrypted*. To read them in the UI, run a codec endpoint where the keys are and point the server at it: ```sh fusee codec-server --keys-file keys.txt --allow-origin https://fusee.example.com fusee server --dir /var/lib/fusee --http 0.0.0.0:7601 --token-file token --codec-url http://127.0.0.1:7602 ``` `keys.txt` holds lines of ` <64 hex digits>`. The browser sends the payloads it shows to the endpoint (`POST /decode` with `{"payloads": [base64, ...]}`, answered with the decoded payloads, `null` for one it cannot decode), so the keys and the plain payloads never reach the Fusee server. Only pages from `--allow-origin` may call it, and it listens on loopback unless it sits behind a proxy that signs people in (`--insecure`). Any service that answers the same call works as an endpoint, for keys held in a KMS. ## Audit Signals, cancellations, terminations and resets record who asked, in the journal record they write (`by`), and the UI lists them under a run's Diagnostics. The HTTP API names the requester: the user a proxy in front signed in (`fusee server --identity-header X-Forwarded-User`), else the token (`token`, `token:client`, `token:namespace:acme`), else `http`. The CLI names its user (`cli:ana`) when it talks to a server or a data directory itself; through an HTTP API the API's name wins. In code: ```ts await asRequester("job:nightly", () => engine.run("order-1").cancel("expired")); ``` ```python with fusee.as_requester("job:nightly"): await engine.run("order-1").cancel("expired") ``` ```rust fusee::audit::as_requester("job:nightly", engine.run::("order-1").cancel("expired")).await?; ``` Schedule changes are signals and terminations of the schedule's workflow, so they are recorded the same way. A request that names nobody writes its record exactly as before. Over the server protocol a client names itself: trust the name as far as you trust the client's token. ## Replay testing ```bash fusee export ./data order-123 -o order-123.fjnl fusee export --sample 1000 ./data -o journals/ fusee show order-123.fjnl ``` ```ts const r = await new Replayer().workflow("checkout", checkout).replayDir("journals"); // [{ workflowId, ok: false, error: "... recorded step 'charge' here but the code called step 'refund'" }] ``` `fusee export` copies the data directory (checkpoints first, so a copy taken while the application runs is consistent), recovers the copy and writes each run's journal. `--sample` takes the most recent runs whose whole journal is available: running and suspended runs, and finished runs of engines with `keepHistory`. `engine.exportJournal(id)` does the same from a running engine. A `Replayer` runs new code against exported journals with no step function or timer running, and reports the first difference by step name. Run it in CI before a deploy. Encrypted journals need the codec: `new Replayer({ codec })`. ## Workflow ID and child policies ```ts await checkout.start("order-123", order, { idPolicy: { running: "use-existing", finished: "allow-if-failed" }, }); ctx.spawnChild("audit", `${ctx.workflowId}/audit`, data, { parentClose: "abandon" }); ``` - Starting an ID that is running fails by default (`AlreadyExists`). It can return the running workflow (`use-existing`) or terminate it and start a new run (`terminate-existing`). - Starting an ID whose last run finished within retention starts a new run by default. It can be allowed only if that run failed (`allow-if-failed`), refused (`refuse`), or answered with that run (`use-existing`). - A child's parent close policy is `terminate` (the default), `request-cancel` or `abandon`, for children the parent waits for and for spawned ones alike. A close that reaches a child's shard before the child's start records the child as closed, so the late start does not run it. - The policies are recorded with the start, so replay and recovery apply them the same way. ## Across threads (TypeScript) One JavaScript thread runs one engine's workflow code. To use more cores in one process, open the engine with worker threads: ```ts // main.ts const engine = await Engine.open({ dir, workers: 4, module: new URL("./workflows.js", import.meta.url) }); const order = engine.workflow("order"); // a handle: the code is in the module const run = await order.start("order-42", { amount: 42 }); console.log(await run.result()); // workflows.ts: each worker imports it and calls the default export export default (engine: Engine) => { engine.workflow("order", async (ctx, input) => ctx.step("charge", () => charge(input))); }; ``` - Each worker thread owns a set of shards (shard `s` belongs to worker `s % workers`), holds its own copy of the workflow code, and runs those shards' workflows. All workers append to the data directory's one shared log, so their steps share flushes as shards on one thread do. - The thread that opened the engine holds no shards: it routes `start`, `signal`, `update`, `query`, `cancel`, `terminate`, `reset`, `result`, `complete` and `fail` to the worker that owns the workflow ID's shard, and asks every worker for `list`, `inspect`, `exportJournal` and `shardMetrics` (these block until each worker answers). A limit set with `engine.limit` holds across all workers. - Children, signals and cancellations between workflows on different workers go through the same durable outbox as between shards, delivered to the owning worker's inbox. Code inside a worker may call its own engine for any ID: a call for an ID another worker owns goes there through the main thread. The synchronous reads are the exception: in a worker, `list` and `shardMetrics` cover that worker's shards, and `inspect` and `exportJournal` refuse another worker's IDs. - A new directory gets one shard per worker unless `shards` is given; the count is then fixed, and there are at most as many workers as shards. A directory opens with any worker count, or on one thread with `shards` set to its count: after a crash every shard recovers in whichever worker (or thread) now owns it. - `engine.close()` closes every worker's shards (with the usual checkpoint) and stops the threads. Query arguments and answers cross threads with the structured clone algorithm; other values cross as JSON or bytes, as they are recorded. `telemetry` is not available with workers; local and memory durability only. Node.js and Bun. ## Across processes In server mode workflow code runs in the worker that holds the server's lease. Other processes connect with `fusee_server::Client` and send `complete`, `fail`, `update`, `query`, `signal`, `start`, `result`, `cancel` and `terminate`; the server relays each call to the worker, which runs it against its engine (`fusee_server::serve_calls`). Listings fan out across the engine's shards. On a failover (server standby or replicated backup) runs resume from the journal: schedules keep firing, accepted updates whose handlers had not finished run again from their records (a retried call with the same update ID gets the result), and persisted limit definitions continue to apply while prior permits expire. # Errors Every SDK raises the same errors, named by the same codes. Each error carries its `code`, the IDs involved, and `docs`, a link to its section here. A failure that application code raised (a step's, a workflow's, an update's) travels as a [failure envelope](#the-failure-envelope), so its type, code and details reach the caller in any language. | Code | Raised when | Retried by a step's policy | | --- | --- | --- | | [`NotFound`](#notfound) | no run has this workflow ID | — | | [`AlreadyExists`](#alreadyexists) | a start's ID is taken and its ID policy refuses | — | | [`UnknownWorkflow`](#unknownworkflow) | no code is registered for the type | — | | [`StepFailed`](#stepfailed) | a step failed after its last attempt | — | | [`WorkflowFailed`](#workflowfailed) | a workflow, update or object handler failed | — | | [`ChildFailed`](#childfailed) | a child workflow failed | — | | [`Cancelled`](#cancelled) | cancellation reached the operation | yes | | [`TimedOut`](#timedout) | a step, wait or workflow deadline passed | yes | | [`Terminated`](#terminated) | the run was terminated | no | | [`Determinism`](#determinism) | the code no longer matches the run's journal | no | | [`Rejected`](#rejected) | a request was refused and nothing was recorded | no | | [`Codec`](#codec) | a value cannot be encoded or decoded | no | | [`Connection`](#connection) | no connection to a Fusee server | yes | | [`Io`](#io) | a disk or network failure | yes | | [`Busy`](#busy) | the shard's queue is full | yes | | [`Closed`](#closed) | the engine is closed | no | ## The failure envelope A failed attempt, step, update, object call or workflow records its failure as JSON, with absent fields left out: ```json { "message": "card declined", "type": "CardDeclined", "code": "card_declined", "details": { "last4": "4242" }, "retryable": false, "stack": "...", "cause": null } ``` - `type` is the error's class or type name, left out for a language's plain error type. Fusee's own errors use their code (`StepFailed`, `ChildFailed`, ...) so every SDK reads them back as the same error. - `code` and `details` are the application's: set them on the error you throw (TypeScript, Python) or on the `Failure` you return (Rust). - `retryable: false` makes the failure final: no retry policy retries it. - `stack` is recorded by default, cut to 4 KiB; turn it off with `recordStacks: false` where stacks must not be journaled. Rust records none unless asked. - `cause` is the failure's cause, with the same fields. A step's failure that ends a workflow is its `StepFailed` envelope with the step's failure as `cause`, so `ChildFailed` → `StepFailed` → the original error survives any number of levels. A successful step records nothing of this: the envelope costs only failures. ## Retries A step's failure is retried under its retry policy (three attempts by default, about 100 ms and 200 ms apart, with jitter) unless: - the failure is marked final: `FatalError` (TypeScript), `fusee.Fatal` (Python), `Failure::fatal` (Rust), or any error with `retryable = false`; - its type, or a cause's, is in the policy's `nonRetryable` list; - the policy's `retryIf(failure, attempt)` returns false; - the next retry would start after the policy's `maxElapsed` from the first failure; - the step's result cannot be encoded ([`Codec`](#codec)), since every attempt would fail the same way. The delay taken is recorded, so a replay waits for the same time. ## NotFound No run with this workflow ID is known: it was never started, or its retention ended. Check the ID; `list` shows the runs that exist. ## AlreadyExists A start named a workflow ID that is taken, and the start's ID policy refuses. The message names the run that holds the ID and whether it is still running. To attach to a running run instead (a client that retries a start should), start with the ID policy `running = use-existing`. To run a finished ID again, use `finished = allow`. ## UnknownWorkflow No code is registered for the workflow type in this process. The message lists the registered types and the nearest one when the name looks like a typo. Register the workflow before starting it, or start it where its code runs. Where another worker may hold the code (server mode, worker threads), the run waits with status `waiting_for_code` instead. ## StepFailed A step failed after its last attempt, or with a failure that is not retried. The error carries the step's name and its failure (`cause`), with the type, code and details the step's error had. ## WorkflowFailed The workflow, an update or an object's handler failed with this failure. Its fields are the error the code raised. ## ChildFailed A child workflow failed. The error carries the child's workflow ID and its failure as `cause`. ## Cancelled Cancellation reached this operation, with the reason given to `cancel`. Workflow code may catch it to compensate. ## TimedOut A timeout passed: a step attempt's `timeout` or `heartbeatTimeout`, a wait's timeout, or the workflow's deadline. Step timeouts are retried under the step's policy. ## Terminated The run was terminated. Nothing runs after it, not even compensation. ## Determinism The run's code makes different operations than its journal records: a step was renamed, added or removed before others, or the code reads the clock or randomness without going through the context. The run stops (status `stuck`), one log line names the difference, and it waits, unchanged, until one of these: 1. deploy code that matches its journal; 2. guard the change with `patched("id")`, so runs that started before it take the old path; 3. reset the run to before the step that differs. Under the `dev` profile and in tests, `result()` rejects with this error at once instead of waiting. ### key() outside a step `key()` names the step whose function is running. Called anywhere else there is no step to key, and every such call would get the same key, so unrelated effects would look like retries of one another. The run stops with this message. Call it inside the step's function; in TypeScript and Python, use the step context's `s.key`. ## Rejected A request was refused and nothing was recorded: an update's validator refused it, a completion token is stale, a schedule's spec is invalid (the message names the schedule), or an argument is out of range. ## Codec A value could not be encoded or decoded. A step whose result cannot be encoded fails at once, without retries, naming the step (and in TypeScript and Python, the path to the value). Return plain data: objects, arrays, strings, numbers, booleans and null. For dates, maps, sets and big integers in TypeScript, open the engine with the `rich` serializer: `Engine.open({ serializer: rich })`, with `import { rich } from "@bitsinfra/fusee/rich"`. ## Connection No Fusee server answers at the address, or the connection was lost. The message names the address. Start one with `fusee server --dir ...`, or check `FUSEE_URL`. ## Io A disk or network failure. The message names the path. After a failed flush the shard stops, since acknowledging work it could not make durable would break the guarantees. A directory that another process has open says so, with that process's pid: one process writes a directory; use server mode for several. ## Busy The shard's queue is full and nothing was recorded. Retry later, or raise the queue size. ## Closed The engine was closed. Open a new one. # Guarantees Stated exactly as they are tested (see [testing.md](testing.md)). 1. **An acknowledged start, signal or step result is never lost** within the durability mode's failure model. "Acknowledged" means the call returned (the promise resolved, the `await` completed). 2. **A recorded step result never changes**, and code after it sees the same value on every run. 3. **A step's function runs at least once.** It runs again only if the process failed before that step's result became durable. `ctx.key()` returns a stable workflow-and-step key; passing it to a service that accepts idempotency keys makes the step's effect exactly once. A step marked [read-only](features.md#read-only-steps) may also run again with different inputs, if the process failed before the records ahead of it were durable; mark only steps that change nothing outside the process. 4. **Nothing leaves the process based on state a crash could take back.** Acknowledgments, results returned to callers, refusals (for example "the workflow has already finished"), messages to other workflows and the inputs a step function sees are all released only after the records behind them are durable. The one exception is declared by the code: a read-only step sees its inputs at once, because it has no effect for a crash to contradict. Fusee does not weaken these guarantees anywhere. Speculating through external calls with rollback is not part of Fusee. ## Failure models | Mode | Survives | Does not survive | | --- | --- | --- | | `local` | process crash (SIGKILL, panic, OOM kill); power loss, if the drive honors flushes | loss of the machine or its disk; a drive that acknowledges flushes it has not done | | server | application crash; failover to a standby application | loss of the server's disk | | `replicated` | loss of a minority of replicas | loss of a majority | | `memory` | nothing | anything | **Drives.** Consumer drives without power-loss protection, and macOS without `F_FULLFSYNC`, can acknowledge data that is not on stable media. On macOS Fusee uses `F_FULLFSYNC`, the only flush macOS guarantees reaches the media; the `fsync` flush option exists for development and is labeled non-durable everywhere it appears. `fusee disk-check ` shows the flush latency of the disk under a directory. **Damaged logs.** Recovery distinguishes a torn, never-acknowledged write at the end of the log (discarded) from damage to acknowledged records (refused with an error naming the record, never silently truncated). In replicated mode a replica repairs damaged records from its peers. # Running it The same workflow code runs in five arrangements. Start with the first and move when you need what the next one gives. | | Where the journal lives | Code runs in | When | | --- | --- | --- | --- | | Embedded | a directory on the app's disk | the app | one process, or one per directory | | Server | `fusee server`'s disk | the app (workers) | several processes, one store | | Replicated | three or five servers | the app (workers) | surviving a lost machine | | Invoked | a server | functions it calls (Lambda, Cloud Run, Vercel) | nothing running while runs sleep | | Kubernetes | a StatefulSet of servers | the app's pods | the above, managed | ## Embedded ```ts const engine = await Engine.open({ dir: "./data", workflows: [checkout] }); ``` The engine is a library: no server, no network. A run's records go to the directory's log with one flush per batch; a crash loses nothing that was acknowledged. One process opens a directory at a time (a second one is told which process holds it). Use [durability modes](../durability.md) to choose the flush (`fsync`, `fullfsync` on macOS) and memory for tests. ## Server ```sh fusee server --dir /var/lib/fusee --listen 0.0.0.0:7600 --token-file /etc/fusee/token --http 0.0.0.0:7601 ``` ```ts const engine = await Engine.open({ durability: "server", address: "fusee.internal:7600?token-file=/etc/fusee/token", workflows: [checkout] }); ``` Workers lease shards from the server and run their runs; when a worker dies its shards go to another. Clients (`Client.connect`, the CLI, the HTTP API) reach any run through the server. Tokens and scopes: [deploying](../../deploy/README.md#who-can-reach-what). Details: [server mode](../server-mode.md). ## Replicated Three or five servers, each with its own disk, agree on every record before it is acknowledged; a majority must be up. Workers connect to the group as to one server and follow its primary. [Replication](../replication.md) covers setup, membership changes and what is measured. ## Invoked The server calls an HTTP endpoint when a shard has work and no worker: the function opens an engine for that shard, runs it until idle, and returns. Nothing runs while workflows sleep for a week. ```sh fusee server --dir /var/lib/fusee --listen 0.0.0.0:7600 --token-file /etc/fusee/token \ --invoke https://worker.example.com/invoke --invoke-secret-file /etc/fusee/invoke-secret ``` Each SDK has a handler that verifies the request and runs the shard ([server mode, invoked workers](../server-mode.md#invoked-workers)). ## Kubernetes `helm install fusee deploy/helm/fusee --set members=3` runs a replicated group as a StatefulSet with its token in a Secret; `deploy/terraform` does the same through Terraform, `deploy/systemd` runs one server on a host, and `deploy/monitoring` has alert rules and a dashboard over `/metrics`. See [deploying](../../deploy/README.md). ## Watching it `/metrics` (Prometheus), the UI on the `--http` address, `fusee runs ls --status stuck`, and the [operations](../operations.md) page: disk checks, backups, upgrades, traces. # Durability modes ## `local` Every shard of a data directory writes to one shared log (log format 3, in `/wal/`): whoever writes next takes every shard's pending records and writes them as one batch with one durable flush, so steps on different shards share writes instead of each paying for one. Directories created before log format 3 keep a log per shard (`Config::shared_log(false)` creates one). A batch is written and made durable with one call: | Platform | Flush | Notes | | --- | --- | --- | | Linux | `O_DIRECT` aligned writes, each carrying `RWF_DSYNC` (default, for the shared log and per-shard logs); `fdatasync`; or `io_uring` (feature `io-uring`, per-shard logs, buffered writes) | Where the filesystem refuses `O_DIRECT` (ramfs, tmpfs before Linux 6.6), or takes the open and then rejects a write with `EINVAL`, the log writes buffered from then on, rewriting that write exactly, and says so once on stderr: a per-shard log then flushes with `fdatasync`, and the shared log keeps `RWF_DSYNC` on its buffered writes. `LocalOptions::direct_io = false` selects buffered writes. `RWF_DSYNC` needs Linux 4.7 or later | | macOS | `F_FULLFSYNC` | plain `fsync` does not flush the drive cache | On a fast disk (last flush at most 200 µs) an idle shard writes the batch on its own thread when no other write is in flight, saving the thread handoffs; otherwise the log thread writes while code keeps running, with up to two writes in flight (each padded to blocks of its own, positions published in order). A shard may have a section in each of them: a busy shard submits its next records while its previous section is still being written, and learns of each position as it becomes durable. A shard writing on its own thread that finds a padded write in flight starts a second one beside it once its records have waited a quarter of a flush, instead of waiting for the first to hand it the next write. On Linux with `O_DIRECT` and `RWF_DSYNC` (and the default `io-uring` feature), the log thread and its I/O threads are replaced by one owner thread: whatever seals a batch queues it, and the owner submits each write through an `io_uring`, up to the in-flight limit, and reaps completions in the same system call, publishing each durable position as it lands. A batch that starts a segment is written after its header, without the ring; a write the kernel refuses with `EINVAL` is written again buffered, as above. Where `io_uring` is unavailable, or with `FUSEE_LOG_OWNER=0`, the threads stay. They stay on tmpfs too, where the kernel runs each such write on a worker thread, one at a time per file (`FUSEE_LOG_OWNER=1` takes the owner there anyway). Docker's default seccomp profile refuses `io_uring`, so a container keeps the threads unless its profile allows the three `io_uring` calls. What is written, and in what order positions publish, is the same either way. A roll to a new segment asks for a checkpoint only from shards whose oldest retained record is in the segment the log retires next (the others checkpoint on their interval), and while the log fills a segment at over 100 MB/s the next segments grow to 256 MiB, so a busy log neither rolls nor checkpoints every shard as often. **What recovery accepts.** Each batch carries its sequence, the checksum of the batch before it, the sequence its writer had acknowledged when it was submitted, and a value unique to that open of the log. Recovery follows the chain from the oldest segment and stops at the first batch that does not continue it; the next open writes right after that point. Bytes left by a torn write, by a write in flight at a crash, or by a segment's earlier life can never join the chain. A valid batch after the end whose acknowledged sequence reaches past the end proves that an acknowledged batch was lost, and the directory refuses to open. The rule is model-checked in `spec/FuseeLog.tla`. On macOS the default is one shard, because `F_FULLFSYNC` flushes the whole drive and shards' flushes serialize. **Batch size.** A batch carries at most 64 MiB, which is also the most recovery reads as one batch. A shard hands the log at most 8 MiB of records at a time (one record at least, whatever its size), and when more than a batch is staged before a write, because a flush stalled or many shards wrote large records at once, the log writes several batches: whole sections while they fit, then the leading records of the next section, cut at a record boundary, with the shards taking turns to go first. A backlog is therefore written and acknowledged in order, in bounded pieces, in `local`, server and replicated mode alike (a replicated entry takes at most 16 MiB of staged sections). **Large payloads.** A record whose canonical encoding is longer than `Config::payload_offload_bytes` (64 KiB by default for `Config::local`; 0 turns it off) does not go into a log batch. Its bytes go to the log's blob area (`/wal/blobs/`), and the batch carries a reference in its place: the record's kind, a blob file, an offset, a length and a CRC-32C of the bytes. A large step result, signal, update argument, start input or stream chunk therefore does not hold up the small records of every other shard behind one long write. - **Ordering.** The shard's own thread finds and checksums its large records before handing the section over, so below the threshold the cost is one comparison. The blob thread writes the bytes and syncs their file. The section that references them, and every later section of that shard, enters a batch only after that sync. The chain therefore never accepts a batch that names bytes that are not durable, and other shards' batches keep going out meanwhile. - **Layout.** Blob files are append-only and never reused. Each open of the log appends to a new file, and a file rolls over past a fixed size. Every blob starts on a 4 KiB block of its own, so whole-block `O_DIRECT` writes never touch another blob. - **Recovery.** Each reference is resolved and checked against its length, checksum and kind. A mismatch refuses to open, naming the record and the blob file. Blob files that no batch of the chain references are deleted at open. - **Retirement.** Blob files go with the log segments that reference them. A segment is retired once every shard's checkpoints (and the backup) cover it, and its header is zeroed durably before any blob file that only retired batches reference is deleted. - **What stays the same.** References exist only in log sections. Journals, run stores, checkpoints, archives, exports, `fusee inspect` and every SDK see the record's canonical encoding, so conformance journals are byte-identical with offload on or off. - **Where it applies.** A continuous backup uploads the blobs that a log chunk's batches reference, as one pack object per chunk, before any manifest names the chunk. A restore writes them back to `wal/blobs/`. `fusee reshard` reads the log with its references resolved. Server and replicated mode refuse a nonzero threshold, because their logs carry large records inside batches; `memory` mode ignores it. **What a backup adds.** `local` survives a process crash and power loss; it does not survive losing the disk. A continuous backup ([operations.md](operations.md#backup-and-restore)) copies the shared log and the checkpoints to object storage as they become durable, so a lost disk (or a bad deploy, or a deleted directory) costs at most what was acknowledged since the last upload: about `upload_interval` (1 s by default) plus the upload's own time, or the active log segment when `upload_interval` is `None`, and nothing after a clean shutdown. A restore to a point in time brings back exactly what was acknowledged at that moment, with the same guarantees as recovery after a crash then: nothing acknowledged by then is missing and nothing acknowledged later is there. What a backup does not protect: - Effects outside Fusee after the restore point. A step whose result was acknowledged after the point runs again after a restore (its idempotency key from `Ctx::key` is the same, so a service that honours it sees a retry). - Acknowledgments within the loss window when the disk is lost. - Directories in log format 2 (`fusee upgrade` converts them), server-mode directories and replicated directories. - The visibility index (`visibility.redb`): it is rebuilt after a restore. - A target written by two directories at once: a manifest write finds the other's manifest and that backup stops. ## Server `fusee server --dir ` holds the logs; applications open an engine with `fusee_server::remote_config("host:port")`. See [server-mode.md](server-mode.md). ## `memory` Nothing is written. For tests. ## `replicated` A primary-backup protocol (Viewstamped Replication style) with protocol-aware recovery: outputs and step gates wait until a quorum holds the records durably. Embedded: the application instances are the replicas and workflows run on the primary (`fusee_server::replicated::ReplicaNode`). See [replication.md](replication.md). # Server mode The same engine runs as `fusee server`. The server holds one log for every shard (log format 3, as in local mode) and each shard's run store and checkpoints on its disk; workflow code still runs in the application. The application's shard runtimes use a remote storage backend, so a flush is durable once the server flushed it: one pipelined round trip plus the server's flush. ```bash fusee server --dir /var/lib/fusee --listen 0.0.0.0:7600 --shards 4 ``` ```rust let engine = Engine::open(fusee_server::remote_config("10.0.0.5:7600")?).await?; ``` On one host, add `--unix /run/fusee.sock` and connect workers to `unix:/run/fusee.sock` to skip the TCP stack, or to `unix:/run/fusee.sock?shm=1` to send log records through shared memory (see [Shared memory](#shared-memory)). **Leases.** Each shard is leased on its own, with an epoch of its own. A worker takes every free shard unless its address says how many it wants at most (`host:port?max-shards=2`); for the shards it does not get it waits as a standby. When a holder's connection closes or goes silent, its shards go to the waiting workers with new epochs; each recovers them from the server's disk and resumes their workflows. Every request carries its shard's epoch; the server refuses stale epochs, so a deposed process can never append again. With `max-shards`, several workers run workflows against one server at once and a hung worker loses only its shards: ```rust // Two processes, each running two of the server's four shards. let (cfg, remote) = fusee_server::remote_worker("10.0.0.5:7600?max-shards=2")?; ``` A worker's engine opens without waiting for the shards another worker holds; calls to workflows on those shards wait until this worker takes the shard over. Send them through a [`Client`](#calls-from-other-machines), which the server routes to the shard's holder. Workers from before per-shard leases wait until every shard is free and take them all. **Messages between workers.** Child starts and results, signals and cancels from one workflow to another, and their acknowledgments, go through the server when the two shards are held by different workers. The server forwards them to the target shard's holder, or keeps them (up to 1,024 frames or 8 MiB per shard, the oldest dropped first) until a worker takes the shard. Delivery is exactly once in effect: the sender's outbox is durable and resends every unacknowledged message each second, the receiver absorbs duplicates as it does after a restart, and a receiver acknowledges only once the message's effect is durable. An acknowledgment names the epoch of the sender's lease; the server forwards it only to that lease, and the worker checks it again, so an acknowledgment that outlives a failover (whose new holder renumbers its outbox) is dropped instead of removing another message; the new holder resends. A worker crash on either side, or a target lease moving while a message is in flight, therefore costs at most a resend. **Liveness.** A worker holding the lease sends `PING` every second from its own thread (over shared memory, it advances a heartbeat counter in the mapping instead); any other request counts too. The server revokes the lease once it has spent five seconds waiting to read from the worker without receiving a byte, or once a write to the worker has made no progress for five seconds. It closes the connection and hands the lease to the next standby with a higher epoch, exactly as when the holder disconnects, so the revoked worker can never append again. The connection's reader does no disk I/O, so a long `fdatasync` or a large recovery read on a shard thread never delays it. Time the server does not read is not silence: while a stalled runtime holds the reader up, or while a shard has more than 32 MiB of requests waiting and the server stops reading until it drains, the worker's pings wait in the socket, and the server reads them before it checks the timeout. `Server::set_liveness` sets the interval and timeout for leases granted afterwards, or turns revocation off. The worker offers the capability in its hello and the lease carries both values; a worker that does not offer it keeps its lease until its connection closes. **Run-store offsets.** Run-store appends are pipelined, and the worker computes each frame's offset itself. Each append carries that offset. The server refuses an append whose offset is not its store's length, then refuses every later write to that shard for the rest of the lease; otherwise it answers with the offset it wrote at, and the worker checks the answer when it arrives. Either mismatch stops the shard with an error before any flush or checkpoint sent after the append is acknowledged. A worker uses this only when the lease advertises it, and otherwise sends the original unverified append. **The shared log.** A worker sends each shard's records as a section of the server's log and does not wait for an answer before it sends the next section of another shard: appends are pipelined. Whoever writes next takes the sections every connection has staged and writes them as one batch with one write (`O_DIRECT` and `RWF_DSYNC` on Linux, as in local mode), so workers and shards share flushes. When a batch is durable, each connection holding one of its shards gets one `DURABLE` frame with the cumulative durable position of each, and the waiting shard spins, then parks, until its position arrives. When nothing is in flight and the disk is fast (last write at most 200 µs), the batch is written by the thread that received it; otherwise by the log's I/O threads, with up to two writes in flight, each padded to the end of its last 4 KiB block so that two writes in flight never share a block. The shards a write made durable usually send their next sections within a round trip of each other, so on that slow path the next batch waits for them: until all have arrived, or three times their usual turnaround (measured, at least 20 µs and at most a quarter of a flush), whichever comes first, so eight shards do not split into groups taking turns on the two writes (with a 25 ms flush, eight shards take 1.1 to 1.2 times one shard's time, as in local mode). A lone shard never waits: it is the whole group. A shard granted to a new worker has what its previous holder staged dropped (none of it was acknowledged), and its recovery waits until that holder's writes in flight are durable, so its records can never follow the new holder's. **Recovery.** The server reads the log and ships each shard its records after its checkpoint (`RECOVER_LOG`, in pages of about 1 MiB); the worker no longer pulls whole segment files. The log is read once when the server opens, and once per failover for every shard granted since the last read. Run-store reads that continue a scan (a mass wake of runs parked one after another) read further ahead each time, up to 1 MiB, so most of them need no round trip. **Threads.** Each shard of a lease has its own server thread. It owns the shard's run store and checkpoints, handles the shard's requests in arrival order, does their blocking I/O and writes the replies to the socket itself, so the server's async runtime never waits on the disk for them. Each shard's replies leave in the order its requests were handled, and every reply to a run-store append leaves before the `DURABLE` frame of a section sent after it. A section whose shard thread has nothing pending is staged (and on a fast disk written) by the connection's reader itself, saving a thread handoff. Between requests a shard thread spins for up to twice the last gap while requests arrive within 100 µs of each other, and never on hosts with two or fewer cores. A directory with a log per shard (format 2) is served as before: each shard thread flushes its own log, inline on a fast disk or through the shard's flusher thread. **Waiting for a flush.** When a worker's shard has nothing else to do and its last flush round trip took at most 200 µs, the shard thread sends the batch and waits for the reply itself: it spins for about twice the recent round trip (at most 200 µs, and never on hosts with two or fewer cores), then parks. The connection's reader thread hands the reply to it directly, without waking the shard's runtime. Otherwise the flush runs in the background and the reader wakes the runtime when it completes. Both paths measure the round trip, so a slow server is never waited for inline. ## Calls from other machines Any process can connect as a client and send calls to the worker holding the lease: ```rust let c = fusee_server::Client::connect("10.0.0.5:7600")?; c.complete(&token, &serde_json::json!(true))?; let n = c.update("cart-1", "addItem", "req-42", &item)?; ``` The worker opts in by serving them next to its engine: ```rust let (cfg, remote) = fusee_server::remote_worker("10.0.0.5:7600")?; let engine = Engine::open(cfg).await?; tokio::spawn(fusee_server::serve_calls(remote, engine.clone())); ``` `complete`, `fail`, `update`, `query`, `signal`, `start`, `result`, `cancel`, `terminate` and `streamRead` are relayed, with JSON arguments, to the worker holding the workflow's shard. `list` and `list_where` go to every worker holding shards, each listing its own, and come back merged in shard order (up to the limit per shard, as an engine's own listing). A call made while no worker holds the shard it needs (for a listing: any shard) fails at once; the client retries. **Objects from other machines.** A call to a key of an [object type](features.md#objects-keyed-state) is an `updateWithStart` on the key's run (`/`), relayed to the worker holding its shard, and its state is the `fusee.state` query, so `Client::object_call(type, key, handler, arg, id)`, `object_send` and `object_state` (and `client.objectRef(type, key)` in TypeScript, `client.object_ref` in Python) need nothing new from the server. A workflow's call to a key on another worker's shard is a child-start message through the server, and its result comes back as a child result. **Streams from other machines.** `streamRead` is one read of a [stream](features.md#streams): the worker answers with the durable chunks from the offset, or waits for one (a long poll of up to a minute; 30 s by default) and answers with none. Readers loop on it, which the clients do: ```rust for chunk in client.stream("q-1", "tokens", 0) { print!("{}", chunk?.json::()?); } ``` In TypeScript `client.run(id).stream(name, { from })` and in Python `client.run(id).stream(name, from_=n)` are async iterators. A read in flight when its worker goes away fails ("the worker went away before answering"); the reader connects again at its offset (`ClientStream::offset`, `reader.offset`), and once a standby holds the shard it reads on from there: chunks are only ever handed out once the server's log holds them, so a failover never takes back one a reader saw, and the step that runs again on the new worker continues after what was recorded. Each waiting read holds one of the server's pending calls (at most 4,096). **Limits.** The server persists named concurrency and rate definitions and renews held permits every second. Leases expire after three seconds without renewal. A restarted server waits out outstanding leases before reissuing permits. The bound applies to valid permits. An external operation that already received its request can continue after the issuing worker loses its lease. Replicated groups record lease decisions in their quorum log. The server also holds the wait queue, with each waiting step's priority and fairness key ([priority and fairness](features.md#priority-and-fairness)), so the order is the same whichever worker asks. **Covers:** container fleets without local disks, failover between application instances, and several instances sharing one server's shards (`max-shards`). **Security:** off by default. Turn on TLS and a token (see [Security](#security)) on any network you do not fully trust. **Measured** on an AWS i8g.2xlarge (instance-store NVMe, one shard, `cargo run --release -p fusee-server --example server_step_cost -- 500 7`, rounds that alternate local, Unix socket, shared memory and TCP): a sequential step over shared memory takes 1.0 µs more than in local mode (15.9 against 14.9 µs at p50), with one voluntary context switch per step (the flush, as in local mode) and almost no socket messages; over a Unix socket a step takes 12 to 18 µs more and over loopback TCP 20 to 21 µs more, with three to four voluntary context switches per step. ## Shared memory A worker on the server's host can send its log records through shared memory instead of the socket: connect to `unix:/path.sock?shm=1`. The worker asks for it in its hello; the server creates the mapping (a memfd on Linux, an unlinked POSIX shared-memory object on macOS) and passes it, with its doorbell page, alongside the lease frame (`SCM_RIGHTS`), and learns the worker's process from the socket's peer credentials. Off by default; a server that cannot set it up (no shared log, another platform) leases over the socket alone, and the worker uses that. ```text mapping header: version, shards, ring size, worker and server pids, heartbeat (worker), closed (server), log position and segment one cache line per shard: durable LSN, lease epoch, a futex word bumped after every store, parked, pins, notify, failed ring: head (server), tail (worker), frames of length, CRC32C, payload (kind, shard, epoch, flags, section) doorbell one page per server: the futex word its log thread sleeps on ``` - **Appends.** A shard thread copies its section into the ring and publishes it with a release store of the tail; if the log thread sleeps, it rings the doorbell. One server thread polls every worker's ring, stages what it finds with the same epoch and LSN checks as the socket path, writes the batch (inline on a fast disk, else on the I/O threads), stores each shard's durable LSN in its line and wakes only waiters that parked. It spins for twice the longer of the recent flush and the workers' recent turnaround, from publishing a position to that shard's next section (at most 100 µs, never on hosts with two or fewer cores), before it sleeps on the doorbell: the next frame is about that far away. (Spinning for the flush time alone left it asleep on a fast disk, and every step paid a futex wake-up.) - **Waiting.** A shard waiting for its own flush spins on its line for about twice the recent round trip (at most 200 µs), then parks on the line's futex word (`futex` on Linux, `__ulock_wait` on macOS). A background flush sets the line's notify flag, and the server then also sends a `DURABLE` frame on the socket, which wakes the shard's runtime. - **What stays on the socket.** Recovery reads, run-store appends and reads, checkpoints, grants, limits, relayed calls and messages. A section is sent through the ring only while no run-store append of its shard awaits its answer, so the server has written every frame a section may reference before it takes the section; otherwise, or when the ring is full, it goes over the socket. - **The ring is volatile.** Only watermarks acknowledge. A worker that dies mid-frame never published it; a frame published and not yet taken when its lease ends is refused by its epoch. A frame that fails its length or checksum checks, or names a shard the ring cannot carry, breaks the protocol: the server revokes the lease. - **Fencing.** The server refuses ring frames with a stale epoch and reports it (`IO_FAILED`, and the line's failed flag), and the shard takes no more ring frames in that lease. When a lease ends, the server stops reading the ring, clears each line's epoch, sets `closed` and wakes every waiter. It also watches the worker's process (a pidfd on Linux, kqueue `EVFILT_PROC` on macOS) and revokes the lease as soon as it exits, even if another process kept the socket open; the heartbeat in the mapping replaces `PING` for liveness. Measured with `cargo run --release -p fusee-server --example server_step_cost -- 500 7` on an AWS i8g.2xlarge (instance-store NVMe): 500 sequential steps, seven rounds that alternate local mode, the Unix socket, shared memory and TCP, per-run medians. | Path | Step p50 | Over local | | --- | ---: | ---: | | Local mode | 14.9 µs | | | Shared memory | 15.9 µs | +1.0 µs | | Unix socket | 27.0–32.5 µs | +12 to +18 µs | | Loopback TCP | 35.1–36.2 µs | +20 to +21 µs | A step over shared memory makes no socket call; it copies its section twice (into the ring, out of it) where the socket path copies it four times (into the frame, into and out of the kernel, out of the decoder), and the voluntary context switch that remains is the flush itself, on the server's log thread instead of the shard's. ## Invoked workers A server can run workflows on workers that exist only while there is work: functions on Lambda, Cloud Run or Vercel, or any HTTP server. When a shard that no worker holds has work due, the server sends an HTTP `POST` to an endpoint; the process that receives it opens an engine in server mode for that shard, runs it until it is idle or its time budget ends, and returns. No process stays up while workflows sleep. ```bash fusee server --dir /var/lib/fusee --listen 0.0.0.0:7600 --token-file /etc/fusee/token \ --invoke https://fusee-worker.example.com/invoke \ --invoke-secret-file /etc/fusee/invoke-secret --invoke-concurrency 16 --invoke-budget-ms 60000 ``` ```rust // In the function: any HTTP framework hands the request to the handler. let handler = fusee_server::invoke::Handler::new( "fusee.example.com:7600?ca=ca.pem&token-file=token", std::fs::read("invoke-secret")?, |engine| { engine.workflow("order", order); }, )?; let reply = handler.handle(signature_header, &body).await; // status and JSON body ``` The TypeScript and Python handlers are in their READMEs; each takes the same address, secret and registration function. In Go, `fusee.NewInvocationHandler(fusee.InvocationOptions{Address, Secret, Register})` is an `http.Handler` (and `Invoke(ctx, signature, body)` for other frameworks); in Java, `new InvocationHandler(address, secret, options, register)` serves the JDK's `HttpServer` through `httpHandler()` and any framework through `invoke(signature, body)`. **When the server invokes.** A shard is *due* when no worker holds it and one of these holds: - a client call for it arrived: `start`, `signal`, `update`, `query`, `result`, `complete`, `fail`, `cancel`, `terminate`, `reset`, `streamRead`, or a listing's share of it; - a message from another shard waits for it at the server (a child start or result, a signal, a cancel); - the time its last holder reported comes: a timer (a sleep, a deadline, a scheduled start, a debounce window), a retry, or the resend of a message another shard has not acknowledged yet; - its last holder left without reporting: it crashed, its lease was revoked, or the server restarted. A server starting with endpoints invokes every shard once, because what its shards' next due times are lives only in the memory of the server that last heard them. Calls wait at the server (up to 60 seconds) until a worker holds their shard, then go to it as usual; a call that waits longer fails and the client retries. Messages wait as they do for any shard without a holder (see *Messages between workers* above). **Which endpoint.** Each endpoint takes every workflow type, or the types it lists. A `start` goes to an endpoint for its type. For other work the server uses the endpoint the shard's last holder named for the types it had runs of and no code for, else the endpoint that last ran the shard, else the first endpoint that takes every type. A shard with work for two endpoints is invoked on one, then on the other. **The request.** `POST` to the endpoint's URL with `content-type: application/json`, `fusee-signature: t=,v1=` where `v1` is HMAC-SHA256 with the endpoint's secret over `.`, and the body ```json {"version":1,"invocation":"8f3c0e5a1b2d4c6e","shards":[2,5],"budgetMs":60000,"reasons":["call","timer"]} ``` Over `https://` the server verifies the endpoint's certificate against the CA file given for it (or the system bundle) and can present a client certificate (mutual TLS, `--invoke-cert` and `--invoke-key`). The handler answers 401 to a request without a valid signature before it connects to anything, so nobody but the server can make it open a connection; an endpoint configured without a secret (the server warns) must check the server's client certificate instead. A handler refuses a signature older than five minutes, and a replayed request is harmless besides: its invocation ID is no longer valid (see below). **What the invoked process does.** The handler checks the signature, then connects to the server as a worker with `invocation=` among its address options. The server grants exactly that invocation's shards, those of them no other worker holds by now, each under a new lease epoch; an invoked worker never waits as a standby, and a connection naming an unknown or finished invocation is refused. Its engine recovers those shards from the server (the checkpoint, the run store's index and the log after the checkpoint: a few round trips per shard, no file transfer) and leaves every other shard closed. Messages to other shards go through the server's relay, calls are relayed to it as to any worker, and it pings to keep its lease. It parks runs 50 ms after they suspend (instead of 5 s), runs until idle or until its budget ends, then closes its engine (flushing, then checkpointing, so the next open replays almost nothing), reports each shard's next due time on the lease connection (`IDLE`), closes the connection and answers `200` with a summary. It keeps nothing between invocations; each one opens a fresh engine. **Idle.** A shard is idle when nothing can make progress there: every loaded run waits on a timer, a signal, a child, or an outside completion (no step is running and no code is replaying), no records or checkpoint wait for the server, every message to another shard has been acknowledged, no relayed call is being answered (waits for a result or for stream chunks do not count), and its next due time is at least 250 ms away. Messages to another shard that are not acknowledged yet keep the shard for up to 3 s: an acknowledgment names the sender's lease, so one that arrives after the sender left is dropped and the message is sent again from its next invocation. Once every shard the worker holds has been idle for 100 ms (so a trickle of calls shares one invocation instead of paying a cold start each), it asks the server to stop sending it work for them (`DRAIN`): calls and messages that arrive from then on wait at the server and make the shard due again the moment the worker leaves. The worker answers what was already on its way and closes once all its shards are idle again. Calls waiting for a result or for stream chunks that are still open when it leaves are not answered: they wait at the server for the shard's next holder and do not invoke one by themselves (a stream read waits up to two minutes there). **Everything due, not a batch.** An invocation runs everything due on its shards, not a bounded number of runs: the shard is the unit of leases, recovery and fencing, and splitting a shard's runs across processes would need leases per run. The server spreads due shards across the endpoint's free concurrency (an invocation carries at most `--invoke-shards` shards, default all of them, divided evenly between the free slots), so a shard is the unit of parallelism, as for long-lived workers. The time budget bounds an invocation's length. **The time budget.** Each endpoint has a budget (`--invoke-budget-ms`, default 60 s, at most the platform's execution limit), sent in the request and in the lease. The worker stops at the budget less a margin (a tenth of it, at most 5 s): it closes its engine without waiting for steps still running and reports those shards as due now, handing their runs back; the server invokes again at once. A step abandoned this way runs again in the next invocation, as after a crash: its effects are recorded once, and its function runs at least once, as always. A step therefore has to fit in one budget; longer work belongs in an [async completion](features.md) that an outside system finishes with `complete`. An invocation that reaches its budget without finishing a step on a shard puts the shard in backoff (1 s, doubling to 5 min) and logs it, so a step that can never fit costs one invocation per backoff period instead of a loop. **Exactly once.** Unchanged. An invoked worker holds ordinary per-shard leases with epochs and the liveness timeout, and every request carries its shard's epoch. The server waits for the HTTP response for the budget plus a grace (30 s by default, `graceMs`); when the request fails or times out it revokes the invocation's lease if it still holds one, and the invocation ID stops being valid, so a process that wakes up later can neither connect with it nor append under the old epoch. Its shards are due again with backoff. A process the platform kills mid-invocation is a crash: its lease ends (connection closed, or five seconds of silence), and the next invocation recovers the shard from the server's log. **Backoff, retries and concurrency.** An endpoint runs at most `--invoke-concurrency` invocations at once (default 16). An invocation that fails (no connection, a status other than 2xx, a timeout, or a 2xx after which the shards were never leased) leaves its shards due and backs the endpoint off: 100 ms, doubling to 30 s, with a fifth of jitter, reset by the next success. A slow endpoint holds its concurrency slots, so at most that many invocations wait on it. A shard whose invoked holder left without reporting is due again with the same backoff, per shard (a long-lived worker's shards are due at once). **Long-lived workers alongside.** A long-lived worker (one that connected without `invocation=`) waiting as a standby takes any shard that frees up, before the server would invoke for it; the server only invokes for shards that no worker holds or waits for. When a long-lived worker leaves, its shards are due at once and invoked workers take over. **Configuration.** Each `--invoke URL` starts an endpoint; the `--invoke-*` options after it apply to it: `--invoke-secret-file`, `--invoke-types a,b`, `--invoke-concurrency`, `--invoke-budget-ms`, `--invoke-shards`, `--invoke-ca`, `--invoke-cert` and `--invoke-key`. `--invoke-config FILE` reads `{"endpoints": [{"url": ..., "secretFile": ..., "types": [...], "concurrency": 16, "budgetMs": 60000, "shards": 0, "graceMs": 30000, "ca": ..., "cert": ..., "key": ...}]}`; unknown keys are refused. In Rust, `Server::set_invokers(vec![invoke::Endpoint::new(url, secret)...])`, or `Endpoint::from_json`. **Metrics.** `Server::invocation_metrics()`, and the `invocations` call any client can make (`client.call("invocations", {})`), give per endpoint the invocations sent and how they ended (succeeded, failed, timed out, answered without taking the lease), those in flight, the backoff left, the time from sending to the answer, shards invoked, steps they ran and the engines' open time; and overall the shards due, invoking and in backoff, handbacks, leases that ended without a report, invocations without progress and calls waiting at the server. `FUSEE_INVOKE_TRACE=1` prints a line per invocation event on standard error, on the server and in the worker. **Protocol.** The worker offers `INVOKED` (512) in its hello, followed by the invocation ID; the lease carries the budget left. `DRAIN` (28) names shards to route nothing more for, answered with `OK` behind every call already routed to the worker. `IDLE` (29) carries, per shard and lease epoch, whether work is left, whether the budget ended, the next due time (relative, so clocks need not agree), steps run, the engine's open time and the workflow types the worker had runs of and no code for. A shard's report counts only under the epoch it names. **Measured** on an Apple M-series laptop (plain `fsync`, 4 shards, server, endpoint and client in one process over loopback, a fresh engine per invocation: `cargo run --release -p fusee-server --example invoke_cost -- 100 200 100 10000`): | What | Result | | --- | --- | | Start for a shard nobody holds, call to the first step's body | p50 2.0 ms, p90 2.6 ms, max 2.9 ms | | Of which: until the engine is open and the workflows registered | p50 1.6 ms (the shard's own recovery 0.8 ms) | | 200 workflows of 4 steps on 4 shards, started at once | 4 invocations, 200 steps each | | The same started one after another | 4 invocations, 200 steps each (the 100 ms idle grace) | | Step after a 1 s sleep, invoked vs long-lived worker | late by 0 ms p50 in both; engine open 0.1 ms and invocation time 15 ms per resumed run (100 runs, 8 invocations) | | Start on a shard holding 10,000 runs asleep for an hour | p50 165 ms, of which recovery 159 ms | The cold start is a few round trips: the server's dispatch, the HTTP request, the lease, the shard's recovery and the step. Timers are invoked slightly early (the server wakes at the shard's earliest timer, which may be early) and the worker waits for them, so the step after a sleep is no later than on a long-lived worker. An invocation stays at least 100 ms after its last work, so invocation time per resumed run falls as more runs share one. Recovery grows with the runs a shard holds, about 16 µs each (the open reads the shard's run store from the server): more shards keep each invocation's open small, since an invocation opens only its own. Platform cold starts (a function's process and runtime) come on top. **Limits.** The next due times live in the server's memory, not on its disk: a restarted server invokes every shard once. A listing invokes every shard nobody holds. A step waiting for a concurrency permit or a retry backoff longer than the budget keeps its invocation open until the budget ends; the shard then backs off as an invocation without progress. ## Security Workers and clients authenticate with TLS 1.3, a shared token, or both. Without either, connections carry the plain protocol, and the server prints a warning at startup when it listens on a non-loopback address. ```bash openssl rand -hex 32 > /etc/fusee/token && chmod 600 /etc/fusee/token fusee server --dir /var/lib/fusee --listen 0.0.0.0:7600 \ --tls-cert server.pem --tls-key server.key --tls-ca workers-ca.pem \ --token-file /etc/fusee/token ``` Workers and clients put connection options after the address: ```rust let addr = "db1.internal:7600?ca=/etc/fusee/ca.pem&cert=w1.pem&key=w1.key&token-file=/etc/fusee/token"; let engine = Engine::open(fusee_server::remote_config(addr)?).await?; let client = fusee_server::Client::connect(addr)?; ``` The same string works as the TypeScript and Python `address`, after `unix:/path.sock` (token only) and after `replicas://a,b,c`. | Option | Meaning | | --- | --- | | `ca=PATH` | Turns TLS on: the CA certificates that sign the server's certificate. | | `cert=PATH`, `key=PATH` | This worker's certificate chain and key (PKCS#8, PKCS#1 or SEC1 PEM), for a server started with `--tls-ca`. | | `name=NAME` | The name to verify in the server's certificate instead of the address's host. | | `token-file=PATH` | The shared token. | | `max-shards=N` | Not a security option: a worker holds at most `N` shards at once (see Leases). | | `shm=1` | Not a security option: a worker on a `unix:` address sends log records through shared memory (see [Shared memory](#shared-memory)). Off by default. | An unknown or repeated option is an error, so a misspelt option never silently turns security off. In Rust, `fusee_server::security::Security` holds the same settings as values: `remote_worker_with`, `Remote::connect_with` and `Client::connect_with` on the worker side, `ServerConfig::security` and `Server::set_security` on the server. **TLS.** TLS 1.3 only (rustls with ring), without session resumption. The server presents `--tls-cert`. A worker verifies it against `ca=` and the host it dialled (or `name=`), so the certificate must carry that DNS name or IP address. With `--tls-ca` the server also requires a client certificate signed by that CA (mutual TLS); any such certificate is accepted. Unix sockets never use TLS: the socket file's permissions decide who connects, and the token still applies. **Token.** A file of at least 16 bytes (surrounding whitespace is not part of it; use 32 random bytes). The server and workers warn when other users can read it. Before its first protocol frame each connection proves the token with HMAC-SHA256, in both directions: ```text worker → HELLO 0xF0 version 1, purpose (1 server, 2 replica), sender u8, nonce [32] server → CHALLENGE 0xF1 version 1, nonce [32] worker → PROOF 0xF2 HMAC(token, "fusee auth v1 initiator" ‖ T) server → ACCEPT 0xF3 HMAC(token, "fusee auth v1 responder" ‖ T) or REFUSED 0xF4 reason (UTF-8); the server closes the connection T = hello body ‖ challenge body ‖ binding length u8 ‖ binding ``` They are ordinary frames (`FSE1` header, CRC32C) of at most 1 KiB. Under TLS, `binding` is 32 bytes of TLS exporter output (RFC 8446 section 7.5, label `EXPORTER-fusee-auth-v1`); without TLS it is empty. The token never crosses the wire. Both nonces are fresh, so a recorded exchange never verifies again, and under TLS a proof cannot be relayed into another session. The server answers only after the worker's proof verifies, and the worker sends no protocol frame before it has verified the server's, so neither side talks to a party without the token. MACs are compared in constant time. TLS and the token exchange must finish within 10 seconds. A worker without a token that reaches a server with one, or the reverse, gets an error naming the option to change; failed handshakes are logged at most once every ten seconds. **A token without TLS** authenticates the start of each connection only: anyone on the network path can still read the traffic and inject into the stream. Use TLS wherever the network is not trusted. **Limits:** workers are not told apart, so any worker with the credentials can take the lease and any client can call the lease holder; there are no revocation lists; certificates and tokens are read at startup, so rotating them means a restart; and a flood of connections can still exhaust the server. ## Server directories A server directory keeps one log for every shard (`/wal/`, log format 3); its layout is that of a local data directory, so `fusee inspect` reads it. `Server::open(cfg, false)` creates a directory with a log per shard (format 2) instead; `fusee upgrade `, with the server stopped, converts such a directory to the shared log. A server with a shared log refuses workers that only speak per-shard logs, with a message saying to upgrade them. ## Reopening a worker With a log per shard (format 2), a replacement worker negotiates paged segment recovery with the server. The server reports the file length and scans backwards for the last nonzero byte. The worker fetches the prefix in bounded pages and reconstructs the exact padding before passing the segment to its normal corruption and torn-tail checks. With the shared log, the server sends the shard's records instead (see Recovery above). ## Dedicated replicated servers Run three or five servers, each with its own directory. List the replication addresses in the same order on every member. Give each server its index in that list and a separate listener for applications. One replication group owns one shard. ```bash fusee server --dir /var/lib/fusee --listen 10.0.1.10:7600 \ --replicas 10.0.1.10:7400,10.0.1.11:7400,10.0.1.12:7400 --replica-id 0 ``` Start members 1 and 2 with their own listen addresses. `--tls-cert`, `--tls-key`, `--replica-tls-ca` and `--token-file` secure both the worker listener and the replica protocol ([replication security](replication.md#security)). An application passes the worker listeners to its ordinary server-mode API. ```ts const engine = await Engine.open({ durability: "server", address: "replicas://10.0.1.10:7600,10.0.1.11:7600,10.0.1.12:7600", }); ``` Rust `remote_worker`, the Python `Engine.open` API and standalone `Client.connect` accept the same address. A new connection probes the members until one commits a quorum record. That primary leases the shard to one worker. Other application processes wait as standbys. Each workflow flush waits for a durable majority. Checkpoints, store compaction, indexed listing and immutable history use the primary’s materialized shard files and its replicated checkpoint mechanism. A replacement worker first commits a new permit session. Outstanding permits from its predecessor keep their expiry time. On a view change or a failed quorum check the server closes its worker and client sockets. Reconnect clients to discover the new primary. A call interrupted before its reply may already have committed. Reuse its workflow or update ID when retrying. Users of a worker engine that lost its connection must reopen it or let a connected standby recover. Replication-group membership is fixed. # Replication ## Protocol `crates/fusee-core/src/vsr.rs`, a pure state machine in the style of Viewstamped Replication Revisited: - 1 to 9 replicas per group (3 or 5 in practice), with member IDs below 32. Membership is fixed within an *epoch*; a view carries its epoch in its high 32 bits, and the primary of view `v` is the epoch's member list at position `(v - first view of the epoch) % n`. A group that never changed is epoch 0 with members `0..n`, whose primary of `v` is `v % n` ([Membership change](#membership-change)). - The primary appends entries and sends them to backups at once (before its own flush). Backups flush and acknowledge how far they are durable. - An entry commits when any quorum holds it durably **in the current view**; the primary need not be one of them. Outputs and step gates use the commit position. - Entries form a hash chain, so a replica can verify entries from any peer. - View change: `StartViewChange`, then `DoViewChange` with each replica's last normal view, op and uncommitted suffix; the new primary takes the log with the highest (last normal view, op), fetches committed entries it lacks, and sends `StartView`. - Protocol-aware recovery: a replica whose log has a damaged record in the middle does not truncate committed work. It takes no part in view changes, fetches the range from peers, verifies it against the chain, and rewrites its log; if its later entries belong to another history it takes the primary's log from below the damage. Rules the protocol keeps, each covered by the simulator's checkers: 1. A replica that restarts during a view change comes back in view-change status, not normal. 2. A restarted primary never leads its old view again: entries a quorum of backups committed may be missing from its disk. 3. Responses that extend a log are accepted only from the current view's primary (or, for a new primary, from the replica whose log it adopts). 4. A backup keeps everything through its own commit; only later entries are replaced. 5. A backup missing committed entries of a new view stays recovering (does not vote) until a full state transfer fills the gap. 6. A log is only ever replaced atomically, once the complete new log is in hand; and the saved "last normal view" is a barrier behind the log it vouches for. 7. Only a replica with holes in its log abstains from voting. 8. A backup that answered its primary's ping takes part in no view change (starting, joining, voting or leading one) for the promise time after, and a restarted replica abstains for that long from its start ([read leases](#step-down-and-read-leases)). 9. A primary proposes nothing after a reconfiguration entry: it is the last entry of its epoch ([membership change](#membership-change)). 10. Only a replica that holds the log through the entry that began its epoch votes in it; one that does not first fetches that committed prefix from any member, verified against the entry. 11. A replica that restarted with a hole in its log installs a checkpoint image only with a suffix that reaches everything it holds after the hole: it may have acknowledged those entries, and be the only member of their quorum still holding them. A primary-elect missing part of its selected donor’s committed prefix pulls verified pages and retries dropped continuation requests. A replica catching up from a primary keeps its old durable log until it has verified the entire replacement. The preferred page limit is 1 MiB or 4,096 operations. A single admissible entry may exceed the preferred byte limit. Full-log rewriting streams to a temporary file before the durable rename. ## Failover timing `Timing { heartbeat_ms, election_ms }` (`ReplicaConfig::timing`; `fusee server --heartbeat-ms --election-ms`), in milliseconds of each member's clock. Defaults: **heartbeat 5 ms, election 40 ms**. An idle primary sends a heartbeat every heartbeat interval; a backup that hears nothing from its primary for the election timeout plus a jitter drawn again for every election from `0..election_ms` starts a view change, so a silent primary is suspected after 40 to 80 ms. The election timeout must be at least four heartbeats. The I/O thread ticks at half the heartbeat (1 to 5 ms), and before it starts an election it lets its readers run once, so heartbeats that waited in a socket while the thread was descheduled count. - **A crashed process is noticed sooner.** Its sockets close at once; a backup whose primary's stream closes (with nothing received since) elects as soon as its promise to the primary lapses (half the election timeout, 20 ms) instead of waiting out the timeout and jitter. - **View changes repeat themselves.** A replica in a view change it started or joined resends its StartViewChange (and its DoViewChange) every 10 ms, so a member that was restarting when they were first sent joins the same view instead of waiting for another timeout; a primary that hears one for its own view sends that member its StartView again (it missed it, often because the primary's stream to it was not up yet), and a dialer retries a refused connection every 10 ms. A replica that restarted as the primary of a view never leads that view again (rule 2). - **SDK members settle first.** A TypeScript or Python member opens its engine once it has been the ready primary of one view for 150 ms: while three processes start, the first primary can hand the role on within a few election timeouts. - **Resends wait for a stall.** A primary resends a backup's missing suffix only once that backup's position has stalled for 10 ms with nothing new proposed. **Lossy or distant networks.** A lost TCP segment stalls a stream until it is retransmitted, at least 200 ms later on Linux; an election timeout shorter than that turns every loss burst into an election. The defaults suit a datacenter or loopback network; raise `--election-ms` (to 150 ms or more, as the partition tests below do for their 2 % loss and 20 ms delay) where packets are lost or round trips take tens of milliseconds. Scheduling gaps count too: on a host with several times more runnable threads than cores, a member can miss heartbeats for longer than the election timeout, so raise `--election-ms` there as well. **Primary kill to resumed work.** On three i4i.large hosts in one placement group, the failovers that completed before the next of 30 primary kills had the next member ready p50 25.0 ms after the kill (at most 30.3 ms) and committing new work p50 30.3 ms after it (at most 35.3 ms). `fusee-crashtest replicated-driver` with `stream` measures the same on any machine: every primary also starts a workflow every 2 ms, and the driver SIGKILLs the primary 30 to 230 ms after it opened its engine and times, from the kill, the next member becoming ready, opening its engine, running workflow code and committing a new workflow. A kill that comes before the previous victim has restarted and caught up is a second failure within the restart window, more than a three-member group tolerates at once: the new primary has only a recovering backup, and its first commit waits for that member's restart. Opening the engine takes 3 to 9 ms of a failover (eight shards, each reopening its directory and replaying its records since the checkpoint); hot standby removes it. **Hot standby.** `ReplicaNode::standby_config()` gives an engine that opens at once on any member: its shard threads, runtimes and open run stores exist and its workflows are registered while the member is a backup, and its shards stand by (calls to them wait). The moment the member becomes the ready primary, each shard takes the committed log batches from the I/O thread's memory (one barrier round trip, 50 to 150 µs per shard) and replays its records since its checkpoint; nothing is opened, except a shard's run store when a checkpoint image installed while it stood by moved the engine directory. A standby engine serves one primary tenure: when the member loses the role it stops like any engine, and the application opens the next one. `fusee-crashtest replica` uses it unless `FUSEE_REPLICA_STANDBY=0`. With it, ready to first new commit takes about 1.5 ms, and kill to commit is dominated by detection and the view change. A promoted shard still replays the records since its checkpoint (from memory). ## Step-down and read leases **The assumption.** Every member's clock advances at a rate within 1 % of real time (`fusee_core::vsr::MAX_CLOCK_DRIFT_PPM`). Delays of messages and of threads are unbounded; only rates are assumed. The protocol clock counts time a host spends suspended: it advances with the monotonic clock plus however far `CLOCK_BOOTTIME` has run ahead of it on Linux (macOS's `CLOCK_MONOTONIC` already counts sleep), and never runs back, even when a tool such as libfaketime moves the boot clock back. A clock that stops while time passes, or a VM restored from a memory snapshot, breaks the assumption. **The rule.** With every heartbeat the primary sends each backup a ping stamped with its clock. A backup normal in the primary's view answers and *promises* not to take part in any view change (start, join, vote in or lead one) until its own clock has advanced `P` (the promise, half the election timeout: 20 ms) past the moment it received the ping; a request to change views that arrives meanwhile waits and is joined once the promise lapses, unless the primary renews it. A restarted replica, which cannot remember its promises, abstains for `P` from its start. The primary holds a **read lease** while its clock is less than `L` past the stamp of the oldest ping among a quorum's freshest answers (itself included), where L = (P - 1)(1 - ρ)/(1 + ρ) - 1 (17 ms for the default timing) the promise as the slowest allowed backup clock measures it, as the fastest allowed primary clock measures it, less a millisecond at each end for millisecond clocks. Every view change needs a quorum that intersects the lease's quorum in a member bound by its promise, or in the primary itself, which loses the lease the moment it leaves its view (before anything of the view change is sent). So while a primary holds a lease no other member can lead a newer view, and its state has every acknowledged write. **Step-down.** A primary that hears from no quorum (acknowledgments or ping answers) for four election timeouts, and at least 200 ms (`Timing::step_down_ms`), starts a view change: its engine stops instead of waiting on a commit that cannot come. Not sooner: a backup blocked in a slow flush, or a process still starting, answers late, not never; by then a backup that lost the primary has elected another, and the lease lapsed long ago. **Reads.** `ReplicaNode::lease_valid()` (also on a cloned `ReadGuard`) says whether reads of the engine are current; `confirm_reads()` returns at once under a lease, else after a no-op commits through a quorum (a read started after it returns sees every write the group acknowledged before the call), and fails without a quorum. The TypeScript and Python SDKs check the lease before every engine call (queries, results, inspect and list included) and, without one, confirm through the protocol or refuse the call. A dedicated server's quorum check (`QUORUM_PROBE` and its idle fence) answers from the lease when it can. A Rust application reading its embedded engine calls `confirm_reads()` before a read that must not be stale. Under a load spike a lease can lapse briefly; reads then pay one round trip. **Checked.** `spec/FuseeLease.tla` models pings, promises, leases, view changes, crashes and per-replica clocks advancing one or two units per tick (drift of 50 %, so the safe lease is `P·1/2`). TLC checks that a lease read sees every committed write and that no replica is normal in a newer view while a lease holds (28,105,017 distinct states without crashes, exhaustive; see `spec/README.md` for the run with a crash); two negative controls fail: without the promise rule (six steps) and with a lease that ignores drift (`L = P`, seven steps). With one crash, TLC exhausted 105,112,688 distinct states with no violation. The simulator gives every member a clock rate within ±0.9 % and checks the same two properties every simulated millisecond for a ready primary under its lease (5,000 seeds, half with the default timing: 21,756,164 ms under a lease, no violation); its negative controls find violations when backups ignore promises (first at seed 74) and when clocks drift up to 90 % (seed 298). The process test `a_cut_off_primary_loses_its_read_lease_then_steps_down` holds the lease in every sample of 500 ms in steady state, and after both backups are killed sees it lapse in 15 ms and the primary step down. ## Membership change **The choice: epochs, a single committed reconfiguration entry per change, no joint consensus.** Joint consensus (Raft §6) needs every decision during the change to count two quorums, and single-server changes (Raft's dissertation) need configurations to take effect when appended, with a fix for a known bug; both change how every view change in VSR picks its log, and VSR's deterministic primary per view would have to be computed from a configuration that differs between members mid-change. Instead, membership is fixed within an epoch, and a change switches epochs at a known log position, as Viewstamped Replication Revisited's reconfiguration does: 1. The primary proposes a reconfiguration entry naming the next epoch's members and addresses, as the last entry of its epoch: it proposes nothing after it (sealed). Shards' sections wait meanwhile; a view change that drops an uncommitted entry abandons the change. 2. The entry commits like any other, on a quorum of the old members. 3. A member that learns it committed enters the next epoch. A member of the new configuration that holds the log through the entry changes to the epoch's first view by an ordinary view change among the new members. Its first view is led by the first member of the new list: the old primary when it stays. Members not in it retire. 4. A member of the new configuration without that prefix (a new member, or one that lagged) fetches it from any member, or a member the change removed, verified against the entry's checksum, or takes a checkpoint image from any of them; it votes only once it holds it. 5. A member that missed the change learns it from any message of a later epoch (it asks), and a member answers a message of an earlier epoch with the entry, so a removed member retires even if it heard nothing. **Restriction.** A quorum of the new members must be members now, so one step adds or removes members (3 → 5, 5 → 3), or replaces a minority; a dead member is replaced by naming its successor in its place. A change also needs a quorum of the old members up (to commit the entry) and, to start the new epoch, a quorum of the new members holding the log, which a removed member can serve. Keep removed members running until the new members report the new epoch. Like a failover, a change stops the primary's engine and reopens it on the first member of the new list. **Why it is safe.** Within an epoch nothing changes. Across the boundary: every committed entry of the old epoch is at or before the reconfiguration entry (rule 9), every voter in the new epoch holds the log through it (rule 10), and the new epoch's first view starts from a quorum of the new members, so its log contains every committed entry; later views of the new epoch are ordinary. **Checked.** `spec/FuseeReconfig.tla` adds the epoch handoff to the replication model (replace a member: 1,662,865 distinct states; grow 3 → 4: 1,515,475; no violation of `CommittedNeverChanges` or `CommittedSurvives`); dropping the seal rule fails in 14 steps, dropping the hold rule in 10. The simulator test `membership_changes_survive_crashes_during_reconfiguration` changes membership under crashes, partitions, lost and duplicated messages, damage and clock drift, with leases checked throughout: 3 → 5 → 3 (5 → 7 → 5 for a third of the seeds), then a member lost for good and replaced, with requests sent to the primary directly or through a backup and idle periods in which only heartbeats run. Every change must complete after the faults stop, removed members retire and every final member holds every committed entry. The process test `membership_changes_three_to_five_to_three_and_replace_a_lost_member` does the same on real nodes, with workflows checked across all four epochs. **Using it.** Start each new member with an empty directory and `--join` (`ReplicaConfig::join`), giving `--replicas` the members' addresses (`id=host:port,...`). Then, on any member: ```bash fusee replicas --dir /var/lib/fusee status fusee replicas --dir /var/lib/fusee set 0=10.0.1.10:7400,1=10.0.1.11:7400,2=10.0.1.12:7400,3=10.0.1.13:7400,4=10.0.1.14:7400 ``` or `ReplicaNode::change_members(members, timeout)`. The membership is saved in `vsr/members`; once a directory has one, it wins over `--replicas`. Under mutual TLS a member's certificate must name the host of its new address, as for the initial members. ## Commit on any durable quorum The primary's own log write runs in the background, and it releases an entry (opens the step gate, lets outputs leave) as soon as the entry commits: when any quorum of the view holds it durably, with or without the primary itself, so a slow primary disk does not slow commits. What the primary keeps to `min(commit, durable)` is materialization: the checkpoint images it builds and, in a dedicated server's groups, the segment files it writes. **Why this is safe.** A committed entry must survive every later view. 1. *Quorum intersection.* The entry is durable on a quorum Q of view v. Any later view starts from a view-change quorum Q', which intersects Q in at least one replica, r. Until v ends, r's log is a prefix-compatible copy of the primary's (rule 3); r is normal in v, so its last normal view is v or later, and the new primary adopts the log with the highest (last normal view, length) among Q', which is at least as long as r's in view v and contains the entry (rule 6 makes the adoption atomic; rule 4 keeps every backup's committed prefix). Induction over views carries the entry forward. Nothing in this argument uses the primary's own durability: the commit quorum may exclude it. 2. *The primary's lost tail.* What the primary released but did not flush is not on its disk after a crash. Rule 2 covers exactly this case: a restarted primary never leads its old view again. It rejoins in view-change status, cannot become primary of a view without a view-change quorum (which, by 1, brings it the entry), and as a backup it fetches what its log lacks. Its recovered commit hint never exceeds what its log held when the hint was written, so it never claims an entry it lost. 3. *Materialization.* A checkpoint image covers only entries that are committed, released and durable on the replica that builds it, so an image never anchors an entry the builder's own log could lose. **Checked.** In `spec/FuseeReplication.tla` the `Commit` action takes any quorum, with the primary's own durable length counted only if the primary is in it. TLC checks it (`MC.cfg`, three replicas, three operations, three views, two crashes): no violation of `CommittedNeverChanges` or `CommittedSurvives`. A witness configuration (`MC-witness.cfg`) adds the invariant that a primary never commits past its own durable prefix; TLC violates it in nine steps, a primary committing an entry only its two backups hold, so the checked state space does contain the behavior. The simulator test `a_slow_primary_disk_never_loses_entries_a_backup_quorum_committed` runs every seed with the primary flushing 10 to 100 times less often than its backups: 5,000 seeds committed 12,843,442 entries, 5,519,235 of them before the primary held them durably, with 66,046 crashes and no violation. `a_slow_primary_disk_does_not_slow_commits_and_loses_nothing` (process test) gives the primary a 40 ms flush: steps still commit in about a millisecond, the primary is killed with its log behind, and nothing acknowledged is lost. ## Where a step's time goes With timings enabled (`ReplicaObserver::enable_timings`; the benchmark harness enables them in replicated mode and adds the primary's to every result line as `replica_stages`), each member records the stages of a step's critical path (`replicated/timings.rs`), and `bench/aws/check_cluster.py` lays them out per trial: | Stage | Member | | --- | --- | | `shard_stage_to_take`: a shard hands its section to the I/O thread, which takes it | primary | | `take_to_prepare_sent`: encode, write the Prepare to every backup's socket | primary | | `prepare_sent_to_ack_read`: until the read that brings the acknowledgment completing a quorum | primary | | `prepare_read_to_write`, `protocol_log_append` (write; with `RWF_DSYNC` the flush too), `protocol_log_sync`, `durable_to_ack_sent`; `prepare_read_to_ack_sent` the whole turnaround | backup | | `ack_read_to_release`: commit published to the shards | primary | | `release_to_shard_resume`: the shard sees its records committed | primary | | `shard_resume_to_stage`: workflow code and the next record (its p50; the mean includes gaps between workflows) | primary | | `io_thread_tick`: the I/O thread's timer work (heartbeats, timeouts, checkpoint images); `io_thread_lock_hold`: each hold of the protocol lock by its main task | any | Network time is `prepare_sent_to_ack_read` less the faster backup's turnaround. **How the path is kept short.** - **The lone shard proposes its own section.** When nothing else is staged or queued and at most two shards worked in the last 2 ms (`fusee::spin::few_busy`), a shard that stages a section takes the protocol state without waiting (it is behind a mutex the I/O thread holds while it works) and runs the round itself: propose, send the Prepare, hand the primary's write to its helper. The I/O thread, woken by the acknowledgment, releases the commit. - **The Prepare goes out before the primary's own write is handed off** (the helper's wake-up is not on the commit path). - **A finished primary write wakes the I/O thread only if a quorum waits for it** (`Replica::commit_waits_on_own_durability`) or more frames wait to be written; otherwise the next acknowledgment or tick finds it. - **Group commit.** With an entry proposed and uncommitted, staged sections wait and go out together in the next entry when it commits; an idle group proposes at once. A second entry (up to four) goes out only while the oldest has waited one and a half times a usual round, at least 40 µs after the last, and a round that reads the acknowledgment committing an entry proposes what staged meanwhile in the same round. Each shard keeps up to four sections in flight, so under load its next records join the next entry instead of waiting for the commit. - **The checkpoint check is off the commit path.** An index grows as entries commit (`log3::Uncovered`) and tells the I/O thread where each shard's records start past its checkpoint; the image check runs when a checkpoint is pinned, a build finishes, or every 25 ms, so `io_thread_tick` stays short however long the log since the last image. - **`O_DIRECT` and `RWF_DSYNC` for `vsr/log`** (Linux, `direct_io`): one system call per write and no page cache. - **Busy-polled receive** (`FUSEE_REPLICA_BUSY_POLL_US`, off by default): the I/O thread keeps polling its sockets that long after any activity. On hosts whose network interface moderates receive interrupts adaptively, turning that moderation off (`ethtool -C adaptive-rx off`) lowers the step's p99 more. **Throughput.** On three i8g.2xlarge in one placement group, a group sustains 1.3 million steps a second with step p99 under 1 ms ([benchmark](benchmark.md#replicated)). ## The protocol log on disk `vsr/log` holds checksummed entry frames in prepared 1 MiB extents. On Linux with `LocalOptions::direct_io` (the default) it is written through `O_DIRECT`, with `RWF_DSYNC` under `FlushMethod::Dsync` (no separate flush), in whole 4 KiB blocks from aligned memory, the shared log's way: the log keeps its last partial block and the next write starts at that block and writes it again, with the same bytes, ahead of its own; two writes in flight never share a block, because a write that shares the previous one's last block starts only after it completed (`replicated/writes.rs`, the shared log's `batch_span` rule). Extents are zero-filled through `O_DIRECT` and flushed when prepared, so writes on the commit path overwrite allocated blocks and change no metadata. A filesystem that refuses `O_DIRECT` (`EINVAL` at the open or at a write) moves the log to buffered writes and `fdatasync` for good, with one line on stderr; the refused write is written again buffered (it transferred nothing or a prefix of the same bytes). A torn rewrite of a shared block can only leave old or new bytes in each sector, and its earlier frames are the same in both. The superblock, commit hint and membership file stay buffered. **Checked.** `writes_keep_every_durable_entry_in_any_crash_state` (unit test) writes 400-entry logs packed and in whole blocks, with up to two writes in flight as the primary's background writes run them, and crashes each of 600 runs at a random point with every write in flight landed, lost, or torn at random 512-byte sectors: every entry of a completed write reads back and nothing read back differs from what was written, including padding shorter than a frame's length field before a block boundary. `a_refused_direct_write_falls_back_to_buffered_writes` and `the_protocol_log_writes_through_o_direct_where_the_filesystem_allows` (which asserts the `O_DIRECT` descriptor on Linux) cover the rest; the replicated tests run with it on Linux. ## Storage and verification - **Simulator** (`FUSEE_VSR_SEEDS=5000 cargo test --release -p fusee-core vsr`, eleven tests, 5,000 seeds each where they take seeds): 12,794,098 entries committed, 65,743 crashes, 13,142 damaged-entry repairs and 585,387 checkpoint publications or installations, with message drops, duplicates, reordering, partitions and torn unflushed writes; slow primary 12,775,055 entries, 3,938,552 of them committed ahead of the primary's own flush; default timing 13,551,619 entries over 112,034 views; drift 21,756,164 ms under a lease; membership changes 9,179,100 entries, 15,000 epochs entered, 8,138 requests through a backup, 47,646 crashes; no violation. - **TLA+:** TLC checked 5,815,792 distinct states of the bounded replication model, and the lease and membership models above. [Configuration and results](../spec/README.md). - **Storage:** `fusee_server::replicated`. The log is replicated, not a shard: one protocol entry is one shared-log batch (`replicated/log3.rs`), a section of canonical records from every shard of the primary's engine that had records, behind a small header naming the group's shard count (the entry's op, view and chain checksum stand in for the batch header). One group therefore serves all of the primary's shards: one per core up to eight for a new group, and afterwards the count its log was written with (`ReplicaNode::engine_config` sets it; keep it). The primary's engine stores each shard through `ReplicatedEnv`: a shard's flush completes once the entry holding its section commits. **One copy on disk:** the protocol log is the only log. A new primary's engine recovers each shard from the committed entries after its checkpoint, as a local engine recovers from the shared log; no replica materializes segment files. Each replica keeps `vsr/log` (checksummed entries in prepared 1 MiB extents, replaced atomically during repair), `vsr/superblock` (saved after the log it vouches for), `vsr/commit` (a 12-byte commit hint rewritten in the background every 200 ms, so a damaged committed tail is repaired rather than truncated) and `engine/shard-NNN/` (each shard's run store and checkpoints). - **Dedicated servers' groups** keep one shard per group, with flush batches materialized into `engine/shard-000/log`, a periodic synced hint and a checkpoint barrier, because their remote workers recover by reading those segments. A group's first engine decides its kind, and a dedicated server refuses a group that an embedded engine already wrote. - **I/O thread:** one current-thread runtime per member reads every peer socket, runs the protocol and owns the protocol log. An inbound frame is handled on the task that read it; there is no inbound channel. Every round ends with at most one socket write per peer (frames coalesced; a Prepare encoded once for all backups) and one log write. Shards hand the I/O thread their sections (`log3::Batcher`) and wake it only if it is parked, or a lone shard runs the round itself ([Where a step's time goes](#where-a-steps-time-goes)); the round proposes everything staged as one entry, several entries in one Prepare when permit commands are waiting too, waiting for the entry in flight to commit unless its round runs slow (group commit, above). On commit it publishes each shard's committed LSN and wakes the shard, which spins briefly before it parks (the engine's spin-then-park). Right after such a release the I/O thread spins in turn for the shards' next section, polling its sockets without blocking meanwhile, within the budget of `fusee::spin::Budget` (twice how soon sections followed recent releases, at most 50 µs, never on hosts with two or fewer cores), then parks; a wait for acknowledgments parks at once and the reactor wakes the reader. The primary's own log write and flush run on two helper threads (`replicated/writes.rs`); while the log is busy every write is padded to the end of its 4 KiB block (`fusee_core::wal::batch_span`, the shared log's rule), so the next can start at once and two writes in flight never share a block. Backups write inline: their flush is on the commit path. Permit commands and a dedicated server's batches and checkpoint barriers reach the I/O thread through bounded queues (1,024 and 16 entries); the I/O thread never sends into them and drains them without waiting for a producer, a peer or a commit, so no cycle of waits exists. Status receivers are woken only when the role, view, readiness or stop state changes; progress fields update in place and are read by polling. - **End-to-end test** (`crates/fusee-server/tests/replicated.rs`): three replicas; the primary runs workflows and is killed; a backup becomes primary, opens the engine from its own log, and the workflows finish with their recorded steps not run again; the old primary restarts and catches up as a backup. Other tests there: every shard fails over from the only copy of the log (no segment file anywhere), a slow primary disk (above), an engine reopened by the same primary, checkpoint images of every shard installed by a lagging member. - **Linux partitions** (`bench/faults/partitions.py`, three containers, primary isolated, ±90 s clock skew, 20 ms delay, 2 % loss, 1 % duplication): all 1,000 acknowledged workflows completed with 3,000 consistent external keys, and every member installed a checkpoint image; under a two-permit limit all 60 completed with at most two concurrent effects across all three members. - **Real processes** (`fusee-crashtest replicated-driver stream`): three replica processes; the primary runs workflows whose steps call an external-service log, and starts another every 2 ms; the driver SIGKILLs whichever process is primary, restarts it, and repeats. 50 runs of 20 kills, 1,000 primary SIGKILLs: 5,000 acknowledged workflows, all finished; 152,085 service calls, no key called with two different inputs. - **Limits:** one group per primary; a dedicated server's groups are single-shard with materialized segments; hot standby keeps a promoted backup's engine open but its shards still replay the records since their checkpoint (from memory, see [Failover timing](#failover-timing)); membership changes one quorum-preserving step at a time, and the TypeScript and Python SDKs do not expose them (the CLI and `ReplicaNode::change_members` do); a group's first checkpoint waits for every reachable configured member to advertise support (see below); a group without TLS and a token ([Security](#security)) requires an isolated trusted network. ```rust let mut node = ReplicaNode::start(ReplicaConfig { id, peers, join: false, dir, local, timing, security }).await?; loop { node.wait_ready().await; // primary, caught up let engine = Engine::open(node.engine_config()).await?; register_workflows(&engine); node.wait_lost().await; // deposed: stop work let _ = engine.shutdown().await; } ``` ## Security Members authenticate each other with mutual TLS 1.3, a shared token, or both (`ReplicaConfig::security`, a `fusee_server::security::Security`; the default is neither, which keeps the plain protocol and prints a warning for a non-loopback replication address). With the CLI: ```bash fusee server --dir /var/lib/fusee --listen 10.0.1.10:7600 \ --replicas 10.0.1.10:7400,10.0.1.11:7400,10.0.1.12:7400 --replica-id 0 \ --tls-cert member-0.pem --tls-key member-0.key --replica-tls-ca group-ca.pem \ --token-file /etc/fusee/token ``` `--tls-cert` and `--tls-key` also secure the worker listener; `--tls-ca` (instead of `--replica-tls-ca`) also requires worker certificates. `--replica-token-file` gives the replicas their own token; it defaults to `--token-file`. An embedded member (the TypeScript and Python `replicated` durability) takes the same settings as the `address` option: `?cert=PATH&key=PATH&ca=PATH&token-file=PATH`. - **Mutual TLS.** Every stream is TLS 1.3 with a certificate in both directions, verified against the group CA. A member's certificate must be valid for the host of its own replication address (its DNS name or IP address) and, if it lists extended key usages, allow both server and client authentication. A member that dials a peer verifies that host. A member that accepts a stream admits it only for the members whose hosts the certificate names, so a certificate the same CA issued to a worker, or for any other name, never speaks for a member. - **Token.** Before its first frame, every stream runs the token exchange of [server mode](server-mode.md#security) with purpose 2 and the dialler's member ID in its hello. The listener then accepts only frames carrying that ID, and under TLS only an ID the certificate names. A proof made for the worker protocol is never accepted here, nor the reverse. - A stream that fails either check is closed before one frame reaches the protocol. A frame that claims another member than its stream's closes the stream. A dialler whose handshake is refused retries every 500 ms; failed handshakes are logged at most once every ten seconds. - TLS handshakes of inbound streams run on the node's general runtime, not on the I/O thread, so connection attempts never take its time. Established streams are read on the I/O thread. Replication traffic counters count plaintext bytes. Members holding the group's credentials are trusted alike: a member's key can speak for every member its certificate names, and a token alone for any member. A token without TLS authenticates each stream's start but does not encrypt or protect the traffic after it. ## Replica checkpoints After a workflow checkpoint is durable, the primary associates its last LSN with a committed protocol entry. It builds an immutable image containing the workflow checkpoint, its run-store prefix, retained history archives, the boundary segment through that entry, and the permit table at the same operation. Image copies and SHA-256 validation run off the protocol thread. A generation pointer selects the image, verified log suffix and applied watermark together. Recovery rebuilds mutable engine files from the image and replays the suffix. A lagging member fetches pages from the established primary or the donor selected by the view-change quorum. It keeps its old generation until the image and every locally committed position above it are accounted for. A healthy backup also keeps prepares it acknowledged in the current view. A member with a damaged image retains its saved view and commit watermark and refrains from voting until it receives a verified replacement. A group that loses every copy of a committed value cannot reconstruct it. Compaction never waits for a member's progress: a member below the new base installs the image. Only a group's first checkpoint waits: the primary waits until every configured member has advertised checkpoint support in its Hello (sent every 500 ms). A member that has neither sent a frame nor kept the primary's connection open for 30 seconds is down or cut off and no longer holds the log; when it returns it takes the offered image and a verified suffix like any lagging member. A reachable member that never says Hello still holds the first checkpoint back. Once a replica holds a generation, its compaction waits for nobody. This rule decides when compaction may start, never what it drops: an image covers only committed entries and is durable before the log is trimmed. A long run of permit or connection records can also advance the checkpoint when no workflow executes. The primary copies the previous immutable engine image, verifies the original SHA-256 while reading it, and substitutes the lease state and protocol position for the new committed prefix. With no earlier workflow records, the image contains only the lease state. A primary creates this empty-engine format only after every member that has said Hello advertises support, under the same reachability rule. An application reopening such a group starts with an empty engine and its previously committed limits. Workflow batches after the checkpoint replay normally. # Deploying Fusee - [`systemd/fusee.service`](systemd/fusee.service): one server on a host, with the token, the HTTP API on :7601 and hardening. Instructions in the file. - [`helm/fusee`](helm/fusee): a StatefulSet. `members: 1` is one server; `members: 3` is a replicated group (each member keeps a copy; a majority must be up). The chart generates the token into a Secret unless you set one. `helm install fusee deploy/helm/fusee --set members=3`. - [`terraform`](terraform): installs the chart with Terraform (Helm provider), generates the token, and outputs the addresses. - [`monitoring`](monitoring): Prometheus alert rules and a Grafana dashboard over `/metrics`. - The [`Dockerfile`](../Dockerfile) at the root builds the server image. Put the data directory on a disk that honours flushes; `fusee doctor ` checks it. A Python worker connects with `Engine.open(address=fusee.connection_url(addr, token_file=path))` and a client with `Client.connect(addr, token_file=path)` (see each SDK's README); tools use the HTTP API with the token as `Authorization: Bearer`. ## Who can reach what `fusee server` serves its own protocol (workers and clients, `--listen`) and, with `--http host:port`, the HTTP API, the UI and `/metrics`. Each token is a file; the HTTP API takes it as `Authorization: Bearer `, workers and clients as `?token-file=PATH` in the address. | Token | Flag | Server protocol | HTTP API | | --- | --- | --- | --- | | worker | `--token-file FILE` | holds shards, any call | everything | | client | `--client-token-file FILE` | calls (start, signal, query, ...), no shards | everything | | read-only | `--read-token-file FILE` | calls that read (list, describe, query, result, ...) | every `GET`; a change answers 403 | | a namespace's | `--namespaces-file FILE` | | runs whose ID starts with `/`; schedules, objects and metrics need a full token | Behind a proxy that signs people in (oauth2-proxy, an identity-aware load balancer), pass `--identity-header X-Forwarded-User` (and `--http-insecure` if the API listens off loopback with no token): calls then need that header or a token, the UI shows who is signed in, and every signal, cancellation, termination and reset records the user ([audit](../docs/features.md#audit)). Only the proxy may reach the API. Server-protocol tokens are at least 16 bytes (`openssl rand -hex 32`). A client or read-only token needs the worker token too. The scope is fixed when a connection is made: a client's token is refused at the worker's hello, and a reader's calls that change something answer `Rejected`. `kill -HUP ` reloads the certificates and every token file (and the namespaces file) for connections made after it: write the new token, send `SIGHUP`, then move the callers over. A file that does not load keeps the old settings and the server says why. The replica protocol's certificates and token need a restart. A namespace is a prefix of the workflow ID. The file holds one namespace and its token per line: ``` # namespace token acme 9c1f0e... globex 47ab2d... ``` With `acme`'s token, starting `acme/order-1` works and starting or reading `globex/order-1` answers 403; `GET /v1/runs` lists only `acme/` runs, and `/v1/info` names the namespace. Payloads the application encrypted show as *encrypted* in the UI; to read them there, run `fusee codec-server` where the keys are and pass its URL as `--codec-url` ([payload encryption](../docs/features.md#payload-encryption)). # Operations ## Data directory ``` /fusee.meta format, shard count and log format /wal/.wal the shared log (log format 3): segments from 1 MiB doubling to 64 MiB, reused /shard-000/log/.seg per-shard log segments (log format 2) /shard-000/runs-.dat run store (parked and finished runs) /shard-000/checkpoint-.fck the two newest checkpoints (a second one of the same position: b.fck) /shard-000/visibility.redb rebuildable search-attribute index /shard-000/archive/ immutable retained journals, when enabled /reshard.new/, reshard.old/ only while `fusee reshard` runs ``` The shard count is set when the directory is created. An engine opened with another count refuses the directory and names the command that changes it (see [Changing the shard count](#changing-the-shard-count)); the TypeScript, Python and Go SDKs pass their `shards` option to the same check. ## Tools - `fusee disk-check [seconds]`: flush latency of the disk under `dir` (`--json` for machine-readable output). - `fusee inspect `: log segments (the chain's batch ranges, records per shard, torn tails, spares), checkpoints and the log position each covers. - `fusee upgrade `: converts a closed log-format-2 directory to the shared log (see Format versions). - `fusee reshard `: changes a closed local directory's shard count (see Changing the shard count). - `fusee restore [--at