Sorry, I encountered an error while processing your request. Please try again.

Great question. In Prefect 3.x, beyond a deployment’s

concurrency_limit

, you’ve got a few options depending on what you want to control: - Work pool concurrency limit (caps all runs fetched by a pool) - Good for capping total concurrent flow runs across many deployments that share a pool - CLI:

prefect work-pool set-concurrency-limit <pool-name> <N>
    prefect work-pool clear-concurrency-limit <pool-name>

- Docs: Work pools and queues - Work queue concurrency limit (caps runs from a specific queue) - Good for tiering/priorities within a pool - CLI:

prefect work-queue set-concurrency-limit <queue-name> <N>
    prefect work-queue clear-concurrency-limit <queue-name>

- Docs: Work pools and queues - Task-level tag-based concurrency limits (caps concurrent task runs by tag) - Add a tag to tasks, then set a limit for that tag - Code + CLI:

python
    from prefect import task

    @task(tags=["db"])
    def query_db(): ...

prefect concurrency-limit create db --concurrency-limit 5
    prefect concurrency-limit ls

- Docs: Concurrency (concepts) - Global concurrency limits (named, reusable; can also act as rate limits) - Create a named limit and explicitly acquire a slot in code where you want to gate concurrency (flow or task) - Create/update via CLI:

prefect global-concurrency-limit create my-limit --limit 5
    prefect global-concurrency-limit update my-limit --limit 10

- Acquire in code:

python
    from prefect.concurrency.sync import concurrency  # or asyncio.concurrency

    @flow
    def my_flow():
        with concurrency("my-limit", strict=True):
            # Only N flows/tasks across your workspace enter here at once
            ...

- For API rate limiting, use

rate_limit("name", occupy=1)

with a decay set on the limit. - Note: Global limits are not auto-bound to deployments; you opt-in by acquiring them in code. - Docs: Concurrency (concepts) - Intra-flow parallelism caps (limit task parallelism within a single run) - Use a task runner with

max_workers

to keep per-run parallelism in check:

python
    from prefect import flow
    from prefect.task_runners import ThreadPoolTaskRunner

    @flow(task_runner=ThreadPoolTaskRunner(max_workers=4))
    def my_flow(): ...

- Docs: Concurrency (concepts) Quick CLI references (verified): -

prefect work-pool --help

(has

set-concurrency-limit

,

clear-concurrency-limit

) -

prefect work-queue --help

(has

set-concurrency-limit

,

clear-concurrency-limit

) -

prefect concurrency-limit --help

(manage tag-based limits) -

prefect global-concurrency-limit --help

(manage named global limits) If you share a bit more about what you want to limit: - total flow runs across a pool or queue, - concurrency of specific tasks (e.g., DB access), - or a global cross-flow throttle/rate limit, I can suggest the cleanest setup and example code.

Short answer: Not with tag-based limits. In Prefect 3.x, task tag concurrency limits always enqueue/wait (tasks enter an AwaitingConcurrencySlot state and retry) — there isn’t a “cancel new” collision strategy at the task level. That option exists only for deployment-level concurrency (flow runs), not tasks. Two practical ways to get “cancel-new” behavior for tasks: - Use a named Global Concurrency Limit with the code-level context manager and fail fast - Create a limit:

prefect global-concurrency-limit create db-proc --limit 5

- Acquire it in your task and short-circuit if no slot is available:

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

    @task(retries=0)
    def run_db_proc(...):
        try:
            # Try to acquire immediately; if full, raise and short-circuit
            with concurrency("db-proc", occupy=1, strict=True, timeout_seconds=0):
                # do the work here
                ...
        except Exception:
            # No capacity -> treat as "cancel new"
            # Option A: return early (no-op)
            return
            # Option B: raise to mark the task as failed (retries=0 avoids requeue)
            # raise RuntimeError("No capacity available for db-proc")

- This gives you cancel-new semantics at the code boundary instead of queuing. - Docs: Concurrency (concepts) - If cancelling entire flow runs is acceptable, use deployment-level concurrency with cancel-new - This cancels new flow runs when the limit is hit (not just tasks):

my_flow.to_deployment(
        name="my-deployment",
        concurrency_limit={
          "limit": 5,
          "collision_strategy": "cancel_new"
        }
    )

- Docs: Deployments If you’d like, share how you want “cancel” to manifest (skip/no-op vs. fail-fast) and I can tailor the snippet to your exact behavior.