# Production Worker

> Run WorkflowWorker: loops, shards, schedules, health probes and shutdown.

Source: https://docs.pyrula.io/workflows/guides/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](https://docs.pyrula.io/workflows/concepts/deployment-modes.md)).

## Minimal worker

```python
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](https://docs.pyrula.io/reference/cli.md#pyrula-worker)) reads the store from `PYRULA_STORE_URL` and handles SIGTERM ([Graceful shutdown](#graceful-shutdown)). See [Deploy a worker](https://docs.pyrula.io/getting-started/deploy-a-worker.md) and [Choosing a store](https://docs.pyrula.io/getting-started/choosing-a-store.md).

## Workflow discovery

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

```python
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](#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](#scaling-across-shards)).

A run whose attempt loses the store is requeued (`run:requeued`, `reason: store_unavailable`), not failed ([Errors and retries](https://docs.pyrula.io/workflows/guides/errors-and-retries.md#when-the-store-fails-under-a-run)).

Set options with `WorkerOptions`:

```python
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](https://docs.pyrula.io/workflows/guides/errors-and-retries.md).

### 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.

```python
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.

```python
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](https://docs.pyrula.io/getting-started/managed-valkey.md#when-the-server-refuses-scripts) has the recipe).

## Fairness across tenants

Tag a run with a tenant by passing `owner`:

```python
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](https://docs.pyrula.io/workflows/guides/flow-control.md).

## 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.

```python
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`](https://docs.pyrula.io/workflows/guides/client.md) 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](https://docs.pyrula.io/getting-started/cloud.md) and takes the runs Cloud submits, including [what-if](https://docs.pyrula.io/workflows/guides/what-if.md) 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.

```python
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](https://docs.pyrula.io/agents/serving/schedules.md).

## 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.

```python
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:

```bash
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:

```python
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.

```yaml
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](#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](https://docs.pyrula.io/workflows/concepts/replay-and-recovery.md#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:

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