Great question. In Prefect 3.x there isn’t a single “fail fast” switch for concurrent tasks. The typical pattern is to drive your orchestration with futures, consume them with

as_completed

, and raise as soon as one fails. That will fail the flow immediately and stop waiting on the rest; the task runner will shut down and attempt to cancel pending work. Here’s a minimal pattern you can drop in:

from prefect import flow, task
from prefect.futures import as_completed
from prefect.task_runners import ThreadPoolTaskRunner

@task
def risky(i: int) -> int:
    # replace with your logic
    if i == 5:
        raise ValueError("boom")
    return i * 2

@flow(task_runner=ThreadPoolTaskRunner(max_workers=8))
def fail_fast_flow(items: list[int]):
    futures = [risky.submit(i) for i in items]

    results = []
    for fut in as_completed(futures):
        try:
            results.append(fut.result(raise_on_failure=True))
        except Exception as exc:
            # Fail fast: propagate the first error to fail the flow
            raise
    return results

Notes: -

as_completed

yields futures as they finish; calling

result(raise_on_failure=True)

will raise immediately on the first failure, which fails the flow and stops awaiting others. - Prefect’s

PrefectFuture

does not expose a public

.cancel()

API; cancellation is handled by the task runner during shutdown. Tasks already running may not stop instantly, but pending work won’t be awaited and the runner will best-effort cancel. If you need “harder” cooperative cancellation (e.g., long-running tasks should stop quickly when any peer fails), pass a flag/event into your tasks and have them check it:

from prefect import flow, task
from prefect.futures import as_completed
from prefect.task_runners import ThreadPoolTaskRunner
import threading
import time

@task
def work(i: int, stop: bool = False, stop_flag: threading.Event | None = None) -> int:
    for _ in range(100):  # simulate chunked work
        if stop_flag and stop_flag.is_set():
            # exit early cooperatively
            return -1
        time.sleep(0.05)
        if i == 5:  # trigger a failure for demo
            raise RuntimeError("boom")
    return i * 2

@flow(task_runner=ThreadPoolTaskRunner(max_workers=8))
def fail_fast_flow(items: list[int]):
    stop_flag = threading.Event()
    futures = [work.submit(i, stop_flag=stop_flag) for i in items]

    results = []
    for fut in as_completed(futures):
        try:
            results.append(fut.result(raise_on_failure=True))
        except Exception:
            stop_flag.set()  # signal others to stop
            raise  # fail the flow immediately
    return results

Alternative APIs you can use depending on your needs: -

prefect.futures.wait(futures, timeout=...)

if you prefer waiting with a timeout and then inspecting

done

vs

not_done

. - Raising a standard exception from the flow (or

prefect.exceptions.Abort

) will end the flow immediately; this is the normal way to “fail fast.” Quick clarifier so I can tailor this precisely: - Are you running many tasks concurrently inside a single flow (using a task runner), or do you mean multiple Prefect workers pulling deployments from a work pool? - Which task runner are you using (threads, processes, or distributed)?

Got it — you’re orchestrating many child deployments from a single flow (they show up as subflows/tasks in the UI), and you want fail‑fast: if one child run fails, cancel the rest and fail the orchestrator. In Prefect 3.x the cleanest way to do this is: - Launch each deployment with

run_deployment(..., timeout=0)

to start it immediately and get its FlowRun ID without blocking. - Poll all of those flow runs concurrently. - On the first failure/crash, programmatically cancel the remaining runs by setting their state to

Cancelling

, then raise to fail the orchestrator. Here’s a ready-to-use async pattern:

import asyncio
from prefect import flow
from prefect.deployments import run_deployment
from prefect.client.orchestration import get_client
from prefect.states import Cancelling

async def wait_for_final(flow_run_id, poll_interval=5):
    async with get_client() as client:
        while True:
            fr = await client.read_flow_run(flow_run_id)
            if fr.state and fr.state.is_final():
                return fr
            await asyncio.sleep(poll_interval)

@flow
async def orchestrate_fail_fast(deployment_names: list[str]):
    # 1) Fire off all child runs without blocking
    flow_runs = await asyncio.gather(
        *[run_deployment(name, timeout=0) for name in deployment_names]
    )
    id_to_name = {fr.id: name for fr, name in zip(flow_runs, deployment_names)}

    # 2) Poll all runs concurrently
    poll_tasks = {asyncio.create_task(wait_for_final(fr.id)): fr.id for fr in flow_runs}

    first_bad = None
    completed = []

    try:
        for task in asyncio.as_completed(poll_tasks):
            fr_final = await task
            completed.append(fr_final)

            if fr_final.state.is_failed() or fr_final.state.is_crashed():
                first_bad = fr_final
                break
    finally:
        # 3) Fail fast: cancel all unfinished runs, then raise
        if first_bad:
            unfinished_ids = [fid for t, fid in poll_tasks.items() if not t.done()]
            async with get_client() as client:
                await asyncio.gather(
                    *[
                        client.set_flow_run_state(
                            flow_run_id=fid,
                            state=Cancelling(message="Fail-fast: a peer run failed"),
                        )
                        for fid in unfinished_ids
                    ],
                    return_exceptions=True,
                )
            raise RuntimeError(
                f"Fail-fast: {id_to_name[first_bad.id]} failed/crashed: {first_bad.state.message}"
            )

    return completed

Why this works: -

timeout=0

returns immediately so you have FlowRun IDs to manage. - Polling lets you detect the first failure quickly. - Setting

Cancelling

tells the worker (your Cloud Run work pool) to terminate those jobs. Once they exit, they’ll transition to

Cancelled

. Notes and options: - If you prefer not to write your own poller, you can

await run_deployment(..., timeout=N)

for each and use

asyncio.gather

. However, for real fail‑fast cancellation you still need the FlowRun IDs early, so the

timeout=0

pattern is best. - Cancellation is cooperative with the infrastructure. With a running worker and deployments launched via your work pool,

Cancelling

will trigger best‑effort termination of the Cloud Run jobs. - If you’ve set

as_subflow=False

on

run_deployment

, the logic still works; they just won’t be linked as subflows in the UI. If you share how you’re assembling the deployment names/params, I can tailor the snippet to your exact structure (and provide a sync-flow version if needed).

Great question. For orchestrating many downstream deployments concurrently in a single flow, Prefect 3.x recommends leaning on its async-friendly orchestration patterns rather than Python’s

concurrent.futures

. In short: use

run_deployment

concurrently (it’s async-compatible), monitor child runs, and on the first failure cancel the rest and fail the orchestrator. Best-practice pattern (fail-fast + cooperative cancellation) - Make your orchestrator flow async to use true I/O concurrency. - Launch deployments with

run_deployment(..., timeout=0)

to return immediately with FlowRun IDs. - Poll all child runs concurrently for terminal state. - On the first failure/crash, set the remaining runs to

Cancelling

and raise to fail the orchestrator. Example you can drop in:

import asyncio
from typing import Iterable
from prefect import flow
from prefect.deployments import run_deployment
from prefect.client.orchestration import get_client
from prefect.states import Cancelling

async def wait_for_final(flow_run_id, poll_interval=5):
    async with get_client() as client:
        while True:
            fr = await client.read_flow_run(flow_run_id)
            if fr.state and fr.state.is_final():
                return fr
            await asyncio.sleep(poll_interval)

@flow
async def orchestrate_fail_fast(deployments: Iterable[str], *, max_launch_concurrency: int = 20):
    # 1) Launch all child runs with bounded concurrency
    sem = asyncio.Semaphore(max_launch_concurrency)
    async def launch_one(name: str):
        async with sem:
            # timeout=0 returns immediately with a FlowRun object
            # idempotency_key helps avoid duplicates on orchestrator retries
            return await run_deployment(name, timeout=0, idempotency_key=name)

    flow_runs = await asyncio.gather(*[launch_one(d) for d in deployments])
    id_to_name = {fr.id: name for fr, name in zip(flow_runs, deployments)}

    # 2) Poll all launched runs concurrently
    poll_tasks = {asyncio.create_task(wait_for_final(fr.id)): fr.id for fr in flow_runs}
    first_bad = None
    completed = []

    try:
        for task in asyncio.as_completed(poll_tasks):
            fr_final = await task
            completed.append(fr_final)

            if fr_final.state.is_failed() or fr_final.state.is_crashed():
                first_bad = fr_final
                break
    finally:
        # 3) Fail fast: cancel unfinished runs and raise
        if first_bad:
            unfinished_ids = [fid for t, fid in poll_tasks.items() if not t.done()]
            async with get_client() as client:
                await asyncio.gather(
                    *[
                        client.set_flow_run_state(
                            flow_run_id=fid,
                            state=Cancelling(message="Fail-fast: a peer run failed"),
                        )
                        for fid in unfinished_ids
                    ],
                    return_exceptions=True,
                )
            raise RuntimeError(
                f"Fail-fast: {id_to_name[first_bad.id]} failed/crashed: {first_bad.state.message}"
            )

    return completed

Why not Python futures here? -

run_deployment

is async-compatible and already does the right orchestration work (links subflows, handles parameters, resumes polling). Using

asyncio

keeps it simple and avoids wrapping in extra threads/tasks that complicate state handling. Additional best practices - Keep

run_deployment

calls in the flow body (don’t wrap in a @task) so the UI shows subflows correctly and you avoid unnecessary task runs. - Cap fan-out: use a semaphore (as above) and/or Prefect concurrency limits at the deployment/work-pool/global level to protect infra. See Concurrency limits. - Use

idempotency_key

on

run_deployment

to avoid duplicate launches if the orchestrator itself retries.

- Tag runs to correlate an orchestration wave (e.g.,

tags=["orchestrator:batch-2024-11-24"]

). - Timeouts per child: either pass

timeout=N

to

run_deployment

(it will stop polling after N seconds) and treat “not final yet” as a failure, or enforce your own timeout in the poller and then set

Cancelling

. - Be considerate with polling: adjust

poll_interval

to reduce API load if you fan out a lot. - Cancellation semantics: setting child runs to

Cancelling

is the correct way to stop them via the worker/infrastructure; they will transition to

Cancelled

once terminated by the work pool’s worker (Cloud Run in your case). - Observability: log child FlowRun IDs and names; optionally emit artifacts summarizing successes/failures for quick triage. Useful references - API: run_deployment - Concepts: Deployments, States, Work pools - Cancellation guide: Cancelling runs If you share how you build the deployment names/parameters (and your desired concurrency cap), I can tailor the snippet to match your structure and add per-deployment parameterization.

Awesome—here’s a practical, Prefect‑specific way to test your async orchestrator flow that launches downstream deployments and fails fast. Core testing best practices for Prefect 3.x - Isolate Prefect state: wrap tests in

prefect_test_harness()

so runs, states, and settings don’t leak across tests. - Use async tests: mark tests with

@pytest.mark.anyio

(or

pytest-asyncio

) and

await

your flow directly. - Make it fast: parameterize your flow so you can pass a tiny

poll_interval

in tests, and use

mock_anyio_sleep

to skip real sleeps. - Don’t hit real infra: monkeypatch

run_deployment

(and/or your polling function) to return fabricated

FlowRun

objects with the states you want to simulate. - Assert cancellation: spy on

client.set_flow_run_state

and assert you set remaining runs to

Cancelling

on first failure. - Capture logs and last run: use

caplog

and

get_most_recent_flow_run()

for assertions. Recommended pytest setup - conftest.py

import pytest
from prefect.testing.utilities import prefect_test_harness

@pytest.fixture(scope="session", autouse=True)
def _prefect_harness():
    with prefect_test_harness():
        yield

- Optional: add Prefect loggers to caplog for log assertions

# In your test use:
from prefect.testing.fixtures import add_prefect_loggers_to_caplog

def test_logs(caplog):
    add_prefect_loggers_to_caplog(caplog)
    # run your flow, then assert in caplog.records

Unit-style test: simulate success

import pytest
import asyncio
from uuid import uuid4
from prefect.client.schemas.objects import FlowRun, State
from prefect.client.schemas.objects import StateType

# Assume your orchestrator is `orchestrate_fail_fast` and it imports run_deployment and wait_for_final
# from my_orchestrator import orchestrate_fail_fast

@pytest.mark.anyio
async def test_orchestrator_all_success(monkeypatch):
    # Fake flow runs
    frs = [FlowRun(id=uuid4()), FlowRun(id=uuid4()), FlowRun(id=uuid4())]

    async def fake_run_deployment(name, timeout=0, **kwargs):
        # Return immediately with a FlowRun (what timeout=0 does)
        return frs.pop(0)

    async def fake_wait_for_final(flow_run_id, poll_interval=5):
        # Pretend each child completes successfully
        return FlowRun(
            id=flow_run_id,
            state=State(type=StateType.COMPLETED, message="done"),
        )

    monkeypatch.setattr("my_orchestrator.run_deployment", fake_run_deployment)
    monkeypatch.setattr("my_orchestrator.wait_for_final", fake_wait_for_final)

    result = await orchestrate_fail_fast(["flow/dep-a", "flow/dep-b", "flow/dep-c"])
    assert all(fr.state.is_completed() for fr in result)

Unit-style test: fail-fast cancels others ``` import pytest from uuid import uuid4 from prefect.client.schemas.objects import FlowRun, State, StateType from prefect.client.orchestration import get_client from prefect.states import Cancelling @pytest.mark.anyio async def test_orchestrator_fail_fast_cancels_others(monkeypatch): ids = [uuid4(), uuid4(), uuid4()] returns = [FlowRun(id=ids[0]), FlowRun(id=ids[1]), FlowRun(id=ids[2])] async def fake_run_deployment(name, timeout=0, **kwargs): return returns.pop(0) # First finishes failed, others would still be running async def fake_wait_for_final(flow_run_id, poll_interval=5): if flow_run_id == ids[0]: return FlowRun(id=flow_run_id, state=State(type=StateType.FAILED, message="boom")) else: # not reached — orchestrator should cancel remaining before awaiting these return FlowRun(id=flow_run_id, state=State(type=StateType.RUNNING)) calls = [] class SpyClient: async def __aenter__(self): return self async def __aexit__(self, *exc): return False async def set_flow_run_state(self, flow_run_id, state): calls.append((flow_run_id, state)) return None

async def read_flow_run(self, flow_run_id): # not used in this unit stub return FlowRun(id=flow_run_id, state=State(type=StateType.RUNNING)) monkeypatch.setattr("my_orchestrator.run_deployment", fake_run_deployment) monkeypatch.setattr("my_orchestrator.wait_for_final", fake_wait_for_final) monkeypatch.setattr("my_orchestrator.get_client", lambda: SpyClient()) with pytest.raises(RuntimeError): await orchestrate_fail_fast(["flow/dep-a", "flow/dep-b", "flow/dep-c"]) # Assert we attempted to cancel the remaining two runs cancelled_ids = {fid for (fid, st) in calls if isinstance(st, Cancelling)} assert cancelled_ids == {ids[1], ids[2]}

Speeding up tests
- Allow your orchestrator to accept `poll_interval` and pass a tiny value in tests (e.g., 0.01).
- Use `mock_anyio_sleep` to skip real sleeps:

from prefect.testing.fixtures import mock_anyio_sleep @pytest.mark.anyio async def test_fast(monkeypatch, mock_anyio_sleep): # any anyio.sleep(...) in your code will be fast-forwarded ...

Integration-ish option (still fast)
- Keep `run_deployment(timeout=0)` patched to avoid spinning real jobs.
- Let your poller call Prefect’s client, and drive state transitions by monkeypatching `read_flow_run` to return RUNNING then FAILED for one ID and RUNNING for others. Verify `set_flow_run_state(..., Cancelling)` gets called for the remaining IDs.
- Use `get_most_recent_flow_run()` if you need to inspect the latest run metadata:

from prefect.testing.utilities import get_most_recent_flow_run fr = await get_most_recent_flow_run() ``` Additional tips - Prefer testing your flow’s orchestration logic in isolation (unit-style) and keep E2E tests minimal; avoid launching real Cloud Run jobs in CI. - Keep cancellation logic in one place (e.g., a helper that sets

Cancelling

for a list of IDs) so you can assert its behavior directly. - Make your orchestrator accept knobs for

max_launch_concurrency

,

poll_interval

, and maybe an injected “client factory” so you can pass a spy client in tests. References - run_deployment API: https://docs.prefect.io/latest/api-ref/deployments/#prefect.deployments.flow_runs.run_deployment - Cancelling runs: https://docs.prefect.io/latest/guides/cancellations/ If you share your orchestrator module name and function signature, I can tailor the test snippets exactly to your code (including the monkeypatch paths).

Great question—timeouts get subtle when you’re awaiting lots of downstream deployments. Here’s how to think about them in Prefect 3.x, the limits to be aware of, and best practices to avoid traps. Where timeouts apply (and what they actually do) 1) run_deployment timeout (client-side polling only) - What it does: Controls how long your orchestrator waits/polls for the child flow to reach a terminal state. - Semantics: - timeout=None (default): poll forever - timeout=0: return immediately (no polling) - timeout>0: poll up to N seconds then return whatever state it’s in - Critical: This does NOT stop the remote run. If timeout expires, the child run continues on the worker. - Tip: Always set a finite timeout in orchestration code so your orchestrator doesn’t hang forever. 2) Flow/task execution timeouts (server-side state, enforced by the engine) - @flow(timeout_seconds=...) and @task(timeout_seconds=...) - What they do: Bound the execution time of the flow/task itself. On timeout, Prefect raises a TimeoutError, transitions to Failed, and the run ends. - Precision: Async code cancels at await points; long CPU-bound or C extensions may overrun slightly in sync code. - No hard max from Prefect; use reasonable values for your case. 3) Infrastructure/work-pool timeouts (hard runtime limits) - Set by your compute platform (e.g., Cloud Run’s max duration, pod activeDeadlineSeconds for K8s, etc.). - For Cloud Run via work pool: pass job_variables with a runtime limit when launching:

await run_deployment(
      "flow/dep",
      timeout=0,
      job_variables={"timeout_seconds": 3600}  # Cloud Run max is typically 1h
  )

- These are the only true “hard” caps that will stop the container/process regardless of app-level timeouts. 4) API/client request timeouts (per HTTP call) - PREFECT_API_REQUEST_TIMEOUT (default 60s) applies to each API call (e.g., polling via read_flow_run). - If your API is slow and you have many concurrent polls, you can hit these timeouts. Tune if needed. Hidden traps to avoid - Polling forever: run_deployment(timeout=None) for dozens/hundreds of runs can hang your orchestrator indefinitely if one never finishes. - Confusing polling timeout with execution timeout: run_deployment(timeout=300) doesn’t cap run duration—just the polling loop in your flow. - Over-polling: Many concurrent runs with a small poll_interval can cause high API QPS and hit request timeouts or pool limits. - Connection pool exhaustion: Spawning many concurrent polling loops without reusing a client can exhaust HTTP connection limits. - Mismatched timeouts: Task timeout and flow timeout are independent; retries reset the timeout clock each attempt, increasing total wall time. Best practices for async orchestration with many downstream deployments - Always set a finite polling timeout:

await run_deployment(name, timeout=600, poll_interval=1.0)

If still Running after 10 minutes, treat as a failure or cancel it. - Bound concurrency and reuse the client: - Use a semaphore to limit how many child runs you launch/poll at once. - Reuse a Prefect client with connection limits when you do your own polling. - Example pattern you can adapt:

import asyncio
  from httpx import Limits
  from prefect.client.orchestration import get_client

  limits = Limits(max_connections=20, max_keepalive_connections=10)

  async with get_client(httpx_settings={"limits": limits}) as client:
      sem = asyncio.Semaphore(20)

      async def launch_and_wait(dep):
          async with sem:
              fr = await run_deployment(dep, timeout=300, poll_interval=1.0)
              return fr

      results = await asyncio.gather(*(launch_and_wait(d) for d in deps))

- Use flow/task timeouts to cap actual work: - Put a guardrail at the flow level:

@flow(timeout_seconds=3600)  # hour cap for the orchestrator itself
    async def orchestrator(): ...

- Put caps on long-running tasks (if any) inside the child flows:

@task(timeout_seconds=900)  # 15 min cap
    def heavy_op(...): ...

- Enforce hard runtime caps at the infrastructure layer: - For Cloud Run via work pool: set job_variables with a max runtime. This guarantees termination regardless of app logic. - Keep infra limits ≥ Prefect flow/task timeouts so the engine can fail gracefully first. - Fail fast and cancel peers: - If a child fails or exceeds your orchestration timeout, set other child runs to Cancelling and raise in the orchestrator so you don’t waste compute. - Tune polling responsibly: - poll_interval=0.5–2.0s is usually plenty; going lower increases API load. - For very large fan-out, consider backoff (poll slower over time) or tiered checks. - Watch API timeouts under load: - If you see request timeouts when polling many runs, increase PREFECT_API_REQUEST_TIMEOUT modestly or reduce concurrency/poll rate. - Idempotency and retries: - Use idempotency_key in run_deployment so orchestrator retries don’t double-launch runs. - Remember retries reset task/flow timeout windows; account for worst-case total runtime when setting infra caps. What’s the max limit? - run_deployment timeout: no hard max in Prefect; it’s a float (seconds). Practically, use minutes-to-hours. - @flow/@task timeout_seconds: no hard max in Prefect; use values aligned to business SLAs. - PREFECT_API_REQUEST_TIMEOUT: configurable float; default 60s. - Cloud Run max runtime: commonly 3600s (1 hour). Respect your platform’s documented limits. If you share: - Your target SLAs, - Typical/peak fan-out size, - Current poll_interval and timeouts, I can suggest concrete numbers and a tuned template for your orchestrator that balances speed, stability, and cost on Cloud Run.

Short answer: yes—what you’re seeing is almost certainly the startup “handshake” and early API calls timing out under load. With ~60 runs launched at once, you can exhaust the HTTP connection pool and hit per-request timeouts before the runs even start executing user code. What’s happening under the hood - Each run needs several API calls at startup (read deployment/flow, propose Pending, resolve blocks/variables, etc.). - Prefect’s client uses httpx with a default connection pool of roughly 16 max connections and per-request timeouts of 60s. - If you launch 60+ runs simultaneously, you can queue hundreds of API calls in a small window. Many end up waiting for a connection and then fail with TimeoutError (PoolTimeout/ReadTimeout) before work starts. How to prevent it (prioritized) 1) Spread out load across multiple workers - Instead of one worker handling 60 runs, start 2–4 workers each with a limit, e.g.: - 3 workers with --limit 20 - This multiplies available connections and smooths the “thundering herd.” - If you’re using a single Cloud Run work pool, you can run multiple worker processes/instances attached to that pool. 2) Cap per-worker concurrency - Use the worker’s concurrency limit to avoid spikes: - prefect worker start -p <pool> --limit 20 - This prevents one worker from attempting to submit 60 runs at once. 3) Increase API request timeout - Give API calls more time during spikes: - export PREFECT_API_REQUEST_TIMEOUT=120 - This helps absorb transient slowdowns but won’t fix pool exhaustion by itself. 4) Add more jitter/spacing - Stagger requests so they don’t all pile up at the same instant: - export PREFECT_CLIENT_RETRY_JITTER_FACTOR=0.5 - Slightly increase query intervals: - export PREFECT_WORKER_QUERY_SECONDS=15 - export PREFECT_WORKER_PREFETCH_SECONDS=5 5) Tune the orchestrator - If your top-level flow is launching all deployments at once, bound the launch rate: - Use an asyncio.Semaphore to limit concurrent run_deployment calls. - Prefer run_deployment(..., timeout=0) to fire-and-forget and poll later, reducing long-lived blocked calls. 6) Be mindful of block/variable resolution - If deployments reference many blocks/variables, startup can involve extra API calls. Consider reducing per-run block lookups or caching shared inputs in parameters where feasible. How to confirm this diagnosis - Check logs for httpx.PoolTimeout, httpx.ReadTimeout, or “Failed to update state of flow run …” right at the top of the run logs. - If you see errors before any user task logs, it’s almost certainly startup API pressure/timeouts. - You can also temporarily boost PREFECT_API_REQUEST_TIMEOUT to 120 and see if the failures drop—good signal that timeouts are the bottleneck. If helpful, I can suggest a concrete worker layout given your typical fan-out size and Cloud Run limits, and provide a sample orchestrator snippet that rate-limits submissions to keep startup smooth.

That warning means the background process that keeps your concurrency “slot” reserved failed to renew it. Your code continues running, but Prefect can no longer guarantee the limit—you may exceed your global concurrency. What’s happening - When you enter a concurrency limit (e.g., async with concurrency("my-limit", occupy=1)), Prefect: - Acquires a lease (reserves a slot) for a TTL (default ~300s) - Starts a background renewal loop that extends the lease every ~75% of the TTL - Releases the lease on exit - If the renewal loop can’t reach the API (timeouts, server hiccup, worker/network issues) or the lease has already expired/revoked, renewal fails. Prefect logs: "Concurrency lease renewal failed - slots are no longer reserved. Execution will continue, but concurrency limits may be exceeded." Common causes - API request timeouts or connection pool exhaustion under heavy load (lots of runs at once) - Temporary Prefect API unavailability or high latency - Lease TTL too short for long-running operations, so it expires before renewal succeeds - Single worker under heavy load doing many concurrent leases/renewals How to prevent it - Use strict mode for critical sections - This fails fast if renewal fails rather than continuing without a slot.

from prefect.concurrency.asyncio import concurrency

  async with concurrency("my-limit", occupy=1, strict=True, lease_duration=600):
      await do_work()

- For sync code:

from prefect.concurrency.sync import concurrency
  with concurrency("my-limit", occupy=1, strict=True, lease_duration=600):
      do_work()

- Increase the lease_duration - Fewer renewals = fewer chances to fail. Typical: 600–1800 seconds for long operations. - Trade-off: if the worker crashes, slots can remain held longer until TTL expires. - Reduce API timeout pressure and spread load - Run multiple workers with per-worker --limit to avoid spikes (e.g., 3 workers × limit 20 instead of 1 × 60). - If you saw earlier startup timeouts, this helps both problems. - Give API calls more headroom and jitter - Increase per-request timeout: - export PREFECT_API_REQUEST_TIMEOUT=120 - Increase retry jitter to stagger retries: - export PREFECT_CLIENT_RETRY_JITTER_FACTOR=0.5 - Prefer persistent lease storage (self-hosted) - If you’re self-hosting, set: - PREFECT_SERVER_CONCURRENCY_LEASE_STORAGE=prefect.server.concurrency.lease_storage.filesystem - Survives brief server restarts better than in-memory. - Monitor and alert - Bump logging to DEBUG on workers to observe lease renewals and failures. - If you see frequent warnings even at low load, check worker→API latency, DNS, and any proxies. Quick checklist - Is this happening during high fan-out? Start more workers and cap per-worker concurrency. - Are operations long-running? Increase lease_duration (e.g., 600–1800s). - Is it critical to respect the limit? Use strict=True to fail fast on renewal failure. - Seeing timeouts elsewhere too? Raise PREFECT_API_REQUEST_TIMEOUT and jitter factor. If you share how you’re using the concurrency context (name, typical duration, fan-out), I can suggest concrete lease_duration values and a worker layout tailored to your workload.

You’re running into a “thundering herd” cold-start: 100+ batches trigger at once, each tries to spin up a deployment, and your concurrency limit isn’t absorbing the burst. The error you’re seeing:

prefect.concurrency._asyncio.ConcurrencySlotAcquisitionError: Unable to acquire concurrency slots on ['xyz']

means the attempt to acquire your global concurrency slots failed (exhausted, timed out, or the limit doesn’t exist when using strict mode). When 100+ start together, acquisition requests collide and many fail immediately or after the acquisition timeout. To fix this, add backpressure across multiple layers. In practice, you want orchestration limits that shape the load before it reaches workers, plus per-deployment and in-flow controls. Recommended multi-layer solution 1) Shape the queue: Work pool and queue concurrency - Set concurrency on the work queue that these runs use (not just “10 workers”): - Work pool queue concurrency limits cap how many flow runs are picked up at once. - Use priority queues (critical/high/batch) if you have mixed importance. - Example (CLI or UI): - Create a work queue for this fanout and set concurrency_limit=10. This guarantees no more than 10 runs are pulled concurrently regardless of how many are scheduled. 2) Cap per-deployment concurrency - Each deployment can have its own limit so a single hot deployment doesn’t starve others: - During deploy, set a per-deployment concurrency limit and choose the collision strategy: - ENQUEUE: new runs wait instead of failing - CANCEL_NEW: new runs are cancelled if at capacity - This prevents a single deployment from allowing 100 active instances at once. 3) Use global concurrency limits inside flows for shared resources - For anything shared (DB pool, external API), gate with Prefect global concurrency: - async with concurrency("xyz", occupy=1, timeout_seconds=..., max_retries=..., strict=True/False) - If many runs compete for "xyz" at cold start: - Increase timeout_seconds and max_retries so acquirers wait instead of fail. - Consider strict=False if you prefer degraded service over hard failure. - Increase lease_duration (e.g., 600–1800s) to reduce renewal frequency under load. 4) Rate-limit the orchestrator’s fanout - Your top-level flow should not submit 100+ deployments at once. Bound it: - Use an asyncio.Semaphore to limit concurrent run_deployment calls (e.g., 10–20). - Use backoff when acquisition fails (AcquireConcurrencySlotTimeoutError or ConcurrencySlotAcquisitionError), then retry: - Exponential backoff with jitter (0.5–2s initial delay) smooths cold-start spikes. - Prefer run_deployment(timeout=0) to fire-and-forget and poll later to reduce long-held client calls. 5) Increase stability under load - Start multiple workers with per-worker --limit instead of one large worker. Example: 3 workers × limit 20 rather than 1 × 60. - Raise API call headroom if you saw timeouts earlier: - export PREFECT_API_REQUEST_TIMEOUT=120 - export PREFECT_CLIENT_RETRY_JITTER_FACTOR=0.5 Concrete patterns - Work queue with concurrency limit: - Set the specific queue used by these deployments to concurrency_limit=10 (via UI or API). This throttles pickup at the source. - Per-deployment limit with enqueue: - When deploying, add a concurrency limit with ENQUEUE so excess runs wait instead of failing. This absorbs bursts cleanly. - In-flow concurrency with backoff: ``` from prefect.concurrency.asyncio import concurrency, AcquireConcurrencySlotTimeoutError, ConcurrencySlotAcquisitionError import asyncio, random async def acquire_with_backpressure(limit_name: str, occupy: int = 1): base = 0.5 for attempt in range(6): try: async with concurrency( limit_name, occupy=occupy, timeout_seconds=10, # wait for a slot per attempt max_retries=2, # retry on 423 with server-provided Retry-After

strict=False # or True if exceeding is unacceptable ): return except (AcquireConcurrencySlotTimeoutError, ConcurrencySlotAcquisitionError): if attempt == 5: raise delay = min(base * (2 ** attempt) + random.uniform(0, 0.5), 30) await asyncio.sleep(delay)

- Orchestrator fanout with a semaphore:

import asyncio from prefect.deployments import run_deployment sem = asyncio.Semaphore(10) # cap concurrent submissions async def submit_one(name, params): async with sem: return await run_deployment(name, parameters=params, timeout=0) ``` Checklist to resolve your current crash - Ensure the concurrency limit “xyz” exists (strict=True will error if it doesn’t). - Increase timeout_seconds and max_retries in your concurrency(...) calls so instances can wait instead of immediately failing. - Set the work queue used by these runs to concurrency_limit=10 (this is separate from “10 workers”). - Add a per-deployment concurrency limit with ENQUEUE so overage waits instead of crashing. - Limit orchestrator fanout to 10–20 concurrent submissions and use backoff on acquisition failures. - Consider increasing lease_duration to 600–1800s to reduce renewal load during cold-start. If you can share: - Where you’ve set the “10 worker” limit (worker flag vs queue concurrency), - How you’re using concurrency(...) in code (strict/timeout), - The deployment(s) involved and which queue they use, I can provide exact CLI/API calls and code adjustments tailored to your setup.