Durable execution

Some work outlives the request that starts it: a multi-step fulfilment that runs for days, retries a flaky payment, waits on a human approval, and must survive a deploy or a crash in the middle. In-process sagas coordinate steps within one process; durable execution runs the orchestration against a store that persists every step — so a crash resumes exactly where it left off, not from the top. That store can be an external engine (Temporal / Inngest) or the self-hosted tier on the database you already operate — Postgres or MongoDB.

The mental model: journaled progress¶

A durable workflow is ordinary code whose progress the engine journals. Each step's result is recorded, so after a crash the engine replays the workflow and skips the steps already done — the slow external calls, the timers, the waits resume rather than repeat. You write the orchestration; the engine owns the durability, retries, and timers. That's the difference from a queue task, which runs once with basic redelivery and no memory of where it was.

Three forms¶

• Workflows — multi-step, long-running, and observable: a start returns a handle immediately, and a query port reads coarse status, the typed result, or in-flight state. Signals and updates push messages into a running workflow.
• Schedules — fire a workflow on a cron or interval; the durable counterpart to a queue's delayed jobs.
• Functions — event-triggered work composed of individually-retried, memoized steps (the Inngest model).

A workflow start returns a handle you observe through the query port:

handle = await workflows.start(FulfilOrder(order_id=order_id), workflow_id=f"fulfil-{order_id}")
run = await queries.describe(handle)          # coarse status: RUNNING / COMPLETED / FAILED / …
result = await queries.result(handle)         # the typed return value, once complete

A stable workflow_id makes start idempotent — the same id won't launch a second run.

Self-hosted on your own database¶

The external engines are operational dependencies — a Temporal cluster or the Inngest service. For the common deployment that runs one database and no engine, the self-hosted tier gives you the functions form (memoized steps + crash recovery) and crash-resumable sagas on that database, with nothing to stand up. Postgres and MongoDB both implement it, behind the same ports and the same runner.

It reuses the journaled-progress model, backed by two app-provided relations: a durable_step memo journal (each step's result recorded so a replay skips it) and a durable_run run store (run instances, claimed for recovery — under FOR UPDATE SKIP LOCKED on Postgres, under a token-stamped batch claim on Mongo, which has no row locks to take). Wire both on your database's module and drive them with the forze_kits runner:

deps = PostgresDepsModule(
    client=client,
    durable_step=PostgresDurableStepConfig(relation=("public", "durable_step")),
    durable_run=PostgresDurableRunConfig(relation=("public", "durable_run")),
)

registry = DurableFunctionRegistry()
registry.register("fulfil-order", fulfil_order)      # async (ctx, input) -> output
runner = DurableFunctionRunner(registry=registry)

await runner.enqueue(ctx, "fulfil-order", {"order_id": str(order_id)})

On MongoDB the wiring is the same shape, over collections instead of tables:

from forze_mongo.execution.deps import MongoDurableRunConfig, MongoDurableStepConfig

deps = MongoDepsModule(
    client=client,
    durable_step=MongoDurableStepConfig(collection=("app", "durable_step")),
    durable_run=MongoDurableRunConfig(collection=("app", "durable_run")),
)

The run collection needs one index — a partial unique index on idempotency_key — or two simultaneous submits of one key can both insert. See MongoDB → durable execution for it and the two optional ones.

A registered function does its work in steps via the step port; each step memoizes, so a re-invocation after a crash replays completed steps and resumes at the first incomplete one:

async def fulfil_order(ctx, input):
    step = resolve_durable_step(ctx)
    charge = await step.run("charge", lambda: charge_card(ctx, input))   # journaled once
    await step.run("ship", lambda: ship(ctx, charge))
    return {"shipped": True}

A background scanner re-claims runs abandoned by a crash and re-invokes them — durable_recovery_background_lifecycle_step(runner=runner). It is multi-worker-safe: concurrent scanners never claim the same run (FOR UPDATE SKIP LOCKED) and a terminal write is fenced against a reclaimed lease, so a stalled worker whose lease expired can't finish a run the new owner already took over. A still-executing run heartbeats its lease alive, so a long body is never reclaimed mid-flight — bounded by the runner's max_run_duration (default 1 hour): past the cap the body is cancelled while the lease is still held (nothing double-executes) and the run lands TIMED_OUT with the deadline reason, so a body hung on a dead peer can't pin a recovery slot forever. TIMED_OUT is its own terminal rather than a flavour of FAILED because the two send you to different places — FAILED to the body's code, TIMED_OUT to the cap, the workload's size, or a hung peer. Set it above your longest body, or None to remove the cap; re-enqueue a timed-out run to retry it. Run the scanner on every replica, or pair it with the singleton lifecycle guard to elect one. max_concurrency bounds how many runs a sweep recovers at once. Enqueue with run_at=<when> for a delayed run — the scan skips it until it's due.

Multi-tenant. The stores resolve their table under the bound tenant. On a tagged shared table (a tenant_id column), an unbound scanner recovers every tenant's runs and the runner re-binds each run's tenant to execute it. On a namespace store (a per-tenant relation resolver, one table per tenant schema) pass tenants=… to the lifecycle step: each sweep binds every assigned tenant in turn and recovers its table — shard the tenant set across instances to parallelize.

Each operation resolves one tenant and uses it for everything it decides — the table, the row's tag and the scoped id together. enqueue(tenant_id=…) and a schedule record carrying a tenant_id are therefore honoured in full, including which table the row lands in, so a control plane bound to nothing can register work for a tenant and that tenant's scanner will find it. Naming a tenant that contradicts a bound one is refused (authentication / tenant_mismatch) rather than applied to some of the three.

What a bound caller reaches splits by what the verb does. The verbs taking a run id — begin, renew, load and every terminal write — reach that tenant's runs and untagged ones; the untagged arm is what keeps a run belonging to no tenant completable, since a terminal write that matched nothing would leave it reclaimed and re-run forever. The verbs that enumerate or control — the recovery scan, list_runs, request_cancel, refuse_cancel — match the tenant exactly. Unbound reaches everything, which is what lets one sweep serve every tenant.

Declaring required_tenant_isolation covers the durable routes: a durable_step, durable_run or durable_schedule wired below the declared floor now fails at wiring rather than passing on the strength of the other planes. Note the coupling that creates — tenant_aware=True on a run store fails closed when nothing is bound, which is how the cross-tenant sweep runs, so a deployment declaring tagged runs per-tenant recovery (tenants=…) instead.

Stopping a run¶

A "Stop" button on a self-hosted run goes through request_cancel:

await runner.request_cancel(ctx, run_id)      # True if the ask was recorded

It is cooperative, and the framework will not pretend otherwise. There is no in-process red button in Python: a thread can't be killed, asyncio cancellation lands only at await points, and a body blocked inside a C extension notices nothing until it returns. So this requests a stop. A PENDING run lands CANCELLED at once — nothing is executing. A RUNNING one keeps going until its lease holder sees the stamp on the next heartbeat (lease_for / heartbeat_divisor, 100 s at defaults — tune those two if you need tighter), at which point the body is cancelled at its next await and the run lands CANCELLED. A body that never awaits is bounded only by max_run_duration; if you need bounded-latency stop, structure the body for it (await regularly, run_cpu with checkpoints at chunk boundaries). True hard kill exists only at process/container granularity and is a deployment concern.

CANCELLED is deliberately not FAILED: nothing is wrong with the code, so it should not page anyone. Completed steps stay journaled, so re-enqueueing a cancelled run replays them rather than redoing the work.

The ask is unfenced and the landing is fenced: anyone may ask, but only the run's current claim holder may write the terminal state, so a stalled replica can't cancel a run its new owner is midway through. If the holder dies with the stamp down, the recovery scan claims the run and lands it without invoking the body.

Cancelling a saga is decided by the pivot. Before it, compensations run (journaled, as always) and the run lands CANCELLED — stopping mid-flight is safe precisely because compensation exists. At or after it, the request is refused and the saga completes forward: honouring it there would manufacture a FORWARD_INCOMPLETE by operator request. The refusal is recorded on the run (cancel_refused_at) so the operator who pressed Stop and watched it finish anyway gets the reason.

A run paused waiting on user input is a different case: its durable run already completed cleanly (it succeeded at producing the question), so there is nothing to request_cancel. Cancelling there is a domain action on your own task record, not a run-control call.

Engine-backed tiers advertise whether they can do any of this (durable_run_control_capabilities(port).supports_cancel); runner.request_cancel fails closed against one that can't, rather than accepting a request it would silently drop.

Observability. Pass DurableTelemetry.create() to the runner and scheduler for OpenTelemetry: a durable.run span per execution plus forze.durable.runs / forze.durable.run.duration (by name + outcome — completed / failed / forward_incomplete / cancelled / timed_out / reclaimed / interrupted (the body never finished — a drain or shutdown) / unrecorded (the body finished but its terminal write went unacknowledged — the row may already be terminal or may still await recovery, and the worker cannot tell), forze.durable.recovered, and forze.durable.schedule.fires metrics. Emits via the global OTel providers — configure the SDK in your app.

The step journal is exactly-once for a step's recorded result — a completed step replays from the journal instead of re-running. It is not exactly-once for arbitrary side effects: a body can run more than once if a worker is reclaimed (its run lease expired mid-body) or crashes before the result is journaled — the same at-least-once step guarantee as Temporal / DBOS / Inngest. Keep step bodies idempotent (use an idempotency key on external calls) for exactly-once effects. The replay guarantee is checked by deterministic simulation: a seeded crash fault kills a run mid-step, the recovery scanner re-invokes it, and the oracle asserts every completed step replays from its journal instead of re-executing. Keep durable bodies deterministic too — read time / ids through utcnow / uuid7 and do work in steps — and the simulator explores the crash-point space for you.

Crash-resumable sagas¶

The self-hosted tier closes the "an in-process saga is not crash-resumable" gap. Swap the saga executor for the durable one and run the saga as a durable function — each step and each compensation is journaled, so a crash mid-saga (or mid-rollback) resumes instead of leaving committed steps un-compensated:

deps = SagaDepsModule(executor=DurableSagaExecutor())      # swap the seam
registry.register(str(saga.name), durable_saga_handler(saga, OrderCtx))
await runner.run_now(ctx, str(saga.name), initial.model_dump(mode="json"))

The saga context must be a serializable pydantic.BaseModel (it is journaled between steps). A step failure classified retryable (infrastructure, throttled, concurrency) is retried in place with backoff before it is journaled — compensation only runs on genuine failures or exhausted retries, never a one-off blip. This tier is self-hosted only (Postgres or MongoDB); a full workflow engine (timers, signals, versioning) is still Temporal/Inngest. The relations come from your migrations; their schema — and the one index the Mongo run store needs — is documented on the adapter classes.

Recurring schedules¶

A durable_schedule table + the DurableScheduler fire a run on a cron cadence. Put a schedule and run the scheduler step (alongside the recovery step, which executes the runs it enqueues):

deps = PostgresDepsModule(client=client, durable_run=…, durable_schedule=…)
# …or MongoDepsModule(client=client, durable_run=…, durable_schedule=…)

await scheduler.put(ctx, "nightly-report", "report", "0 3 * * *", tz="Europe/Berlin")

lifecycle = [
    durable_scheduler_background_lifecycle_step(scheduler=scheduler),
    durable_recovery_background_lifecycle_step(runner=runner),
]

await scheduler.remove(ctx, "nightly-report") unregisters a schedule so it never fires again — distinct from pausing it (put it back with enabled=False, which keeps the row).

Or declare the cadence on the function and let it wire itself: a DurableFunctionSpec with a DurableFunctionCronTrigger auto-registers its schedule when you pass the specs to the step — no manual put:

spec = DurableFunctionSpec(
    name="report",
    run=DurableFunctionInvokeSpec(args_type=ReportArgs),
    triggers=(DurableFunctionCronTrigger(expression="0 3 * * *"),),
)
durable_scheduler_background_lifecycle_step(scheduler=scheduler, specs=[spec])

Auto-registration is idempotent — a restart re-uses an unchanged schedule (so it never resets next_fire_at and skips a due fire) and re-registers only when the cron changes.

Firing is fire-once / skip-missed: if the scheduler was down across several occurrences it fires once and advances to the next future one (no backfill). It's exactly-once across replicas — each fire enqueues a run keyed {schedule_id}:{fire_epoch} and the next fire is a compare-and-set — so, like recovery, it's safe to run on every node. "Now" comes from the TimeSource seam, so schedules are deterministic under simulation. Recurring schedules are self-hosted here; a full engine's timers/signals/versioning are still Temporal/Inngest.

When to reach for it¶

To start a workflow reliably from a request — only if the write commits — stage it through the outbox instead of starting it directly.

The ports and dep keys are the durable reference; the worked flows are the background work and scheduled jobs recipes.