In Prefect 3.x, the usual way is: *Use a deployment schedule that only creates runs during your allowed window*, and if you need a hard guarantee, add a time check inside the flow. ### Example: only run on weekdays from 9am–5pm

python
from prefect import flow
from prefect.schedules import Cron

@flow
def my_flow():
    print("Doing work")

if __name__ == "__main__":
    my_flow.serve(
        name="business-hours",
        schedule=Cron(
            "0 9-17 * * MON-FRI",
            timezone="America/New_York",
        ),
    )

That schedule creates runs at the top of each hour, Monday–Friday, between 9:00 and 17:00 in the given timezone. ### Important caveat This controls when runs are scheduled, not a strict execution gate. If a run is scheduled at

4:55pm

but a worker only picks it up at

5:10pm

, it can still run. ### If you need a strict time window Add a guard inside the flow:

python
from prefect import flow
from datetime import datetime
from zoneinfo import ZoneInfo

@flow
def my_flow():
    now = datetime.now(ZoneInfo("America/New_York"))

    in_window = (
        now.weekday() < 5 and   # Mon-Fri
        9 <= now.hour < 17
    )

    if not in_window:
        print("Outside allowed time window, skipping.")
        return

    print("Doing work")

### For more complex windows You can also use: -

Cron

for cron-style schedules -

RRule

for more advanced recurrence rules -

Interval

for every N minutes/hours ### Recommendation For most cases: 1. Use `Cron`/`RRule` so Prefect only schedules runs in the desired window 2. Add an in-flow time check if it must never execute outside that window ### Notes - Timezones are supported directly on schedules - Work pools / concurrency limits do not enforce time windows - Pausing/resuming schedules manually is possible, but usually not the best solution for daily windows Useful docs: - Create schedules - Schedules concept docs - Serve flows If you want, I can give you an exact example for: - every 15 minutes between two hours - multiple daily windows - only one run allowed in that window

What you’re seeing is consistent with queueing behind a deployment concurrency limit: - one old run is still in Cancelling - because your deployment limit is

1

, newer runs cannot start - with the default behavior, they pile up in AwaitingConcurrencySlot For your goals, the best steady-state setup is: 1. Deployment concurrency limit = 1 2. Collision strategy =

CANCEL_NEW

instead of queueing 3. Limit how many future runs are scheduled 4. Manually clear the currently stuck run/backlog once ### Why this matches your requirements You want: 1. Only 1 instance running at once -> deployment concurrency limit

1

2. Run every 5 minutes, but if one is late, skip the next one instead of queueing ->

collision_strategy: CANCEL_NEW

3. If a run gets into weird state like

Cancelling

, next run should eventually be able to start -> force-finish the stuck run, then

CANCEL_NEW

prevents backlog buildup going forward --- ## Recommended deployment config ###

prefect.yaml

yaml
deployments:
  - name: my-flow
    entrypoint: path/to/flow_file.py:my_flow
    work_pool:
      name: my-pool
    schedules:
      - interval: 300
        max_scheduled_runs: 1
    concurrency_limit:
      limit: 1
      collision_strategy: CANCEL_NEW
      grace_period_seconds: 300

### What this does -

interval: 300

-> every 5 minutes -

limit: 1

-> only one active run at a time -

CANCEL_NEW

-> if a run is already active, the next scheduled run is cancelled instead of waiting in

AwaitingConcurrencySlot

-

max_scheduled_runs: 1

-> don’t create a large backlog of future scheduled runs This is the closest match to “run every 5 minutes, but skip missed occurrences”. --- ## Immediate recovery steps Right now, you likely need to clean up the stuck run and backlog. ### 1. Force the stuck

Cancelling

run to

Cancelled

If that run never leaves

Cancelling

, it may keep the concurrency slot occupied. You can do this with the Python client:

python
from prefect.client.sync import get_client
from prefect.states import Cancelled

client = get_client()

client.set_flow_run_state(
    flow_run_id="YOUR_FLOW_RUN_ID",
    state=Cancelled(message="Force-cancelled to release stuck concurrency slot"),
    force=True,
)

### 2. Clear the waiting backlog After the stuck run is released, you may still have hundreds of old waiting runs that you no longer want. You can bulk set them to cancelled or delete them via API. If you want, I can help you with a script for that based on: - deployment name - self-hosted server vs Prefect Cloud --- ## Important behavior note With

ENQUEUE

behavior: - run A is active - run B, C, D... go to

AwaitingConcurrencySlot

- if A gets stuck, everything blocks With

CANCEL_NEW

behavior: - run A is active - run B, C, D... are cancelled immediately - when A finishes, the next scheduled occurrence can run normally That is usually the right pattern for polling jobs / periodic sync jobs. --- ## About

Cancelling

Cancelling

is not terminal. It means Prefect is trying to stop infrastructure / clean up. If a worker or infrastructure got into a bad state, a run can remain there too long. In that case, manually forcing it to

Cancelled

is the usual recovery action. --- ## My recommendation for your case I’d strongly suggest this setup: - deployment concurrency

1

- collision strategy

CANCEL_NEW

- schedule every

5 min

-

max_scheduled_runs: 1

That gives you: - never more than one active run - no buildup of

AwaitingConcurrencySlot

- skipped intervals instead of backlog - much easier recovery if one run misbehaves --- ## Docs - Deployment concurrency limiting - Schedules - Flow run states If you want, I can give you one of these next:

1. a ready-to-run cleanup script to cancel/delete all

AwaitingConcurrencySlot

runs for one deployment 2. the equivalent Python

deploy()

code instead of

prefect.yaml

3. a safe recovery checklist for this exact outage situation

Yes — here’s a safe cleanup script for your self-hosted Prefect 3.x server. It will: - find deployment

scanner1

- force-cancel runs in

Cancelling

- force-cancel runs in

AwaitingConcurrencySlot

- optionally delete normal

Scheduled

runs too - print a summary I strongly recommend running it once with

--dry-run

first. ``` #!/usr/bin/env python3 import os import sys from prefect.client.orchestration import get_client from prefect.client.schemas.filters import ( DeploymentFilter, DeploymentFilterName, FlowRunFilter, FlowRunFilterDeploymentId, FlowRunFilterState, FlowRunFilterStateName, FlowRunFilterStateType, ) from prefect.client.schemas.objects import StateType from prefect.exceptions import ObjectNotFound from prefect.states import Cancelled def find_deployment(client, deployment_name: str): # If user passed full name like "flow_name/scanner1", try exact match first if "/" in deployment_name: try: return client.read_deployment_by_name(deployment_name) except ObjectNotFound: return None # Otherwise search by partial deployment name deployments = client.read_deployments( deployment_filter=DeploymentFilter( name=DeploymentFilterName(like_=deployment_name) ), limit=20, ) if not deployments: return None if len(deployments) > 1: print(f"Multiple deployments matched '{deployment_name}':") for d in deployments: flow_name = getattr(d.flow, "name", "<unknown-flow>") print(f" - {flow_name}/{d.name}") print("Please rerun with full deployment name: flow_name/deployment_name") sys.exit(1) return deployments[0] def cleanup_backlog(deployment_id, dry_run=False, delete_scheduled=False): cancelled_count = 0 deleted_count = 0 with get_client(sync_client=True) as client: # 1. Cancelling runs cancelling_runs = client.read_flow_runs( flow_run_filter=FlowRunFilter( deployment_id=FlowRunFilterDeploymentId(any_=[deployment_id]), state=FlowRunFilterState( type=FlowRunFilterStateType(any_=[StateType.CANCELLING]) ), ), limit=None, ) print(f"Found {len(cancelling_runs)} run(s) in Cancelling") for run in cancelling_runs: print(f" Cancelling -> {run.id}") if not dry_run: client.set_flow_run_state( flow_run_id=run.id, state=Cancelled(message="Force-cancelled by backlog cleanup script"), force=True, ) cancelled_count += 1 # 2. AwaitingConcurrencySlot runs # In Prefect these are SCHEDULED type with state name AwaitingConcurrencySlot awaiting_runs = client.read_flow_runs( flow_run_filter=FlowRunFilter( deployment_id=FlowRunFilterDeploymentId(any_=[deployment_id]), state=FlowRunFilterState( type=FlowRunFilterStateType(any_=[StateType.SCHEDULED]), name=FlowRunFilterStateName(any_=["AwaitingConcurrencySlot"]), ), ), limit=None, ) print(f"Found {len(awaiting_runs)} run(s) in AwaitingConcurrencySlot") for run in awaiting_runs: print(f" AwaitingConcurrencySlot -> {run.id}") if not dry_run: client.set_flow_run_state( flow_run_id=run.id, state=Cancelled(message="Cancelled by backlog cleanup script"), force=True, ) cancelled_count += 1 # 3. Optionally delete ordinary Scheduled runs too if delete_scheduled: scheduled_runs = client.read_flow_runs( flow_run_filter=FlowRunFilter(

deployment_id=FlowRunFilterDeploymentId(any_=[deployment_id]), state=FlowRunFilterState( type=FlowRunFilterStateType(any_=[StateType.SCHEDULED]), name=FlowRunFilterStateName(not_any_=["AwaitingConcurrencySlot"]), ), ), limit=None, ) print(f"Found {len(scheduled_runs)} ordinary Scheduled run(s)") for run in scheduled_runs: print(f" Scheduled -> {run.id}") if not dry_run: client.delete_flow_run(run.id) deleted_count += 1 return cancelled_count, deleted_count def main(): if len(sys.argv) < 2: print("Usage:") print(" python cleanup_prefect_backlog.py scanner1 [--dry-run] [--delete-scheduled]") print(' python cleanup_prefect_backlog.py "flow_name/scanner1" [--dry-run] [--delete-scheduled]') sys.exit(1) deployment_name = sys.argv[1] dry_run = "--dry-run" in sys.argv delete_scheduled = "--delete-scheduled" in sys.argv print("Prefect backlog cleanup") print(f"Deployment: {deployment_name}") print(f"PREFECT_API_URL: {os.environ.get('PREFECT_API_URL', '<not set>')}") print(f"Dry run: {dry_run}") print(f"Delete ordinary Scheduled runs: {delete_scheduled}") print("") with get_client(sync_client=True) as client: deployment = find_deployment(client, deployment_name) if deployment is None: print(f"No deployment found matching '{deployment_name}'") sys.exit(1) flow_name = getattr(deployment.flow, "name", "<unknown-flow>") print(f"Matched deployment: {flow_name}/{deployment.name}") print(f"Deployment ID: {deployment.id}") print("") cancelled_count, deleted_count = cleanup_backlog( deployment_id=deployment.id, dry_run=dry_run, delete_scheduled=delete_scheduled, ) print("") print("Summary:") print(f" Cancelled runs: {cancelled_count}") print(f" Deleted scheduled runs: {deleted_count}") if dry_run: print("") print("Dry run only: no changes were made.") if name == "__main__": main()

## How to use it

### 1. Set your API URL
For self-hosted Prefect, set `PREFECT_API_URL` first:

export PREFECT_API_URL=http://YOUR-PREFECT-SERVER:4200/api

If you run it on the same host as the server, it may be:

export PREFECT_API_URL=http://localhost:4200/api

### 2. Dry run first

python cleanup_prefect_backlog.py scanner1 --dry-run

### 3. Actually clean `Cancelling` + `AwaitingConcurrencySlot`

python cleanup_prefect_backlog.py scanner1

### 4. If you also want to remove ordinary future `Scheduled` runs

python cleanup_prefect_backlog.py scanner1 --delete-scheduled ``` --- ## Recommendation for your case After cleanup, update the deployment to avoid this happening again: - concurrency limit:

1

- collision strategy:

CANCEL_NEW

- schedule interval:

300

-

max_scheduled_runs: 1

That will make missed 5-minute ticks get skipped instead of piling up. ## Docs - Deployment concurrency limiting - States - Schedules If you want, I can also prepare: - a one-time script to update the deployment config - a version that targets deployment ID directly - a script that only deletes old runs older than N hours

Yes — for

RunnerDeployment

, adjust both the schedule and the concurrency settings. Your current version: - uses

concurrency_limit=1

-> good, but this alone defaults to queueing behavior - uses

schedules=[Interval(300)]

-> this should be replaced with a deployment schedule object so you can set

max_scheduled_runs

- should add

concurrency_options

with

CANCEL_NEW

Here is the adjusted version:

python
from datetime import timedelta

from prefect.client.schemas.actions import DeploymentScheduleCreate
from prefect.client.schemas.objects import (
    ConcurrencyLimitStrategy,
    ConcurrencyOptions,
)
from prefect.client.schemas.schedules import IntervalSchedule
from prefect.deployments.runner import RunnerDeployment
from prefect.runner.storage import LocalStorage

deployment = RunnerDeployment(
    name="scanner1",
    flow_name="simple_scanner",
    entrypoint="scanner.main:simple_scanner",
    work_pool_name="default_pool",
    storage=LocalStorage(path=env_config.working_directory),
    tags=["scanner", "orchestrator"],
    description=description,
    version=get_version("scanner"),
    schedules=[
        DeploymentScheduleCreate(
            schedule=IntervalSchedule(interval=timedelta(minutes=5)),
            active=True,
            max_scheduled_runs=1,
        )
    ],
    concurrency_limit=1,
    concurrency_options=ConcurrencyOptions(
        collision_strategy=ConcurrencyLimitStrategy.CANCEL_NEW,
    ),
)

## Why these changes ###

concurrency_limit=1

Keeps only one active run at a time. ###

collision_strategy=CANCEL_NEW

This is the important part for your case. Instead of putting new runs into

AwaitingConcurrencySlot

, Prefect will cancel new scheduled runs when the existing one is still active. That matches your requirement: - one scanner only - every 5 minutes - skip missed intervals instead of creating backlog ###

max_scheduled_runs=1

This prevents Prefect from creating a large number of future scheduled runs for this deployment. --- ## Expected behavior after this change If one run is still active at the next 5-minute tick: - old behavior: next run waits in

AwaitingConcurrencySlot

- new behavior: next run is cancelled immediately So you won’t get hundreds of waiting runs anymore. --- ## One practical note This does not magically solve a run stuck forever in

Cancelling

— that still may require manual cleanup, like the script I gave you. But after you clean the current backlog and redeploy with this config, the deployment should behave much closer to what you want. --- ## Optional improvement If you want, you can also set a grace period:

python
    concurrency_options=ConcurrencyOptions(
        collision_strategy=ConcurrencyLimitStrategy.CANCEL_NEW,
        grace_period_seconds=300,
    ),

That can help with infrastructure startup / lease handling. If your runs are normal local process runs, leaving it unset is also fine. --- ## Docs - Deployment concurrency limiting - Schedules If you want, I can also rewrite your full deployment creation function so it includes: - this config - comments - a safe

apply()

/ deploy step