Production Worker

WorkflowWorker is the worker process for SEPARATE deployments. It runs per-workflow claim loops, a timer loop (wakes sleeping runs), and an orphan recovery loop, plus schedule, trigger, reconcile, and projection loops when the store supports them.

It is not only for a separate tier. InlineWorkflowRunner is the test and script path: it has no background loops, so sleeping runs never wake, a resumed run stalls at the next step, and schedules never fire. Any app whose workflows sleep, wait for signals, use interrupts, or run on a schedule wants a WorkflowWorker even when it runs in the same process - see Deployment modes for the embedded pattern.

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 ArchivingStore (Valkey plus a Postgres archive) are the durable stores for production. 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 these concurrently. The first three are always on; the rest start only if the store supports them:

• 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. This loop is also what continues a run after a signal, a resume, or a wake: the store re-queues the run and the next poll picks it up.
• 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.
• Schedule loop: fires due schedules. Missed occurrences are not backfilled - on recovery a schedule fires once and moves to the next slot.
• Trigger, reconcile, and projection loops: batch triggers, archival reconciliation, and projection updates respectively.
• Shard lease loop (ValkeyStore only): renews the shards this worker owns and rebalances toward an even share of them across the live workers. Runs every shard_lease_interval seconds. See Scaling across shards.

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,
        shard_lease_interval=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, shard_lease_interval=10.0.

Scaling across shards

ValkeyStore splits each workflow’s queue into shard_count shards (default 64). Shards let one workflow spread across cluster nodes and are the unit fairness is computed over.

worker = WorkflowWorker(
    store=ValkeyStore(url="redis://localhost:6379", shard_count=64),
)

Use the same shard_count on every worker for a name. The first store to use a name records its count; a worker with a different count raises ConfigurationError. Changing the count for a name that has run means draining it first (stop submitting, let it empty, restart with the new count). There is no live resharding, so pick a number once; 64 is plenty for a six-node cluster.

Shard leases

Each worker leases unowned shards (shard_lease_ttl, default 30 s) and renews them every shard_lease_interval (default 10 s, must be shorter than the TTL), rebalancing toward shard_count / N each. A worker claims from its own shards first, then from any shard, so a lease spreads load but doesn’t fence it. The per-run claim lock stops double runs.

• Clean stop (worker.stop()): leases are released at once.
• Crash: new work keeps flowing through other workers’ fallback claims. The crashed worker’s in-flight runs recover on the orphan scan (lock_ttl plus up to orphan_scan_interval); its shards are re-leased after shard_lease_ttl.

Key namespace

Every ValkeyStore has a namespace - there is no un-namespaced layout. Pass namespace (or set PYRULA_VALKEY_NAMESPACE) to run several independent stores on one Valkey instance or cluster - the only way to separate stores on a cluster, which has one database. Leaving both unset gets you "default", not “no namespace”:

store = ValkeyStore(url="redis://localhost:6379", namespace="staging")

Every key the store writes is prefixed f"{namespace}:{key}", validated so two namespaces’ prefixes can never collide (even a namespace equal to a key family, like cache or run). Point every worker, CLI invocation and cloud-reporting process for one environment at the same namespace - a mismatch behaves like a different store, not an error, since there is nothing to detect.

A Valkey ACL restricting a runtime’s credentials to the key pattern ~<namespace>:* stops that runtime reading or writing another namespace’s data - but it does not fully confine it. Construction calls FUNCTION LOAD, which needs the +function command category; that category also grants FUNCTION FLUSH/FUNCTION DELETE, which touch the Lua function library’s definitions, server-wide, not scoped by any key pattern. A runtime with +function can still delete every namespace’s functions even if its keys are fenced to its own. Scope +function only to processes meant to manage the library (deploys, prune_stale_function_libraries), not to every worker.

The prune tombstone itself (_pyrula:{functions}:pruned_libraries) is one more key every runtime’s ACL must reach regardless of namespace - it is server-wide by design (see Cloud), so a ~<namespace>:* pattern alone does not cover it; grant ~_pyrula:* or the specific key alongside it.

Fairness across tenants

Tag a run with a tenant by passing owner:

await client.submit("sync_account", {"account": "a-1"}, owner="tenant-42")

Runs without an owner share one default tenant. Claims are fair by design:

• Across tenants on a shard: deficit round robin. The least-served tenant goes next, so one tenant queuing thousands of runs doesn’t delay another’s. Idle tenants don’t bank credit, and a newly active tenant doesn’t jump the queue.
• Within a tenant: strict FIFO by submission time.
• Across shards: a quiet tenant waits about one pass over the shards for its first claim, not for another tenant’s backlog. At 64 shards, a tenant queuing 10 runs behind another tenant’s 10,000 got its first claim after 68 claims (the same with 1,000).

To cap a tenant’s concurrency or rate, use flow control.

Running on a Valkey cluster

Pass cluster=True and any one node’s URL; the client discovers the rest. Every key for a shard shares a hash tag, so each store call lands on one slot.

worker = WorkflowWorker(
    store=ValkeyStore(
        url="valkey://cluster-node-1:6379",
        cluster=True,
        shard_count=64,
    ),
)

A local six-node cluster for testing:

docker compose --profile valkey-cluster up -d --wait valkey-cluster

valkey-py’s cluster client adds a COMMAND GETKEYS round trip before every call, roughly doubling per-call latency against a single node.

Capacity

Claim latency doesn’t grow with backlog: p50 stays at 134-141 µs from an empty queue to 20,000 queued runs. More nodes add throughput, not lower latency. At ~221k ops/s per node and ~5.1 commands per event:

cluster	ops/s	concurrent runs @ 5 events/s each	@ 1 event/s each
1 node	221k	~8,000	~43,000
3 nodes	662k	~25,000	~130,000
6 nodes	1.32M	~50,000	~260,000

Measured in a Docker Desktop VM; rerun on your hardware before planning capacity.

Submitting runs

Submit through a WorkflowClient on the same store, usually in another process. In the worker’s own process, use worker.client:

run = await worker.client.submit("my_workflow", {"x": 1})

Schedules

A schedule submits a run on a cron expression. It lives in the store, so every worker serving that workflow sees it without a restart.

from pyrula.workflows.schedules import Schedule

store.schedule_create(Schedule.create("nightly_report", "0 2 * * *", {"window": "24h"}))

The worker checks for due schedules every 30 s, so a run starts within that window of its cron time. Missed occurrences are not backfilled: after downtime a schedule fires once and moves on. Use store.schedule_list, schedule_update (set enabled=False to pause) and schedule_delete to manage them. Agents can use the pyrula schedules CLI.

Postgres archive

Valkey holds live runs. To keep finished runs queryable for longer, wrap it in an ArchivingStore with a PostgresArchive; terminal runs are copied to Postgres and expire from Valkey live_ttl_after_archive seconds later (default 300).

from pyrula.workflows import ArchivingStore, PostgresArchive, ValkeyStore

store = ArchivingStore(
    live=ValkeyStore(url="redis://localhost:6379"),
    archive=PostgresArchive(dsn="postgresql://..."),
)

Apply the schema once, and check it before starting workers:

pyrula db migrate --postgres-dsn "$PYRULA_POSTGRES_DSN"
pyrula db check --valkey-url redis://localhost:6379 --postgres-dsn "$PYRULA_POSTGRES_DSN"

A missing schema, or one migrated by a different pyrula version, fails check, migrate and the worker’s first run with ArchiveSchemaMismatch. There is no upgrade path between versions: recreate the database and migrate again.

Rolling deploys on Valkey

ValkeyStore runs its atomic operations as a Lua function library loaded into the server. The library is content-addressed: each build loads its own copy under a name derived from the Lua source, so workers on two builds can share one Valkey during a rolling deploy and each keeps calling the functions it loaded. Old libraries are never removed automatically. Once no worker on the previous build is running, prune them:

from pyrula.workflows.stores.valkey.functions import upgrade_scripts

upgrade_scripts("redis://localhost:6379", prune_stale=True)

Pruning while an old worker is still up makes that worker’s next store call fail with StoreUnavailable. That’s why the store never prunes on its own; you run it once the old workers are gone.

A node that lost its libraries without being pruned - restarted without persistence, or added to a cluster after a worker started - answers “Function not found” for the slots it serves. The store loads its own library again and retries that call once, so the worker carries on without a restart. Pruning records the libraries it removed, so a worker on a pruned build still fails rather than loading its library back.

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). On a sharded store it also releases the shards it holds as part of this shutdown, so the remaining workers pick them up right away instead of waiting for shard_lease_ttl to lapse - see Shard leases above.

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())