Struct CheckpointingRunner 

pub struct CheckpointingRunner<B> { /* private fields */ }

A single-process workflow runner with checkpointing for crash recovery.

CheckpointingRunner executes an entire workflow within one process, saving snapshots after each task. Fork branches run concurrently as tokio tasks. If the process crashes, the workflow can be resumed from the last checkpoint.

When to Use

• Single-node execution: One process runs the entire workflow
• Crash recovery: Resume from the last completed task after restart
• Simple deployment: No coordination between workers needed

For horizontal scaling with multiple workers, use PooledWorker.

Example

let backend = InMemoryBackend::new();
let runner = CheckpointingRunner::new(backend);

let ctx = WorkflowContext::new("my-workflow", Arc::new(JsonCodec), Arc::new(()));
let workflow = WorkflowBuilder::new(ctx)
    .then("step1", |i: u32| async move { Ok(i + 1) })
    .build()?;

// Run workflow - snapshots are saved automatically
let status = runner.run(&workflow, "instance-123", 1u32).await?;

// Resume from checkpoint if needed (e.g., after crash)
let status = runner.resume(&workflow, "instance-123").await?;

Implementations

impl<B> CheckpointingRunner<B>

where B: PersistentBackend,

pub fn new(backend: B) -> Self

Create a new checkpointing runner with the given backend.

pub fn from_shared(backend: Arc<B>) -> Self

Create a runner from a shared backend reference.

Useful when the same backend is shared with a WorkflowClient.

pub fn with_conflict_policy(self, policy: ConflictPolicy) -> Self

Set the conflict policy for duplicate instance IDs.

pub fn backend(&self) -> &Arc<B>

Get a reference to the backend.

impl<B> CheckpointingRunner<B>

where B: PersistentBackend + 'static,

pub async fn run<C, Input, M>( &self, workflow: &Workflow<C, Input, M>, instance_id: impl Into<String>, input: Input, ) -> Result<WorkflowStatus, RuntimeError>

where Input: Send + 'static, M: Send + Sync + 'static, C: Codec + EnvelopeCodec + EncodeValue<Input> + DecodeValue<Input> + 'static,

Run a workflow from the beginning, saving checkpoints after each task.

The instance_id uniquely identifies this workflow execution instance. The ConflictPolicy (set via with_conflict_policy) controls behaviour when a snapshot with this ID already exists.

Errors

Returns an error if the workflow cannot be executed, if snapshot operations fail, or if the conflict policy rejects a duplicate.

pub async fn resume<'w, C, Input, M>( &self, workflow: &'w Workflow<C, Input, M>, instance_id: &str, ) -> Result<WorkflowStatus, RuntimeError>

where Input: Send + 'static, M: Send + Sync + 'static, C: Codec + EnvelopeCodec + DecodeValue<Input> + EncodeValue<Input> + 'static,

Resume a workflow from a saved snapshot.

Loads the snapshot for the given instance ID and continues execution from the last checkpoint.

Errors

Returns an error if:

• The snapshot is not found
• The workflow definition hash doesn’t match (workflow definition changed)
• The workflow cannot be resumed