Replay and Recovery

Before a durable step runs, the worker writes step:pending. After it succeeds, it writes step:done. That pair is the whole recovery mechanism.

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.

Doing it by hand

The replay primitives are public, though workers call them for you:

from pyrula.workflows import replay_all_events, reconstruct_state, ReplayState

reconstruct_state folds a run’s events into a ReplayState. replay_all_events walks the full stream.

What this asks of your code

A step interrupted between “side effect done” and “step:done written” runs again on recovery. So durable steps need to be idempotent, or guarded. That constraint is the subject of Determinism.

Step results

A step’s result is persisted in step:done and replayed later, so it has to be JSON-serializable. ctx.step raises if it isn’t. data_step widens this by encoding pyrula values (IList, @case classes) into the stream, but they come back as plain list/dict on replay, since the original classes may not be importable when a worker rebuilds the run. Convert at the boundary if you need the typed form back.