@Marvin: I am running a deployment on a schedule of 5 mins. How can I make sure no other flow starts before the current has finished execution?

To avoid starting a new flow run before the current one has finished, you can leverage tags and concurrency limits in Prefect. Here's how you can do it: 1. Tag your Flow Runs: Assign a unique tag to your flow runs. 2. Set Up Concurrency Limits: Use concurrency limits based on the tag to ensure only one run starts at a time. ### Step-by-step Implementation: ### 1. Tagging Flow Runs Ensure you assign a unique tag to your flow runs:

python
from prefect import Flow, task

@task
def my_task():
    pass

with Flow("my flow", tags=["my_unique_tag"]) as flow:
    my_task()

### 2. Setting Up Concurrency Limits You need to create a concurrency limit through the Prefect Cloud UI or using the Prefect Python API. The following is an example using the API:

python
from prefect.client import Client

client = Client()

# Set a concurrency limit of 1 for the tag 'my_unique_tag'
client.create_concurrency_limit(
    tag="my_unique_tag",
    limit=1
)

With the above setup, Prefect will ensure that only one flow run with the tag

my_unique_tag

can run at any given time. ### Links to Documentation - Tagging Flow Runs - Setting Concurrency Limits That should prevent overlapping runs for your scheduled deployments. If you need further assistance, feel free to ask.

@Marvin: Yes but it still somehow adds the next scheduled flows in running state and messes up.

There appears to be an existing issue where Prefect schedules or duplicates flow runs unexpectedly. While there might not be an immediate, specific fix available from the searches, I can suggest another approach that might mitigate this problem. ### Suggested Approach Instead of solely relying on tags and concurrency limits, you can create a more robust check within the flow itself to ensure no other instance is running. Here is an example approach using the Prefect Client: ### 1. Check for Existing Runs Add a task that checks for existing runs of the same flow that are currently running before executing the main flow logic.

python
from prefect import task, Flow, Parameter
from prefect.client import Client
from prefect.engine.state import Running

@task
def check_running_flows(flow_name):
    client = Client()
    current_runs = client.get_flow_run_info(flow_name=flow_name)
    running_runs = [run for run in current_runs if run.state == Running]
    if running_runs:
        raise RuntimeError("Another instance of the flow is currently running.")

@task
def my_task():
    print("Flow task is running.")

with Flow("my flow") as flow:
    flow_name = Parameter("flow_name", default="my flow")
    check_running_flows(flow_name)
    my_task()

flow.register(project_name="my-project")

### 2. Retry Mechanism To handle flows that fail to start due to collision, you can add a retry mechanism:

python
from prefect.tasks.control_flow import case
from prefect.engine import signals

@task(max_retries=3, retry_delay=timedelta(minutes=1))
def check_running_flows(flow_name):
    client = Client()
    current_runs = client.get_flow_run_info(flow_name=flow_name)
    running_runs = [run for run in current_runs if run.state == Running]
    if running_runs:
        raise signals.TRIGGERFAIL("Another instance of the flow is currently running.")

### Further Reading - Prefect's Control Flow documentation - Prefect's Retry Logic documentation For further troubleshooting, it might be helpful to look through Prefect's issues on GitHub. Hopefully, these steps will help manage and prevent overlaps in your flow runs effectively.

@Marvin: Is it possible to skip a new scheduled flow run if a previous flow run of the same flow is still in progress?

It seems there is no built-in feature directly addressing this in Prefect. However, you can implement a custom check within your flow to skip execution if a previous run is in progress. Here’s how you can achieve that: ### 1. Check for Existing Runs Create a task to check if there are any running instances of the flow. If any are found, skip the current flow run.

python
from prefect import task, Flow
from prefect.client import Client
from prefect.engine.state import Running
from prefect.engine import signals

@task
def check_existing_runs(flow_id):
    client = Client()
    flow_runs = client.get_flow_runs(flow_id=flow_id, state=Running)
    if flow_runs:
        raise signals.SKIP("Another run is still in progress. Skipping this run.")

@task
def my_task():
    print("Executing my_task")

with Flow("my_flow") as flow:
    flow_id = "your-flow-id"  # Replace with your Flow ID
    check_existing_runs(flow_id)
    my_task()

flow.register(project_name="my-project")

### Explanation 1. Check Existing Runs Task: This task queries the Prefect backend for any running flow instances with the same flow ID. If any are found, it raises a SKIP signal to skip the current flow run. 2. Main Flow Logic: The flow proceeds to execute tasks only if no running instances are detected. ### Deployment Ensure the deployment is properly set up and registered. This implementation ensures that if a run is already in progress, the new scheduled run will be skipped. ### Further Reading - Prefect API Client - Task Signals By incorporating this approach, you can effectively manage and skip new flow runs if a previous one is still running.