Short answer: Set deployed child subflows to Cancelling, and ad‑hoc child subflows (no deployment) to Cancelled. Why: In Prefect 2.19.3, Cancelling signals the server/worker to stop the run and clean up infrastructure; it will then transition to Cancelled. If you try to set a RUNNING deployed flow directly to Cancelled with

force=False

, the transition is usually rejected by orchestration rules. For non‑deployed subflows (no infra), you can set Cancelled directly. Here’s a small tweak to your hook to follow that pattern and fix paging while we’re at it (incrementing

offset

by 1 will skip/duplicate results; use a

limit

and advance by

len(runs)

):

from prefect import get_client
from prefect.logging import get_logger
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterState
from prefect.states import Cancelling, Cancelled

async def cancel_subflows(flow, flow_run, state):
    logger = get_logger()
    <http://logger.info|logger.info>(f"Flow run information\nFlow: {flow}\nFlowRun: {flow_run}\nState: {state}")

    async with get_client() as client:
        offset = 0
        limit = 200
        while True:
            runs = await client.read_flow_runs(
                flow_run_filter=FlowRunFilter(
                    parent_flow_run_id=dict(any_=[flow_run.id]),
                    state=FlowRunFilterState(
                        name=dict(
                            any_=[
                                "Pending",
                                "Scheduled",
                                "Running",
                                "Paused",
                                "AwaitingRetry",
                                "Retrying",
                                "Late",
                                "Suspended",
                                "Cancelling",
                            ]
                        )
                    ),
                ),
                offset=offset,
                limit=limit,
            )
            if not runs:
                break

            for run in runs:
                # Deployed subflows -> Cancelling (let infra cleanup happen)
                # Ad-hoc subflows (no deployment) -> Cancelled directly
                target_state = (
                    Cancelling(message="Parent flow was cancelled")
                    if getattr(run, "deployment_id", None)
                    else Cancelled(message="Parent flow was cancelled")
                )
                await client.set_flow_run_state(run.id, target_state, force=False)

            offset += len(runs)

Notes: - If you set Cancelled on a running deployed subflow with

force=False

, it’ll typically be rejected; Cancelling is the correct signal. After cleanup, it becomes Cancelled automatically. - Tasks don’t have deployments; if you cancel tasks explicitly, set them directly to Cancelled. - You can inspect the returned

OrchestrationResult

from

set_flow_run_state

to log accept/reject outcomes if you want extra visibility. References (source code for 2.19.3): - Cancellation service logic for child flows/tasks: cancellation_cleanup.py - State classes (Cancelled, Cancelling): states.py - Orchestration client `set_flow_run_state`: orchestration.py - Orchestration rules (cancelling -> cancelled): core.py If you share how these subflows are launched (deployed vs dynamic/ad‑hoc), I can tailor the filter/state-setting logic even more precisely.

Got it — if a flow enters Cancelling and never reaches Cancelled in 2.19.3, it usually means the worker (or infra) never completed the kill and/or never marked the run as Cancelled. A few targeted checks and likely fixes: First, a quick clarifier - Is it the parent flow stuck in Cancelling, the subflows, or both? - Are the stuck subflows deployed (have a

deployment_id

) or ad‑hoc (no deployment)? Why this happens in 2.19.3 - CANCELLING → CANCELLED is not automatic; a worker must perform the kill and then set Cancelled. Orchestration rules block any other transition from CANCELLING. - Rules: core.py (EnforceCancellingToCancelledTransition) - States: states.py - If a subflow has no deployment/infrastructure (ad‑hoc) and you set it to Cancelling, there may be no worker to finish cancellation — it can sit in Cancelling forever. Those should be set directly to Cancelled. Most common root causes and quick checks 1) Ad‑hoc subflows set to Cancelling - Symptom: No

deployment_id

, no

infrastructure_pid

, stays Cancelling. - Fix: Set those directly to

Cancelled()

(not

Cancelling()

). - Example patch to your hook:

target_state = (
      Cancelling(message="Parent flow was cancelled")
      if getattr(run, "deployment_id", None)
      else Cancelled(message="Parent flow was cancelled")
  )
  await client.set_flow_run_state(run.id, target_state, force=False)

- Source showing this pattern: cancellation_cleanup.py 2) No worker online for that work pool/queue - Check a worker is running for the same work pool as the stuck run: - Start/confirm:

prefect worker start --pool <pool_name> --log-level DEBUG

- Look for logs like “Found X flow runs awaiting cancellation” and “Killing …” - CLI help (verified):

prefect worker start --help

3) Missing or unusable infrastructure PID - If

infrastructure_pid

is absent or invalid, the worker can’t perform the kill. - Inspect one stuck run:

async with get_client() as c:
      fr = await c.read_flow_run("<flow_run_id>")
      print("deployment_id:", fr.deployment_id)
      print("infra_pid:", fr.infrastructure_pid)
      print("work_pool:", fr.work_pool_name, "queue:", fr.work_queue_name)
      print("state:", fr.state.type, fr.state.name, fr.state.message)

- If ad‑hoc (no deployment_id) and no infra_pid, set directly to

Cancelled()

. 4) Infra kill cannot complete (permissions or signals) - Process: worker must run on same host; ensure the PID exists and isn’t ignoring SIGTERM. - Docker: worker needs permission to stop containers; check Docker socket access. - Kubernetes: ensure RBAC can delete/stop pods; terminationGracePeriodSeconds/preStop hooks can delay termination. - ECS: worker needs IAM permission to StopTask; check CloudWatch logs. 5) Worker/pool mismatch or offline heartbeat - Ensure the stuck flow’s

work_pool_name

matches a running worker. - If worker is marked OFFLINE (missed heartbeat), it won’t process cancellations; restart it. Remediations - If these are ad‑hoc subflows: update your hook to set them to

Cancelled()

(as above). For already-stuck runs, you can unstick them:

async with get_client() as c:
      # ONLY if you’ve verified the infra is stopped or there was never any infra (ad-hoc)
      await c.set_flow_run_state("<flow_run_id>", Cancelled(message="Manual finalize"), force=True)

- If they’re deployed subflows: make sure a worker for that work pool is running and can reach/stop the infrastructure. Once the worker completes

kill_infrastructure

, it will set the run to Cancelled. Handy references - State rules: core.py

- States: states.py - Cancellation cleanup service behavior: cancellation_cleanup.py - Orchestration client `set_flow_run_state`: orchestration.py If you can share one stuck flow_run_id (and whether it’s deployed or ad‑hoc), I can pinpoint which of the above is the culprit and suggest the exact fix.

Thanks for the detail — I hear you. In 2.19.3, deployed subflows should reliably move from Cancelling to Cancelled, but that only happens when the worker for the subflow’s work pool successfully kills the infra and then marks the run as Cancelled. If anything in that chain fails, the run can sit in Cancelling. Since your subflows do have a deployment_id and you have a worker running, here’s a focused checklist that usually surfaces the culprit: 1) Confirm the right worker is doing the cancelling - Each subflow’s deployment must have an ONLINE worker in the same work pool. It’s common for the parent’s pool to differ from the subflows’ pools. - Quick check you can run to print a stuck subflow’s essentials (infra_pid, pool, queue, state message):

from prefect import get_client

async def inspect_run(run_id: str):
    async with get_client() as c:
        fr = await c.read_flow_run(run_id)
        print("flow_run_id:", fr.id)
        print("deployment_id:", fr.deployment_id)
        print("work_pool:", fr.work_pool_name, "queue:", fr.work_queue_name)
        print("infrastructure_pid:", fr.infrastructure_pid)
        print("state:", fr.state.type, fr.state.name, fr.state.message)

2) Check worker logs at cancellation time - Start the worker with DEBUG to see cancellation steps:

prefect worker start --pool <pool_name> --log-level DEBUG

- You should see messages like: - “Found X flow runs awaiting cancellation” - “Killing <infra> <pid> for flow run …” - “Cancelled flow run …” - If you see exceptions in

kill_infrastructure

or “no infrastructure pid”, the worker will never mark the run as Cancelled, and the run will stay in Cancelling. 3) Verify infrastructure_pid is present and valid - For deployed subflows, the worker sets

infrastructure_pid

when it submits the run (e.g., container ID, pod name, ECS task ARN, or host:pid). - If it’s missing or wrong, the worker can’t kill anything, and the run won’t be finalized. 4) Infra-specific gotchas to double-check - Kubernetes worker: - Ensure the worker’s service account can delete/stop pods (RBAC). If it can’t, you’ll see permission errors in worker logs and runs remain Cancelling. - Confirm pods actually terminate on cancellation:

kubectl get pods

and

kubectl describe pod <name>

(look at termination/Finalizers). - Docker worker: - Ensure the worker can talk to the Docker daemon (DOCKER_HOST/socket perms). If it can’t stop containers, runs won’t finalize. - ECS worker: - Worker IAM must allow

ecs:StopTask

on the correct cluster/ARN. Permission errors here leave runs in Cancelling. 5) Parent cancels ≠ subflow finalized unless the subflow’s own worker acts - The server’s cleanup service sets subflows to Cancelling when parents cancel, but it’s the subflow’s worker that must perform the kill and set Cancelled. If no worker for that subflow’s pool is online (or it lacks permission), the subflow will sit in Cancelling. 6) Temporary mitigation when you’ve verified infra is dead - If infra has already stopped (or never came up) and the run is just stuck in Cancelling, you can unstick it:

from prefect import get_client
from prefect.states import Cancelled

async def force_finalize(run_id: str):
    async with get_client() as c:
        await c.set_flow_run_state(run_id, Cancelled(message="Manual finalize after stuck Cancelling"), force=True)

- I recommend only doing this once you’re sure infra is gone; otherwise you can create orphans. Relevant sources - Orchestration rules (Cancelling → only → Cancelled): core.py - State classes: states.py - Server cleanup behavior for subflows/tasks: cancellation_cleanup.py

- Orchestration client `set_flow_run_state`: orchestration.py - Worker cancellation path (kill infra then mark Cancelled): workers/base.py - Worker start (help verified):

prefect worker start --help

If you can share: - One stuck subflow’s flow_run_id, - Its work pool type (Kubernetes/Docker/ECS/Process) and pool name, - A short snippet of worker logs around “cancellation” for that pool, …I can pinpoint whether it’s a permission/infra_pid/worker-pool mismatch or the worker’s kill step failing, and give you an exact fix.

Great question. Short answer for Prefect 2.x: run fewer workers per machine and control concurrency with limits. Crashes on long-running batch jobs are most often memory-related (OOM) rather than a need for more workers. Practical recommendations - Prefer 1 worker per machine and set a per‑worker concurrency limit with

--limit

. Running 7 workers on the same host tends to compete for RAM/CPU and can make OOM issues more likely. - Set a work pool concurrency limit to cap total concurrent runs across all workers pulling from that pool. - Set explicit memory limits at the infrastructure level (Docker/Kubernetes) for each flow run. For process workers on a VM, estimate memory per run and cap concurrency accordingly. Good starting points by worker type - Process worker (VM/bare metal): - 1 worker per machine; start with

--limit 1-2

for heavy, long-running batch jobs. - If jobs are lighter and memory allows, increase to

--limit 3-5

. Use the formula below. - Docker worker: - 1 worker per machine; start with

--limit 2-5

. - Set container memory limits per run (e.g., 4–8 GiB depending on your job). - Kubernetes worker: - 1–3 workers per cluster are usually enough; K8s does the heavy lifting. - Higher

--limit

is fine (e.g., 10–30), but set pod

resources.requests/limits

appropriately. Capacity planning quick formula - Headroom = total_machine_RAM × 0.7 (leave ~30% for OS/overhead) - Max concurrent runs per machine ≈ floor(Headroom / peak_mem_per_run) - Example: 32 GiB box, jobs peak at ~6 GiB -> floor(22.4 / 6) = 3 -> set a single worker with

--limit 3

Why the crashes on long-running jobs? - Most likely OOM: - Linux/Docker: container exits 137 (SIGKILL). - K8s: pod shows OOMKilled. - Process worker: subprocess return code -9/-15 in logs. - Contributing factors: - Too many concurrent runs (multiple workers or high

--limit

). - No infra memory limits (Docker/K8s). - In-flow parallelism creating multiple large objects in memory. - In-memory accumulation (e.g., large dataframes/lists kept around). What to change right now 1) Consolidate workers and cap concurrency - Replace many workers with one per machine and set a limit:

prefect worker start --pool "your-pool" --type process --limit 2

- Cap total pool concurrency (so multiple machines don’t overwhelm infra):

prefect work-pool set-concurrency-limit "your-pool" 6

2) Set memory limits per run - Docker worker: add a hard memory limit in deployment job variables:

# in your flow.deploy(...) or deployment config:
job_variables:
  mem_limit: "6g"
  # optionally:
  memswap_limit: "8g"

- Kubernetes worker: set pod resources in your job template:

resources:
  requests:
    memory: "4Gi"
    cpu: "2"
  limits:
    memory: "6Gi"
    cpu: "2"

3) Reduce per-run memory pressure - If your flow uses parallel tasks and memory spikes, consider a more conservative task runner for that flow:

from prefect import flow
from prefect.task_runners import SequentialTaskRunner

@flow(task_runner=SequentialTaskRunner())
def long_running_flow():
    ...

- Persist intermediate results to disk/object storage instead of holding large objects in memory. - Free references to large objects once done. 4) Observe and iterate - Check whether failures are worker crashes vs. flow run OOM: - Docker: look for exit code 137 in container logs. - K8s: check pod status “OOMKilled.” - Process: worker logs will show subprocess exit codes (-9/-15). - Increase/decrease

--limit

and memory per run based on observed peak usage. A couple of CLI references we verified - Start a worker:

prefect worker start --help

- Set a pool-wide concurrency cap:

prefect work-pool set-concurrency-limit "my-pool" 10

A few clarifying questions to tailor this precisely - Which worker type are you using (process, docker, or kubernetes)? - Are the 7 workers on one host or spread across machines? - What’s the machine/cluster size (CPU/RAM)?

- When a “worker crashes,” do you see OOMKilled (K8s), exit 137 (Docker), or a worker process dying with SIGKILL (-9) in logs? - Roughly how much memory does a single long-running batch job use at peak? If you can share those details (and a snippet of the crash log), I can give you a concrete configuration—exact

--limit

values and memory limits per run for your setup.