Replay and Recovery

Before a durable step runs, the worker writes step:pending. After it succeeds, it writes step:done. Recovery uses this pair to distinguish completed work from the first unfinished step.

After a crash

When a worker dies or loses its lock, another one picks up the run:

1. Claims it.
2. Replays the event stream in order, rebuilding in-memory state.
3. Resumes from the first step with no matching step:done.

Steps that already wrote step:done don’t run again. Their results come back from the stream. Only the first unfinished step and everything after it re-executes.

@workflow(name="fulfill_order")
async def fulfill_order(ctx, order_id: str) -> str:
    charge_id = await ctx.step("charge", lambda: charge_card(order_id))
    await ctx.step("email_receipt", lambda: send_receipt(order_id, charge_id))
    return charge_id

If the worker dies right after charge succeeds, the run’s stream looks like this, and another worker picks it up:

run:init
step:pending   charge
step:done      charge          result="ch_1a2b"     # worker crashes here
--- another worker claims the run and replays the stream ---
step:done      charge          # result comes from the stream; the card is NOT re-charged
step:pending   email_receipt
step:done      email_receipt
run:complete   result="ch_1a2b"

charge already has a step:done, so replay returns its recorded ch_1a2b without calling charge_card again. Only email_receipt actually runs.

The window where a step can run twice

The crash that matters is the one between charge_card returning and step:done being written:

run:init
step:pending   charge
                               # charge_card() has charged the card...
                               # ...worker dies here, before step:done
--- another worker claims the run and replays the stream ---
step:pending   charge          # no step:done, so charge runs AGAIN
step:done      charge

No engine can close this on its own: that would need the payment provider to join the run store’s transaction. The guarantee is exactly-once processing, at-least-once side effects. In a hard-kill test across 3,000 runs, 3 re-ran a completed write on recovery (~0.2% of write steps); a graceful stop produced none.

Close it with an idempotency key the downstream system dedupes on. run_id and the step name are stable across replays:

charge_id = await ctx.step(
    "charge",
    lambda: charge_card(order_id, idempotency_key=f"{ctx.run_id}:charge"),
)

ctx.once is for duplicate submissions, not crashes

ctx.once(key, fn) stores fn’s result under your key and returns it on any later attempt, across runs: the same order submitted twice, a resubmitted run, a retried enqueue.

charge_id = await ctx.once(f"charge:{order_id}", lambda: charge_card(order_id))

It writes its record after fn returns, so it has the same crash window as a step (measured: the same ~0.2%). Retries inside one ctx.step aren’t journaled either; a crash mid-retry re-runs the whole step. For the crash window, pass the key through to the system doing the write.

Cloud delivery guarantees

The store is the source of truth; Pyrula Cloud is a replica. Every critical event is eventually stored by the cloud, or the runtime reports that it can’t be. Within these bounds:

• Critical events: lifecycle (run_start, run_complete, run_error, cancel), step:done / step:error, tool results, usage, park and resume, gap markers. Token deltas, llm:start, tool:start, step:pending and heartbeats are best-effort.
• Keyed events: every run-scoped event has an idempotency key; only periodic runtime telemetry is excluded.
• Within the runtime’s control: a run refused over quota is purged deliberately and the refusal is visible.
• Within store retention: a maxlen-trimmed ValkeyStore run bounds what can be recovered (ArchivingStore extends it). A run trimmed before delivery arrives in the cloud starting mid-run, and the runtime logs cloud_sweep_truncated with the count.

Critical events are spooled to disk before sending and removed only on a cloud ack; a per-run ledger in the run store records what the cloud actually stored; and a sweep at worker start and on each orphan scan re-sends any terminal run the ledger can’t confirm. Delivery has no latency bound; see Connect to Pyrula Cloud for freshness and what happens when the cloud stops accepting.

History limits

ValkeyStore keeps up to maxlen events per run (default 100,000) and trims the oldest beyond that. A trimmed run can’t be replayed safely (its first steps would look unrun), so replay raises TruncatedHistoryError with the number of missing events. With ArchivingStore, a complete archived copy is used instead.

Use ctx.continue_as_new() to end the run and start a fresh one that carries your state forward, so no single history grows without bound. Raising maxlen only moves the ceiling.

Step results

A step’s result is stored as JSON in step:done. ctx.step raises when a result cannot be serialized. An annotated @step restores its declared result type during replay, including Pyrula sum and list types.

Replay primitives

Workers replay for you, but the primitives are public:

from pyrula.workflows import replay_all_events, reconstruct_state, ReplayState

replay_all_events reads a run’s full stream; reconstruct_state folds it into a ReplayState. What replay asks of your code is covered in Determinism.