How to reproduced deployment concurrency_limit CANCEL_NEW using global concurrency limits?

[tonal]

Due to error #19068, I would like to reproduce the deployment concurrency_limit using global concurrency limits.
How can this be done?

A construction that leads to an error:

await flow_tk.to_deployment(
  name='wp-tk-upd', job_variables=dict(env=job_override), schedules=[shedule],
  concurrency_limit=ConcurrencyLimitConfig(
    limit=1, collision_strategy=ConcurrencyLimitStrategy.CANCEL_NEW)
)

How can I replace it?

[br413]

@tonal — short answer: global concurrency limits (GCL) are not a drop-in replacement for deployment concurrency_limit with CANCEL_NEW. They solve different problems.

Mechanism	Scope	Behavior when limit is full
Deployment concurrency_limit	Per deployment's flow runs	CANCEL_NEW cancels the incoming run before it starts
Global concurrency limits (concurrency() / rate_limit())	Any Python block you wrap	Blocks until a slot is available (closer to ENQUEUE)

Issue #19068 is about lease renewal / flow-timeout interaction while a deployment slot is held for the entire flow duration — not that ConcurrencyLimitConfig is unsupported. If to_deployment(..., concurrency_limit=ConcurrencyLimitConfig(...)) is failing on your version, try defining the limit in prefect.yaml or via flow.deploy() instead of inline on to_deployment.

Keep CANCEL_NEW at the deployment layer (recommended if that's the semantics you need)

prefect.yaml:

deployments:
  - name: wp-tk-upd
    entrypoint: flows/tk.py:flow_tk
    work_pool:
      name: my-pool
    concurrency_limit: 1
    concurrency_options:
      collision_strategy: CANCEL_NEW

Or flow.deploy():

from prefect.client.schemas.objects import (
    ConcurrencyLimitConfig,
    ConcurrencyLimitStrategy,
)

await flow_tk.deploy(
    name="wp-tk-upd",
    work_pool_name="my-pool",
    job_variables={"env": job_override},
    schedules=[schedule],
    concurrency_limit=ConcurrencyLimitConfig(
        limit=1,
        collision_strategy=ConcurrencyLimitStrategy.CANCEL_NEW,
    ),
)

That preserves "reject new runs while one is active" without moving the limit into task code.

If #19068 is crashing long runs: mitigate lease renewal

For flows that can exceed the default lease window, the GCL docs recommend tolerating transient renewal failures on long operations:

from prefect.concurrency.sync import concurrency

with concurrency(
    "my-gate",
    occupy=1,
    strict=True,
    raise_on_lease_renewal_failure=False,
):
    ...

At the deployment level, you can also increase grace_period_seconds inside ConcurrencyLimitConfig so slow-starting infrastructure doesn't release slots prematurely (see deployment concurrency docs).

When GCL does help

Use GCL if you want to cap resource usage inside a flow (DB connections, API calls, etc.) rather than cap how many deployment runs start:

prefect gcl create wp-tk-upd-gate --limit 1

from prefect import flow
from prefect.concurrency.sync import concurrency


@flow
def flow_tk():
    with concurrency("wp-tk-upd-gate", occupy=1):
        # only one execution enters this block cluster-wide
        ...

Note: this waits for a slot; it does not cancel the triggering run the way CANCEL_NEW does.

Practical recommendation

1. If you need CANCEL_NEW semantics → keep deployment-level concurrency (yaml/deploy()), and tune grace_period_seconds / upgrade past the Concurrency lease renewal failed - slots are no longer reserved. Terminating execution to prevent over-allocation. #19068 fix (closed in 3.x).
2. If you need to protect a shared resource inside the flow → add a named GCL around that section.
3. Avoid stacking both a deployment-wide slot and a GCL gate for the same limit unless you really need layered control — it can double-bookkeep slots and worsen lease-timeout crashes.

References:

• Global concurrency limits concept
• How to apply global concurrency limits
• Deployment concurrency limiting

If you can share your Prefect version and whether the failure happens at deploy-time vs run-time, we can narrow this further.