Crate dbos

Expand description

DBOS Transact for Rust: lightweight durable workflows on Postgres.

Pre-production release. DBOS Transact for Rust 0.6 is a pre-production release, and isn’t yet supported for production use. Its API may still change in breaking ways between releases.

DBOS checkpoints each step of a workflow in a Postgres database. When a process stops, for whatever reason, its workflows recover from those checkpoints and continue from the last completed step. It is only a library: there is no separate service or orchestrator, and Postgres is the only other thing it needs. Its system database has the same schema as the other DBOS Transact SDKs’, so a Rust application can run alongside applications written in their languages.

§Getting started

A workflow is an async fn that takes one input and returns a Result, and a step is a function whose result the workflow checkpoints. Set DBOS_SYSTEM_DATABASE_URL to a Postgres connection string (the database is created if it doesn’t exist), then:

use dbos::{Config, DBOS};

async fn workflow(_: ()) -> dbos::Result<String> {
    dbos::step("step_one", step_one).await?;
    dbos::step("step_two", step_two).await?;
    Ok("success".to_owned())
}

async fn step_one() -> dbos::Result<()> {
    println!("Step one completed");
    Ok(())
}

async fn step_two() -> dbos::Result<()> {
    println!("Step two completed");
    Ok(())
}

#[tokio::main]
async fn main() -> dbos::Result<()> {
    let mut config = Config::from_env("dbos-starter");
    config.app_version = Some("0.1.0".to_owned());
    let dbos = DBOS::new(config);

    // Register workflows before launching, so that recovery can find them.
    let workflow = dbos.register_workflow("workflow", workflow)?;
    dbos.launch().await?;

    let handle = workflow.start(()).await?;
    println!("Workflow result: {}", handle.result().await?);

    dbos.shutdown().await;
    Ok(())
}

The programming guide is the place to start, with tutorials alongside it. There are also runnable demo applications.

§Where to look

Everything is at the crate root, apart from the system database layer in sysdb, whose types some calls take and return. Free functions such as send and cancel_workflow are for workflow bodies, and fail with Error::NotInWorkflow elsewhere. Outside a workflow, call the same operation on DBOS or on a Client.

To…	Use
Configure and run an application	`Config`, `DBOS`
Define and start workflows	`DBOS::register_workflow`, `WorkflowRef`, `StartOptions`, `WorkflowHandle`
Checkpoint work	`step`, `step_with`, `StepOptions`, `PendingStep`
Race or wait for durable calls	`select_step!`, `select_workflow`, `join_workflows`
Wait and communicate	`sleep`, `send`, `recv`, `set_event`, `get_event`
Stream values from a workflow	`write_stream`, `close_stream`, `read_stream`
Limit concurrency and rate	`DBOS::register_queue`, `QueueOptions`, `QueueRequest`, `Debouncer`
Run workflows on a schedule	`Schedule`, `ScheduleSpec`
Manage workflows	`DBOS::list_workflows`, `DBOS::cancel`, `DBOS::resume`, `DBOS::fork`, `DBOS::rewind`, `DBOS::delete`; from a workflow body, `list_workflows`, `cancel_workflow`, `resume_workflow`, `fork_workflow`, `rewind_workflow`, `delete_workflow`
Change a workflow’s code	`patch`, `deprecate_patch`
Choose how values are stored	`SerializationFormat`, `CustomSerializer`
Reach an application from outside it	`Client`
Handle failures	`Error`, `DurableError`
Use the system database directly	`sysdb`

§Determinism in workflows

A workflow body must be deterministic. A recovery runs it again from the start and replays what each durable call recorded, so it has to reach the same durable calls, in the same order, with the same inputs. Anything non-deterministic belongs in a step, whose result is recorded, or has to be one of this crate’s durable primitives. The workflow tutorial covers the same ground with more examples.

None of this applies inside a step: a step’s body runs once per attempt and only its result is recorded, so plain races, timeouts, sleeps and spawns are all fine there. A durable call still has to be awaited where it was built, though, so one spawned from a step body fails. Nor does any of this apply outside a workflow.

§Non-deterministic values

The clock (SystemTime::now, Instant::now, chrono::Utc::now), random numbers (rand, Uuid::new_v4), environment and file reads, network calls and any other I/O all go in a step, so the value is recorded and replayed:

async fn workflow(_: ()) -> dbos::Result<String> {
    // Not `if rand_bool()`: a recovery could draw the other value and take the other branch.
    let choice = dbos::step("choose", || async { Ok(rand_bool()) }).await?;
    if choice {
        dbos::step("charge", charge).await
    } else {
        dbos::step("refund", refund).await
    }
}

Branching on such a value outside a step is the commonest way a recovery diverges. When a recovery reaches a different step than the one recorded at that position, the call fails with Error::SystemDatabase holding sysdb::Error::UnexpectedStep. That isn’t recorded as the workflow’s outcome: the workflow stays PENDING, and recovery tries it again until it is parked.

Most divergence reports nothing, so don’t rely on that error. A recovery compares step names and not inputs, and only at positions that have a record. If charge above were chosen outside a step and the process stopped before its result was recorded, a recovery that chose differently would run refund as well, with no error, and both would have happened.

§Concurrency that makes a choice nothing records

The following are not supported inside a workflow body, on any runtime. What matters is the behaviour rather than the crate it comes from, so each form is listed under the runtime that provides it.

Races over durable calls. The winner isn’t recorded, so a recovery can take another branch. Use select_step! instead.
- tokio: tokio::select!
- futures: futures::select!, select_biased!, futures::future::select, try_select, select_all and select_ok, and futures::stream::select and select_all
- smol and futures-lite: future::or, future::race, stream::or and stream::race, and FutureExt::or and FutureExt::race (smol::future and smol::stream re-export futures_lite::future and futures_lite::stream)
Timeouts around durable calls. A timeout is a race against a timer. Use select_step! with a durable sleep, or the call’s own timeout option where it has one.
- tokio: tokio::time::timeout and timeout_at
- smol and async-global-executor: an async_io::Timer (also smol::Timer) raced with or or race, and timeout crates built on it
Non-durable sleeps. tokio::time::sleep, sleep_until or interval, or a smol::Timer::after or async_io::Timer::at, used as a delay isn’t recorded, so it runs again in full on every recovery. Use sleep.
Completion-order streams over durable calls. The order concurrent calls finish in comes from their latency, which isn’t recorded. On a recovery they come out in an order unrelated to the first run, so steps a loop runs for each completion can be handed each other’s recorded results, and nothing reports an error. There is no durable replacement yet: run such a fan-out as child workflows on a queue, or sequentially.
- futures::stream::FuturesUnordered, StreamExt::buffer_unordered and StreamExt::for_each_concurrent
- any other combinator that yields results in completion order, including futures-lite’s
Spawning a durable call. The running workflow is found through a task-local, which a spawned task doesn’t inherit on any runtime. What happens then depends on the call and on where it was built, and the commonest cases are silent:
- Built inside the spawned task, a step or a sleep runs as a plain call and records nothing, and so do WorkflowHandle::result and the methods on DBOS.
- Built inside the spawned task, WorkflowRef::start and WorkflowRef::run start a new top-level workflow with no parent, so each recovery that repeats the spawn starts another.
- Built inside the spawned task, send, recv, set_event, get_event and the management functions fail with Error::NotInWorkflow.
- Built in the workflow and awaited in the spawned task, any of them fails with Error::StepBuiltElsewhere.
The ways to spawn include:
- tokio: tokio::spawn, tokio::task::spawn_blocking, spawn_local, LocalSet and JoinSet
- smol: smol::spawn, smol::unblock and Executor::spawn
- async-global-executor: async_global_executor::spawn, spawn_local and spawn_blocking
- std::thread::spawn

These are supported in a workflow body:

tokio::join!, futures::join!, futures::future::join_all and futures_lite::future::zip, on every runtime, when each element is one durable call. Each waits for all of its elements and returns results in input order, and each call took its step ID when it was built, which is what PendingStep explains.
- An async block isn’t one durable call. The calls in it are built when it is polled, so several such blocks take IDs in the order they happen to run, which a recovery doesn’t reproduce. Work needing several calls in one branch is a child workflow or a single step.
- Some calls don’t belong in a join. patch and deprecate_patch claim their positions when awaited, and Debouncer::debounce can claim more when awaited: await each of those on its own.
tokio::try_join!, futures::try_join!, futures::future::try_join_all and futures_lite::future::try_zip, under the same rules, only when the error goes straight out of the workflow. These stop at the first failure and drop the rest, which makes them a race on the error path: nothing records which error came back, so code that branches on it can take a different path on a recovery. To handle a failure, use an all-wait and inspect the results.
select_step!, select_workflow and join_workflows, and the select_workflow! and join_workflows! macros.
sleep, for durable delays.

§Changing a workflow’s code

A recovered workflow replays the step history it recorded, position by position, so code that moves, adds or removes steps can no longer replay executions the old code started. patch is the way to make such a change while those executions are in flight: new executions take the new path and record that they did, and old ones take the old path. Once no old execution can run again, deprecate_patch replaces it and the old code goes:

// Deployed first.
async fn v1(_: ()) -> dbos::Result<u32> {
    Ok(step_one().await? + step_two().await?)
}

// Deployed while runs of `v1` may still be recovered: they take the old path.
async fn v2(_: ()) -> dbos::Result<u32> {
    let first = if dbos::patch("use-step-three").await? {
        step_three().await?
    } else {
        step_one().await?
    };
    Ok(first + step_two().await?)
}

// Deployed once no run of `v1` can run again.
async fn v3(_: ()) -> dbos::Result<u32> {
    dbos::deprecate_patch("use-step-three").await?;
    Ok(step_three().await? + step_two().await?)
}

Old and new code share executions only if they run under one application version, so the deployments that patch keep Config::app_version the same. The other way to change a workflow’s code is a new version per deployment, with the old version’s executors kept running until its executions finish; that needs no patch. See patch.

§Cargo features

Feature	Default	What it adds
`engine`	yes	The durable execution engine: workflows, steps, queues, messaging and the client. Without it, what remains is the system database layer.
`macros`	yes	`select_step!`, the durable race. A procedural macro, so it adds `syn` and `quote` to the build, though neither ends up in the binary. Implies `engine`.
`runtime-tokio`	yes	Run on tokio.
`runtime-smol`	no	Run on smol.
`runtime-async-global-executor`	no	Run on async-global-executor.

The crate needs at least one runtime feature, for spawning its background tasks and setting its timers. Several may be enabled: tokio is used when the caller is in a tokio runtime, and otherwise async-global-executor, then smol.

Re-exports§

pub use sysdb::types::Change;
pub use sysdb::types::IdleTransactionTimeout;
pub use sysdb::types::PollingConcurrency;
pub use sysdb::types::QualifiedWorkflowName;
pub use sysdb::types::RateLimit;
pub use sysdb::types::ScheduleFilter;
pub use sysdb::types::ScheduleStatus;
pub use sysdb::types::WorkflowDelay;
pub use sysdb::types::WorkflowIdReusePolicy;

Modules§

sysdb: The DBOS system database.

Macros§

join_workflows: Waits for all of these workflows and returns what each of them returned.
select_stepmacros: Races these durable calls and runs the arm belonging to the one that finishes first.
select_workflow: Races these workflow handles and runs the arm belonging to the one that finishes first.

Structs§

CancellationToken: A token which can be used to signal a cancellation request to one or more tasks.
Client: A connection to a DBOS application’s system database, from outside the application.
ClientConfig: Everything a Client needs.
Config: Everything a DBOS instance needs.
DBOS: A DBOS instance: the whole public surface of the library.
DebounceOptions: What one debounce says beyond its key, period and input — see Debouncer::debounce_with and DebouncerClient::debounce_with.
Debouncer: A registered workflow, debounced.
DebouncerClient: A workflow debounced through a Client, named as an enqueue names it.
EnqueueOptions: What an enqueue by name may say, beyond the input: Client::enqueue_with’s, and DBOS::enqueue_with’s.
Executor: The running half of DBOS, created by DBOS::launch and dropped by DBOS::shutdown.
Fork: One fork: the workflow it is forked from, the step it restarts at, and the id it gets.
ForkOptions: What every fork of a call inherits, and where it goes.
ListWorkflowStepsOptions: What a step listing reads, and which page of it.
Payload: A value for a CustomSerializer to encode: whatever the application handed the engine, seen through Serialize.
PayloadSlot: Where a CustomSerializer puts the value it decodes.
PendingStart: A call that sets a workflow running, has claimed its step id, and has done nothing yet: a start, or a fork, resume or rewind such as fork_workflow.
PendingStep: A durable call that has taken its step id and has not run.
PendingWorkflow: A call that creates a workflow, has claimed the step ids that costs, and has started nothing.
PortableWorkflowArgs: A portable workflow input, as the envelope every DBOS SDK writes it in.
Queue: A registered queue, as the database holds it.
QueueChange: A change to a registered queue’s limits, naming only what moves.
QueueOptions: What a queue allows, and how often it is polled.
QueueRequest: A queue to leave a workflow on, and what to ask of it.
ReadStreamOptions: How a stream read waits: how often it looks, and how long it waits for each value.
ResumeOptions: Where a resumed workflow goes.
RewindOptions: Where a rewind cuts a workflow’s history, and where the rewound workflow goes.
RunOptions: What a caller may say about a run, beyond the input.
Schedule: A schedule, as the database holds it.
ScheduleChange: A change to a schedule’s definition, leaving what it does not name.
ScheduleSpec: A schedule’s definition: what create_schedule creates and apply_schedules reconciles.
ScheduledWorkflowInput: What a scheduled workflow is called with: the tick it is running for, and the schedule’s context.
SendMessage: One message: a payload, the workflow it is for, and how to address it.
SendOptions: How a send behaves, as opposed to what it sends.
SetEventOptions: What set_event_with may ask for, beyond the key and the value.
ShutdownOptions: What DBOS::shutdown_with does with the workflows still running.
StartOptions: What a caller may say about a start, beyond the input.
StepOptions: What a caller may say about a step, beyond its name and body.
StepStatus: What a step body can learn about the attempt it is running as.
StreamReader: A read of one stream, value by value, in order.
WorkflowHandle: A running — or finished — workflow, by id.
WorkflowOptions: How a workflow is registered, beyond its name and its body.
WorkflowRef: A registered workflow, with its argument and result types kept.
WriteStreamOptions: What write_stream_with may ask for, beyond the key and the value.

Enums§

Children: Whether an operation reaches a workflow’s descendants.
DuplicationPolicy: What an enqueue does when its deduplication_id is already held.
EngineOnly: The error type of a workflow or step with no failure of its own.
Error: Everything a durable function can fail with.
Forks: Whether a message also reaches the workflows forked from its destination.
QueueConflict: What registering a queue that already exists does to the stored limits.
SerializationFormat: Which encoding a workflow, an enqueue, or a single write uses.
Serializer: How payloads are encoded on their way into the system database, by default.
Timeout: How long a workflow may take, and — started from inside another one — what it does about the deadline its parent is under.

Constants§

APP_ID_ENV: Environment variable carrying the DBOS Cloud application id.
APP_VERSION_ENV: Environment variable carrying the application version.
CLOUD_APP_NAME_ENV: Environment variable naming the application on DBOS Cloud.
CLOUD_ENV: Environment variable saying this process is running on DBOS Cloud.
DEFAULT_STREAM_POLLING_INTERVAL: How long a reader waits before looking again when nothing wakes it, unless ReadStreamOptions::polling_interval says otherwise.
EXECUTOR_ID_ENV: Environment variable identifying the VM this process runs on.
MIN_STREAM_POLLING_INTERVAL: The shortest ReadStreamOptions::polling_interval a reader accepts. A shorter one would have a waiting reader spin on the database.
SYSTEM_DATABASE_URL_ENV: Environment variable naming the system database, which Config::from_env and ClientConfig::from_env read.

Traits§

CustomSerializer: An application’s own serializer, configured with Serializer::Custom.
DurableError: What an application may fail with, inside Error::Application.

Functions§

cancel_workflow: Marks a workflow cancelled, from inside another workflow, without a handle.
cancel_workflows: Cancels workflows, and optionally everything descended from them, from inside a workflow.
cancellation_token: A token that fires when the step this code is running inside is abandoned.
close_stream: Closes stream key of the current workflow, so its readers end once they reach the close.
delete_workflow: Removes a workflow and everything recorded against it, from inside another workflow.
delete_workflows: Deletes workflows, and optionally everything descended from them, from inside a workflow.
deprecate_patch: Honours a patch’s markers already recorded, without recording new ones.
fork_workflow: Forks a workflow from start_step, from inside another workflow, and hands back the fork’s handle.
fork_workflow_with: fork_workflow, choosing the fork’s id, what it inherits and where it runs.
fork_workflows: Forks workflows from inside a workflow, handing back a handle to each fork in the order given.
get_event: Reads a key a workflow published, waiting up to timeout for it to appear.
join_workflows: Waits until every one of these workflows has finished.
list_workflow_steps: Reads one workflow’s steps, in execution order, from inside a workflow.
list_workflow_steps_with: list_workflow_steps, choosing what each step carries and which page to read.
list_workflows: Reads the workflows matching a filter, from inside a workflow.
merge_workflow_attributes: Merges keys into a workflow’s attributes, in one statement, from inside another workflow.
patch: Whether this execution should take the new code path at a patch point.
read_stream: Reads stream key of workflow workflow_id from the beginning, from a workflow body.
read_stream_from: read_stream, starting at offset rather than the beginning.
read_stream_value: Reads the one value at offset of stream key of workflow workflow_id, waiting for it to be written. From a workflow body, where the read is a step.
recv: Takes the oldest message sent to this workflow, waiting up to timeout for one to arrive.
resume_workflow: Puts a workflow back on a queue, from inside another workflow, and hands back a handle to it.
resume_workflow_with: resume_workflow onto a queue of the caller’s choosing.
resume_workflows: Resumes workflows from inside a workflow, handing back one handle per id, in the order given.
retrieve_workflow: A handle to a workflow this workflow did not start, from inside a workflow.
rewind_workflow: Re-runs a finished workflow from its first step under the same id, from inside another workflow.
rewind_workflow_with: rewind_workflow, choosing the step it re-runs from and where it runs.
select_workflow: Waits until one of these workflows finishes, and reports which.
send: Sends a message to a workflow, for it to recv when it is ready.
send_bulk: Sends many messages in one transaction, for their destinations to recv when they are ready.
send_bulk_with: send_bulk, with SendOptions rather than the defaults.
send_with: send, with a SendMessage that may carry a topic or an idempotency key, and SendOptions rather than the defaults.
set_event: Publishes a key/value on the current workflow, for anyone to read.
set_event_with: set_event, with SetEventOptions.
set_workflow_delay: Holds a queued workflow back, or lets it go sooner, from inside another workflow.
sleep: Waits for duration, resuming at the original wake time after a crash.
step: Runs body once per workflow, recording what it returned.
step_id: The id of the step this code is running inside, or None when it is not inside one.
step_status: What the step this code is running inside knows about its own attempt, or None outside one.
step_with: Runs body as a step, retrying it as options allows.
update_workflow_attributes: Replaces a workflow’s attributes, or clears them with None, from inside another workflow.
workflow_id: The id of the workflow this code is running inside, or None outside one.
write_stream: Appends a value to stream key of the current workflow.
write_stream_with: write_stream, with WriteStreamOptions.

Type Aliases§

PendingRun: A run that has claimed its two step ids and has started nothing.
Result: The result of a durable function, or of an engine operation.
ShouldRetry: A step’s retry predicate: given a failure, whether to try again.

Crate dbos

Crate dbos Copy item path

§Getting started

§Where to look

§Determinism in workflows

§Non-deterministic values

§Concurrency that makes a choice nothing records

§Changing a workflow’s code

§Cargo features

Re-exports§

Modules§

Macros§

Structs§

Enums§

Constants§

Traits§

Functions§

Type Aliases§

Crate dbos