Skip to main content

Crate durare

Crate durare 

Source
Expand description

§durare

A DBOS-style durable execution library for Rust.

Write ordinary async code; wrap each side-effecting unit in a step. Every step’s result is checkpointed to a StateProvider (Postgres, SQLite, or in-memory). If the process crashes or restarts, call DurableEngine::recover and every unfinished workflow resumes exactly where it stopped — completed steps are served from their checkpoints instead of re-running.

There is no separate server: the engine is a library that runs inside your worker and talks directly to the database, using the same dbos system schema as the DBOS Transact SDKs for Python, Go, and TypeScript — the SDKs interoperate on one database.

The problem this solves: any multi-step operation has crash windows — after the card is charged but before the receipt is sent, the process dies, and a naive retry charges twice. In durare the workflow id is an idempotency key and every step is checkpointed, so a duplicate trigger (a retried webhook, a double-click, a crashed-and-rerun caller) attaches to the same run instead of repeating its effects. This example is a test — the assertions at the bottom hold on every commit:

use durare::{DurableContext, DurableEngine, InMemoryProvider, Result, WorkflowOptions};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};

/// Stand-in for a payment API we must never call twice for the same order.
static CHARGES: AtomicU32 = AtomicU32::new(0);

#[durare::step]
async fn charge_card(ctx: &DurableContext, order_id: String) -> Result<String> {
    CHARGES.fetch_add(1, Ordering::SeqCst);
    Ok(format!("ch_{order_id}"))
}

#[durare::step]
async fn send_receipt(ctx: &DurableContext, charge_id: String) -> Result<()> {
    // A crash between the two steps does NOT re-charge: on restart,
    // `charge_card` is served from its checkpoint and the workflow
    // resumes right here.
    println!("emailing receipt for {charge_id}");
    Ok(())
}

#[durare::workflow]
async fn process_order(ctx: DurableContext, order_id: String) -> Result<String> {
    // Reads like ordinary async code; each step checkpoints once.
    let charge_id = charge_card(&ctx, order_id).await?;
    send_receipt(&ctx, charge_id.clone()).await?;
    Ok(charge_id)
}

let engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?;
engine.recover().await?; // after a crash: resume every unfinished workflow

let handle = engine
    .start_with(ProcessOrder, "1001".into(), WorkflowOptions::with_id("order-1001"))
    .await?;
assert_eq!(handle.await?, "ch_1001");

// The same trigger arrives again — same workflow id, no second charge.
let duplicate = engine
    .start_with(ProcessOrder, "1001".into(), WorkflowOptions::with_id("order-1001"))
    .await?;
assert_eq!(duplicate.await?, "ch_1001");          // served from the checkpoint
assert_eq!(CHARGES.load(Ordering::SeqCst), 1);    // charged exactly once

Swap InMemoryProvider for PostgresProvider and the same guarantees hold across processes, restarts, and a fleet of workers. For the full crash-and-recover demo — a process killed mid-workflow, restarted, and finishing without repeating work — run examples/order.rs.

§What’s in the crate

§Guides

Five module-level guides explain the concepts in depth, std-style, each with tested examples: start with durability (checkpoints, replay, and the determinism contract — read this first), then its companion determinism (the rules for writing a correct workflow body — deterministic control flow, durable-safe data, and dependencies), and queues, messaging, and transactions. Eleven runnable, end-to-end examples live in examples/.

§Cargo features

Backends are compiled behind features, all on by default; enable just one to drop the other’s driver. At least one backend is required — a zero-backend build is a compile error. InMemoryProvider is always available (no feature).

  • postgres (default) — the PostgresProvider backend.
  • sqlite (default) — the SqliteProvider backend (a bundled C library; drop it with default-features = false, features = ["postgres"] for a pure-Postgres build that needs no C toolchain).
  • conductor (off by default) — the DBOS Conductor client (Conductor, ConductorConfig, AlertHandler): a websocket client for the DBOS control plane, behind a feature because it pulls in a TLS websocket stack and gzip framing.
  • admin (off by default) — the AdminServer HTTP control surface (health, recovery, and workflow management for the DBOS console/conductor and health probes), behind a feature because it pulls in the axum/hyper/tower HTTP stack.

Modules§

determinism
Writing a correct workflow body: determinism, durable-safe data, and dependencies.
durability
How durable execution works: checkpoints, replay, and the determinism contract.
messaging
Messaging and events: how workflows talk to each other and to the outside world.
queues
Durable queues: decouple submitting work from running it, with fleet-wide control over parallelism.
transactions
Transactional steps: SQL writes and the step checkpoint in one commit — exactly-once, even across a crash.

Macros§

params
Build a Vec<Param> for a transactional-step query: params![amount, id].

Structs§

AdminServeradmin
A running admin HTTP server. Drop or shutdown to stop it.
ApplySchedule
One entry for DurableEngine::apply_schedules, which creates each schedule or replaces an existing one of the same name.
AuthContext
The identity a workflow runs under: the user it was started on behalf of, the role assumed for this run, and the full set of roles available to that user. It is persisted with the workflow and flows into any work the workflow starts, so an audit trail and authorization decisions stay consistent across a workflow tree and across recovery.
Client
A registry-less, out-of-process handle over the system database.
Conductorconductor
A running conductor connection. Call shutdown to stop it.
ConductorConfigconductor
Configuration for Conductor::start.
Debouncer
Debounces a target workflow. Create one with DurableEngine::debouncer.
DebouncerClient
Debounces a target workflow from an out-of-process Client. Create one with Client::debouncer. Identical to Debouncer except the producer runs outside any engine — a launched engine (which auto-registers the internal debouncer and the target) still executes the coalesced run.
DequeueRequest
Parameters for one dequeue iteration, computed by the engine’s dispatcher from a crate::WorkflowQueue’s configuration. Plain scalars so the storage layer stays decoupled from the queue type.
DurableContext
Handle passed into every workflow function. It carries the workflow id, the state backend, the identity the workflow runs under, and a deterministic per-execution step counter.
DurableEngine
The durable execution engine.
DurableEngineBuilder
Builds a DurableEngine with a complete, conflict-checked registry.
EngineConfig
Engine-level identity settings, resolved at construction.
ExportedWorkflow
One workflow’s full durable state in a portable, backend-agnostic form: the workflow_status row plus every dependent operation_outputs, workflow_events, workflow_events_history, and streams row, each kept as a column-keyed JSON object. Produced by StateProvider::export_workflow and consumed by StateProvider::import_workflow; the conductor ships it between environments as gzipped, base64-encoded JSON. The keys match the other DBOS SDKs’ portable schema, so a workflow exported by one can be imported by another.
ForkParams
Parameters for StateProvider::fork_workflow. The fork is created directly ENQUEUED on queue_name, in the same transaction that copies the original’s checkpoints, so a fork is never observable half-made.
InMemoryProvider
In-memory StateProvider for tests and quick starts (no database needed).
ListFilter
Filter for StateProvider::list_workflows. All fields are ANDed; empty/None fields are ignored. Times are epoch milliseconds.
PortableWorkflowArgs
The cross-language workflow-input envelope: {"positionalArgs":[…],"namedArgs":{…}}.
PortableWorkflowError
The cross-language workflow-error envelope: {"name":…,"message":…,"code"?,"data"?}.
PostgresProviderpostgres
Postgres-backed StateProvider, built on sqlx and the canonical DBOS schema (workflow_status / operation_outputs).
RateLimiter
Rate limit for workflow starts on a queue. At most limit workflows may start within any trailing period window.
RegisteredWorkflow
A workflow registered on a DurableEngine, as reported by DurableEngine::list_registered_workflows. The name is the identifier the workflow is registered and persisted under.
Row
One row from a transactional-step query. Read columns by name with Row::get / Row::try_get.
ScheduleFilter
Filters for DurableEngine::list_schedules. An empty filter returns every schedule.
ScheduleOptions
Optional settings for DurableEngine::create_schedule.
ScheduledInput
The input a scheduled workflow receives on each cron tick: the tick’s wall-clock instant plus any user value attached to the schedule via ScheduleOptions::context.
SqliteProvidersqlite
SQLite-backed StateProvider.
StepAggregate
One aggregate group from StateProvider::get_step_aggregates.
StepAggregateQuery
Grouping, selected aggregates, and filters for StateProvider::get_step_aggregates: aggregate operation_outputs rows grouped by function name and/or derived status and/or a completed_at time bucket.
StepInfo
One recorded operation of a workflow.
StepOptions
Retry policy for a durable step.
TransactionOptions
Options for a transactional step: its checkpoint name, isolation level, whether the transaction is read-only, and the user-facing retry policy for application errors raised by the body.
Tx
A handle to the in-progress transaction, handed to a transactional step’s body. Runs against either Postgres or SQLite; the step’s checkpoint commits in this same transaction.
VersionInfo
A registered application version (a row of application_versions). The “latest” version is the one with the most recent version_timestamp.
WorkflowAggregate
One aggregate group from StateProvider::get_workflow_aggregates. Each aggregate is Some only when the query selected it (an unselected aggregate is None, serialized as null, matching the other SDKs).
WorkflowAggregateQuery
Grouping and filters for StateProvider::get_workflow_aggregates: count workflows grouped by one or more workflow_status columns and/or a created_at time bucket.
WorkflowHandle
A reference to a workflow execution.
WorkflowOptions
Per-invocation options for starting or enqueuing a workflow — its idempotency key, queue and priority, deduplication, timeout, app version, and authenticated identity. Build with the chained setters (e.g. WorkflowOptions::with_id then queue); all fields default to unset.
WorkflowQueue
A named durable queue.
WorkflowRegistration
A compile-time workflow registration emitted by #[durare::workflow].
WorkflowSchedule
A persisted cron schedule for a registered workflow.
WorkflowStatus
A persisted workflow instance.

Enums§

ChangeWait
A condition a blocked recv/get_event wants to be nudged about, so it can re-check the database promptly instead of waiting out its poll interval. A backend with push signalling (Postgres LISTEN/NOTIFY) maps each variant to its channel + payload; others ignore it and simply sleep.
DeduplicationPolicy
Per-workflow start options.
Error
The crate-wide error type.
ErrorCode
A stable, programmatic classification of an Error, returned by Error::code. Lets callers branch on what kind of failure occurred without matching every concrete variant. Non-exhaustive: new codes may be added as the SDK grows, so always include a _ arm when matching.
IsolationLevel
Isolation level for a transactional step. ReadCommitted is the default; RepeatableRead/Serializable give stronger guarantees but can fail with a serialization conflict, which the transactional step retries automatically. SQLite runs every transaction serializably, so the level is advisory there.
Param
A bound parameter for transactional-step SQL. Build these with params! rather than by hand in the common case.
ScheduleStatus
Lifecycle state of a persisted schedule. Stored as the cross-SDK strings ACTIVE / PAUSED.
Serializer
A serialization format for workflow data. Cheap to clone; held by each provider as the format it encodes with (decoding is format-directed).

Constants§

PORTABLE_ERROR_NAME
Generic name stored for an error that carries no cross-language type, written when an untyped error is serialized in portable mode.
STATUS_CANCELLED
Terminated by an operator; replay is refused.
STATUS_DELAYED
Enqueued with a delay; transitions to STATUS_ENQUEUED once it is due.
STATUS_ENQUEUED
Enqueued and waiting to be claimed by a dispatcher.
STATUS_ERROR
Terminal: the workflow failed and its error is recorded.
STATUS_MAX_RECOVERY_ATTEMPTS_EXCEEDED
Recovered too many times; parked until manually resumed.
STATUS_PENDING
Claimed by an executor and currently running.
STATUS_SUCCESS
Terminal: the workflow completed and its output is recorded.

Traits§

RowValue
A value readable from a transactional-step Row column, via Row::get / Row::try_get.
SerializerCodec
A user-supplied serialization codec, plugged in via Serializer::custom.
StateProvider
The pluggable durable-state backend.
Stream
Re-exported so callers can consume the asynchronous stream returned by read_stream_values (StreamExt::next) without depending on futures directly. A stream of values produced asynchronously.
StreamExt
Re-exported so callers can consume the asynchronous stream returned by read_stream_values (StreamExt::next) without depending on futures directly. An extension trait for Streams that provides a variety of convenient combinator functions.
WorkflowDef
A compile-time, typed reference to a registered workflow.

Functions§

erase
Erase a typed async fn(DurableContext, Input) -> Result<Output> into the JSON-in / JSON-out WorkflowFn the engine stores.
is_terminal
true if status is terminal (no further execution will occur).

Type Aliases§

AlertHandlerconductor
Handler invoked when the conductor delivers an alert message: receives the alert’s name, message, and metadata. Registered via ConductorConfig::alert_handler; a panic inside it is caught and reported back to the conductor as a failure rather than tearing down the connection.
Result
The crate-wide result type: Result<T, Error>.
RetryPredicate
Predicate deciding whether a step error is retryable — see StepOptions::retry_if. Returning false stops retries at once.
TxBody
The type-erased body a transactional step runs: it borrows the Tx and resolves to the step’s JSON output. Produced by DurableContext::transaction; consumed by the provider, which supplies the Tx and the surrounding transaction.
WorkflowFn
A type-erased workflow handler: takes a context + JSON input, returns JSON output.

Attribute Macros§

step
The #[step] attribute macro. Annotate an async fn(&DurableContext, args..) -> Result<T> to have its body run as a durable DurableContext::step — checkpointed once, replayed thereafter — so it reads like an ordinary async call. Turn an async fn(&DurableContext, args..) -> Result<T> into a durable step: the body is checkpointed on first run and served from the checkpoint on replay — exactly like calling ctx.step("name", || async move { ... }) by hand, but without the closure, the Box::pin, or the Ok::<_, Error> annotation.
transaction
The #[transaction] attribute macro. Annotate an async fn(&DurableContext, &mut Tx, args..) -> Result<T> to have its body run as a durable DurableContext::transaction — the SQL writes and the checkpoint commit atomically — without the |tx| Box::pin(..) wrapper. Turn an async fn(&DurableContext, &mut Tx, args..) -> Result<T> into a durable transaction: the body’s SQL writes and the step checkpoint commit in one database transaction, without the |tx| Box::pin(async move { ... }) wrapper.
workflow
The #[workflow] attribute macro. Annotate an async fn(DurableContext, Input) -> Result<Output> to have it auto-registered with every DurableEngine in the binary. Register an async fn(DurableContext, Input) -> Result<Output> as a durable workflow.