<@ULVA73B9P> I wonder how to set lease duration an...
# ask-marvin
e
@Marvin I wonder how to set lease duration and strict mode for concurrency management in those two cases : • concurrency limiting on deployment • tag-based task concurrency limiting
m
thought for 396.1 seconds
Great question. In Prefect 3.x, “lease duration” and “strict mode” are knobs on the global concurrency API (the
prefect.concurrency.concurrency
context manager) — they are not configurable for deployment-level concurrency or tag-based task concurrency. Here’s how each case works and what you can configure instead: - Deployment-level concurrency - There is no lease duration or strict mode here. - You control: - `concurrency_limit`: maximum concurrent runs for that deployment - `concurrency_options.collision_strategy`: what to do when the limit is hit (ENQUEUE or CANCEL_NEW) - `concurrency_options.grace_period_seconds`: buffer for infra startup before a run “consumes” a slot (not a lease) - Example (Python deploy):
Copy code
from prefect import flow
    from prefect.server.schemas.core import ConcurrencyOptions, ConcurrencyLimitStrategy

    @flow
    def my_flow():
        ...

    my_flow.from_source(
        source="<https://github.com/org/repo.git>",
        entrypoint="flows.py:my_flow",
    ).deploy(
        name="my-deployment",
        work_pool_name="my-pool",
        concurrency_limit=5,  # simple cap on concurrent runs
        concurrency_options=ConcurrencyOptions(
            collision_strategy=ConcurrencyLimitStrategy.ENQUEUE,  # or CANCEL_NEW
            grace_period_seconds=300,
        ),
    )
- Tag-based task concurrency - There is no user-facing lease duration or strict mode. - You set a limit per tag, and Prefect automatically delays tasks when the limit is reached. The server returns a Retry-After and the client backs off. - The main “knob” is the server-side cap on the wait time for tag-based slots: -
PREFECT_SERVER_TASKS_TAG_CONCURRENCY_SLOT_WAIT_SECONDS
(default 30s) - Usage: - Tag tasks:
Copy code
from prefect import task

      @task(tags=["db"])
      def query(...):
          ...
- Manage limits via CLI (verified):
Copy code
prefect concurrency-limit create db 10
      prefect concurrency-limit inspect db
      prefect concurrency-limit ls
      prefect concurrency-limit reset db
      prefect concurrency-limit delete db
- Or via Python client:
Copy code
from prefect import get_client

      async with get_client() as client:
          await client.create_concurrency_limit(tag="db", concurrency_limit=10)
          limit = await client.read_concurrency_limit_by_tag(tag="db")
          await client.reset_concurrency_limit_by_tag(tag="db")
          await client.delete_concurrency_limit_by_tag(tag="db")
If you specifically need to set lease duration and strict mode, use global concurrency around the code that should be gated. That’s where these options live: - Global concurrency (for completeness) - Available knobs on the context manager: -
lease_duration
(seconds; default 300) -
strict
(default False) - Signature (verified):
Copy code
from prefect.concurrency.sync import concurrency

    # Parameters:
    # names, occupy=1, timeout_seconds=None, max_retries=None,
    # lease_duration=300, strict=False, holder=None

    with concurrency("resource-name", occupy=1, lease_duration=600, strict=True):
        # protected section
        ...
Want help mapping these to your specific flow(s) and deployment(s)? If you share how you’re currently setting your deployment or tags, I can suggest the exact changes.