Resonate basic durable world usage python

Resonate Basic Durable World Usage — Python

Overview

Durable functions are the payload of Resonate. They are async def coroutines that the SDK treats as a sequence of durable checkpoints. Every await ctx.run(...) or await ctx.rpc(...) is a save point: if the process crashes, the function resumes from the last successful checkpoint on another process or a restart.

This skill covers the Context API (ctx.*) that you use inside those functions. The ephemeral-world counterpart (Resonate class, registration, top-level invocation) lives in resonate-basic-ephemeral-world-usage-python.

The contract

Every durable function:

1. Is registered with r.register(fn) in the ephemeral world.
2. Takes ctx (the Context) as its first parameter; any other parameters after that are the invocation args.
3. Uses await to interact with ctx. Each awaited ctx.* call is a durable checkpoint.
4. Is deterministic when replayed: no direct time.time(), no random.random(), no raw network I/O outside ctx.run.

Minimal shape:

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from resonate.context import Context

async def process_order(ctx: Context, order_id: str) -> dict:
    # step 1 — durable: its result is persisted
    order = await ctx.run(load_order, order_id)

    # step 2 — durable: runs locally in the same worker
    charge = await ctx.run(charge_card, order)

    # step 3 — ordinary Python control flow; free to branch, loop, raise
    if not charge["ok"]:
        raise RuntimeError(f"payment failed: {charge['reason']}")

    return {"order_id": order_id, "status": "paid"}

If this function crashes after step 1 but before step 2, it resumes at step 2 on restart — load_order is NOT re-run. Its result is read from the promise store.

Same-process invocation: ctx.run

ctx.run invokes another function in the same process durably. The durable function awaits it and the result is checkpointed:

async def foo(ctx: Context, arg: str) -> str:
    result = await ctx.run(bar, arg)
    return result

async def bar(ctx: Context, arg: str) -> str:
    # bar can be any async def; it runs as a durable leaf
    return arg.upper()

To fire off multiple children and await them in parallel, launch them before awaiting:

async def enrich_batch(ctx: Context, order_ids: list[str]) -> list[dict]:
    # Launch all children (returns futures)
    futures = [ctx.run(enrich_one, oid) for oid in order_ids]
    # Await them all
    results = [await f for f in futures]
    return results

Do NOT use asyncio.gather here. ctx.run(...) returns a Resonate future, not an asyncio future; asyncio.gather(ctx.run(...), ctx.run(...)) will not work correctly. The pattern above — collect futures in a list, then await each — is the correct idiom.

Note: unawaited ctx.run(...) futures are still tracked by the runtime via structured concurrency — the parent cannot settle until all spawned children complete. But explicitly awaiting them gives you the results.

Remote invocation: ctx.rpc and ctx.detached

ctx.rpc dispatches a registered function to a remote worker process by name:

async def orchestrator(ctx: Context, batch_id: str) -> dict:
    # "worker_fn" must be registered in a reachable worker group
    result = await ctx.rpc("worker_fn", batch_id)
    return result

ctx.detached is fire-and-forget — the parent does NOT wait for the result:

async def place_order(
    ctx: Context, customer: str, sku: str, qty: int, amount: int
) -> tuple[str, str, str]:
    stock_ref = await ctx.run(reserve_stock, sku, qty)
    charge_ref = await ctx.run(charge_card, customer, amount)

    # Dispatch audit as a detached workflow; get back the promise id only
    audit_future = ctx.detached(
        "write_audit_log",
        customer=customer,
        sku=sku,
        amount=amount,
        stock_ref=stock_ref,
        charge_ref=charge_ref,
    )
    audit_id = await audit_future.id()   # await the id; do NOT await the result
    return stock_ref, charge_ref, audit_id

Use detached when the side effect should survive independently of the parent's success — post-commit hooks, notifications, audit logs.

Options

Options configure an individual call. For Context invocations, options come first via ctx.options(...):

from resonate.retry import Exponential, Never

async def foo(ctx: Context, arg: str) -> str:
    result = await ctx.options(
        retry_policy=Exponential(),
        version=2,
    ).run(bar, arg)
    return result

To target a specific worker group for rpc:

result = await ctx.options(target="db-workers").rpc("query_orders", user_id)

External promises: ctx.promise

ctx.promise creates a durable promise the function suspends on until externally resolved. This is the primitive for human-in-the-loop workflows.

async def fulfill_order(ctx: Context, order_id: str, amount: int) -> str:
    # Create the promise; its id is deterministic (derived from the workflow id)
    approval = ctx.promise()
    approval_id = await approval.id()

    # Notify whoever needs to resolve (via a leaf — side effects in leaves only)
    await ctx.run(notify_reviewer, order_id, amount, approval_id)

    # Suspend here until an external party resolves or rejects
    decision_raw = await approval

    if decision_raw.get("approve"):
        return await ctx.run(ship_order, order_id)
    return await ctx.run(cancel_order, order_id)

The external resolver calls await r.promises.resolve(approval_id, Value(data=decision)) from outside the worker (a webhook handler, a CLI, a separate process).

ctx.sleep

ctx.sleep suspends for a timedelta. The Resonate server holds the continuation; no process needs to stay alive during the sleep:

from datetime import timedelta

async def daily_digest(ctx: Context, user_id: str) -> None:
    while True:
        await ctx.sleep(timedelta(hours=24))
        await ctx.rpc("send_digest", user_id)

Pass a timedelta, not a raw float. For fake async work inside a leaf, use await asyncio.sleep(...) directly instead.

Dependencies

ctx.get_dependency(MyType) retrieves a dependency registered via r.with_dependency(obj):

import psycopg

async def write_to_db(ctx: Context, record: dict) -> None:
    db = ctx.get_dependency(psycopg.Connection)
    # Wrap the actual I/O in a helper via ctx.run so the side effect is checkpointed
    await ctx.run(_insert_row, db, record)

async def _insert_row(ctx: Context, db: psycopg.Connection, record: dict) -> None:
    db.execute("INSERT INTO orders VALUES (...)", record)

Never instantiate DB clients or HTTP sessions inside a durable function body — they would be recreated on every replay.

Side effects and the ctx.run envelope

Any side effect — HTTP call, DB write, file write, print — belongs inside a function that the durable function reaches via ctx.run. This ensures the effect is executed exactly once: the orchestrator awaits the checkpoint, the leaf runs, the result is persisted, and on replay the stored result is returned instead of re-executing.

Bad (replays re-execute the side effect):

async def foo(ctx: Context, arg: str) -> str:
    requests.post(WEBHOOK_URL, json=arg)  # runs on every replay
    return "done"

Good (side effect is checkpointed exactly once):

async def send_webhook(ctx: Context, arg: str) -> str:
    requests.post(WEBHOOK_URL, json=arg)
    return "sent"

async def foo(ctx: Context, arg: str) -> str:
    await ctx.run(send_webhook, arg)
    return "done"

Determinism: time and randomness

Never call time.time() or random.random() directly in a durable function — on replay, the values change and the function branches differently. Wrap them in a leaf via ctx.run.

v0.7.0 note: ctx.time.time() and ctx.random.random() were removed in v0.7.0. If you are migrating from v0.6.x, replace any ctx.time.time() / ctx.random.random() calls with a leaf function dispatched via ctx.run, as shown below.

import time, random

async def _get_time(ctx: Context) -> float:
    return time.time()

async def _get_random(ctx: Context) -> float:
    return random.random()

async def foo(ctx: Context) -> None:
    t = await ctx.run(_get_time)
    r = await ctx.run(_get_random)

The leaf settles once and its value is persisted; replay returns the stored value.

Common shapes

Sequential pipeline with retries:

from resonate.retry import Exponential

async def ingest(ctx: Context, source_url: str) -> dict:
    data = await ctx.options(retry_policy=Exponential()).run(fetch, source_url)
    parsed = await ctx.run(parse, data)
    await ctx.run(persist, parsed)
    return {"count": len(parsed)}

Parallel fan-out within a single function:

async def enrich_batch(ctx: Context, ids: list[str]) -> list[dict]:
    futures = [ctx.run(enrich_one, i) for i in ids]
    return [await f for f in futures]

Recursive workflow (dispatching itself via rpc):

async def crawl(ctx: Context, url: str, depth: int) -> dict:
    page = await ctx.run(fetch, url)
    if depth > 0:
        for link in page["links"]:
            ctx.detached("crawl", link, depth - 1)
    return {"url": url, "id": page["id"]}

Related skills

• resonate-basic-ephemeral-world-usage-python — the Client API companion
• resonate-basic-debugging-python — diagnosing stuck workflows, missing registrations, non-determinism regressions
• durable-execution — the foundational concept
• resonate-philosophy — "do not build these things" mindset