Flow control

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

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:

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.

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:

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 workflow can’t take any of these (ConfigurationError). Throttle with max_concurrency and the partition count.