Limiting DAG Fan-Out With Concurrency Groups

A fan-out has two widths and only one of them is usually configured. The first is how many tasks the orchestrator will run at once, which protects your workers; the second is how many of those tasks may touch a particular shared resource, which protects everybody else’s. Setting only the first is how a fan-out that is perfectly polite to its own memory limit takes down a national WFS endpoint at nine in the morning. Concurrency groups are the mechanism for the second, and they belong on the resource, not on the task.

When to Use This Pattern#

• The fan-out reads a shared external service — a WMS, a WFS, a geocoder, an object store with a request quota.
• Several flows read the same resource, so a per-flow limit cannot bound the total.
• A downstream database has a connection ceiling that is smaller than the fan-out width.
• The graph’s width is decided at runtime from a footprint, which means it can be much larger than anyone tested with.

Complete Working Example#

Two limits, declared separately, applied to the same task.

from __future__ import annotations

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


@task(retries=2, retry_delay_seconds=[15, 60], timeout_seconds=900, tags=["tile"])
def build_tile(tile: Tile, source: SourceSpec, dest: Path) -> Path:
    # Limit 1 — the upstream service. Global across every flow that names this slot.
    with concurrency(f"source:{source.host}", occupy=1, timeout_seconds=1800):
        window = fetch_window(source, tile)

    # Limit 2 — the database. Held for the write only, not for the whole task,
    # because a connection blocked on a warp is a connection nobody else can use.
    warped = warp_in_memory(window, tile)
    with concurrency("postgis-writers", occupy=1):
        write_footprint(tile, warped.stats)

    warped.save(dest)
    return dest


@task(retries=3, retry_delay_seconds=[5, 20, 60])
def geocode(address: str) -> Point:
    # A rate limit, not a concurrency limit: the constraint is requests per second,
    # not simultaneous requests, and holding a slot would not express it.
    rate_limit("nominatim", occupy=1)
    return call_geocoder(address)


@flow(name="tile-build", timeout_seconds=7200)
def tile_build(footprint_wkt: str, zoom: int, source: SourceSpec, dest: Path) -> int:
    tiles = plan(footprint_wkt, zoom)
    get_run_logger().info("fan-out width: %d tiles", len(tiles))
    states = build_tile.map(tiles, source=source, dest=dest, return_state=True)
    return sum(s.is_completed() for s in states)

The limits themselves are declared once, outside the flow, so they are shared by everything that names them:

# Six simultaneous reads of this host, across every flow in the deployment.
prefect gcl create "source:geodata.example.gov" --limit 6 --slot-decay-per-second 0.2

# The database has 40 connections; the pipeline may have 12 of them.
prefect gcl create postgis-writers --limit 12

# A per-tag limit as a backstop, so a runaway manifest cannot schedule 40 000 at once.
prefect concurrency-limit create tile 64

Where the slot is acquired matters as much as its size. In the example the source slot wraps only the fetch, and the database slot wraps only the write, with the expensive warp sitting outside both. Wrapping the whole task body in a single slot is the tempting simplification and it converts a concurrency limit into a throughput limit: with six slots and a ninety-second warp, the pipeline makes six requests every ninety seconds regardless of how fast the source can actually answer. Holding a slot for the shortest interval that preserves the guarantee is the difference between a limit that protects a service and one that throttles your own pipeline to its slowest stage.

Parameter & Option Reference#

Verification & Testing#

The property to test is that the limit is honoured under load, which needs a counter rather than an assertion on configuration.

import threading

from prefect.testing.utilities import prefect_test_harness


def test_source_slot_is_respected(monkeypatch) -> None:
    live, peak, lock = 0, 0, threading.Lock()

    def fake_fetch(source, tile):
        nonlocal live, peak
        with lock:
            live += 1
            peak = max(peak, live)
        time.sleep(0.05)
        with lock:
            live -= 1
        return object()

    monkeypatch.setattr("mymodule.fetch_window", fake_fetch)
    with prefect_test_harness():
        tile_build(SERVICE_AREA_WKT, zoom=12, source=SPEC, dest=Path("/tmp"))
    assert peak <= 6, f"source limit breached: {peak} concurrent reads"


def test_slot_is_not_held_across_the_warp(monkeypatch) -> None:
    held: list[float] = []
    monkeypatch.setattr("mymodule.warp_in_memory", lambda *a: time.sleep(0.2))
    ...
    assert max(held) < 0.1, "slot held for the whole task, not just the fetch"

In production the number to watch is not the limit but the wait for it. A queue that is always empty means the limit is not binding and something else is the bottleneck; a queue that grows without bound means the limit is below the pipeline’s required throughput and the run will not finish in its window. The healthy signal is a short, non-zero wait — the limit is doing work and the pipeline is still moving.

Common Pitfalls#

• One limit for both jobs. Worker capacity and upstream capacity are unrelated numbers; a single figure is wrong for at least one of them.
• Naming the slot after the flow. Two flows reading the same host then get one limit each, and the host gets both.
• Holding the slot for the whole task. The limit becomes a throughput cap set by the slowest stage rather than a bound on simultaneous requests.
• Setting the limit at the measured knee. Retries add load precisely when the service is degraded, so the limit must sit below the point where errors begin.
• Using a concurrency limit where a rate limit is meant. Six simultaneous requests and six requests per second are different constraints, and geocoders almost always mean the second.
• No timeout on slot acquisition. A jammed slot then turns a bounded fan-out into a flow that hangs until its own timeout, with nothing in the logs explaining why.

Frequently Asked Questions#

Related#

• DAG design principles for spatial ETL — fan-out is bounded twice, and why
• How to structure a DAG for raster processing — the stage the limit is applied to
• Circuit breakers for external WMS services — what to do once a service has failed
• Implementing retry logic for slow WFS endpoints — the retry policy that sits inside the limit
• Right-sizing workers for raster mosaic jobs — where the worker-side number comes from

← DAG Design Principles for Spatial ETL