@Marvin I want a specific scheduled deployment to have concurrency = 1. In other words, I don't want a prior late deployment to overlap a currently scheduled deployment. How to I achieve this?

To set a concurrency limit for a specific scheduled deployment in Prefect, ensuring that no prior late deployment overlaps with a currently scheduled deployment, you typically need to use scheduling features that allow for concurrency settings. However, the provided excerpts from the documentation do not explicitly describe how to set such concurrency configurations. It may be necessary to consult the documentation further or review sample configuration files or deployment commands. If Prefect's standard scheduling features do not support your exact concurrency requirements, as a workaround, you could potentially implement a custom mechanism that locks a resource when a flow starts running and only allows a new run when the previous one has released the lock. This can be done using a database or a distributed lock system like Redis or Zookeeper. For a precise solution, I recommend looking deeper into the Prefect documentation, checking the

prefect deployment

CLI command and its options, or even reaching out to Prefect's community support for guidance on how to achieve concurrency control that meets your specific needs. If you would like, I can continue to search for a more definitive answer within Prefect's documentation or by searching through Prefect GitHub issues for similar problems and solutions. Would you like me to do that?

In the documentation I cannot find a flow or deployment level concurrency limit. There are task concurrency limits available via tags, however no such tag limits for flows. I cannot use a work-pool or work-queue concurrency limit I think, because my deployment has concurrent async task group flows as subflows. My deployed flow should run only of no other runs from that deployment are PENDING or RUNNING.

do other deployments run in the same work pool / queue?

I haven't tried, I'm trying to figure out a solution before testing one. What did you find?

in the situation you described, are these

run_deployment

calls?

starts multiple flows as part of async TaskGroup

yup:

@flow
async def run_from_spec(deployment_list):
    async with asyncio.TaskGroup() as tg:
        flow_runs = []
        for deployment in deployment_list:
            task = tg.create_task(
                run_deployment(
                    name=deployment.get('slug')
                    , parameters=deployment.get('params')
                    , tags=tag_from_deploy_object(deployment)
                    , timeout=TIMEOUT_SECONDS
                    )
                )
            await asyncio.sleep(1)
            flow_runs.append(task)