Concurrency and Rate Limiting

Back in the dynamic execution post, fan-out stopped being a design problem and became a method call: load_file.map(files) submits one task run per file and runs them concurrently through a thread pool, no executor conversation required. That’s the feature. It’s also the loaded gun. The morning the extract drops forty regional files instead of three, that same clean line opens forty concurrent connections to the warehouse — and the warehouse, or the third-party API behind one of those tasks, was never asked whether it wanted forty. Prefect made the fan-out free. Nothing made it safe. That’s this chapter’s job.

In Airflow you had one tool for this and you knew its name: the pool. You created a pool with some number of slots, tagged tasks into it with pool="warehouse", and the scheduler refused to run more than that many pooled tasks at once, across every DAG. One concept, one ceiling, applied by the scheduler. It was blunt and it was reliable, and its bluntness was the point — you rarely had to think about which mechanism, because there was only one.

Prefect fragments that single idea into several distinct knobs, each operating at a different layer of the stack, and getting production right means knowing which one you’re reaching for. There’s a tag-based limit that looks almost exactly like an Airflow pool. There’s a global concurrency limit you can wrap around an arbitrary block of code, not just a task. There’s a rate limit for “N per second” instead of “N at once.” And there are two infrastructure-level ceilings — on work pools and on deployments — that count whole flow runs rather than tasks. They are not interchangeable, and the reason there are several is that Airflow’s one pool was quietly answering four different questions at once.

Tag-based limits: the Airflow pool, ported

Start with the one that maps cleanest, because if you’re coming from pools it’s the mechanism you already understand. A tag-based task concurrency limit says: across every flow run, everywhere, no more than N tasks carrying this tag may be Running at the same time. That is the Airflow pool, almost line for line — a named ceiling on concurrent task execution, enforced globally rather than per flow.

You create the limit against a tag name and a slot count:

prefect concurrency-limit create warehouse 5

Then you tag the tasks that should share those five slots:

from prefect import flow, task

@task(tags=["warehouse"])
def load_file(path: str) -> int:
    rows = read_csv(path)
    write_to_warehouse(rows)          # holds a warehouse connection
    return len(rows)

@flow(log_prints=True)
def daily_load():
    files = list_order_files()
    counts = load_file.map(files)     # fans out to N runs...
    total = sum(f.result() for f in counts)
    print(f"loaded {total} rows across {len(files)} files")

Now the .map() can produce forty load_file runs and Prefect will not care. Every one of those runs carries the warehouse tag, and the limit lets exactly five hold the Running state at once; the other thirty-five sit in a concurrency-gated pending state and get promoted as slots free up. The fan-out count is still whatever the data says it is — you didn’t cap the work, you capped how much of it touches the warehouse simultaneously. That distinction is the whole game, and it’s identical to what the Airflow pool bought you.

The tag is the join key, and it’s cheap to slice finely. Give your warehouse writes tags=["warehouse"] and your Stripe calls tags=["stripe_api"], create a limit for each, and the two ceilings are independent — a backfill saturating the warehouse can’t also starve the payments API, because they draw from different pools. A task can carry several tags and will then be governed by every matching limit at once, which is occasionally what you want (a task that both writes the warehouse and calls Stripe waits on whichever slot is scarcer) and occasionally a footgun (a broad tag like db accidentally throttling half your codebase). Tag deliberately.

Two operational notes that bite people. First, these limits are global server state, not code — the limit for warehouse lives in the Prefect API and persists until you change it, so prefect concurrency-limit ls, inspect, and delete are how you see and manage them, and a limit set to 0 is a valid, deliberate way to pause every task with that tag without touching a deployment. Second, a task that dies without cleanly releasing its slot — a hard crash, a killed worker — can leave a slot occupied; Prefect reclaims these, but if you ever see a limit wedged below its ceiling, prefect concurrency-limit inspect <tag> shows you which run IDs are holding slots, and that’s your first stop.

Global concurrency limits and the `concurrency()` block

The tag limit is elegant right up until the thing you need to protect isn’t a task. Maybe the contended resource is guarded inside a helper function three calls deep, or a single task makes several protected calls in a loop, or the resource is shared by code that isn’t a Prefect task at all. Tagging can’t reach in there. This is where Prefect goes somewhere Airflow’s pool never could: the global concurrency limit and its concurrency() context manager, which put a ceiling around an arbitrary block of code.

You create the limit up front, same shape as before but this is a distinct object type:

prefect gcl create warehouse-writes --limit 5

(gcl is the short alias; prefect global-concurrency-limit create is the full form, and prefect gcl ls / inspect / update round it out.) Then you acquire slots inline, wherever the contention actually lives:

from prefect import flow, task
from prefect.concurrency.sync import concurrency

@task
def load_file(path: str) -> int:
    rows = read_csv(path)
    with concurrency("warehouse-writes", occupy=1):
        write_to_warehouse(rows)      # only 5 of these run at once, globally
    return len(rows)

Only five threads, flows, or processes — anywhere holding this context — can be inside that with block simultaneously; the sixth blocks at the with line until one exits and releases a slot. Notice what moved: the ceiling no longer wraps the whole task, only the write. read_csv still runs at full fan-out width; the throttle applies exactly to the four lines that touch the warehouse. That surgical scope is the reason to reach past a tag limit.

The trap: a limit that doesn’t exist is not an error

Now the part that will bite you, and it is exactly the kind of silent failure this book keeps hunting. The limit and the code that uses it are created in two different places — one is a CLI command you ran once, the other is a string in your source. Nothing checks that they match. And if they don’t:

with concurrency("warehouse-wrties", occupy=1):   # typo. or a fresh environment
    write_to_warehouse(rows)                      # ...runs at full fan-out. no error.

concurrency() against a limit that doesn’t exist does nothing at all — it enters the block, grants you imaginary slots, and returns. No exception, no warning, no log line. Your throttle is now a comment. Typo the name, or deploy to a staging environment where nobody ran prefect gcl create, and the protection you carefully designed silently isn’t there — which you discover when the warehouse falls over under a load you thought was capped at five.

The fix is one keyword, and you should treat it as mandatory in anything that isn’t a scratch script:

with concurrency("warehouse-writes", occupy=1, strict=True):
    write_to_warehouse(rows)
# ConcurrencySlotAcquisitionError: Concurrency limits ['warehouse-writes']
# must be created before acquiring slots

strict=True refuses to proceed against a limit that was never created. It turns “silently unprotected” into “loudly broken at the first call,” which is the trade you always want from a guardrail. rate_limit() has the identical default and the identical trap — pass strict=True there too. The default is False presumably so that a limit you haven’t created yet doesn’t block local development; that’s a defensible default and a terrible production posture, and knowing which one you’re in is your job.

Three things the block gives you that the tag can’t. occupy lets a single acquisition claim more than one slot — a bulk load that’s worth three normal writes can take occupy=3, so the limit measures weight, not just count. A timeout_seconds turns “wait forever for a slot” into “raise after this long,” which you want anywhere an unbounded wait would be worse than a clean failure. And because it’s just a context manager, it composes with anything — it protects code inside a task, code shared across tasks, and code that runs outside a flow entirely. There’s an async twin, from prefect.concurrency.asyncio import concurrency, with the identical shape for async flows, so an async API client gets the same guard without blocking the event loop.

The mental model worth carrying: a tag limit governs task runs by their tag; a global limit governs whoever enters the block, by name. The first is declarative and coarse and reads like an Airflow pool. The second is imperative and precise and has no Airflow analogue at all — it’s the tool for when the resource you’re protecting doesn’t line up with a task boundary.

The async block, inside a `.map()` fan-out

The place this earns its keep is the exact scenario that opened the chapter: a .map() that fans out wide against a resource that hates being hit wide. Do it with async tasks and the guard has to hold across an await — the slot must stay claimed while a coroutine is parked on a network call, not just while it’s on-CPU. The async concurrency() does precisely that. Here’s the whole thing, end to end:

import httpx
from prefect import flow, task, unmapped
from prefect.concurrency.asyncio import concurrency

@task
async def enrich(order: dict, client: httpx.AsyncClient) -> dict:
    async with concurrency("warehouse-writes", occupy=1, timeout_seconds=30):
        resp = await client.post("https://internal/enrich", json=order)
        resp.raise_for_status()
    return resp.json()

@flow(log_prints=True)
async def enrich_all(orders: list[dict]):
    async with httpx.AsyncClient(timeout=30) as client:
        futures = enrich.map(orders, client=unmapped(client))
        enriched = [f.result() for f in futures]
    print(f"enriched {len(enriched)} orders, at most 5 in flight at the warehouse")

Two things are doing the work. unmapped(client) tells .map() not to iterate the shared HTTP client — every mapped run reuses the same connection pool while orders is the axis that fans out (the .map() broadcast rule from the dynamic-execution post, and it matters here because you want one pooled client, not a thousand). And the async with concurrency(...) means that when a thousand enrich runs are all suspended on their POST, only five are actually inside the block holding a slot; the other 995 are suspended at the async with line, consuming no slot and not blocking the event loop. Swap the import to prefect.concurrency.sync and the identical shape works for threaded tasks — the async version exists so the wait yields the loop instead of pinning a thread. The timeout_seconds=30 is the seatbelt: if the warehouse wedges and no slot frees in thirty seconds, the acquisition raises rather than hanging the whole fan-out forever.

`rate_limit()`: throttling per unit of time

Concurrency limits answer “how many at once.” A whole class of external systems doesn’t care about that question — they care about “how many per second.” A partner API rated at 100 requests/minute doesn’t mind whether you make them one connection or ten; it minds the rate, and it will 429 you the moment you cross it regardless of your concurrency. Capping simultaneous calls is the wrong instrument here: five concurrent tasks in a tight loop can blow a per-second budget while never exceeding five at once.

The right instrument is rate_limit(), which throttles the frequency of entry rather than the count of occupants. It’s built on the same global-limit object, with one extra property: a slot decay that returns slots to the pool at a fixed rate over time. You configure the decay when you create the limit:

prefect gcl create partner-api --limit 100 --slot-decay-per-second 1.67

A limit of 100 with slots decaying at ~1.67 per second is, in effect, a budget of about 100 calls per minute that refills continuously. Then you call rate_limit() before the protected call:

from prefect import task
from prefect.concurrency.sync import rate_limit

@task
def enrich_order(order: dict) -> dict:
    rate_limit("partner-api")         # blocks until a slot has decayed free
    return partner_client.enrich(order)

Unlike concurrency(), rate_limit() isn’t a context manager and doesn’t wrap a block — it’s a single gate you pass through. It acquires a slot and doesn’t hold it; the slot is simply gone until decay replenishes it. When you fan out enrich_order.map(orders) over a thousand orders, they don’t all fire at once and they don’t trickle through a fixed five-wide door — they pace themselves to the decay rate, and the partner API sees a smooth ~100/min stream instead of a thundering herd followed by a wall of 429s. That’s the behavior you actually wanted from “don’t hammer the API,” and no amount of pure concurrency capping produces it. There’s an async rate_limit under prefect.concurrency.asyncio too, same signature.

The math underneath is worth making explicit, because it’s the whole difference from a concurrency cap. Model the limit as a bucket holding at most --limit slots. Every rate_limit() call spends one slot and doesn’t return it; independently, --slot-decay-per-second slots trickle back into the bucket every second, up to the ceiling. Two numbers, two behaviors. The decay rate sets the sustained throughput — at 1.67/s the bucket hands out ~1.67 calls per second on average, ~100/min, no matter how long you run. The limit sets the burst allowed before pacing bites: a full bucket of 100 lets the first 100 calls rush through, then the 101st waits ~0.6s for a slot to decay back and you settle into the steady 100/min. Want a strict trickle with no burst? --limit 1 --slot-decay-per-second 1.67, and every call waits its ~0.6s turn. Contrast a concurrency cap of 5, where slots return the instant a call finishes — fast calls refill it in milliseconds and let five more fire at once, which is exactly why a tight loop of quick requests stays under “5 at once” while blowing past “100 per minute.” Concurrency slots track occupancy; decay slots track the clock, and only the clock enforces a rate.

Work-pool and work-queue caps: counting flow runs

Everything so far counts tasks (or code blocks). The next two mechanisms count flow runs, and they live at the infrastructure layer we built in the work pools post rather than in your flow code. The distinction matters: a task-tag limit can’t stop ten copies of the same flow from all starting, because each copy independently respects the tag ceiling for its own tasks. If the problem is “too many whole runs at once,” you need a ceiling on runs.

The coarsest is the work-pool concurrency cap. A pool is a typed queue of scheduled runs; capping it limits how many of its runs can be executing simultaneously, no matter how many workers are polling or how much the scheduler wants to launch:

prefect work-pool set-concurrency-limit k8s-pool 10

Now k8s-pool runs at most ten flow runs concurrently. The eleventh scheduled run stays in the pool in a pending state until one of the ten finishes — the runs aren’t lost, they’re queued at the infrastructure door. This is your backstop against a cluster meltdown: whatever your deployments schedule, the pool won’t let more than ten pods exist at once, so a runaway schedule or a flood of event-triggered runs can’t scale your Kubernetes bill to the moon. Set it to the concurrency your infrastructure can actually afford, and treat it as a safety rail rather than a routing tool.

Work queues give you the same lever one level finer. A pool can hold several named queues with their own priorities and their own caps, so you can carve the pool’s total capacity between classes of work:

prefect work-queue set-concurrency-limit high 3 --pool k8s-pool

Cap the high queue at three and the default queue at, say, seven, and you’ve reserved headroom so an urgent run isn’t stuck behind a backfill that filled the pool. Priorities decide drain order; per-queue caps decide how much of the pool any one class may occupy. Reach for queues when one pool legitimately serves work of different urgency; skip them until you feel that specific pain, exactly as in the pools post.

Deployment concurrency and the collision strategy

The last knob is the most specific and the one with a genuinely new decision attached. A deployment-level concurrency limit caps how many runs of one particular deployment can be active at once — and crucially, it makes you answer a question the other mechanisms dodge: when a new run wants to start but the limit is full, what happens to it?

The classic case is a deployment that must never overlap itself. Your hourly incremental load reads a watermark, processes everything since, and advances the watermark; if the 3:00 run is still going when the 4:00 run fires, two runs race on the same watermark and you get duplicated or skipped data. The flow isn’t idempotent under overlap, so overlap must be forbidden. A limit of one enforces it:

from prefect.client.schemas.objects import ConcurrencyLimitConfig
from flows import incremental_load

if __name__ == "__main__":
    incremental_load.deploy(
        name="bookshop-incremental",
        work_pool_name="k8s-pool",
        concurrency_limit=ConcurrencyLimitConfig(
            limit=1,
            collision_strategy="ENQUEUE",
        ),
    )

limit=1 means one run of bookshop-incremental at a time. The interesting part is collision_strategy, which decides the fate of a run that arrives while the limit is full:

ENQUEUE holds the new run in a pending state and lets it start once the slot frees. If the 3:00 run runs long, the 4:00 run waits and then executes — nothing is dropped, work just serializes. This is what you want when every scheduled run represents data that must eventually be processed. The cost is that a persistently slow run backs up a queue behind it, and you can drift ever further from wall-clock schedule.
CANCEL_NEW does the opposite: the new run is cancelled outright the moment it collides. If the 3:00 run is still going at 4:00, the 4:00 run never happens. This is right when a fresh run would be redundant — a “sync latest state” job where a run already in progress will pick up the newest data anyway, so a second one is pure waste. The cost is silent skips: runs vanish from the schedule, and you need to be certain that’s harmless.

There’s no universal default worth memorizing, because the choice is the semantics of your pipeline. Ask one question: if a run gets skipped, does data go unprocessed? Yes → ENQUEUE, every run matters. No, a later run subsumes it → CANCEL_NEW, drop the redundant one. Getting this backwards is a real incident — CANCEL_NEW on a load that needed every window silently loses data, and ENQUEUE on a “latest state” sync builds a pointless backlog that runs stale work for hours.

Which knob for which problem

Five mechanisms is a lot after one pool, so here’s the decision guide, framed against the bookshop ETL fan-out that’s run through the whole series. First the whole set on one screen:

Mechanism	Counts	Where it lives	Airflow analogue	Reach for it when
Tag concurrency limit	Task runs, by tag	API / global	`pool`	The contended resource is a task
Global limit + `concurrency()`	Occupants of a code block	Inline in code	none	Contention is inside a helper / non-task code
`rate_limit()` (decaying GCL)	Entries per second	Inline in code	none	The system meters by time, not count
Work-pool / work-queue cap	Whole flow runs	Infrastructure	parallelism config	Capping runs hitting your infra
Deployment limit + collision strategy	Runs of one deployment	Deployment	`max_active_runs`	Stopping a deployment lapping itself

And the same choices as prose, since the table can only gesture at the why:

Protecting a shared resource that lines up with a task — the warehouse, an API, “no more than 5 of these at once, everywhere.” Reach for a tag-based task concurrency limit. It’s the direct port of the Airflow pool, it’s declarative, and it’s the right first instinct. load_file writes the warehouse → tags=["warehouse"] + prefect concurrency-limit create warehouse 5.
Protecting a resource that doesn’t line up with a task — a guarded call inside a helper, a bulk write worth extra weight, code outside any flow. Reach for a global concurrency limit with the concurrency() block, scoped to exactly the lines that contend. It’s the surgical version, and it’s the only one that reaches inside a task.
Protecting something that meters by time, not count — a partner API rated per minute, anything that 429s on rate. Reach for rate_limit() with a decaying global limit. Concurrency capping is the wrong shape here and will still get you throttled.
Limiting how many whole runs hit your infrastructure — stopping a runaway schedule from launching a hundred pods, or reserving cluster headroom for urgent work. Reach for the work-pool cap (total infrastructure ceiling) and, when one pool serves mixed urgency, work-queue caps (dividing that ceiling). These count flow runs, not tasks, and they’re a safety rail, not routing.
Stopping one deployment from overlapping itself — the non-idempotent incremental load, the sync that mustn’t double-run. Reach for the deployment-level limit with a deliberate collision strategy: ENQUEUE when every run’s data matters, CANCEL_NEW when a running instance already covers the new one.

The useful way to hold all five: they answer different questions Airflow’s single pool answered by accident. “How many tasks touch this resource?” is the tag limit. “How many at once inside this exact code?” is the block. “How many per second?” is the rate limit. “How many runs on my infrastructure?” is the pool cap. “Can this deployment overlap itself, and if not, wait or drop?” is the deployment limit. A serious pipeline usually wants more than one — a warehouse tag limit and a work-pool cap and an ENQUEUE deployment limit on the incremental load are three different ceilings guarding three different failure modes, and none substitutes for the others.

Final thoughts

Airflow gave you one pool and made you clever about squeezing every concurrency problem through it. Prefect gives you a matched set of instruments and makes you name the problem before you pick one — which feels like more to learn, and is, right up until you hit the case the single pool never fit: throttling a rate-metered API, guarding a resource that isn’t a task, or forbidding a deployment from lapping itself. Then the fragmentation reads as precision rather than sprawl. The through-line is that concurrency control is part of correctness, not an afterthought you bolt on when something falls over. The .map() that fans out to forty is only as good as the ceiling standing behind it; retrying faster than a source recovers, or running ten copies of a non-idempotent flow, isn’t reliability — it’s the same failure, louder. Pick the knob that fits the resource, and the fan-out you got for free stays free and safe.

Next: Variables, Settings, and Profiles

Tag-based limits: the Airflow pool, ported

Global concurrency limits and the concurrency() block