durable-workflow 0.1.8

# Durable Workflow Rust SDK

`durable-workflow` is the first-party Rust SDK for Durable Workflow workers
and clients. It can register workflow and activity handlers, long-poll the
worker protocol, start, signal, query, cancel, terminate, and await workflow
executions, start and await durable child workflows, expose named read-only
query handlers, heartbeat
workers and activities, and exchange JSON-native payloads through the
platform's generic Avro wrapper. Workflow code can also wait on server-backed
durable time without blocking a Rust executor thread.

## Install

Add the exact crates.io release with Cargo:

```sh
cargo add durable-workflow@0.1.8 --exact
```

Or add the same exact requirement directly to `Cargo.toml`:

```toml
[dependencies]
durable-workflow = "=0.1.8"
```

Version `0.1.8` requires Rust `1.86` or newer. Snapshot inspection queries were
introduced in `0.1.1`; replayed workflow-instance state queries are available
from `0.1.2`, deterministic durable timers are available from `0.1.4`, and
durable child workflows are available from `0.1.5`. Durable activity retry,
timeout, and typed terminal options are available from `0.1.7`. Workflow
cancellation, termination, selected-run safety, and typed workflow outcomes
are available from `0.1.8`.

## Compatibility

| SDK releases | Durable Workflow server | Worker protocol | Control plane |
| --- | --- | --- | --- |
| `0.1.0` | `>=0.2,<0.3` | `1.2` | `2` |
| `0.1.1` | `>=0.2,<0.3` | `1.2` (snapshot queries require `1.8`) | `2` |
| `0.1.2`–`0.1.3` | `>=0.2,<0.3` | `1.2` (replayed queries require `1.8`) | `2` |
| `0.1.4` | `>=0.2,<0.3` | `1.2` (timers; replayed queries require `1.8`) | `2` |
| `0.1.5`–`0.1.6` | `>=0.2,<0.3` | `1.2` (timers and child workflows; replayed queries require `1.8`) | `2` |
| `0.1.7` | `>=0.2,<0.3` | `1.2` (activity options, timers, and child workflows; replayed queries require `1.8`) | `2` |
| `0.1.8+` | `>=0.2,<0.3` | `1.2` (workflow lifecycle, activity options, timers, and child workflows; replayed queries require `1.8`) | `2` |

The machine-readable values live in `[package.metadata.durable-workflow]` in
`Cargo.toml` as `supported-server-versions`, `worker-protocol-version`, and
`control-plane-version`. Query-capable releases also publish `query-tasks`,
`query-task-minimum-worker-protocol-version`, `replayed-instance-state-queries`,
`query-state-model`, `snapshot-inspection-queries`, and `payload-codecs`.
Timer-capable releases additionally publish `durable-timers`, `timer-command`,
and `timer-replay-validation`. Child-capable releases additionally publish
`child-workflows`, `child-workflow-command`, and
`child-workflow-failure-reasons`. Activity-options releases publish
`activity-options`, `activity-retry-policy`, `activity-timeouts`, and
`activity-failure-reasons`. Lifecycle releases publish
`workflow-lifecycle-commands`, `workflow-lifecycle-run-targeting`, and
`workflow-terminal-outcomes`. Existing worker operations retain the `1.2`
baseline; only query-task poll, complete, and fail requests use the additive
`1.8` feature floor. The server's advertised protocol manifests remain
authoritative when checking compatibility during deployment.

## Durable timers

`WorkflowContext::sleep` waits on server-backed durable wall time. It emits a
`start_timer` workflow command and yields the workflow future; it never calls
`tokio::time::sleep` or keeps a process-local deadline. Durations are rounded
up to whole seconds, so a timer cannot be scheduled earlier than requested.

`WorkflowContext::start_timer` is an alias for `sleep`. On every workflow task,
the SDK reconstructs one shared sequence-ordered command stream for activities,
timers, and signal waits. A replayed timer must have a matching
`TimerScheduled` event, and it resolves only when the same sequence and timer
identity has one `TimerFired` event. The recorded delay must equal the current
workflow call. Worker and server restarts therefore preserve the original
deadline, while replay and repeated polling do not append or consume the timer
twice. Changed command order, changed delay, unpaired events, duplicate durable
sequences, and duplicate fires return `Error::NonDeterministicReplay`; its
`ReplayFailure` exposes stable `reason`, `sequence`, `expected`, and `actual`
fields. Durations too large to round up without shortening the requested wait
return `Error::TimerDurationOverflow` without emitting a command.

```rust
# use std::time::Duration;
# use durable_workflow::{json, Client, Worker};
# fn configure(client: Client) {
let mut worker = Worker::new(client, "reminder-workers");
worker.register_workflow("send-reminder", |ctx, _input| async move {
    ctx.sleep(Duration::from_secs(30)).await?;
    let receipt = ctx.activity("deliver-reminder", json!([])).await?;
    Ok(receipt)
});
# }
```

Use `WorkflowContext::sleep` only inside workflow code. Ordinary
`tokio::time::sleep` remains appropriate for worker-process concerns such as
local polling or shutdown coordination, but it is not durable workflow state.

Running and completed executions remain available through `WorkflowHandle`'s
`describe`, `query`, and `result` methods. Server protocol incompatibilities
remain `Error::Protocol(ProtocolFailure)`, including stable `reason`, `status`,
and requested/supported version fields. Activity settlement rejections are
typed `Error::ActivityTaskRejected`; other rejected worker requests remain
`Error::Http` values with the response status and body.

## Workflow cancellation, termination, and outcomes

Cancellation is cooperative. Use it when workflow and activity code should run
cleanup; use termination only as a forced operator stop. Instance-targeted
commands resolve the server's current run at command time:

```rust
# use durable_workflow::{Client, Result, WorkflowCommandOptions};
# async fn stop(client: &Client) -> Result<()> {
client.cancel_workflow(
    "order-42",
    WorkflowCommandOptions::new()
        .reason("customer withdrew the order")
        .request_id("cancel-order-42"),
).await?;
# Ok(())
# }
```

A `WorkflowHandle` offers the same `cancel` and `terminate` methods. When the
handle's original run identity matters, use `cancel_selected_run` or
`terminate_selected_run`. These call the run-targeted endpoint and fail with
`Error::WorkflowCommandRejected(WorkflowCommandRejection)` if that run is no
longer current. Automation can match `reason ==
"historical_run_command_rejected"`; the rejection also retains `workflow_id`,
`run_id`, `target_scope`, HTTP status, and the original response body.

`WorkflowHandle::result` still returns the decoded `Value` on success. Other
terminal states are distinct typed errors. Handles returned by `start_workflow`
carry a run ID, and `result` automatically describes that selected run rather
than a newer current run that may reuse the same workflow ID:

```rust
# use durable_workflow::{Error, Result, WorkflowHandle, WorkflowResultOptions};
# async fn wait(handle: WorkflowHandle) -> Result<()> {
match handle.result(WorkflowResultOptions::default()).await {
    Ok(value) => println!("completed: {value}"),
    Err(Error::WorkflowCancelled(outcome)) => {
        println!("cancelled {} / {:?}: {}", outcome.workflow_id, outcome.run_id, outcome.reason);
    }
    Err(Error::WorkflowTerminated(outcome)) => {
        println!("terminated: {}", outcome.reason);
    }
    Err(Error::WorkflowFailed(outcome)) => {
        println!("failure {:?}: {:?}", outcome.failure_id, outcome.exception_class);
    }
    Err(Error::WorkflowTimedOut(outcome)) => {
        println!("timeout category: {:?}", outcome.failure_category);
    }
    Err(error) => return Err(error),
}
# Ok(())
# }
```

Every `WorkflowTerminalOutcome` carries workflow/run identity and a stable
kind and reason. It also exposes failure category and identity, exception type
and class, non-retryable state, message, exception payload, and the raw
description whenever the server supplies them. A local result-wait deadline
uses the same typed timeout with reason `result_wait_timeout` and category
`client_timeout`; it is distinguishable from a server-terminal `timed_out`
run without parsing display text.

## Durable activity options

`WorkflowContext::activity_with_options` adds routing, durable retries, and
server-enforced timeouts while the existing `activity` and `activity_on_queue`
convenience methods remain unchanged. `ActivityRetryPolicy` accepts explicit
backoff intervals or generates integer exponential intervals. All options are
recorded on the single `schedule_activity` command; transport retry settings
on `Worker` and `Client` are separate.

```rust
# use durable_workflow::{json, ActivityOptions, ActivityRetryPolicy, Error, Result, WorkflowContext};
# use std::time::Duration;
# async fn charge(ctx: WorkflowContext) -> Result<durable_workflow::Value> {
let options = ActivityOptions::new()
    .task_queue("payments")
    .retry_policy(
        ActivityRetryPolicy::new(4)
            .exponential_backoff(Duration::from_secs(1), 2, Some(Duration::from_secs(30)))
            .non_retryable_error_type("PaymentDeclined"),
    )
    .start_to_close_timeout(Duration::from_secs(60))
    .schedule_to_start_timeout(Duration::from_secs(10))
    .schedule_to_close_timeout(Duration::from_secs(180))
    .heartbeat_timeout(Duration::from_secs(15));

match ctx
    .activity_with_options("charge-card", options, json!([{"order_id": "order-42"}]))
    .await
{
    Ok(receipt) => Ok(receipt),
    Err(Error::ActivityFailed(failure)) => Ok(json!({
        "kind": format!("{:?}", failure.kind),
        "reason": failure.reason,
        "category": failure.failure_category,
        "activity_execution_id": failure.activity_execution_id,
        "timeout_kind": failure.timeout_kind,
    })),
    Err(error) => Err(error),
}
# }
```

Timeouts use `Duration` and round up to whole protocol seconds.
`start_to_close_timeout` limits one attempt, `schedule_to_start_timeout` limits
queue wait, `schedule_to_close_timeout` includes all attempts and retry
backoff, and `heartbeat_timeout` limits the gap between heartbeats. Invalid
positive values, ordering, retry bounds, blank error types, and empty policies
return `Error::InvalidActivityOptions(ActivityOptionsError)` before any command
is emitted.

Completed activities still return their decoded value. Terminal
`ActivityFailed`, `ActivityCancelled`, and `ActivityTimedOut` history returns
`Error::ActivityFailed(ActivityFailure)`. Match `ActivityFailureKind` and the
stable `reason`, `failure_category`, and `timeout_kind` fields; durable activity
and attempt identities are retained when the server provides them. A retry
history with no terminal event remains pending during replay and does not emit
a second activity schedule after a worker restart.

## Child workflows

`WorkflowContext::start_child_workflow` records a named child on a mandatory,
explicit task queue and waits for a terminal `ChildRun*` history event. The
same history is replayed after a Rust worker restart: a committed result is
decoded using its recorded payload codec and no second child command is
emitted. A recorded child that has not settled remains pending without another
start command on redelivery. This also permits a Rust parent to call a PHP or
Python child (or the reverse) without changing payload shape.

```rust
# use durable_workflow::{json, ChildWorkflowOptions, ParentClosePolicy, Result, WorkflowContext};
# async fn parent(ctx: WorkflowContext) -> Result<durable_workflow::Value> {
let child = ctx
    .start_child_workflow(
        "python.fulfil-order",
        ChildWorkflowOptions::new("python-workers")
            .parent_close_policy(ParentClosePolicy::RequestCancel)
            .execution_timeout_seconds(600)
            .run_timeout_seconds(120),
        json!([{"order_id": "order-42"}]),
    )
    .await?;

assert_eq!(child.parent, ctx.workflow_identity()?);
println!("child workflow={:?} run={:?}", child.child.workflow_id, child.child.run_id);
Ok(child.result)
# }
```

Success returns `ChildWorkflowResult`, including parent and child workflow/run
identities. Failure, cancellation, and termination return
`Error::ChildWorkflowFailed(ChildWorkflowFailure)` inside workflow code. Match
its `kind` or stable `reason` (`child_workflow`, `cancelled`, or `terminated`),
not its message. An uncaught error becomes a durable `fail_workflow` command
whose structured exception retains those fields.

`ParentClosePolicy::Abandon` leaves an open child running when the parent
closes. `RequestCancel` requests child cancellation, and `Terminate` closes it
immediately. Retry and timeout options are recorded server-side with the child
call; they are not SDK HTTP retry limits.

## Worker

```rust,no_run
use durable_workflow::{json, Client, Result, Worker};

#[derive(Clone, Default)]
struct HelloState {
    started_by: Option<String>,
}

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::builder("http://127.0.0.1:8080")
        .token(std::env::var("DURABLE_WORKFLOW_TOKEN").ok())
        .namespace("default")
        .build()?;

    let mut worker = Worker::new(client.clone(), "rust-workers");

    worker.register_activity("hello.activity", |ctx, args| async move {
        ctx.heartbeat(json!({"stage": "started"})).await?;
        let name = args
            .as_array()
            .and_then(|arguments| arguments.first())
            .and_then(|value| value.as_str())
            .unwrap_or("world");
        Ok(json!(format!("hello, {name}")))
    });

    worker.register_replayed_workflow("hello.workflow", HelloState::default, |ctx, _input, state| async move {
        let signal = ctx.wait_signal("start").await?;
        let name = signal.first().and_then(|value| value.as_str()).unwrap_or("world");
        state.update(|current| current.started_by = Some(name.to_string()))?;
        let greeting = ctx.activity("hello.activity", json!([name])).await?;
        Ok(json!({"greeting": greeting}))
    });

    worker.register_replayed_query::<HelloState, _, _>("hello.workflow", "started-by", |_ctx, state, _args| async move {
        Ok(json!(state.started_by))
    });

    worker.run().await
}
```

## Client

```rust
# use durable_workflow::{json, Client, Result};
# async fn example(client: Client) -> Result<()> {
let handle = client
    .start_workflow("hello.workflow", "rust-workers", "hello-rust-1", json!([]))
    .await?;

client
    .signal_workflow(&handle.workflow_id, "start", json!(["Rust"]))
    .await?;

let started_by = handle.query("started-by", json!([])).await?;
assert_eq!(started_by, json!("Rust"));

let output = handle.result(Default::default()).await?;
# println!("{output}");
# Ok(())
# }
```

## Queries

`Worker::register_replayed_workflow` gives ordinary workflow execution a typed
`WorkflowInstance<S>`. Put transitions after activity and signal resolution in
that workflow closure. `Worker::register_replayed_query` re-runs the same
closure over committed durable history, then invokes the named query with an
immutable, detached `Arc<S>`. This is the recommended workflow-instance query
API: query code does not parse history or duplicate transition logic.

Replay-generated commands are discarded. A query handler has no command API,
and its detached state is never retained, so successful and failed queries do
not append history, advance execution, or change a later query. The same query
serves running, restarted, and successfully completed workflows:

```rust
# use durable_workflow::{json, Client, Worker};
# #[derive(Clone, Default)]
# struct CounterState { count: i64 }
# fn configure(client: Client) {
let mut worker = Worker::new(client, "counter-workers");
worker.register_replayed_workflow("counter", CounterState::default, |ctx, _input, state| async move {
    let signal = ctx.wait_signal("increment").await?;
    let amount = signal.first().and_then(|value| value.as_i64()).unwrap_or_default();
    state.update(|current| current.count += amount)?;
    state.read(|current| Ok(json!(current.count)))?
});
worker.register_replayed_query::<CounterState, _, _>("counter", "current", |_ctx, state, _args| async move {
    Ok(json!(state.count))
});
# }
```

`Worker::register_query` remains the lower-level snapshot-inspection API. Its
`QueryContext` exposes normalized workflow input, raw committed history, and
decoded signals. Use it for transport-level inspection when replayed typed
state is not appropriate; snapshot handlers must reduce history themselves and
are not workflow-instance query parity.

Client-side rejections are `Error::QueryFailed(QueryFailure)`. Match the
public `reason` and `status` fields for automation; the original response is
retained in `body`. Stable reasons include `rejected_unknown_query`,
`invalid_query_arguments`, `query_handler_unavailable`,
`query_payload_decode_failed`, `query_workflow_state_unavailable`, and
`query_worker_unavailable`. Protocol negotiation failures use
`Error::Protocol(ProtocolFailure)` and retain supported/requested versions.

## Heartbeats

`Worker::run` and `Worker::run_until` register the worker and then send worker
heartbeats automatically. The registration response supplies the preferred
cadence; `Worker::heartbeat_interval` is the fallback when the server does not
advertise one. Use `Worker::on_worker_heartbeat` to observe successful server
acknowledgements for metrics or structured logging.

Activity handlers report progress with `ActivityContext::heartbeat`. The
returned `ActivityHeartbeatResponse` exposes `heartbeat_recorded` and
`cancel_requested`, plus `can_continue`, `reason`, and the run close state, so
long-running work can stop and clean up without treating cancellation as a
transport or codec failure:

```rust
# use durable_workflow::{json, Client, Result, Worker};
# fn configure(client: Client) {
let mut worker = Worker::new(client, "rust-workers")
    .on_worker_heartbeat(|observation| {
        println!("worker heartbeat acknowledged at {}", observation.acknowledged_at_unix_millis);
    });

worker.register_activity("batch.process", |ctx, _args| async move {
    let acknowledgement = ctx.heartbeat(json!({"completed": 25})).await?;
    if acknowledgement.should_stop() {
        cleanup_temporary_files();
        return Ok(json!({"cleanup": "complete"}));
    }
    Ok(json!({"completed": 100}))
});
# }
# fn cleanup_temporary_files() {}
```

The server remains authoritative after cancellation: if the handler finishes
late, its completion is refused and cannot turn the run into success. Direct
client settlement calls return `Error::ActivityTaskRejected` with a stable
reason such as `run_cancelled`, `run_terminated`, or `stale_attempt`. The
managed worker treats these definitive late-settlement responses as terminal
for that leased attempt and continues polling, including after a worker
restart during cancellation. `Client::poll_activity_task_response` and
`Client::poll_workflow_task_response` preserve drain and poll-stop metadata for
lower-level worker integrations; `Client::poll_query_task_response` does the
same for query tasks. Call `outcome()` on any full poll response and match
`WorkerPollOutcome::Stop` instead of parsing a display string. In particular,
the server's HTTP `409` / `worker_draining` response decodes as a normal stop
outcome. Managed workers honor it by ceasing new polls and draining cleanly.

Lower-level integrations can call `Client::heartbeat_worker` and
`Client::heartbeat_activity_task` directly.

## Worker liveness and errors

Workflow, activity, and query polls advertise the configured poll timeout to
the server. An empty response at that boundary is normal: `Worker::run` and
`Worker::run_until` keep every poller and worker heartbeats running, so the same
worker can accept work after an idle period.

Replaying a workflow that is still blocked on a recorded activity, timer,
child workflow, or signal wait can also produce no new commands. The worker
acknowledges that task as waiting for scheduled history instead of submitting
an invalid empty completion. Workflow and query pollers therefore remain live,
and worker heartbeats continue while unrelated signals are recorded.

Poll acquisition and worker-heartbeat transport failures, HTTP 408/429
responses, and server errors use capped exponential backoff. Configure the
bound with `Worker::retry_policy`; the default retries five times from 100 ms
up to 5 seconds. Retries wrap only acquisition and heartbeat requests, never a
leased task's handler or settlement request, so an ambiguous completion is not
re-executed by the retry loop. Once the retry bound is exhausted, the transport
or HTTP error is returned.

Authentication failures remain `Error::Http` with their status and response
body, and protocol incompatibilities remain
`Error::Protocol(ProtocolFailure)` with stable reason and version fields.
Codec, handler, and other non-retryable failures are returned immediately and
are never retried indefinitely.

## Example

`examples/hello_world.rs` contains a complete round trip: it registers a Rust
worker, starts a workflow, sends a signal, runs an activity, heartbeats that
activity, exposes a named query, and waits for the completed result.
`examples/activity_options.rs` is an executable retry scenario with activity
heartbeats, a heartbeat timeout, and typed terminal failure handling.

With a Durable Workflow server running locally:

```sh
DURABLE_WORKFLOW_SERVER_URL=http://127.0.0.1:8080 \
DURABLE_WORKFLOW_TOKEN=your-token \
cargo run --example hello_world
```

Run the activity-options scenario with the same environment using
`cargo run --example activity_options`.

`TASK_QUEUE` optionally overrides the default `rust-workers` task queue.

## API documentation

The complete API reference is published at
[rust.durable-workflow.com](https://rust.durable-workflow.com/). Documentation
for `main` is rebuilt and deployed automatically.

## Ownership and versioning

The Durable Workflow project owns and maintains the crate. This repository is
the authoritative source for the `durable-workflow` crate and its Rust API
documentation.

Crate releases follow semantic versioning and are tagged with the exact crate
version, such as `0.1.1`. Rust SDK versions are independent from Durable
Workflow server image versions. A compatible server range is declared in
package metadata instead of coupling crate publication to a server release.