Great questions — here’s how Prefect 3.x handles crash and cancellation by default. Summary - Crashed - A flow or task run is marked Crashed when the executing process dies unexpectedly or stops heartbeating for a period of time (“zombie” detection). - One crashed task or subflow does not automatically crash the parent flow. The parent only becomes Crashed if its own execution process/heartbeat is lost. - Details: Detect zombie flows - Cancelled - A run becomes Cancelled when cancellation is explicitly requested (UI/API/automation) and the run acknowledges it. You’ll see a transient Cancelling state first. - A cancelled task or subflow does not automatically cancel the parent flow. The parent may continue unless your code awaits that result and propagates the cancellation. - Cancellation propagates top-down: cancelling the flow will cancel outstanding tasks/background tasks/subflows. - Details: Cancel workflows, States - Scheduled/Pending auto-cancel - Prefect does not auto-cancel Scheduled or Pending runs by default. - Runs that miss their scheduled start can be marked Late by a service after a configurable threshold (see env var like

PREFECT_API_SERVICES_LATE_RUNS_AFTER_SECONDS

), but they are not auto-cancelled unless you add an Automation to do so. - If a run starts then disappears (no heartbeat), it can be detected as a zombie and marked Crashed, but there’s no “cancel after N minutes in Scheduled/Pending” default. - Details: Detect zombie flows, States A bit more color for your background-task usage - When you cancel a flow run (UI/API), Prefect will propagate cancellation to all running tasks, including background tasks started via

.delay()

. Calls to

.result_async()

will raise cancellation (e.g., `CancelledError`/`CancelledRun`) so you can handle/cleanup. - If one background task crashes/fails/cancels, your flow will only fail/cancel if you await its result and let that exception propagate; otherwise you can handle it and continue. - Pattern with asyncio.gather:

# fan out
  futures = [
      my_task.with_options(name=f"job-{i}").delay(i)
      for i in items
  ]

  try:
      results = await asyncio.gather(
          *(f.result_async() for f in futures),
          return_exceptions=True,   # important: collect per-task outcomes
      )
      for r in results:
          # handle per-task outcomes however you prefer
          # - Cancelled: treat as skipped/cancelled
          # - Exception: handle failure
          # - Normal value: success
          ...
  except Exception:
      # If you want to let one failure cancel the whole flow, re-raise here.
      raise

Common follow-ups you might consider - If you want “auto-cancel if not started within X minutes”, create an Automation that triggers on Late (or still Scheduled after N minutes) and transitions the run to Cancelled. - If you want “auto-cancel if a task takes longer than X”, set a

timeout_seconds

on the task (via decorator or

.with_options(timeout_seconds=...)

) so Prefect cancels it. Docs for reference - States: States - Cancellation: Cancel workflows - Zombie/crash detection: Detect zombie flows If you can share whether you’re on Cloud vs OSS and whether you want time-based auto-cancellation, I can suggest a concrete Automation or configuration that matches your fan-out pattern.