Great question. In Prefect 3.x, when a flow is cancelled, your running tasks will receive a cancellation and you’re responsible for cleaning up any external resources inside the task. Here’s how to do it safely and predictably. Key points - Cancellation raises asyncio.CancelledError in user code (it’s a BaseException). Generic except Exception will NOT catch it. - Use try/finally for guaranteed cleanup; re-raise the cancellation after cleanup. - Prefer context managers for resources (files, DB conns, clients). - For async tasks, you can shield critical cleanup with anyio.CancelScope to ensure it completes. - Tasks do not have on_cancellation hooks; flows do. Use flow hooks for higher-level cleanup. - Cancelled tasks will not retry; concurrency slots are released by Prefect automatically. Async task patterns 1) Try/finally + explicit CancelledError handling

from asyncio import CancelledError
from prefect import task

@task
async def load_and_process_async():
    resource = None
    try:
        resource = await acquire_async_resource()
        return await use_resource(resource)
    except CancelledError:
        # Clean up promptly, then re-raise to propagate cancellation
        if resource:
            await cleanup_resource(resource)
        raise
    finally:
        # Safety net for any resource that should always be cleaned up
        if resource:
            await ensure_closed(resource)

2) Async context manager (preferred when available)

from asyncio import CancelledError
from prefect import task

@task
async def with_managed_resource():
    try:
        async with managed_resource() as r:
            return await do_work(r)
    except CancelledError:
        # Optionally perform extra cleanup, then re-raise
        await extra_cleanup_if_needed()
        raise

3) Shield critical cleanup so it cannot be cancelled mid-cleanup

import anyio
from asyncio import CancelledError
from prefect import task

@task
async def do_work_with_critical_cleanup():
    try:
        await long_running_step()
    except CancelledError:
        # Ensure critical cleanup completes even under cancellation
        with anyio.CancelScope(shield=True):
            await flush_to_storage()
            await release_lock()
        raise

Sync task patterns - Prefect raises asyncio.CancelledError in sync tasks too; catch it explicitly or rely on finally blocks. - Blocking I/O in worker threads may not be interrupted immediately; cleanup will run once control returns to Python.

from asyncio import CancelledError
from prefect import task

@task
def process_file_sync(path: str):
    f = None
    try:
        f = open(path, "rb")
        do_cpu_or_io_work(f)
    except CancelledError:
        # Clean up then re-raise cancellation
        if f:
            try: f.close()
            except Exception: pass
        raise
    finally:
        if f:
            try: f.close()
            except Exception: pass

Flow-level cleanup on cancellation (for orchestration-wide actions) - Tasks do not support on_cancellation hooks; flows do. Use them for coordinated cleanup (e.g., revoke external jobs, notify systems).

from prefect import flow

@flow
def my_flow():
    # your flow logic
    ...

@my_flow.on_cancellation
def flow_cancel_cleanup(flow, flow_run, state):
    # Flow-level cleanup, e.g., revoke external run, delete temp buckets, notify
    revoke_external_job(flow_run.id)

Behavior notes and FAQs - Exception type: asyncio.CancelledError (a BaseException). Always catch it explicitly and re-raise after cleanup. - Retries: Cancelling a task or flow will not trigger task retries. If you need retries for non-cancel scenarios, keep them on the task as usual—cancellation short-circuits them. - Concurrency limits: Prefect releases its own concurrency tokens/slots even on cancellation. You still must clean up your external resources (locks, temp files, sessions). - Async vs. sync:

- Async: Cancellation is observed at await points; finally blocks run reliably. - Sync: In worker threads, blocking calls may not be interrupted immediately; finally blocks still run once control returns. References - Workflow cancellation: Cancel workflows (Prefect 3.x) - States: States reference - State change hooks (flow on_cancellation): State change hooks If you share a quick snippet of what your task is doing (e.g., DB, files, remote jobs), I can tailor a cleanup pattern for that resource.