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 async fn cancel( &self, instance_id: &str, reason: Option<String>, cancelled_by: Option<String>, ) -> Result<(), RuntimeError>

Request cancellation of a workflow.

This requests cancellation of the specified workflow instance. The workflow will be cancelled at the next task boundary.

Parameters

• instance_id: The workflow instance ID to cancel
• reason: Optional reason for the cancellation
• cancelled_by: Optional identifier of who requested the cancellation

Errors

Returns an error if the workflow cannot be cancelled (not found or in terminal state).

pub async fn pause( &self, instance_id: &str, reason: Option<String>, paused_by: Option<String>, ) -> Result<(), RuntimeError>

Request pausing of a workflow.

The workflow will be paused at the next task boundary.

Errors

Returns an error if the backend fails to store the pause request.

pub async fn unpause( &self, instance_id: &str, ) -> Result<WorkflowSnapshot, RuntimeError>

Unpause a paused workflow and return the updated snapshot.

Transitions the workflow from Paused back to InProgress.

Errors

Returns an error if the backend fails to unpause the workflow.

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 + 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. If a snapshot with this ID already exists, it will be overwritten.

Errors

Returns an error if the workflow cannot be executed or if snapshot operations fail.

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 + 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