Struct Client 

pub struct Client(/* private fields */);

A connection to a DBOS application’s system database, from outside the application.

It holds a connection and nothing else. Not an executor: an executor is the machinery for running workflows — an id, a version and an application name to stamp claims with, a runtime to spawn onto, a registry to resolve names through, task sets to shut down — and a client does none of that. The operations both surfaces share are methods on the connection, so calling one here costs no executor and implies none.

Cheap to clone — it is an Arc internally — so a clone is another handle to the same connection rather than a second one. Clone it into whatever holds application state.

Dropping the last clone stops the listener and the notifier and leaves the pool to its own teardown, which closes the connections but says nothing about when. That is a safety net — without it the listener would hold a LISTEN connection for the life of the process — and not a shutdown: nothing waits for either task, and whatever the notifier had inside its coalescing window may go out after the drop returns or not at all. close is how a client is put away deliberately, and is what a process that connects repeatedly wants.

use dbos::{Client, ClientConfig, WorkflowHandle};

let client = Client::connect(ClientConfig::from_env()).await?;

// The type parameter says what the workflow returns; nothing here can check it.
let handle: WorkflowHandle<String> = client.enqueue("process", "tasks", 42).await?;
println!("{}", handle.result().await?);

client.close().await;

Implementations

impl Client

pub async fn connect(config: ClientConfig) -> Result<Self>

Connects to the system database.

Eager: an unreachable database is reported here rather than on the first call that needed it, because a client is usually built at start-up where a failure is cheap and diagnosable. The price is one round trip at construction.

Migrates nothing, and creates nothing. It verifies instead: a schema that is missing or behind what this build’s queries are written against fails here, naming the version it found. A client is a guest in the application’s database, and an operator’s tool that can rewrite the schema it is inspecting is a tool that can break the application it was pointed at.

pub async fn close(&self)

Closes the connection.

Idempotent, and safe to call while clones are still alive: what it closes is the pool they share, so a later call through one of them fails as an unreachable database rather than doing something surprising. Drop cannot do this — closing is asynchronous — which is the same reason DBOS::shutdown is a method rather than a destructor; dropping the last clone stops the tasks but waits for nothing.

What this adds over the drop is the waiting: the notifier’s queued wakeups go out, the pool’s connections are closed, and the listener has ended rather than merely been told to, so afterwards nothing of this client is still running.

pub fn app_name(&self) -> Option<&str>

The application this client acts for, or None if it is nameless.

pub fn retrieve_workflow<R, E>(&self, workflow_id: &str) -> WorkflowHandle<R, E>

A handle to a workflow that already exists, by id.

Always a polling handle — nothing runs here — and it is not checked: the id is not read until the handle is used, so this cannot fail and does not go to the database. The two halves of the handle then answer an unknown id differently, deliberately: status reports Error::WorkflowNotFound, while result waits — a client is the caller most likely to hold an id before whoever owns it has committed the enqueue, and this is the one handle in the crate that waits for a row to appear rather than reporting its absence.

A caller who wants an id that names nothing refused calls workflow_status before taking the handle, at the cost of a round trip — or bounds the wait instead, since dropping the future ends it and tokio::time::timeout is all a bounded wait needs.

pub async fn workflow_status( &self, workflow_id: &str, ) -> Result<Option<WorkflowStatus>>

A workflow’s status, or None if there is no such workflow.

The one-shot read behind WorkflowHandle::status, for a caller who has an id and does not want a handle. Absence is a value here rather than Error::WorkflowNotFound, because a client asking after an id it was given has every reason to be told plainly that it does not exist.

pub async fn register_queue( &self, name: &str, options: QueueOptions, on_conflict: QueueConflict, ) -> Result<Queue>

Registers a queue, or reports the one already registered under this name.

The same operation DBOS::register_queue performs, and the reason a client has it is that a queue is a row: a fleet may be configured by the tool that deploys it rather than by the code that drains it.

The limits and the policy are the same QueueOptions and QueueConflict an application states — but UpdateIfLatestVersion is Error::Config here rather than a registration: a client runs none of the application’s code, so there is no version of it to be the latest of. Ask for AlwaysUpdate or NeverUpdate.

A client with no application name of its own registers over a peer’s queue rather than being refused, replacing its stored limits: the ownership check lets a nameless writer through. Whether it should is an open question shared with the other SDKs on the same database. Give the client an app_name to get the refusal QueueConflict describes.

let queue = client.register_queue("fleet", dbos::QueueOptions {
    worker_concurrency: Some(3),
    ..Default::default()
}, dbos::QueueConflict::AlwaysUpdate).await?;

pub async fn queue(&self, name: &str) -> Result<Option<Queue>>

The queue registered under this name, or None if there is none.

pub async fn list_queues(&self) -> Result<Vec<Queue>>

Every queue this client’s application can see: its own, plus the unclaimed ones.

A nameless client sees every queue, which is the read half of what namelessness means.

pub async fn update_queue( &self, name: &str, change: QueueChange, ) -> Result<Queue>

Changes a registered queue’s limits, leaving what the change does not name.

The whole fleet picks the change up within a poll, without a restart — which is what makes this worth having on a client rather than only in the application: the process that decides a queue should slow down is rarely one of the processes draining it.

pub async fn delete_queue(&self, name: &str) -> Result<()>

Removes a queue’s registration, leaving whatever was enqueued on it where it stands.

impl Client

pub async fn enqueue<P, R, E>( &self, workflow: impl Into<QualifiedWorkflowName>, queue: &str, input: P, ) -> Result<WorkflowHandle<R, E>>

where P: Serialize,

Enqueues a workflow, onto a queue, and hands back a handle to it.

The client’s signature operation. The workflow is recorded ENQUEUED and nothing runs it here; whichever executor next polls that queue claims it, and only that executor needs to have the code. The handle is a polling one, because the process that asks is by definition not the one that runs it.

workflow is the QualifiedWorkflowName it was registered under. A free function’s name alone converts into one; a workflow registered under a class, or a class and a configured instance, is named by QualifiedWorkflowName::method or QualifiedWorkflowName::instance, because the registration is identified by all three parts.

let handle: dbos::WorkflowHandle<String> =
    client.enqueue("send_email", "email-queue", "hello").await?;
let sent = handle.result().await?;

let billing = dbos::QualifiedWorkflowName::instance("charge", "Billing", "eu");
let charged: dbos::WorkflowHandle<u64> = client.enqueue(billing, "billing", 42).await?;

The type parameters are the caller’s assertion, not a checked fact: there is no registry here to look the name up in, so R is what the caller knows the workflow returns and a mismatch surfaces when the result is decoded. That is the price of naming code this process does not have.

Called from inside a workflow body this is not a checkpoint, unlike DBOS::enqueue, which records the launch against the parent and re-derives the same child on a replay. A client knows nothing of the ambient workflow (it is a connection, not an execution), so a replayed body enqueues again. Reach for DBOS::enqueue when the caller is a workflow, and for this when it is not.

pub async fn enqueue_with<P, R, E>( &self, workflow: impl Into<QualifiedWorkflowName>, input: P, options: EnqueueOptions<'_>, ) -> Result<WorkflowHandle<R, E>>

where P: Serialize,

enqueue, with something to say about how.

See EnqueueOptions for what may be said — among it a deduplication id, a priority, a partition key, a delay, and how the input is encoded.

For a workflow in another language, enqueue it portable: the input then goes into the envelope every SDK reads, a tuple of two or more as that many positional arguments, () as none, and anything else as the one argument. A client has no registration to defer to, so SerializationFormat::Default is this client’s serializer here. A callee that takes named arguments needs enqueue_portable.

pub async fn enqueue_portable<R, E>( &self, workflow: impl Into<QualifiedWorkflowName>, args: PortableWorkflowArgs, options: EnqueueOptions<'_>, ) -> Result<WorkflowHandle<R, E>>

Enqueues a workflow with its arguments as written: the portable envelope, as given.

For a workflow in another language that takes arguments by name, which no Rust value is written as: enqueue_with writes a tuple’s elements by position. Always portable, whatever EnqueueOptions::serialization says.

use dbos::{EnqueueOptions, PortableWorkflowArgs};
let args = PortableWorkflowArgs {
    positional_args: vec![serde_json::json!("ada@example.com")],
    named_args: serde_json::Map::from_iter([("urgent".to_owned(), serde_json::json!(true))]),
};
let handle: dbos::WorkflowHandle<String> = client
    .enqueue_portable("send_email", args, EnqueueOptions::new("email-queue"))
    .await?;

impl Client

pub async fn get_event<T: DeserializeOwned>( &self, workflow_id: &str, key: &str, timeout: Duration, ) -> Result<Option<T>>

Reads a key a workflow published, waiting up to timeout for it to appear.

The reader a client is built for. An application publishes progress under a key with set_event and anything outside it — an HTTP handler answering “how far along is my order?”, a test waiting for a workflow to reach a known point — reads it here. Nothing about it is checkpointed, because a client has no workflow to checkpoint against and no replay to protect: the read is exactly one wait on the database.

Ok(None) means the key was not there when the deadline passed — absence is a value, not an error, and Duration::ZERO makes this a poll: look once, do not wait.

The wait looks again every second, unless the client was connected with ClientConfig::use_listen_notify turned on. Then it is woken by the application’s notification, and looks again only once a minute, so against executors that don’t push notifications each wait can take up to a minute. On CockroachDB it looks every second whatever the setting.

impl Client

The same surface from outside the application, where an operator’s tool usually stands.

Every method here is DBOS’s, and the differences are the two a client always has. There is no launch check, because a client is connected or it does not exist — connect hands back a usable client or an error, so none of these can fail with Error::NotLaunched. And nothing is checkpointed: called from inside a workflow, a client’s management call runs again on replay, where the same call on DBOS would replay its recorded step. A client has no step counter of its own to agree with the workflow’s, and the ambient context belongs to an instance this client is not — the line WorkflowHandle::result already draws for a client’s handle awaited inside a workflow.

Scope follows the client’s application name. A nameless client reads and writes across every application sharing the database, which is what a cross-application operator’s tool wants and what a client acting for one application must not do by accident — see ClientConfig::app_name.

pub async fn cancel(&self, workflow_id: &str) -> Result<()>

Stops a workflow, and any executor running it.

See DBOS::cancel. A cancelled workflow stays in the database and can be resumed.

client.cancel("stuck-workflow").await?;

pub async fn cancel_all( &self, workflow_ids: &[&str], children: Children, ) -> Result<Vec<String>>

Cancels workflows, and optionally everything descended from them.

See DBOS::cancel_all for what the cascade does and why it is one transaction. Returns the ids that actually moved.

pub async fn resume<R, E>( &self, workflow_id: &str, ) -> Result<WorkflowHandle<R, E>>

Puts a workflow back on a queue, and hands back a handle to watch it.

See DBOS::resume. The handle polls, as every handle a client holds does.

pub async fn resume_with<R, E>( &self, workflow_id: &str, options: ResumeOptions<'_>, ) -> Result<WorkflowHandle<R, E>>

resume, naming the queue the workflow goes back on.

pub async fn resume_all<R, E>( &self, workflow_ids: &[&str], options: ResumeOptions<'_>, ) -> Result<Vec<WorkflowHandle<R, E>>>

Resumes workflows, handing back a handle to each in the order given.

pub async fn fork<R, E>( &self, workflow_id: &str, start_step: i32, ) -> Result<WorkflowHandle<R, E>>

Forks a workflow from start_step, enqueueing the fork and handing back a handle to it.

See DBOS::fork. This is what an operator’s tool exists for: a workflow that failed against broken code is forked onto a fixed deployment, and ForkOptions::app_version is how it names one.

let steps = client.list_workflow_steps("failed-workflow").await?;
let failed = steps.iter().rev().find(|s| s.error.is_some()).map_or(0, |s| s.step_id);
let handle = client
    .fork_with::<u32, dbos::EngineOnly>(
        dbos::Fork::new("failed-workflow", failed),
        dbos::ForkOptions { app_version: Some("1.4.0"), ..Default::default() },
    )
    .await?;

pub async fn fork_with<R, E>( &self, fork: Fork<'_>, options: ForkOptions<'_>, ) -> Result<WorkflowHandle<R, E>>

fork, with the chosen id, version, queue, timeout and replacement children spelled out. See DBOS::fork_with.

pub async fn fork_all<R, E>( &self, forks: &[Fork<'_>], options: ForkOptions<'_>, ) -> Result<Vec<WorkflowHandle<R, E>>>

Forks workflows, handing back a handle to each fork in the order given.

See DBOS::fork_all: every source must exist, or the call fails and no fork is written. A fan-out re-run against fixed code is exactly the job an operator reaches for a client to do.

pub async fn rewind<R, E>( &self, workflow_id: &str, ) -> Result<WorkflowHandle<R, E>>

Re-runs a finished workflow from its first step, under the same id.

See DBOS::rewind. The handle polls, as every handle a client holds does.

pub async fn rewind_with<R, E>( &self, workflow_id: &str, options: RewindOptions<'_>, ) -> Result<WorkflowHandle<R, E>>

rewind, choosing the step it re-runs from and where it runs.

See DBOS::rewind_with for what is discarded and what is kept.

let handle = client
    .rewind_with::<u32, dbos::EngineOnly>(
        "failed-workflow",
        dbos::RewindOptions { start_step: 2, ..Default::default() },
    )
    .await?;

pub async fn delete(&self, workflow_id: &str) -> Result<()>

Removes a workflow and everything recorded against it.

See DBOS::delete. Anyone awaiting the deleted workflow goes on waiting — deleting a row and never creating it are indistinguishable to a waiter, which WorkflowHandle::result records.

pub async fn delete_all( &self, workflow_ids: &[&str], children: Children, ) -> Result<u64>

Deletes workflows, and optionally everything descended from them. Returns the rows removed.

pub async fn set_workflow_delay( &self, workflow_id: &str, delay: WorkflowDelay, ) -> Result<()>

Moves a delayed workflow’s release time.

See DBOS::set_workflow_delay. Only a DELAYED row moves: a workflow already released is running or queued, and pushing its delay out would not recall it.

pub async fn update_workflow_attributes( &self, workflow_id: &str, attributes: Option<&Map<String, Value>>, ) -> Result<()>

Replaces the attributes attached to a workflow, or clears them.

See DBOS::update_workflow_attributes. None clears.

pub async fn merge_workflow_attributes( &self, workflow_id: &str, attributes: &Map<String, Value>, ) -> Result<()>

Merges keys into a workflow’s attributes, in one statement.

See DBOS::merge_workflow_attributes.

pub async fn list_workflows( &self, filter: &WorkflowFilter<'_>, ) -> Result<Vec<WorkflowRecord>>

Reads the workflows matching a filter.

See DBOS::list_workflows. Queued workflows are listed with WorkflowFilter::queues_only rather than a second method.

use dbos::sysdb::types::{WorkflowFilter, WorkflowStatus};

let stuck = client
    .list_workflows(&WorkflowFilter {
        status: vec![WorkflowStatus::Pending],
        ..WorkflowFilter::default()
    })
    .await?;

pub async fn list_workflow_steps( &self, workflow_id: &str, ) -> Result<Vec<StepRecord>>

Reads the steps a workflow has recorded, in the order it recorded them, each with its output or error.

list_workflow_steps_with can leave the outputs out, or read one page of a long workflow’s steps.

pub async fn list_workflow_steps_with( &self, workflow_id: &str, options: ListWorkflowStepsOptions, ) -> Result<Vec<StepRecord>>

Reads the steps a workflow has recorded, choosing what each step carries and which page to read.

See DBOS::list_workflow_steps_with.

impl Client

pub async fn send<T: Serialize>( &self, destination_id: &str, message: &T, ) -> Result<()>

Sends a message to a workflow, for it to recv when it is ready.

The message waits in the database until the destination reads it, so sending to a workflow that has not reached its receive — or is not running at all — is normal rather than an error. Sending to a workflow that does not exist is Error::SystemDatabase carrying the system database’s non-existent-workflow error: the foreign key catches it, so a message is never left addressed to nothing.

A client’s send is not a step, and that is the difference from a workflow’s. A workflow’s send is checkpointed, so a replay does not send twice; a client has no replay and no step sequence, and SendMessage::idempotency_key is the mechanism it has instead — the key becomes the row’s identity, so a retried request delivers one message.

pub async fn send_with<T: Serialize>( &self, message: SendMessage<'_, T>, options: SendOptions, ) -> Result<()>

send, with a SendMessage that may carry a topic or an idempotency key, and SendOptions rather than the defaults.

pub async fn send_bulk<T: Serialize>( &self, messages: &[SendMessage<'_, T>], ) -> Result<()>

Sends many messages in one transaction.

All or none, which is the reason to prefer this over a loop of send: the batch is one insert, so a failure halfway through delivers nothing rather than a prefix.

A topic and an idempotency key travel on each SendMessage, and only the call’s behaviour is in SendOptions. The key especially belongs to one message, since it becomes that row’s identity and a batch sharing one would collide with itself; the fork fan-out and the encoding are properties of the call.

One payload type for the whole batch, which is what typing it costs. A batch of genuinely different shapes is a batch of serde_json::Value, or two calls.

pub async fn send_bulk_with<T: Serialize>( &self, messages: &[SendMessage<'_, T>], options: SendOptions, ) -> Result<()>

send_bulk, with SendOptions rather than the defaults.

impl Client

The same surface for a process outside the application.

DBOS’s methods with a client’s two differences: nothing is checkpointed, and nothing checks the workflow or the queue against a registration, because a client has neither — see ScheduleSpec::for_workflow. apply_schedules, backfill_schedule and trigger_schedule are not refused inside a workflow either: a client’s call is never recorded, so there is nothing for a replay to disagree with.

pub async fn create_schedule(&self, spec: &ScheduleSpec) -> Result<()>

Creates a schedule. See DBOS::create_schedule.

pub async fn apply_schedules(&self, specs: &[ScheduleSpec]) -> Result<()>

Creates or replaces a set of schedules, in one transaction. See DBOS::apply_schedules.

pub async fn list_schedules( &self, filter: &ScheduleFilter<'_>, ) -> Result<Vec<Schedule>>

The schedules matching filter. See DBOS::list_schedules.

pub async fn get_schedule(&self, name: &str) -> Result<Option<Schedule>>

The schedule with this name, or None.

pub async fn update_schedule( &self, name: &str, change: &ScheduleChange<'_>, ) -> Result<()>

Changes a schedule’s definition. See DBOS::update_schedule.

pub async fn pause_schedule(&self, name: &str) -> Result<()>

Stops a schedule firing. See DBOS::pause_schedule.

pub async fn resume_schedule(&self, name: &str) -> Result<()>

Lets a paused schedule fire again.

pub async fn delete_schedule(&self, name: &str) -> Result<()>

Deletes a schedule. Deleting one that does not exist is not an error.

pub async fn backfill_schedule<R, E>( &self, name: &str, start: SystemTime, end: SystemTime, ) -> Result<Vec<WorkflowHandle<R, E>>>

Enqueues every tick of a schedule strictly between start and end. See DBOS::backfill_schedule.

pub async fn trigger_schedule<R, E>( &self, name: &str, ) -> Result<WorkflowHandle<R, E>>

Enqueues one run of a schedule’s workflow now. See DBOS::trigger_schedule.

impl Client

pub fn read_stream<T: DeserializeOwned>( &self, workflow_id: &str, key: &str, options: ReadStreamOptions, ) -> StreamReader<T>

Reads stream key of workflow workflow_id from the beginning, from outside the application.

Nothing is checkpointed: a client has no workflow to record a read against.

pub fn read_stream_from<T: DeserializeOwned>( &self, workflow_id: &str, key: &str, offset: i32, options: ReadStreamOptions, ) -> StreamReader<T>

read_stream, starting at offset rather than the beginning.

pub async fn read_stream_value<T: DeserializeOwned>( &self, workflow_id: &str, key: &str, offset: i32, options: ReadStreamOptions, ) -> Result<T>

Reads the one value at offset of a stream, waiting for it to be written.

See read_stream_value for the answer when the stream ends first.

impl Client

pub async fn list_app_versions(&self) -> Result<Vec<VersionInfo>>

Every application version this client can see, newest first: the ones its own application registered, plus the unclaimed. A nameless client sees every one, which is the read half of what namelessness means — the same scoping list_queues has.

What an operator reads to find out which versions of the code have ever announced themselves, and which one a workflow row’s app_version refers to. A named client answers that question for its own application only — a peer’s versions are not in the listing, and nothing here reads them, which is why a tool that has to see every application leaves app_name unset.

pub async fn latest_app_version(&self) -> Result<Option<VersionInfo>>

The version a rolling deploy currently prefers, or None if none is registered.

Scoped to this client’s application, or across all of them when it is nameless. Latest by its registered timestamp rather than by creation, which is how a rollback is expressed.

pub async fn set_latest_app_version(&self, version_name: &str) -> Result<()>

Promotes an already-registered version to be the latest, by moving its timestamp to now on the database’s clock, and past every other version’s.

The write half of latest_app_version, and how a rolling deploy is steered from outside the fleet: the latest version is chosen by timestamp rather than by creation order, so promoting an older one is a rollback — the executors running it start dequeuing again, and an enqueue that names no version goes to it.

The version must already exist; this does not register one. Registration is an executor’s business, because a version is a claim about code that is running somewhere. A name that matches nothing is not an error, though: the write moves no row and reports success, so a misspelled rollback reads as a rollback: the row count is discarded. Whether it should be reported is an open question shared with the other SDKs on the same database.

Scoped to this client’s application, as every write here is. Promoting a version another application registered fails rather than moving it — a timestamp is what a peer’s fleet is rolling on, so it is not this client’s to move without saying so. Saying so is set_latest_app_version_for.

That scoping needs the client to have a name. A client configured without an app_name has nothing to be refused under, so it promotes whatever version it names, a peer’s included. Whether that is what namelessness is for or a gap is an open question shared with the other SDKs on the same database.

// Roll the fleet back to the version before the bad deploy.
client.set_latest_app_version("1.4.2").await?;

pub async fn set_latest_app_version_for( &self, version_name: &str, app_name: &str, ) -> Result<()>

Promotes a version on behalf of the application that registered it.

set_latest_app_version promotes within this client’s own application; this one names the application to act as, which is what lets a single operator tool roll several applications on a shared system database without connecting a client per application.

The setter takes a name where the reader does not. Promotion is the operation that claims: the write also adopts a version left unclaimed, which would otherwise read as every peer’s latest. Reading claims nothing, so it has nothing to say a name about.

A version some third application registered is still refused — naming an application is how a caller says which fleet it means, not a way around ownership.

// One operator tool, rolling a peer application back.
client.set_latest_app_version_for("1.4.2", "billing").await?;

impl Client

The same two waits from outside the application.

Identical to DBOS’s, with the two differences a client always has: there is no launch check, because a client is connected or it does not exist; and nothing is checkpointed, because a client has no step counter of its own to agree with a workflow’s. Called from inside a workflow body, a client’s wait runs again on replay — the line WorkflowHandle::result already draws for a client’s handle awaited there.

Both waits are here because this is the surface an operator’s tool wants most: a client is the caller most likely to hold a set of ids it did not start.

pub async fn select_workflow(&self, workflow_ids: &[&str]) -> Result<String>

Waits until one of these workflows finishes, and reports which.

See DBOS::select_workflow. Nothing is checkpointed, because a client has nothing to checkpoint against.

pub async fn join_workflows(&self, workflow_ids: &[&str]) -> Result<()>

Waits until every one of these workflows has finished.

See DBOS::join_workflows, which a client reaches by the same body: an all-wait is checkpointed nowhere, so this surface differs from that one only in where it finds a connection.

Trait Implementations

impl Clone for Client

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Client

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.