How to limit concurrency and overlap¶

Three controls, from narrow to wide: a cap on one flow, named resources shared by many, and the engine pool.

Cap one flow¶

from cereyan import flow

@flow(max_concurrent=1, on_overlap="skip")
def every_ten_minutes() -> None:
    ...

@flow(max_concurrent=2)
def exports() -> None:
    ...

assert every_ten_minutes.options["on_overlap"] == "skip"

max_concurrent limits how many runs of the flow are Pending or Running at once. When the cap is reached, on_overlap decides: enqueue (default) waits as AwaitingResource, skip ends the new run Skipped, cancel_new ends it Cancelled, cancel_old cancels the flow's other runs (a queued one at once, a running one as a user cancel would, grace period included) and lets the new run take the slot they free, and buffer_one waits unless another run is already waiting, in which case the new one ends Skipped with reason buffered. Pick skip for polling flows where a missed tick does not matter, enqueue for flows that must not lose work, cancel_old when only the latest request matters, and buffer_one when one pending refresh is all that is ever needed.

A run can also give up on waiting: start_deadline=300 on the flow, or on a schedule, skips a run that has not started five minutes after it was due (or, for a run started by hand, after it was created) with reason missed_start_deadline. A schedule's deadline wins over the flow's.

A run that finds no slot is still created for its tick, so every scheduled time appears in history. Under skip and cancel_new it ends at once with the message "previous run still active" (and a run.skipped event for skip); under enqueue it stays Scheduled, turns Late once its time passes, and starts in scheduled-time order as soon as a run of the flow ends. The reference for this behaviour is the overlap soak, just soak in Contributing, which runs fifteen such flows for an hour and checks that caps hold, controls never skip, and queues drain in order.

Declare totals in cereyan.toml or on the Settings page, and claim units on flows or tasks:

[resources]
db = 4
gpu = 1

from cereyan import flow, task

@task(resources={"db": 1})
def query(sql: str) -> str:
    return sql

@flow(resources={"gpu": 1})
def train() -> str:
    return query("select 1")

assert train() == "select 1"

A run or task run waits as AwaitingResource until every resource it declares has capacity, and a resource.exhausted event records the first wait. Units are released on every terminal state, on pause, and when an engine dies. Resources are global to the machine, so two projects declaring db share the same four units. A resource that is not declared in configuration has a total of one.

Keyed resources¶

A resource name can be a template over the run's parameters, and a total can cover a family of names:

[resources]
"api:*" = 1        # one call per tenant at a time
"tag:gpu" = 1      # one run tagged gpu at a time, whatever its flow

from cereyan import flow, task

@flow(resources={"api:{{ tenant }}": 1})
def call(tenant: str) -> str:
    return tenant

@task(resources={"db:{shard}": 1})
def load(shard: str) -> str:
    return shard

{{ tenant }} and {tenant} both render from the run's parameters when the run is admitted (flow resources) or the task starts (task resources), offline and served alike; an unknown name renders empty. A total whose key contains * applies to every rendered name it matches that has no total of its own, so api:acme and api:globex each get one slot and never wait on each other; such instances appear in the resources snapshot while in use and go away when idle. tag:<name> limits apply to every run carrying that tag, on top of the flow's own resources.

Keep runs unique¶

Overlap policies act on a flow as a whole; a unique key acts on which run it is. @flow(unique=Unique(key="{day}")) says that at most one run of load for a given day exists at a time: while such a run is Scheduled, Pending, Running or Paused, submitting another for the same day creates nothing and answers with the run that holds the key. The check applies wherever a run is created: POST /api/runs answers 200 {"conflict": true, "run": ...} instead of 201, Client.run returns the run with conflict: True, the MCP run_flow tool reports conflict, a rule's run_flow records it in its outcome, and an offline flow call returns None without executing.

from cereyan import Unique, flow

@flow(unique=Unique(key="{day}"))
def load(day: str) -> str:
    return day

@flow(unique=Unique(period=3600, states=("Scheduled", "Pending", "Running", "Completed")))
def hourly_report(region: str) -> str:
    return region

@flow(unique=Unique(on_conflict="replace"))
def sync() -> None:
    ...

Without key, every parameter is the key. period buckets the key by fixed clock windows, and adding Completed to states keeps the key taken after the run ends, so hourly_report runs at most once per region per hour whatever asks for it. on_conflict="replace" cancels the run that holds the key and creates the new one; a run that is already winding down keeps it.

Two more policies compose a unique key with a delayed start. on_conflict="debounce" with debounce=30 creates the run thirty seconds ahead and, while it has not started, every further submission for the same key moves its start to thirty seconds from now and replaces its parameters, so a document saved ten times in a minute renders once, with the last version; max_wait caps how far the run can be pushed from its creation. on_conflict="throttle" with period=60 keeps the first run of a key in any sliding minute and answers the rest with it, so a bursty webhook starts one run per minute.

from cereyan import Unique, flow

@flow(unique=Unique(key="{doc}", on_conflict="debounce", debounce=30, max_wait=300))
def render(doc: str, version: int) -> str:
    return f"{doc}:{version}"

@flow(unique=Unique(key="{source}", on_conflict="throttle", period=60))
def ping(source: str) -> str:
    return source

A request can carry its own key instead: idempotency_key (and idempotency_ttl, a day by default) on POST /api/runs, POST /api/flows/{id}/runs and Client.run makes a repeated call, a webhook delivered twice for instance, answer with the run first created whatever its state. Runs started by a dependency are keyed by their upstream run, so an upstream completion processed twice starts one downstream run.

Order the queue¶

priority (higher first) decides which waiting run gets the next free engine or unit. It never preempts a running run. A negative priority also lowers the engine's OS scheduling priority on Linux and macOS, useful for background reprocessing:

from cereyan import flow

@flow(priority=-10)
def reprocess_history() -> None:
    ...

Bound the pool¶

max_engines in [server] or --max-engines caps engine processes, and therefore runs executing at once, machine-wide. The default is 1 and the most is the CPU count; the Queue page changes it while the server runs.

Stop a failing flow¶

from cereyan import flow

@flow(disable_after=(3, 3600, 86400))
def nightly() -> None:
    ...

Three failures within an hour pause the flow's schedules for a day, with flow.disabled and flow.enabled events marking the window. Manual runs still work.

Check what is waiting¶

The Runs page filter AwaitingResource, the run's Details tab (which names the resource), and GET /api/counts show what is blocked. The Settings page shows each resource's total and in-use count.