Got it — you want to keep your per-store schedules, but ensure that at most two of these dbt-running deployments execute at the same time. Assuming Prefect 3.x: you cannot enforce a shared “max concurrent deployments” purely in prefect.yml. Per-deployment

concurrency_limit

is only for that single deployment. For a shared cap across many deployments, use one of the two patterns below. Option A (recommended): Global Concurrency Limit (GCL) shared across deployments - Create a single global limit (e.g., dbt-runs) with capacity 2. - In your flow code, acquire a slot before you run dbt. Every deployment will contend for the same limit, so only two will proceed at once while others wait; your schedules still trigger runs normally. 1) Create the limit

prefect gcl create dbt-runs --limit 2

2) Wrap your dbt call in the flow or task that executes dbt (sync example)

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

@task
def run_dbt(database, models, full_refresh, threads=3):
    # your dbt invocation here (CLI, adapter, etc.)
    ...

@flow
def clickstream_dbt_run(database: str, models: str, full_refresh: bool, store_timezone_offset_minutes: str):
    # Acquire a shared slot for the duration of the critical section
    with concurrency("dbt-runs", occupy=1, strict=True, lease_duration=7200):
        # Only two flows across all deployments can be here at once
        run_dbt.submit(database, models, full_refresh, threads=3)

Notes - Place the concurrency context as early as you want to limit the run. If you want to limit the entire flow, acquire at the top. If you only want to limit the dbt step, acquire only around dbt. - Schedule remains in effect — flows will start on schedule but will wait at the concurrency gate if needed. - If you prefer “cancel new if busy” behavior, you can implement it by adding a timeout and handling it:

from prefect.concurrency.sync import concurrency
from contextlib import ExitStack

with ExitStack() as stack:
    try:
        stack.enter_context(concurrency("dbt-runs", timeout_seconds=5, strict=True))
    except TimeoutError:
        # decide to exit early / mark as skipped / raise to fail
        return

Option B: Tag-based task concurrency - Tag the task that performs the dbt run (e.g., tags=["dbt"]) and set a concurrency limit for that tag. This will also enforce a shared limit across deployments, but only for tasks, not the entire flow.

from prefect import task

@task(tags=["dbt"])
def run_dbt(...):
    ...

prefect concurrency-limit create dbt 2

Which option should you use? - Want to cap entire flow runs: use Global Concurrency Limit and acquire at the top of the flow (Option A). - Only want to limit the dbt step: tag-based task concurrency (Option B) or wrap just the dbt call with the GCL (Option A). Important details for your scenario - If dbt uses 3 threads per run and you allow 2 simultaneous runs, your DB will see up to ~6 concurrent dbt queries. Adjust either the GCL limit or dbt threads to match instance capacity. - The

concurrency_limit

field in prefect.yml applies per-deployment only; it will not create a shared budget across multiple deployments. CLI references (verified) - GCL management: - List limits:

prefect gcl ls

- Inspect:

prefect gcl inspect dbt-runs

- Update:

prefect gcl update dbt-runs --limit 3

- Tag-based limits: - Create:

prefect concurrency-limit create dbt 2

- Inspect:

prefect concurrency-limit inspect dbt

Docs - Global Concurrency Limits concept: Global concurrency limits - How-to (GCL): Apply global concurrency and rate limits - CLI for GCL: prefect gcl CLI

- Python APIs (context managers): Sync prefect.concurrency.sync, Async prefect.concurrency.asyncio - Deployment YAML (per-deployment limit reference): prefect.yaml guide - Tag-based limits: Tag-based concurrency limits A couple of clarifying questions so I can tailor the snippet precisely: - Do you want to cap the entire flow run (i.e., only two flows active at a time), or just the dbt execution step? - Are your dbt calls made in a single task, or multiple tasks? - Do you want “wait until a slot is free” semantics, or “skip/cancel if busy” when more than two are active?

Great idea—using Prefect’s concurrency context manager at the top of your flow will ensure only a limited number of these dbt runs execute at once across all workers. Assuming Prefect 3.x and a synchronous flow, here’s how to wrap your flow body in a global concurrency limit: - Import the sync context manager:

from prefect.concurrency.sync import concurrency

- Choose a clear limit name (e.g., a per-database limit like

dbt:{database}

) - Set capacity via the CLI (

prefect gcl create ...

) - Optionally use

strict=True

for guaranteed enforcement, and

timeout_seconds

to bound wait time Example integrating directly into your flow:

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

@flow
def clickstream_dbt_run(
    database: str,
    models: str = "",
    full_refresh: bool = False,
    store_timezone_offset_minutes: str = "330"
):
    logger = get_run_logger()

    # Use a per-database concurrency limit (adjust name as needed)
    limit_name = f"dbt:{database}"

    # Acquire N slots (occupy) from the limit for the duration of the block.
    # strict=True -> fail fast if the limit doesn't exist or cannot be enforced.
    # timeout_seconds -> how long to wait for a slot before raising.
    with concurrency(limit_name, occupy=1, timeout_seconds=900, strict=True):
        # If you want the env-load to be gated too, keep it inside the block.
        load_env_vars_dynamic(
            database=database,
            store_timezone_offset_minutes=store_timezone_offset_minutes
        )

        <http://logger.info|logger.info>(f"Prefect-Dbt Environment is Ready For Database: {database}")
        <http://logger.info|logger.info>(f"Running For Database: {database}")
        <http://logger.info|logger.info>(f"Models: {models}")
        <http://logger.info|logger.info>(f"Full refresh: {full_refresh}")

        # Build the dbt command
        dbt_command = ["run", "--target", "dynamic-target"]

        # Add model selection
        if models:
            dbt_command.extend(["-s", models])

        # Add full refresh if specified
        if full_refresh:
            dbt_command.append("--full-refresh")

        <http://logger.info|logger.info>(f"Executing dbt command: {' '.join(dbt_command)}")

        # Run dbt
        try:
            result = PrefectDbtRunner().invoke(dbt_command)
            <http://logger.info|logger.info>("dbt run completed successfully")
            return result
        except Exception as e:
            logger.error(f"dbt run failed: {str(e)}")
            raise

How to set capacity for the limit (run this once to create/update limits):

# Create a per-database limit with capacity 1
prefect gcl create dbt:my_database --limit 1

# Update capacity later
prefect gcl update dbt:my_database --limit 2

# List and inspect limits
prefect gcl ls
prefect gcl inspect dbt:my_database

Notes: - “Capacity” is the limit’s total slots (set via

prefect gcl create --limit N

). - “Slot scope” is the

with concurrency(...)

block; slots are held for the duration of that block. - “Occupy” is how many slots your flow consumes (e.g.,

occupy=2

). - No special enabling is required—global concurrency limits are available by default in Prefect 3.x (Server and Cloud). - If you prefer to only gate the dbt invocation (and not env loading), you can move the

with concurrency(...)

block to wrap just the

PrefectDbtRunner().invoke(...)

. Docs: - Concept guide: Global concurrency limits - How-to: Using global concurrency limits

That pattern works — you can acquire a global concurrency slot up front with ExitStack, bail out on timeout, and only proceed if a slot is secured. A couple of details to get right: - The timeout exception you want is AcquireConcurrencySlotTimeoutError, which subclasses TimeoutError (so your bare TimeoutError handler will work, but it’s broader). - With strict=True, if the limit doesn’t exist or cannot be enforced you’ll get ConcurrencySlotAcquisitionError — catch it if you want a different outcome than generic failure. Here’s a drop-in version for your flow using ExitStack:

from contextlib import ExitStack
from prefect import flow, get_run_logger
from prefect.concurrency.sync import concurrency
# Specific exception classes (optional, but recommended for precise handling)
from prefect.concurrency.asyncio import (
    AcquireConcurrencySlotTimeoutError,
    ConcurrencySlotAcquisitionError,
)

@flow
def clickstream_dbt_run(
    database: str,
    models: str = "",
    full_refresh: bool = False,
    store_timezone_offset_minutes: str = "330",
):
    logger = get_run_logger()

    with ExitStack() as stack:
        try:
            # Acquire a slot up front for the entire flow run
            stack.enter_context(
                concurrency(
                    "dbt-runs",          # or f"dbt:{database}" if you want per-db isolation
                    timeout_seconds=5,   # wait up to 5s for a slot
                    strict=True,         # fail if limit is missing or cannot be enforced
                    # lease_duration=900 # optionally extend if your dbt runs are long
                )
            )
        except AcquireConcurrencySlotTimeoutError:
            # More precise than bare TimeoutError; subclass of TimeoutError
            <http://logger.info|logger.info>("No dbt-runs slots available within 5s; exiting early")
            return
        except ConcurrencySlotAcquisitionError as e:
            # Limit missing (strict=True) or other acquisition failure
            logger.error(f"Failed to acquire dbt-runs slot: {e}")
            raise
        # If you prefer the broader catch, your original handler is okay:
        # except TimeoutError:
        #     <http://logger.info|logger.info>("No dbt-runs slots available within 5s; exiting early")
        #     return

        # Everything below runs only if we successfully acquired a slot
        load_env_vars_dynamic(
            database=database,
            store_timezone_offset_minutes=store_timezone_offset_minutes
        )

        <http://logger.info|logger.info>(f"Prefect-Dbt Environment is Ready For Database: {database}")
        <http://logger.info|logger.info>(f"Running For Database: {database}")
        <http://logger.info|logger.info>(f"Models: {models}")
        <http://logger.info|logger.info>(f"Full refresh: {full_refresh}")

        dbt_command = ["run", "--target", "dynamic-target"]
        if models:
            dbt_command.extend(["-s", models])
        if full_refresh:
            dbt_command.append("--full-refresh")

        <http://logger.info|logger.info>(f"Executing dbt command: {' '.join(dbt_command)}")

        try:
            result = PrefectDbtRunner().invoke(dbt_command)
            <http://logger.info|logger.info>("dbt run completed successfully")
            return result
        except Exception as e:
            logger.error(f"dbt run failed: {str(e)}")
            raise

Create/manage the limit before running:

# Create a limit for all dbt runs
prefect gcl create dbt-runs --limit 1

# Later updates/inspection
prefect gcl update dbt-runs --limit 2
prefect gcl ls
prefect gcl inspect dbt-runs

Notes: - Bare

except TimeoutError:

will catch the timeout case because

AcquireConcurrencySlotTimeoutError

subclasses

TimeoutError

. If you want to avoid catching unrelated timeouts, catch

AcquireConcurrencySlotTimeoutError

specifically. - With

strict=True

, you must create the limit first or you’ll get

ConcurrencySlotAcquisitionError

. Docs: - Concept guide: Global concurrency limits

- How-to: Using global concurrency limits

Here’s how lease_duration and strict apply to your setup: Your setup - Global concurrency limit (GCL) name (e.g., dbt-runs) has capacity 2 - You acquire a slot at flow start (ExitStack + concurrency(...)) - You set lease_duration=5 minutes - A single flow run takes ~10 minutes What lease_duration does - When a slot is acquired, the server starts a countdown for the lease_duration (default 5 minutes; min 1 minute, max 60 minutes). - The client auto-renews the lease periodically while your with concurrency(...) block is active. - As long as renewals succeed, your flow can run arbitrarily long (10 minutes is fine with a 5-minute lease), because the client keeps renewing before expiration. - If your process crashes or can’t renew, the slot is automatically freed when the last lease expires. This prevents deadlocks. What strict does - strict=False (default): - If the limit is missing at acquisition time, Prefect logs a warning and proceeds without enforcement. - If lease renewal later fails (network blip, server down), Prefect logs a warning and continues running. Your code keeps going but may be running without an active lease until it exits. - strict=True: - If the limit is missing or can’t be enforced at acquisition, an error is raised immediately. - If lease renewal fails during the run, execution stops immediately with an error. Prefect attempts to release the slot; worst case, the server frees it when the current lease expires. How this plays out in your case (capacity 2, lease 5 minutes, run ~10 minutes) - Normal operation (no failures): - Two flows (A, B) start, each acquires one slot, and both run concurrently. - A third flow (C) tries to start: - If you set

timeout_seconds=5

and no slot frees within 5s, C gets a timeout (AcquireConcurrencySlotTimeoutError). - If A and B keep running past 5 minutes, their leases are renewed repeatedly. They hold their slots for the full ~10 minutes until they exit the context; C keeps waiting or times out, depending on your timeout. - A or B process crashes unexpectedly: - Their current lease will expire in at most 5 minutes, automatically freeing that slot. - Practical effect: capacity is “stuck” for up to the lease duration after a crash. A shorter lease reduces this “stuck” time; a longer lease increases it. - Lease renewal fails mid-run: - strict=False: A or B continues running; their lease may lapse and release its slot. This means another flow could start and you might temporarily exceed the intended concurrency (the still-running flow is no longer “counted”). - strict=True: A or B errors immediately and exits. Prefect attempts to release its slot right away; if that’s not possible, the slot frees when the current lease expires (≤ lease_duration). - Limit missing: - strict=False: your flow runs without any concurrency enforcement (warning logged). - strict=True: your flow fails fast at acquisition time. Choosing a lease_duration - 5 minutes with 10-minute runs is fine; renewals keep the slot for the entire run. - Shorter lease (e.g., 1–2 minutes): quicker cleanup after crashes, but more frequent renewals. - Longer lease (e.g., 10–30 minutes): fewer renewals, but longer “stuck” capacity if a process dies. - Constraints: minimum 1 minute, maximum 60 minutes. Recommended pattern for your flow - Keep strict=True if exceeding concurrency would cause problems (e.g., warehouse connection limits). - Keep a short

timeout_seconds

if you prefer to skip instead of queue when capacity is full. - Use a lease_duration that balances renewal churn vs. crash cleanup. Default 5 minutes is a good starting point. Example with your ExitStack pattern and lease_duration ``` with ExitStack() as stack: try: stack.enter_context( concurrency( "dbt-runs", timeout_seconds=5, # skip quickly if full strict=True, # enforce strictly lease_duration=300 # 5 minutes )

) except TimeoutError: # No slot became available within 5s return # proceed with the flow; lease will be renewed until this block exits ``` Docs for reference: - Concept guide: Global concurrency limits - How-to: Using global concurrency limits

By “limit missing,” I mean the named global concurrency limit (GCL) you’re trying to acquire does not have any active record the server can enforce. This happens in two practical situations: - The limit was never created (or was deleted) — truly “missing” - The limit exists but is disabled (inactive) — operationally treated the same as missing What happens at runtime - strict=False (default) - If the limit is missing or disabled, Prefect logs a warning and continues without acquiring any slots. Your code runs unbounded. - strict=True - If the limit is missing or disabled, Prefect raises

ConcurrencySlotAcquisitionError

immediately and stops execution. In your ExitStack pattern, with

strict=True

, both “never created” and “disabled” will hit the same except branch. Example handling

from contextlib import ExitStack
from prefect.concurrency.sync import concurrency
from prefect.concurrency.asyncio import (
    AcquireConcurrencySlotTimeoutError,
    ConcurrencySlotAcquisitionError,
)

with ExitStack() as stack:
    try:
        stack.enter_context(
            concurrency("dbt-runs", timeout_seconds=5, strict=True)
        )
    except AcquireConcurrencySlotTimeoutError:
        # No slot became available in 5s
        return
    except ConcurrencySlotAcquisitionError as e:
        # Limit is missing or disabled (or otherwise unenforceable)
        # Decide: exit early, mark skipped, or raise
        return
    # proceed with your flow work under an active lease

How to avoid “missing” at run time - Create the limit once (per name) and set capacity:

prefect gcl create dbt-runs --limit 2

- If it was disabled, enable it:

prefect gcl enable dbt-runs

- Inspect and list limits:

prefect gcl ls
  prefect gcl inspect dbt-runs

Key takeaway - “Missing” = not enforceable. Both “never created” and “disabled” behave the same for the client: - strict=False: warn and continue without enforcement - strict=True: fail fast Docs: - Concept guide: Global concurrency limits - How-to: Using global concurrency limits

You’re seeing “Completed” because you

return

after the timeout. To end the run as Cancelled, return a Prefect

Cancelled

state instead of a normal value. Here’s how to adapt your ExitStack block:

from contextlib import ExitStack
from prefect.concurrency.sync import concurrency
from prefect.concurrency.asyncio import (
    AcquireConcurrencySlotTimeoutError,
    ConcurrencySlotAcquisitionError,
)
from prefect.states import Cancelled

slot_timeout_seconds = 5

with ExitStack() as stack:
    try:
        stack.enter_context(
            concurrency(
                "dbt-runs",
                timeout_seconds=slot_timeout_seconds,
                strict=True,
                # lease_duration=300,  # optional
            )
        )
    except AcquireConcurrencySlotTimeoutError:
        # More precise than bare TimeoutError; subclass of TimeoutError
        return Cancelled(
            message=f"No dbt-runs slots available within {slot_timeout_seconds}s; exiting early"
        )
    except ConcurrencySlotAcquisitionError as e:
        # Limit missing/disabled or otherwise unenforceable (strict=True)
        return Cancelled(
            message=f"Failed to acquire dbt-runs slot: {e}"
        )

    # proceed with the flow; lease will be renewed until this block exits

Notes: - Returning

Cancelled(...)

from a flow sets the flow run to the Cancelled state immediately. - If you use this inside a task, that task will be Cancelled; to cancel the whole flow, return the cancelled state from the flow as well (e.g., check a task’s state with

return_state=True

and propagate the cancellation). - If you prefer a broader catch, a bare

except TimeoutError:

will also work since

AcquireConcurrencySlotTimeoutError

subclasses

TimeoutError

, but using the specific exception is cleaner. References: - Concurrency concepts: Global concurrency limits - How-to guide: Using global concurrency limits - States overview: States in Prefect

ExitStack is a utility from Python’s

contextlib

that lets you manage a dynamic or conditional set of context managers with a single

with

block. Why it’s useful - Dynamic contexts: When you don’t know ahead of time how many contexts you’ll need (e.g., add a concurrency guard only sometimes), you can add them programmatically. - Clean, centralized cleanup: Everything you enter is guaranteed to be exited in LIFO order when the

with ExitStack()

block ends, even if an exception occurs midway. - Partial acquisition safety: If entering a later context fails, ExitStack automatically exits any earlier contexts that were already entered, preventing leaks. - Fine-grained error handling on enter: You can try to enter a context (like a concurrency slot) and, if it fails, handle it immediately before any of the rest of your code runs. Key methods - `enter_context(cm)`: Enters a context manager and returns its

__enter__

value; ExitStack will later call its

__exit__

. - `callback(func, *args, **kwargs)`: Register arbitrary cleanup code to run on exit (handy when you don’t have a context manager). - `push(exit_func)`: Push a raw

__exit__

-like function to the stack (advanced). - `pop_all()`: Detach all cleanup callbacks from the current stack and return a new stack that owns them (advanced). Basic patterns - Multiple contexts, dynamic count:

from contextlib import ExitStack

with ExitStack() as stack:
    files = ["a.txt", "b.txt"]
    opened = [stack.enter_context(open(f)) for f in files]
    # use opened files
# all files are closed here in reverse order

- Conditional contexts:

with ExitStack() as stack:
    if use_lock:
        lock = stack.enter_context(acquire_lock())
    if need_tmpdir:
        tmpdir = stack.enter_context(tempfile.TemporaryDirectory())
    # work
# any entered contexts are cleaned up here

With Prefect’s concurrency - Your pattern acquires a global concurrency slot at the top; if acquisition times out or fails, you exit early. - If acquisition fails during

enter_context(...)

, you never enter the concurrency context, so there’s nothing to release. - If acquisition succeeds, the slot stays held for the entire

with ExitStack()

block and is reliably released when the block exits (even if an exception happens inside). Example with your flow’s concurrency guard and cancellation:

from contextlib import ExitStack
from prefect.concurrency.sync import concurrency
from prefect.concurrency.asyncio import (
    AcquireConcurrencySlotTimeoutError,
    ConcurrencySlotAcquisitionError,
)
from prefect.states import Cancelled

slot_timeout_seconds = 5

with ExitStack() as stack:
    try:
        stack.enter_context(
            concurrency(
                "dbt-runs",
                timeout_seconds=slot_timeout_seconds,
                strict=True,
                # lease_duration=300,  # optional
            )
        )
    except AcquireConcurrencySlotTimeoutError:
        return Cancelled(
            message=f"No dbt-runs slots available within {slot_timeout_seconds}s; exiting early"
        )
    except ConcurrencySlotAcquisitionError as e:
        return Cancelled(
            message=f"Failed to acquire dbt-runs slot: {e}"
        )

    # ... do work under the held lease ...
# lease is released here automatically

Async note - For async code, use

contextlib.AsyncExitStack

with

async with

and

async with concurrency(...)

(from

prefect.concurrency.asyncio

). Docs: Python contextlib.ExitStack

You’ve got a few clean ways to scope your GCL to only the 5 every-30-min deployments so the midnight run isn’t blocked. Option A: Use different limit names (simplest if you control code per deployment) - Apply the concurrency context only in the 5 deployments using a distinct limit name (e.g.,

dbt-runs:batch

). - Do not acquire a limit in the midnight deployment. Example:

# For the 5 batch deployments only
with ExitStack() as stack:
    stack.enter_context(
        concurrency("dbt-runs:batch", timeout_seconds=5, strict=True)
    )
    # ...rest of work...

Create the limit:

prefect gcl create dbt-runs:batch --limit 2

Your midnight deployment runs with no concurrency context at all, so it’s never blocked by

dbt-runs:batch

. Option B: Conditional acquisition based on a parameter (works well if all 6 share the same code) - Add a parameter like

enforce_gcl: bool

and turn it on for the 5, off for the midnight deployment.

from contextlib import ExitStack
from prefect.states import Cancelled
from prefect.concurrency.sync import concurrency
from prefect.concurrency.asyncio import (
    AcquireConcurrencySlotTimeoutError,
    ConcurrencySlotAcquisitionError,
)

def run_dbt(enforce_gcl: bool = True, limit_name: str = "dbt-runs:batch", slot_timeout_seconds: int = 5):
    with ExitStack() as stack:
        if enforce_gcl:
            try:
                stack.enter_context(
                    concurrency(
                        limit_name,
                        timeout_seconds=slot_timeout_seconds,
                        strict=True,
                    )
                )
            except AcquireConcurrencySlotTimeoutError:
                return Cancelled(message=f"No slots available within {slot_timeout_seconds}s; exiting early")
            except ConcurrencySlotAcquisitionError as e:
                return Cancelled(message=f"Failed to acquire slot: {e}")

        # ... do dbt work ...

- For the 5 deployments: set

enforce_gcl=True

(default),

limit_name="dbt-runs:batch"

. - For the midnight deployment: set

enforce_gcl=False

. Option C: Conditional acquisition based on deployment name (no parameter changes needed) - Use runtime metadata to check which deployment is running and only enforce for the 5.

from contextlib import ExitStack
from prefect.runtime import deployment as rt_deployment
from prefect.states import Cancelled
from prefect.concurrency.sync import concurrency
from prefect.concurrency.asyncio import (
    AcquireConcurrencySlotTimeoutError,
    ConcurrencySlotAcquisitionError,
)

with ExitStack() as stack:
    dep_name = rt_deployment.name()
    enforce_for = {"dep-a", "dep-b", "dep-c", "dep-d", "dep-e"}  # 5 batch deployments

    if dep_name in enforce_for:
        try:
            stack.enter_context(
                concurrency("dbt-runs:batch", timeout_seconds=5, strict=True)
            )
        except AcquireConcurrencySlotTimeoutError:
            return Cancelled(message="No slots available within 5s; exiting early")
        except ConcurrencySlotAcquisitionError as e:
            return Cancelled(message=f"Failed to acquire slot: {e}")

    # ... dbt work ...

Option D: If you want the midnight run to still be limited but never “missed” - Keep the same limit name across all 6, but for the midnight deployment set

timeout_seconds=None

so it waits for a slot instead of cancelling. This guarantees it eventually runs, even if delayed:

# Midnight deployment only
with concurrency("dbt-runs", timeout_seconds=None, strict=True):
    # waits indefinitely for a slot
    ...

- The 5 deployments can keep a short timeout so they skip quickly when full. Operational alternative: reserve capacity via work pools - Put the midnight deployment in its own work pool (or dedicated worker) so it always has compute available regardless of the batch runs. This is orthogonal to GCL and ensures scheduling reliability. CLI you’ll need (verified): ```

# Create a limit used only by the 5 deployments prefect gcl create dbt-runs:batch --limit 2 # Observe/adjust prefect gcl ls prefect gcl inspect dbt-runs:batch prefect gcl update dbt-runs:batch --limit 3 ``` Docs: - Global concurrency limits: Concepts - How-to guide: Using global concurrency limits If you share your deployment names or how you build them, I can sketch the exact conditional snippet for your setup.

Great question. Short answer: with a GCL of 2 and three deployments scheduled at the exact same time, the two slots will be taken by whichever two runs the server hands to workers first. The third will wait or time out depending on your acquisition logic. There’s no “fairness” by deployment built into GCLs. More detail: - Work-pool/queue ordering: Workers fetch eligible runs ordered primarily by next_scheduled_start_time ascending. If multiple runs share the exact same scheduled time (anchor-aligned), there is no explicit tiebreaker in the server query; the DB’s natural/insertion order tends to decide which two get dispatched first. - GCL arbitration: Slot acquisition is server-side and atomic. When multiple runs attempt to acquire at once, one succeeds, others get a 423 Locked and retry based on Retry-After headers. There’s no per-deployment priority within a single GCL. What you can do if you need deterministic or protected behavior: - Separate limits per deployment group - Example: use

dbt-runs:group-a

for two deployments and

dbt-runs:group-b

for the third; set

--limit 2

for group-a and no limit for group-b (or a different capacity). - Use deployment-specific limit names - Each deployment acquires

concurrency(f"dbt-runs:{deployment_name}")

with its own capacity. This guarantees isolation. - Stagger schedules slightly - Offset one or two deployments by 1–2 minutes from the common anchor to avoid all three competing simultaneously. - Adjust acquisition behavior - For the “less critical” runs, keep

timeout_seconds

small so they gracefully skip when slots are busy. - For the “must run” deployment, set

timeout_seconds=None

so it waits for a slot instead of cancelling. - Use separate work queues or pool priorities - Put the higher-priority deployment in a higher-priority work queue within the same pool. Workers respecting queue priorities will fetch those runs first, making them more likely to acquire slots before others. If you share how critical each deployment is and whether they share code, I can propose an exact limit-naming and parameter pattern so the right two always win or so a specific one is guaranteed to run.

You’ve got two viable ways to achieve “don’t cancel, just wait for a slot” — with slightly different behavior/visibility: Option 1: Keep the global concurrency context and block until a slot is free - Set

timeout_seconds=None

and remove the timeout handler. The run will wait inside the context manager until a slot is available, then proceed. - Note: the flow run will remain in Running while it waits; there is no special “AwaitingConcurrencySlot” state at the flow level for GCL. Example update to your logic:

from contextlib import ExitStack
from prefect.concurrency.sync import concurrency
from prefect.concurrency.asyncio import ConcurrencySlotAcquisitionError

with ExitStack() as stack:
    try:
        stack.enter_context(
            concurrency(
                PREFECT_ALL_FLOWS_GCL,
                timeout_seconds=None,   # wait indefinitely for a slot
                strict=True,
                lease_duration=PREFECT_ALL_FLOWS_GCL_LEASE_DURATION,
            )
        )
    except ConcurrencySlotAcquisitionError as e:
        # Limit missing/disabled or otherwise unenforceable
        # Decide whether to raise or return a Failed/Cancelled state
        raise

    # ... proceed: env, dbt, etc. ...

Pros: - Minimal code change; preserves your current pattern. Cons: - Flow state remains Running during the wait (no explicit “AwaitingConcurrencySlot” state). Option 2: Use tag-based concurrency on a task to get “AwaitingConcurrencySlot” state If you want Prefect to surface “AwaitingConcurrencySlot” while waiting, move the dbt invocation into a task and apply a tag-based concurrency limit. When capacity is full, the task will be put in an awaiting state by the server and resumed when slots free up — without holding a worker thread open. - Create a tag concurrency limit:

prefect concurrency-limit create dbt-runs 2

- Wrap the dbt invocation in a task tagged with that tag:

from prefect import flow, task

@task(tags=["dbt-runs"])
def run_dbt_cli(database: str, models: str, full_refresh: bool, store_timezone_offset_minutes: str):
    load_env_vars_dynamic(
        database=database,
        store_timezone_offset_minutes=store_timezone_offset_minutes
    )
    dbt_command = ["run", "--target", "dynamic-target"]
    if models:
        dbt_command.extend(["-s", models])
    if full_refresh:
        dbt_command.append("--full-refresh")
    return PrefectDbtRunner().invoke(dbt_command)

@flow
def clickstream_dbt_run(database: str, models: str = "", full_refresh: bool = False, store_timezone_offset_minutes: str = "330"):
    return run_dbt_cli(database, models, full_refresh, store_timezone_offset_minutes)

- Now, when capacity is full, the task state becomes “AwaitingConcurrencySlot” and will resume automatically when a slot opens. The flow will remain running with a waiting task (visible in the UI). If you only want the 5 deployments to be throttled (and not the midnight one): - Use runtime-run tags for the throttled deployments only. Tasks inherit flow run tags at creation, so you can add the tag conditionally:

from prefect.runtime import deployment as rt_deployment
from prefect.runtime.flow_run import tags as flow_run_tags

@flow
def clickstream_dbt_run(...):
    dep_name = rt_deployment.name()
    throttle_deployments = {"dep-a", "dep-b", "dep-c", "dep-d", "dep-e"}

    if dep_name in throttle_deployments:
        with flow_run_tags("dbt-runs"):   # apply tag only for the 5 deployments
            return run_dbt_cli(...)
    else:
        # midnight deployment runs without throttling
        return run_dbt_cli(...)

Operational notes - For Option 1 (GCL): You’ll wait inside the worker process; it’s simple and robust, but the state will show as Running. - For Option 2 (tag concurrency): Prefect will transition the task to “AwaitingConcurrencySlot” and re-schedule efficiently; great observability and doesn’t tie up a worker thread.

- If you must keep GCL for other reasons but still want an “awaiting” state, Prefect does not currently expose a flow-level “AwaitingConcurrencySlot” state via GCL. In that case, prefer tag-based concurrency on a task. Docs: - Global concurrency limits: Concepts and How-to - Tag-based concurrency limits: Tag-based limits and CLI prefect concurrency-limit