Production Worker

WorkflowWorker claims runs, wakes sleeping ones, recovers orphans and fires schedules. InlineWorkflowRunner has no background loops, so anything that sleeps, waits for signals, interrupts or runs on a schedule needs a worker (Deployment modes).

Minimal worker

import asyncio
from pyrula.workflows import WorkflowWorker, ValkeyStore

# Import @workflow 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())

Without a file, pyrula worker --workflow myapp.workflows:process_order (CLI reference) reads the store from PYRULA_STORE_URL and handles SIGTERM (Graceful shutdown). See Deploy a worker and Choosing a store.

Workflow discovery

The worker serves every imported @workflow. To restrict it, pass workflows:

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 first three loops are always on; the rest start only if the store supports them.

• Claim loops, one per workflow, poll every claim_poll_interval and run up to max_concurrent_runs runs at once across workflows.
• Timer loop wakes sleeping runs every timer_poll_interval_s, at most timer_max_batch per tick.
• Orphan recovery loop requeues runs whose worker crashed, every orphan_scan_interval. A workflow name no worker serves is recovered only if its continue_as_new successor’s name is served; otherwise call store.recover_pending_runs(name).
• Schedule loop fires due schedules (Schedules).
• Trigger, reconcile and projection loops handle batch triggers, archival reconciliation and projections.
• Shard lease loop (ValkeyStore) renews this worker’s shards (Scaling across shards).

A run whose attempt loses the store is requeued (run:requeued, reason: store_unavailable), not failed (Errors and retries).

Set options with 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,
        orphan_scan_interval=30.0,
        shard_lease_interval=5.0,
    ),
)

Option	Default
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
event_buffer_size	64
step_boundary_durability	True
reconcile_interval	60.0
projection_interval	60.0
journal_cache_events	50000

journal_cache_events bounds, in events, a per-worker cache of the journals of runs this worker parked. A run resumed on the same worker then reads only the events appended since it parked; a run resumed elsewhere, or after the cache dropped it, reads its whole journal as before. 0 turns the cache off. The bound counts events, not bytes: at a few KB an event the default holds on the order of 150 MB, and a worker whose steps return large payloads should lower it. Set PYRULA_JOURNAL_CACHE_VERIFY=1 while testing to compare every cache hit with a fresh read.

poison_max_retries / poison_retry_ttl_s and stranded_policy / stranded_timeout are covered in Errors and retries.

Journal durability

The journal flushes when a step finishes and on every terminal event, so a recorded step is durable before the next starts (one store round trip per step).

step_boundary_durability=False writes behind, flushing at event_buffer_size events or at run end. It is cheaper for many short steps, but a crash re-runs every step in the lost batch, so use it only when every step is idempotent.

Blocked event loops

A running attempt renews its run lock on its own thread. A body that blocks the event loop keeps the lock for up to max(3 × lock_ttl, timeout); past that, the lock lapses and the orphan scan re-dispatches the run.

Every run on that loop stands down, but only the run whose code was running when the block was found is charged. Its stand-down counts toward poison_max_retries (3) and is logged as run_attempt_loop_blocked. The next count past the limit quarantines the run: run:error code max_retries, error_type EventLoopBlocked, with cause, blocked_for_s and liveness_bound_s in detail.

• A C call that holds the GIL (a catastrophic regex, a huge json.loads) also stops the renewal thread, so the block is found only when the call returns.
• A block in a task the run started (a gather leg, an agent’s tool call) is charged to the run on Python 3.12 and later; on 3.11 only the run’s own task is.
• pyrula dev does not count these stand-downs (WorkerOptions(poison_counts_loop_blocked=False)), so a debugger at a breakpoint does not quarantine the run.

The run’s timeout and a step’s timeout= are enforced by the event loop, so neither fires while it is blocked. Call blocking code from a sync step, or wrap it in asyncio.to_thread.

Scaling across shards

ValkeyStore splits each workflow’s queue into shard_count shards (default 64), the unit of spreading across cluster nodes and of fairness.

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

Use the same shard_count on every worker for a name; a different count raises ConfigurationError. Drain a name that has run before changing its count.

Shard leases

Each worker leases unowned shards (shard_lease_ttl, default 30 s), renews them every shard_lease_interval and rebalances toward shard_count / N. It claims from its own shards first, then any.

• worker.stop() releases leases at once.
• After a crash, other workers keep claiming new work. In-flight runs recover on the orphan scan (lock_ttl plus up to orphan_scan_interval), and the shards are re-leased after shard_lease_ttl.

Key namespace

Every ValkeyStore has a namespace, "default" when unset. Pass namespace (or set PYRULA_VALKEY_NAMESPACE) to run independent stores on one instance. On a cluster, which has one database, it is the only way.

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

Point every worker, CLI invocation and cloud-reporting process of one environment at the same namespace; a mismatch acts like a different store, with no error.

A Valkey ACL on ~<namespace>:* fences the store’s data, scripts included. The store needs +eval +evalsha +script|load and sends no FUNCTION command (Managed Valkey and Redis has the recipe).

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. Across tenants on a shard, claims use deficit round robin: the least-served tenant goes next, so one tenant’s queued runs don’t delay another’s. Within a tenant, claims are FIFO by submission time. To cap a tenant’s concurrency or rate, use flow control.

Running on a Valkey cluster

Give the store any one node’s URL; the client discovers the rest. cluster defaults to None: the store reads INFO once at start and builds the cluster client when the server reports a cluster.

from pyrula.workflows import ValkeyStore, WorkflowWorker

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

cluster=True (the CLI’s --cluster) overrides detection, and cluster=False forces the plain client.

• A cluster that hides its mode, behind a proxy or from a user whose ACL denies INFO, needs True. Otherwise the store reads it as standalone and the first call that meets a redirect raises a ConfigurationError naming the flag.
• An ElastiCache Serverless host always gets the cluster client without a probe.
• A server that does not answer INFO on any attempt raises StoreUnavailable.

Local six-node cluster: docker compose --profile valkey-cluster up -d --wait valkey-cluster.

Submitting runs

Submit through a WorkflowClient on the same store, or worker.client in the worker’s process.

Reporting to Cloud

With PYRULA_CLOUD_API_KEY set, the worker reports its runs to Cloud and takes the runs Cloud submits, including what-if runs. Pass cloud=False to switch that off. A key that does not parse, a missing pyrula-cloud-client, or an unusable PYRULA_CLOUD_* setting (such as PYRULA_CLOUD_DRAIN_DEADLINE_S=5s) raises ConfigurationError at start.

Schedules

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

from pyrula.workflows.schedules import Schedule

store.schedule_create(
    Schedule.create(
        "nightly_report", "0 2 * * *", {"window": "24h"},
        schedule_id="nightly_report",  # stable, so a restart does not add a second schedule
    )
)

schedule_create overwrites: the same schedule_id replaces the record, re-enables a paused schedule and resets next_run_at. Without one, every call mints a new schedule. The pyrula schedules create CLI derives its own id from name, cron and params, so mixing the two paths makes two schedules.

A schedule has exactly one of a cron expression or an interval (every=timedelta(...), at least 60 s). Interval occurrences sit on a grid from the creation time. Cron runs on wall-clock time in its timezone (IANA name, default "UTC"):

• When both day-of-month and day-of-week are restricted, a day matches if either does, as in standard cron.
• A wall time skipped by spring forward fires once after the gap. A time that occurs twice in fall back fires once.

The worker checks for due schedules every 30 s. Each occurrence has idempotency key schedule:{id}:{epoch}, so two workers never fire it twice.

overlap decides what happens when an occurrence is due while the previous run is unfinished:

• "skip" (default) does not submit it, logs schedule_skipped_overlap and counts pyrula.workflows.schedules.skipped. A run parked for days therefore suppresses its schedule.
• "allow" submits anyway.

After downtime a schedule fires once and moves to the first occurrence after now; missed ones aren’t backfilled. catchup_window=timedelta(...) (at least 60 s) fires that one only if a missed occurrence is within the window.

Manage schedules with store.schedule_list and schedule_delete, and pause_schedule, resume_schedule and run_schedule_now from pyrula.workflows.schedules.control. Agents can use the pyrula schedules CLI.

Postgres archive

Valkey holds live runs. To keep finished runs queryable 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). Install pip install "pyrula[workflows,postgres]" ("pyrula[agents,postgres]" for an agents app) first.

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 --store redis://localhost:6379 --postgres-dsn "$PYRULA_POSTGRES_DSN"

A missing schema, or one from a different baseline, fails check and the worker’s first run with ArchiveSchemaMismatch; recreate the database and migrate again. An additive migration keeps older workers running; a contracting one makes them refuse the database, so migrate after every worker is upgraded.

Rolling deploys on Valkey

ValkeyStore loads one Lua script per build with SCRIPT LOAD and calls it by SHA-1 with EVALSHA. Each build’s script is its own entry in the server’s script cache, so workers on two builds can share one Valkey during a rolling deploy. A deploy step can load the new build’s script before its workers start:

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

upgrade_scripts("redis://localhost:6379")

Old scripts are not pruned. Each build adds one cache entry of roughly 140 KB and about 0.5 µs to every call (11 µs with 20 builds cached). After a deploy retires a build, run SCRIPT FLUSH, on every primary in a cluster. It empties the cache for every application sharing the server, so another application that calls EVALSHA without a NOSCRIPT fallback fails. The store itself reloads its script and retries once on NOSCRIPT, so flushing during a deploy is safe.

Health probes

Set WorkerOptions(health_port=8081) (or PYRULA_WORKER_HEALTH_PORT; the option wins) to serve two unauthenticated JSON routes. 0 picks a free port; worker.health_address returns the bound (host, port).

Route	Answers 200 when	Otherwise
/livez	the event loop is turning and no background loop has failed for 2 minutes or (on an agent worker) died	no answer (the probe times out), or 503 loop_failed
/readyz	the store answers ping() within 2 seconds, its binding matches this project and environment, no loop has died, and the worker is not shutting down	503 with status starting, draining, loop_failed, store_binding_mismatch or store_unreachable

A ready body also lists shards, the shard count held per workflow. /readyz turns 503 as soon as stop() is called; /livez answers until the process exits.

A loop failing on a store or network error is logged once per outage and restarted with a jittered delay doubling from 0.5 to 30 s (counted in pyrula.workflows.loop.restarts). Any other exception stops the worker and is re-raised from run(). /livez recovers on the next completed tick, so a loop with an interval over 2 minutes (say reconcile_interval=300) fails it from 2 minutes after a failed tick until the next completes.

The server binds 127.0.0.1. In a container, set WorkerOptions(health_host="0.0.0.0") or PYRULA_WORKER_HEALTH_HOST=0.0.0.0 and expose it to the cluster network only.

containers:
  - name: worker
    env:
      - { name: PYRULA_WORKER_HEALTH_PORT, value: "8081" }
      - { name: PYRULA_WORKER_HEALTH_HOST, value: "0.0.0.0" }
    livenessProbe:
      httpGet: { path: /livez, port: 8081 }
      periodSeconds: 10
      failureThreshold: 3
    readinessProbe:
      httpGet: { path: /readyz, port: 8081 }
      periodSeconds: 5
    # Longer than the drain: shutdown_grace + 3 x cancel_timeout + the Cloud connector's 5 s.
    terminationGracePeriodSeconds: 60

Platforms that take an HTTP health-check path use /readyz.

Graceful shutdown

Call worker.stop() from a signal handler or another coroutine. In order, the worker:

1. Stops claiming.
2. Waits up to shutdown_grace seconds for in-flight runs.
3. Cancels the rest together, within one cancel_timeout.
4. Waits up to cancel_timeout for store calls, sync steps and sync tools still in a thread, so the store can be closed once run() returns.
5. On a sharded store, releases its shards so others pick them up without waiting for shard_lease_ttl (Shard leases).

Runs still in flight are recovered as after a crash.

A hung sync step or tool delays run() by up to cancel_timeout, then is logged (worker_thread_calls_still_running) and left running on a daemon thread. Its run goes back in the queue, and the worker that claims it runs the step again from the start while the old thread may still be running (The window where a step can run twice). When every thread of the sync-step pool (min(32, CPUs + 4)) is busy, further sync steps and tools wait (logged once as user_code_threads_saturated).

pyrula dev drains faster (--shutdown-grace 3, --cancel-timeout 2), inside docker compose stop’s 10 seconds. With pyrula dev or pyrula worker, a second Ctrl+C or SIGTERM while draining quits at once with 130 or 143. A C call that holds the GIL still delays the second signal until it returns; use SIGKILL there.

The Cloud connector gets PYRULA_CLOUD_DRAIN_DEADLINE_S (default 5 seconds) to send what it holds. With a spool, critical events that miss it go out on the next start and only live-stream detail is lost. With no spool (PYRULA_CLOUD_OUTBOX_PATH=off, or it failed to open), undelivered events are lost.

The following example stops the worker on SIGTERM and SIGINT:

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