# Flow control

> Per-key concurrency limits, rate limits, and batching or debouncing events into runs.

Source: https://docs.pyrula.io/workflows/guides/flow-control/

Declared on the workflow, enforced by the store when a worker claims a run. The body doesn't change.

```python
from pyrula.workflows import Limit, Rate, workflow

@workflow(
    name="sync_account",
    concurrency=Limit(key="tenant_id", max=2),
    rate_limit=Rate(key="tenant_id", per_second=5, burst=10),
)
async def sync_account(ctx, tenant_id: str, account: str) -> None:
    ...
```

Two `sync_account` runs per tenant at once, 5 starts per second per tenant (bursts of 10). A run
over the limit stays queued, never dropped. A suspended run frees its `Limit` slot.
`InlineWorkflowRunner` obeys the same limits.

A queued run holds its place until it runs or is cancelled. Inline, a gated run raises
`RunGatedError` after `gate_timeout` seconds (300 default, `None` for no limit), with a log line
after 10 s naming the gate.

| Policy | Arguments | Gate |
|---|---|---|
| `Limit` | `key`, `max` | At most `max` runs per key executing at once |
| `Rate` | `key`, `per_second`, `burst=None` | Token bucket per key; `burst` defaults to `ceil(per_second)` |
| `Collect` | `key`, `max_events`, `window_seconds` | Buffer events per key; one run at `max_events` or after `window_seconds` |
| `Debounce` | `key`, `window_seconds` | Buffer events per key; one run once quiet for `window_seconds` |

## Keys

`key` is a parameter name or a function of the params:

```python
Limit(key="tenant_id", max=2)                               # params["tenant_id"]
Limit(key=lambda p: f"{p['org']}:{p['region']}", max=2)     # computed
```

A missing parameter resolves to `""`, so runs without it share one bucket.

Runs sharing a key land on one shard, so the counter is exact, and one hot key gets one shard's
throughput. See [Fairness across tenants](https://docs.pyrula.io/workflows/guides/worker.md#fairness-across-tenants).

The policy is stamped at submit, whatever path submits (client, HTTP, `ctx.run_child`, schedule,
trigger). Changing the decorator affects new runs only.

A process that submits by name without importing the workflow (a web or Celery process feeding a
separate worker) reads the policy from the store, where each `WorkflowWorker` records the `Limit`
and `Rate` of the workflows it serves at start. A key given as a function resolves only where the
workflow is imported, so submitting it by name raises `ConfigurationError` (and `pyrula runs retry`
exits 2).

### Out-of-date gates

Every run's `run:init` names the gates it was stamped with. A run submitted by name before any
worker recorded the policy is stamped as unknown, and a run stamped by an older build names that
build's gates. If the claiming worker's own `Limit` and `Rate` differ, it hands the run on before
its body runs: it submits one successor stamped with its own gates and version and ends the
original `run:continued`. A run whose body has started keeps its gates.

A hand-on only goes forward: a worker hands on a run only when its workflow version ranks at
least as new as the one that stamped it, so an older build mid-deploy never takes a limit away.
Runs Pyrula Cloud mints and waits on (`case`, `replay` and `eval` triggers) and retries submitted
by name are never handed on.

The id you hold keeps working: `result`, `status`, `signal`, `cancel` and `terminate` on the
original id act on the successor, which records the original as `provenance.source.gated_from`.
`client.events(run_id)` returns only that run's own journal, so the work is in the successor's.

## Batching and debouncing events

`Collect` and `Debounce` turn many events into fewer runs. Send events with `client.trigger` or
`POST /{name}/events`; the run gets `key`, `events` and `count`:

```python
from pyrula.workflows import Collect, WorkflowClient, workflow

@workflow(name="digest", trigger=Collect(key="user_id", max_events=50, window_seconds=60))
async def digest(ctx, key: str, events: list[dict], count: int) -> None:
    ...

client = WorkflowClient(store)
handle = await client.trigger("digest", {"user_id": "u1", "kind": "comment"})   # None until the batch fires
```

`trigger` returns a `RunHandle` when the event completes a `Collect` batch, else `None`; a
worker's trigger loop submits time-based fires. A triggered workflow can also take
`concurrency=` and `rate_limit=`, keyed on `"key"`.

A [Kafka-bound](https://docs.pyrula.io/workflows/guides/kafka.md) workflow can't take any of these (`ConfigurationError`).
Throttle with `max_concurrency` and the partition count.