Struct PooledWorker 

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

A pooled worker that claims and executes tasks from a shared backend.

PooledWorker is designed for horizontal scaling: multiple workers can run across different machines/processes, all polling the same backend for tasks. Task claiming with TTL prevents duplicate execution while allowing automatic recovery when workers crash.

When to Use

• Horizontal scaling: Multiple workers process tasks concurrently
• Fault tolerance: Failed workers’ tasks are automatically reclaimed
• Load balancing: Tasks distributed across available workers

For single-process execution with checkpointing, use CheckpointingRunner.

Example

let backend = InMemoryBackend::new();
let registry = TaskRegistry::new();
let worker = PooledWorker::new("worker-1", backend, registry);

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

// Spawn the worker and get a handle for lifecycle control
let handle = worker.spawn(Duration::from_secs(1), workflows);
// ... later ...
handle.shutdown();
handle.join().await?;

Implementations

impl<B> PooledWorker<B>

where B: PersistentBackend + TaskClaimStore + 'static,

pub fn new( worker_id: impl Into<String>, backend: B, registry: TaskRegistry, ) -> Self

Create a new worker node.

Parameters

• worker_id: Unique identifier for this worker node
• backend: The persistent backend to use
• registry: Task registry containing all task implementations

Heartbeat

Heartbeats are derived automatically from claim_ttl (TTL / 2). With the default 5-minute TTL, heartbeats fire every 2.5 minutes.

pub fn with_claim_ttl(self, ttl: Option<Duration>) -> Self

Set the TTL for task claims.

pub fn with_aging_interval(self, interval: Duration) -> Self

Set the aging interval for priority-based scheduling.

Lower-priority tasks that have been waiting longer than this interval get their effective priority boosted, preventing starvation. Default: 5 minutes (300 seconds).

Panics

Panics if interval is zero.

pub fn with_batch_size(self, size: NonZeroUsize) -> Self

Set the number of tasks to fetch per poll (default: 1).

With batch_size=1, the worker fetches one task, executes it, then polls again. Other workers can pick up remaining tasks immediately.

Higher values reduce polling overhead but may cause workers to hold task IDs they won’t process immediately (though other workers can still claim them).

pub fn with_tags(self, tags: Vec<String>) -> Self

Set affinity tags for this worker.

When tags are set, the worker only picks up tasks whose tags are a subset of the worker’s tags (or tasks with no tags). When no tags are set (the default), the worker accepts all tasks.

pub async fn cancel_workflow( &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. Running tasks will complete, but no new tasks will be started.

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_workflow( &self, instance_id: &str, reason: Option<String>, paused_by: Option<String>, ) -> Result<(), RuntimeError>

Request pausing of a workflow.

This requests pausing of the specified workflow instance. Running tasks will complete, but no new tasks will be started.

Parameters

• instance_id: The workflow instance ID to pause
• reason: Optional reason for the pause
• paused_by: Optional identifier of who requested the pause

Errors

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

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

Get a reference to the backend.

pub fn spawn<C, Input, M>( self, poll_interval: Duration, workflows: WorkflowRegistry<C, Input, M>, ) -> WorkerHandle<B>

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

Spawn the worker as a background task and return a handle.

Consumes self, creates an internal command channel, and spawns the actor loop on the Tokio runtime. Returns a cloneable WorkerHandle for lifecycle control — call WorkerHandle::shutdown to request graceful shutdown and WorkerHandle::join to await completion.

The worker runs until:

• WorkerHandle::shutdown is called, or
• All clones of the handle are dropped, or
• A fatal backend error occurs.

Parameters

• poll_interval: How often to poll for new tasks
• workflows: Map of workflow definition hash to workflow

pub fn spawn_with_executor( self, poll_interval: Duration, workflows: WorkflowIndex, executor: ExternalTaskExecutor, ) -> WorkerHandle<B>

Spawn the worker with an external executor and return a handle.

Like spawn but instead of executing tasks via typed Workflow closures, delegates all task execution to the provided executor. This is used by language bindings (Python, Node.js) where task functions live in the host language.

Parameters

• poll_interval: How often to poll for new tasks
• workflows: Workflow definitions (hash + continuation tree)
• executor: Closure that executes a task by ID given input bytes

pub async fn execute_task<C, Input, M>( &self, workflow: &Workflow<C, Input, M>, available_task: AvailableTask, ) -> Result<WorkflowStatus, RuntimeError>

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

Execute a single task from an available task.

This claims the task, executes it, updates the snapshot, and releases the claim.

Errors

Returns an error if:

• The task cannot be claimed
• The workflow definition hash doesn’t match
• Task execution fails
• Snapshot update fails

pub fn builder(backend: B, registry: TaskRegistry) -> PooledWorkerBuilder<B>

Create a builder with sensible defaults.

By default, the worker ID is derived from {hostname}-{pid}. Override with PooledWorkerBuilder::worker_id.

Example

// Auto-generated worker ID from hostname + PID
let worker = PooledWorker::builder(InMemoryBackend::new(), TaskRegistry::new()).build();

// Or override with explicit ID
let worker = PooledWorker::builder(InMemoryBackend::new(), TaskRegistry::new())
    .worker_id("custom-worker-1")
    .build();