Production Worker

WorkflowWorker is the production worker process for SEPARATE deployments. It runs per-workflow claim loops, a timer loop (wakes sleeping runs), and an orphan recovery loop. InlineWorkflowRunner stays the embedded/test path.

Minimal worker process

import asyncio
from pyrula.workflows import WorkflowWorker, ValkeyStore

# Import your @workflow-decorated modules so they register before run()
import myapp.workflows  # noqa: F401

async def main() -> None:
    worker = WorkflowWorker(store=ValkeyStore(url="redis://localhost:6379"))
    await worker.run()  # blocks until worker.stop() is called

asyncio.run(main())

ValkeyStore and PostgresStore are the production-grade stores. MemoryStore is in-process only and loses data on restart.

Workflow discovery

By default, the worker picks up every @workflow that has been imported into the process. Import your workflow modules before calling run() and they register automatically.

To restrict the worker to a specific set:

from myapp.workflows import process_order, send_invoice

worker = WorkflowWorker(
    store=ValkeyStore(url="redis://localhost:6379"),
    workflows=[process_order, send_invoice],
)

Concurrency and background loops

The worker runs three things concurrently:

• Claim loops: one per workflow. Each polls every claim_poll_interval seconds and executes up to max_concurrent_runs runs at once across all workflows.
• Timer loop: wakes sleeping runs on schedule. Polls every timer_poll_interval_s seconds, processing at most timer_max_batch per tick.
• Orphan recovery loop: scans for stale claimed runs (worker crashed mid-run) and requeues them. Runs every orphan_scan_interval seconds.

All options are set via WorkerOptions:

from pyrula.workflows import WorkflowWorker, WorkerOptions, ValkeyStore

worker = WorkflowWorker(
    store=ValkeyStore(url="redis://localhost:6379"),
    options=WorkerOptions(
        max_concurrent_runs=100,
        claim_poll_interval=0.05,
        timer_poll_interval_s=2.0,
        timer_max_batch=200,
        orphan_scan_interval=30.0,
        shutdown_grace=30.0,
        cancel_timeout=5.0,
    ),
)

Defaults: max_concurrent_runs=50, claim_poll_interval=0.1, timer_poll_interval_s=5.0, timer_max_batch=100, orphan_scan_interval=60.0, shutdown_grace=30.0, cancel_timeout=5.0.

Submitting runs

The worker can also submit runs before or after starting the loop:

run_id = worker.submit("my_workflow", {"x": 1})

In production, the submitter is typically a separate process (an HTTP handler or pipeline step) writing to the same shared store.

Graceful shutdown

Call worker.stop() from a signal handler or another coroutine. The worker stops accepting new claims, waits up to shutdown_grace seconds for in-flight runs to finish, then cancels any that remain (with a cancel_timeout deadline each).

import asyncio, signal
from pyrula.workflows import WorkflowWorker, ValkeyStore

async def main() -> None:
    worker = WorkflowWorker(store=ValkeyStore(url="redis://localhost:6379"))
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGTERM, worker.stop)
    loop.add_signal_handler(signal.SIGINT, worker.stop)
    await worker.run()

asyncio.run(main())