Short answer: with that code, all mapped clearup runs will wait for every mapped process_attachment run to finish. There’s a global “barrier,” not pairwise waiting. What happens step-by-step: 1) process_attachment.map(unmapped(output_dir), total_files) submits one task per item in total_files (output_dir is constant for each). These can run concurrently. 2) clearup.map(total_files, wait_for=process_futures) submits one cleanup per item, but each cleanup run will not start until all process_futures are in a final state. After the last process_attachment finishes, all clearup runs are free to start (and can run concurrently). If you want pairwise ordering (cleanup[i] waits only for process_attachment[i]): - Best: pass the upstream future as a mapped argument so Prefect creates an index-aligned data dependency.

processed = process_attachment.map(unmapped(output_dir), total_files)
cleanup = clearup.map(total_files, processed)  # each cleanup[i] depends on processed[i]

- Or use submit in a zip to set per-item wait_for:

processed = process_attachment.map(unmapped(output_dir), total_files)
cleanup = [clearup.submit(f, wait_for=[pf]) for f, pf in zip(total_files, processed)]

Note: If any process_attachment fails, the dependent clearup runs will not start and will reflect the upstream failure.

Great question. If by “double deployments” you mean “the same flow being scheduled to start at the exact same time and producing two flow runs,” there are two parts: detecting when it happens and preventing it going forward. First, a quick note on causes: - Multiple active deployments for the same flow (each with a schedule) can generate two runs at the same timestamp. - Multiple schedules that align (cron/rrule/interval) can collide. - External triggers plus a schedule can also create overlaps. How to detect duplicates (same scheduled time) - In Prefect 3.x, each flow run stores its expected (scheduled) start time. You can query for runs that share the same expected_start_time and flag duplicates. Example: find duplicate runs for a given deployment within a time window

import asyncio
from collections import Counter
from datetime import datetime, timedelta, timezone

from prefect.client import get_client
from prefect.client.schemas.filters import DeploymentFilter, FlowRunFilter

async def find_duplicate_scheduled_runs(deployment_id: str, since: datetime, until: datetime):
    async with get_client() as client:
        runs = await client.read_flow_runs(
            deployment_filter=DeploymentFilter(id={"any_": [deployment_id]}),
            flow_run_filter=FlowRunFilter(
                expected_start_time={"after": since, "before": until},
                # consider active runs; include all if you want full history
                # state_type={"values": ["SCHEDULED", "PENDING", "RUNNING", "COMPLETED", "CANCELLED", "FAILED"]}
            ),
            limit=5000,  # adjust as needed
        )

    # Group by exact expected_start_time
    counts = Counter(r.expected_start_time for r in runs if r.expected_start_time)
    duplicates = {ts: c for ts, c in counts.items() if c > 1}
    return duplicates, runs

# Usage
since = datetime.now(timezone.utc) - timedelta(days=7)
until = datetime.now(timezone.utc) + timedelta(days=1)
duplicates, runs = asyncio.run(find_duplicate_scheduled_runs("<DEPLOYMENT_ID>", since, until))
for ts, c in sorted(duplicates.items()):
    print(f"{ts.isoformat()} -> {c} runs")

Example: check a single timestamp for duplicates

import asyncio
from datetime import datetime, timedelta, timezone
from prefect.client import get_client
from prefect.client.schemas.filters import DeploymentFilter, FlowRunFilter

async def runs_at_time(deployment_id: str, ts: datetime, window_seconds: int = 1):
    start = ts - timedelta(seconds=window_seconds)
    end = ts + timedelta(seconds=window_seconds)
    async with get_client() as client:
        return await client.read_flow_runs(
            deployment_filter=DeploymentFilter(id={"any_": [deployment_id]}),
            flow_run_filter=FlowRunFilter(
                expected_start_time={"after": start, "before": end},
                state_type={"values": ["SCHEDULED", "PENDING", "RUNNING"]},
            ),
        )

timestamp = datetime(2026, 3, 10, 12, 0, 0, tzinfo=timezone.utc)
runs = asyncio.run(runs_at_time("<DEPLOYMENT_ID>", timestamp, window_seconds=2))
print(f"Found {len(runs)} runs near {timestamp.isoformat()}")

Tip: If you suspect the root cause is multiple deployments for the same flow, list all deployments for that flow and check which have schedules enabled. In the UI, go to the Flow’s Deployments tab and look for multiple active deployments with schedules. In code, you can read all deployments and filter by

d.flow_id

or

d.flow_name

and inspect

d.schedules

. How to prevent overlaps going forward - Use a deployment-level concurrency limit of 1. This enforces that only one run from that deployment can execute at a time. - Choose the right collision strategy: - ENQUEUE: later runs wait for a slot (default). - CANCEL_NEW: later runs at the same time are immediately cancelled (useful if you must never overlap executions). Example: set a single-active-run policy when deploying ```

from prefect.client.schemas.objects import ConcurrencyLimitConfig, ConcurrencyOptions, ConcurrencyLimitStrategy my_flow.deploy( name="my-deployment", work_pool_name="my-pool", # Only 1 run at a time; cancel newcomers if they collide concurrency_limit=ConcurrencyLimitConfig( limit=1, options=ConcurrencyOptions( collision_strategy=ConcurrencyLimitStrategy.CANCEL_NEW, grace_period_seconds=600, # allow infra startup time ), ), # prefer a single deployment with multiple schedules if needed: # schedules=[...] ) ``` Additional best practices - Prefer a single deployment with multiple schedules over multiple separate deployments for the same flow. This centralizes concurrency and reduces accidental duplicate scheduling. - If you programmatically create runs (e.g., via the API), use idempotency keys to deduplicate creation at the API level. If you can share: - Flow name and the deployment names you expect to be active - Whether you’re using Prefect Cloud or OSS - Rough time range where you observed duplicates …I can tailor the exact query/filter and help pinpoint the cause (e.g., two deployments with overlapping schedules vs. external triggers).

Here’s the CLI way to list all scheduled runs for a single flow (across all its deployments): - List scheduled runs by flow name

prefect flow-run ls --flow-name "<FLOW_NAME>" --state SCHEDULED --limit 200

- If you’re not sure which deployments exist for that flow (to explain duplicates), list them:

prefect deployment ls --flow-name "<FLOW_NAME>"

Notes: -

--flow-name

aggregates runs from all deployments of that flow. - Increase

--limit

as needed to see more future scheduled runs. - If you want to narrow by a specific deployment, the flow-run CLI doesn’t expose a deployment filter—list runs and then grep for the deployment name shown in the output. Do you want only future scheduled runs, or all scheduled runs (past and future)?

You can toggle a deployment’s schedules off (pause) in Python by setting the schedule’s active flag to False via the Prefect client. Pause or resume a specific schedule (async)

import asyncio
from uuid import UUID
from prefect.client.orchestration import PrefectClient

async def pause_schedule(deployment_id: str, schedule_id: str):
    async with PrefectClient() as client:
        await client.update_deployment_schedule(
            deployment_id=UUID(deployment_id),
            schedule_id=UUID(schedule_id),
            active=False,  # False = paused, True = resumed
        )

asyncio.run(pause_schedule("<DEPLOYMENT_ID>", "<SCHEDULE_ID>"))

Synchronous version

from uuid import UUID
from prefect.client.orchestration import SyncPrefectClient

def pause_schedule_sync(deployment_id: str, schedule_id: str):
    with SyncPrefectClient() as client:
        client.update_deployment_schedule(
            deployment_id=UUID(deployment_id),
            schedule_id=UUID(schedule_id),
            active=False,
        )

List schedules and pause by slug/name

from uuid import UUID
from prefect.client.orchestration import SyncPrefectClient

def pause_by_slug(deployment_id: str, slug: str):
    with SyncPrefectClient() as client:
        dep_id = UUID(deployment_id)
        schedules = client.read_deployment_schedules(dep_id)
        for s in schedules:
            if s.slug == slug:
                client.update_deployment_schedule(
                    deployment_id=dep_id,
                    schedule_id=s.id,
                    active=False,
                )
                print(f"Paused schedule: {slug}")

Pause all schedules on a deployment

from uuid import UUID
from prefect.client.orchestration import SyncPrefectClient

def pause_all_schedules(deployment_id: str):
    with SyncPrefectClient() as client:
        dep_id = UUID(deployment_id)
        schedules = client.read_deployment_schedules(dep_id)
        for s in schedules:
            client.update_deployment_schedule(
                deployment_id=dep_id,
                schedule_id=s.id,
                active=False,
            )

Create a deployment with a schedule already paused

from prefect import flow
from prefect.client.schemas.actions import DeploymentScheduleCreate
from prefect.client.schemas.schedules import CronSchedule

@flow
def my_flow():
    ...

my_flow.deploy(
    name="my-deployment",
    work_pool_name="default",
    schedules=[
        DeploymentScheduleCreate(
            schedule=CronSchedule(cron="0 9 * * *", timezone="UTC"),
            active=False,  # start paused
            slug="morning"
        )
    ],
)

Do you want to pause all schedules for a deployment or just a specific one? If you share the deployment name/ID (and schedule slug if applicable), I can tailor a snippet exactly to your case.

Short answer: because the task hit a cache before your code ran. In Prefect 3.x, caching is checked and applied before the task function executes. If there’s a cache hit (from a prior successful run with the same cache key), Prefect will: - Skip executing your task code entirely (your

try/except

never runs) - Return the previously persisted result - Mark the state as Completed with the name “Cached” (i.e., “Completed (Cached)”) That’s why your

return states.Failed(...)

in the

except

doesn’t change the outcome when the result is coming from cache—the exception block is never reached. How to make your failure logic take effect: - Prefer raising an exception to fail a task - And/or bypass/disable caching so your code actually runs Options: - Raise instead of returning a State (recommended) - Disable caching for this task - Force a refresh to ignore cache on this run - Change the cache key (e.g., add an input that varies) Examples: 1) Raise to fail, and disable caching

from prefect import task, get_run_logger
from prefect.cache_policies import NO_CACHE

@task(cache_policy=NO_CACHE)  # or use persist_result=False
def rsync_task(source, destination, credentials, subfolder, task_name):
    logger = get_run_logger()
    try:
        cmd = (
            f"rsync -avz {source} {credentials['user']}@{credentials['host']}:{destination} && "
            f"rsync -rlvz {source} {credentials['user']}@{credentials['host']}:{destination}"
        )
        fabric_run(cmd)
        <http://logger.info|logger.info>(f"Folder {subfolder} rsync completed successfully")
    except Exception as e:
        logger.error(f"Folder {subfolder} rsync failed: {e}")
        raise  # fail the task

2) If you really want to return a Failed state, be sure to bypass cache so the code runs:

from prefect import task
from prefect.cache_policies import NO_CACHE
from prefect import states

@task(cache_policy=NO_CACHE)  # or @task(refresh_cache=True)
def my_task():
    try:
        ...
    except Exception as e:
        return states.Failed(message=str(e))

Notes: - Prefect only serves cached results from prior Completed runs. Failed states are not used as cache hits. - In your snippet,

return ...; raise

means the

raise

is unreachable—pick one. If you want failure, use

raise

. If helpful, here’s more background on results and caching in Prefect 3: - Results and result storage: Results and caching (Prefect docs) - Task states overview: States (Prefect docs) If you can share your task decorator settings (

persist_result

,

cache_policy

,

refresh_cache

) or confirm you’re seeing “Completed (Cached)” in the UI, I can suggest the minimal change for your setup.