Resonate External System Of Record Pattern Rust

Resonate External System of Record Pattern — Rust

SDK note (0.6.0). ctx.get_dependency::<T>() and resonate.with_dependency::<T>(value) are real pub fns in the Rust SDK source (resonate-sdk-rs:resonate/src/context.rs, resonate.rs) but are not yet covered in docs/develop/rust.mdx. The APIs are safe to use; cite source paths when reviewers ask.

Overview

When a Rust workflow touches an external system with its own durability (a database, a ledger, a message broker), that system often is or should be the system of record (SoR). Resonate coordinates the workflow and guarantees at-least-once execution; the SoR enforces consistency via its own primitives (transactions, idempotency keys, CAS operations).

Rust's type-dispatched dependency injection makes this pattern particularly clean: register typed resources once at process start with resonate.with_dependency<T>(value), retrieve inside durable functions with ctx.get_dependency::<T>().

For the language-agnostic framing, see resonate-external-system-of-record-pattern-typescript. Rust's shape adds: Arc<T> returns from DI, serde-derived input types, and idiomatic sqlx / ledger-crate usage.

Core principle

Write to the system of record first. Read from it as ground truth. Never let Resonate's promise state contradict the SoR.

Resonate stores workflow state (what step succeeded, what the result was). The SoR stores business state (the account balance, the order status). When they disagree, the SoR wins; Resonate's job is to converge toward it.

Dependency setup (ephemeral world)

use resonate::prelude::*;
use sqlx::PgPool;
use std::sync::Arc;

#[tokio::main]
async fn main() -> Result<()> {
    let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
    let stripe = stripe::Client::new(std::env::var("STRIPE_SECRET_KEY")?);

    let resonate = Resonate::new(ResonateConfig::default())
        .with_dependency(pool)
        .with_dependency(stripe);

    resonate.register(create_order).unwrap();
    resonate.register(charge_card).unwrap();

    tokio::signal::ctrl_c().await?;
    resonate.stop().await?;
    Ok(())
}

Dependencies are retrieved inside durable functions by type via ctx.get_dependency::<T>(), which returns Arc<T>.

Basic shape

use resonate::prelude::*;
use serde::{Serialize, Deserialize};
use sqlx::PgPool;
use std::sync::Arc;

#[derive(Clone, Serialize, Deserialize)]
struct OrderInput {
    order_id: String,
    customer_id: String,
    amount_cents: i64,
}

#[resonate::function]
async fn create_order(ctx: &Context, input: OrderInput) -> Result<String> {
    // SoR write first (idempotent against retries via ON CONFLICT)
    ctx.run(insert_order_row, input.clone()).await?;

    // dependent effects only after the SoR write succeeds
    ctx.run(send_confirmation_email, input.customer_id.clone()).await?;
    ctx.rpc::<()>("enqueue_fulfillment", input.order_id.clone()).await?;

    Ok(input.order_id)
}

#[resonate::function]
async fn insert_order_row(info: &Info, input: OrderInput) -> Result<()> {
    let pool: Arc<PgPool> = info.get_dependency::<PgPool>();

    sqlx::query(
        "INSERT INTO orders (id, customer_id, amount_cents, status)
         VALUES ($1, $2, $3, 'created')
         ON CONFLICT (id) DO NOTHING",
    )
    .bind(&input.order_id)
    .bind(&input.customer_id)
    .bind(input.amount_cents)
    .execute(&*pool)
    .await
    .map_err(|e| Error::Application { message: e.to_string() })?;

    Ok(())
}

The leaf uses info: &Info (which also has get_dependency) instead of ctx: &Context because a pure write doesn't need sub-invocation capabilities. Returning Ok(()) lets the parent's ctx.run(...).await? record the checkpoint.

Idempotency keys — the external side

Use Resonate's deterministic invocation IDs (or derivations of them) as the idempotency key in the external system:

#[resonate::function]
async fn charge_card(info: &Info, order_id: String) -> Result<String> {
    let pool: Arc<PgPool> = info.get_dependency::<PgPool>();
    let stripe: Arc<stripe::Client> = info.get_dependency::<stripe::Client>();

    // skip if we already charged
    let existing: Option<(String,)> = sqlx::query_as(
        "SELECT charge_id FROM orders WHERE id = $1 AND charge_id IS NOT NULL",
    )
    .bind(&order_id)
    .fetch_optional(&*pool)
    .await
    .map_err(|e| Error::Application { message: e.to_string() })?;

    if let Some((charge_id,)) = existing {
        return Ok(charge_id); // already charged; Stripe dedupes on its side too
    }

    // create with idempotency key — Stripe dedupes by this
    let idempotency_key = format!("order:{}:charge", order_id);
    let charge = stripe
        .post(&format!("/charges"))
        .header("Idempotency-Key", &idempotency_key)
        .send_json(&serde_json::json!({ "amount": 100, "currency": "usd" }))
        .await
        .map_err(|e| Error::Application { message: e.to_string() })?;

    // update SoR with the charge result
    sqlx::query(
        "UPDATE orders SET charge_id = $1, status = 'paid' WHERE id = $2",
    )
    .bind(&charge.id)
    .bind(&order_id)
    .execute(&*pool)
    .await
    .map_err(|e| Error::Application { message: e.to_string() })?;

    Ok(charge.id)
}

The ctx.run(charge_card, order_id) that calls this leaf is durably checkpointed; Stripe's idempotency key dedupes on its side. Both layers see exactly one charge even if Resonate retries the step.

Reading from the SoR on resumption

Checkpointed ctx.run calls return the stored value on replay — the external call is not re-executed. When workflow logic needs the current SoR state (not the checkpointed snapshot from first-run), read it explicitly inside a leaf:

#[derive(Serialize, Deserialize)]
struct OrderRow {
    id: String,
    sku: String,
    quantity: i64,
}

#[resonate::function]
async fn fulfill_order(ctx: &Context, order_id: String) -> Result<String> {
    // checkpointed; returns stored value on replay
    let order: OrderRow = ctx.run(load_order, order_id.clone()).await?;

    // current SoR read — use for fresh data even across replay
    let current_inventory: i64 = ctx
        .run(check_inventory_now, order.sku.clone())
        .await?;

    if current_inventory < order.quantity {
        ctx.run(backorder_flag, order_id).await?;
        return Ok("backorder".into());
    }

    ctx.run(reserve_inventory, (order_id.clone(), order.quantity)).await?;
    Ok("fulfilling".into())
}

#[resonate::function]
async fn load_order(info: &Info, order_id: String) -> Result<OrderRow> {
    let pool: Arc<PgPool> = info.get_dependency::<PgPool>();
    let row = sqlx::query_as::<_, (String, String, i64)>(
        "SELECT id, sku, quantity FROM orders WHERE id = $1",
    )
    .bind(&order_id)
    .fetch_one(&*pool)
    .await
    .map_err(|e| Error::Application { message: e.to_string() })?;
    Ok(OrderRow { id: row.0, sku: row.1, quantity: row.2 })
}

#[resonate::function]
async fn check_inventory_now(info: &Info, sku: String) -> Result<i64> {
    let pool: Arc<PgPool> = info.get_dependency::<PgPool>();
    let (qty,): (i64,) = sqlx::query_as(
        "SELECT quantity FROM inventory WHERE sku = $1",
    )
    .bind(&sku)
    .fetch_one(&*pool)
    .await
    .map_err(|e| Error::Application { message: e.to_string() })?;
    Ok(qty)
}

Both are ctx.run calls → both are checkpointed. The difference is intent: load_order is the workflow's snapshot of the order; check_inventory_now is a fresh read each retry. The SoR is the tie-breaker.

Ledger-as-SoR (TigerBeetle shape)

For financial workflows, a dedicated ledger (TigerBeetle, double-entry tables) is a stronger SoR choice:

struct Ledger(tigerbeetle_rs::Client);

#[derive(Serialize, Deserialize)]
struct Transfer {
    from_account: u128,
    to_account: u128,
    amount: i64,
    transfer_id: u128,
}

#[resonate::function]
async fn transfer_funds(ctx: &Context, t: Transfer) -> Result<String> {
    // the ledger rejects duplicate transfer_ids as part of its contract
    ctx.run(post_ledger_transfer, t.clone()).await?;

    // after ledger commit, these are safe
    ctx.run(notify_both_parties, t.clone()).await?;

    Ok(format!("transfer:{}", t.transfer_id))
}

#[resonate::function]
async fn post_ledger_transfer(info: &Info, t: Transfer) -> Result<()> {
    let ledger: Arc<Ledger> = info.get_dependency::<Ledger>();
    ledger
        .0
        .create_transfer(t.from_account, t.to_account, t.amount, t.transfer_id)
        .await
        .map_err(|e| Error::Application { message: e.to_string() })?;
    Ok(())
}

async fn notify_both_parties(t: Transfer) -> Result<()> {
    // ...
    Ok(())
}

On replay, post_ledger_transfer returns the stored checkpoint value — the ledger is hit exactly once across all retries, and the ledger's own idempotency on transfer_id is the belt-and-suspenders.

Distinct Rust idioms

• Type-dispatched DI: ctx.get_dependency::<PgPool>() or info.get_dependency::<Ledger>() — no string keys. Newtype-wrap when you need multiple values of the same logical type (struct PrimaryDb(PgPool), struct AnalyticsDb(PgPool))
• Arc<T> return — dependencies come back as Arc<T>; deref with &*arc when passing to query functions that take &Pool
• sqlx::query_as + tuple/struct derivation — typed query results without a separate ORM
• .map_err(|e| Error::Application { message: e.to_string() }) — convert sqlx/stripe/etc. errors into Resonate's Error for ? propagation. The user-facing error variant in 0.6.0 is Error::Application { message: String } (resonate/src/error.rs); verify the variant against the SDK version you depend on
• ON CONFLICT DO NOTHING / ON CONFLICT DO UPDATE (Postgres) — natural idempotency on inserts
• info: &Info for leaves that only need DI — skip &Context when the leaf doesn't orchestrate sub-tasks; signals intent (pure write) and lets the SDK optimize

Avoid

• Instantiating DB clients inside a durable function — each replay would reconstruct the client; dependencies must be created once in ephemeral-world and shared via with_dependency
• Writing to two systems without a clear SoR — if both fail partially, who wins? Pick one SoR; make the other a compensatable side-effect or best-effort secondary write
• In-memory mutations outside ctx.run — not durable; replay loses the mutation. Always wrap state changes in a checkpointed leaf
• Mixing ctx: &Context on a pure leaf — signals intent poorly; leaves that only hit the SoR should use info: &Info

Related skills

• resonate-basic-durable-world-usage-rust — ctx.get_dependency::<T>(), info: &Info leaves, builder options
• resonate-basic-ephemeral-world-usage-rust — with_dependency::<T>(value) registration
• resonate-saga-pattern-rust — when the SoR doesn't cover all steps and you need compensation
• resonate-human-in-the-loop-pattern-rust — when the SoR update waits on external decision
• durable-execution — foundational replay semantics; this pattern is checkpoint-centric