Flow control

Flow control is declared on the workflow and enforced by the store when a worker claims a run. Nothing in the body changes.

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

Here at most two sync_account runs per tenant execute at once, and each tenant starts at most 5 per second (bursts of 10). A run over the limit stays queued until a slot or token frees up; it is never dropped.

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; start one run at max_events or after window_seconds
Debounce	key, window_seconds	buffer events per key; start one run once they’ve been quiet for window_seconds

Keys

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

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 every run without it shares one bucket.

Runs that share a Limit or Rate key always land on the same shard, so the counter is exact. The flip side: a tenant that gates everything on one key gets one shard’s worth of throughput for that gate. See Fairness across tenants for how claims are shared between tenants.

The policy is stamped when the run is submitted, by whichever path submits it: a WorkflowClient, HTTP, ctx.run_child, a schedule or a trigger. Changing the decorator affects new runs, not queued ones.

Batching and debouncing events

Collect and Debounce turn a stream of events into fewer runs. Send events with client.trigger or POST /{name}/events; the run receives key, events and count:

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)
await client.trigger("digest", {"user_id": "u1", "kind": "comment"})   # None until the batch fires

trigger returns the new run’s id when an event completes a Collect batch, else None. Time-based fires (a Collect window, any Debounce) are submitted by the worker’s trigger loop.

A triggered workflow can also take concurrency= and rate_limit=. Key those on "key", the batch key, since the run’s params are the batch and not the raw event.

A Kafka-bound workflow can’t take concurrency, rate_limit or trigger: records are processed as they’re consumed, so binding one raises ConfigurationError. Throttle a stream with max_concurrency and the partition count.