Great question. Assuming you’re on Prefect 3.x and using the Kubernetes worker via a work pool, here’s how to ensure a flow is retried when its Kubernetes Job fails or crashes: What to configure - Prefer Prefect flow-level retries (recommended) - Add retries on the flow decorator. Prefect will schedule a new attempt when the run ends in Failed or Crashed (e.g., pod OOM, eviction, node preemption, ImagePullBackOff, etc.). - Example:

from prefect import flow

    @flow(retries=3, retry_delay_seconds=60)
    def my_flow():
        ...

- This gives you clear visibility in Prefect of each attempt and avoids hidden restarts at the K8s layer. - Avoid relying on Kubernetes Job backoff for orchestration - Kubernetes Jobs have

spec.backoffLimit

which restarts pods automatically. That happens outside Prefect’s orchestration and can lead to confusing visibility and duplicate side effects. - If you want Prefect to control retries, set a low backoff (often 0 or 1) in your work pool’s base job template:

apiVersion: batch/v1
    kind: Job
    spec:
      backoffLimit: 0
      template:
        spec:
          restartPolicy: Never
          containers:
            - name: flow
              ...

- Let Prefect’s

retries

re-queue a new attempt instead. - Keep your Kubernetes worker running - The worker must be up to pick up the re-scheduled retry attempts. If it’s down, retries will queue until the worker is available. Optional: cover edge cases or add a safety net - Automations: If you want to re-run the deployment when a run ends in Crashed/Failed and you didn’t set

retries

, create an Automation that triggers on those states and runs the deployment again. You can do this in the UI. - Task-level retries: If certain tasks are flaky and idempotent, you can also add retries at the task level for finer control. Notes on behavior - With flow-level

retries

set, Prefect will automatically retry on both failure and crash states produced by the Kubernetes execution (e.g., pod eviction, container exit non-zero, etc.). - K8s job backoff and Prefect retries can both trigger re-execution. To avoid double-retries and improve observability, choose one (preferably Prefect). If you can share: - Which failures you’re seeing (e.g., ImagePullBackOff vs. container exit code vs. pod eviction) - Your current work pool job template’s

backoffLimit

…we can tailor the exact settings. Helpful docs - Prefect Kubernetes worker collection: prefect-kubernetes docs - Prefect retries overview: Retries guide - Work pools and workers: Work pools and workers

Short answer: - Cancelling a run will not trigger Prefect’s built-in flow retries. Retries are invoked when a run ends in Failed or Crashed (and the flow has

retries

configured). -

ttlSecondsAfterFinished

only cleans up Jobs after they’ve finished (Complete/Failed). It has no effect on running pods or Prefect’s retry behavior. How to make your “zombie” protection actually retry - Best option: add flow-level retries and timeouts so Prefect makes the decision for you.

from prefect import flow

  @flow(retries=3, retry_delay_seconds=300, timeout_seconds=3600)  # 1h timeout
  def my_flow():
      ...

- If the pod hangs past 1h, the flow raises a timeout (Failed) and Prefect will schedule a retry. - If the pod crashes (OOM/evicted/etc.), Prefect will mark the run Crashed and retry as well (if retries remain). - If you want to use an Automation to deal with zombies: - Don’t set state to Cancelled if you expect built-in retries to kick in—they won’t. - Either: - Use the “Change flow run state” action to mark the run as Crashed or Failed (which will trigger flow retries if configured), or - Use a separate Automation triggered on Cancelled to “Run deployment” (this starts a fresh run since Cancelled will not auto-retry). Kubernetes settings - You’re correct to keep

backoffLimit: 0

to avoid Kubernetes doing its own hidden retries. -

ttlSecondsAfterFinished

does not impact retries or running pods—set it to your desired cleanup horizon; it only affects completed Jobs. - Consider adding an execution time cap at the Kubernetes layer as a safety net:

apiVersion: batch/v1
  kind: Job
  spec:
    backoffLimit: 0
    activeDeadlineSeconds: 3600   # kill the pod after 1h if still running
    ttlSecondsAfterFinished: 600  # clean up finished jobs after 10m (optional)
    template:
      spec:
        restartPolicy: Never
        containers:
          - name: flow
            ...

If

activeDeadlineSeconds

kills the pod, Prefect will see the run as Crashed and retry (again, only if the flow has retries configured). Recommendations for your case - Add

retries

and a sensible

timeout_seconds

on the flow. - Keep

backoffLimit: 0

. - Optionally add

activeDeadlineSeconds

to catch true zombies at the cluster level. - If you stick with an Automation, use “Change flow run state” to Crashed/Failed (or add a second Automation to re-run the deployment on Cancelled). Helpful docs - Retries: Retries guide - Automations: Automations overview - Kubernetes worker/collection: prefect-kubernetes docs If you share your current flow decorator (do you already have

retries

set?) and your job template, I can suggest exact values for

timeout_seconds

and

activeDeadlineSeconds

to match your SLA.

Thanks for sharing the code — you’re very close. A few important notes and tweaks to make this zombie reaper reliable and to ensure it triggers flow retries: What’s good - Proactive posture, per–flow-run scoping via

for_each={"prefect.resource.id"}

, and using

ChangeFlowRunState(..., state=StateType.CRASHED)

are all correct. - Keeping

backoffLimit: 0

on the Job is correct.

ttlSecondsAfterFinished

does not affect retries or running pods; it only garbage-collects finished Jobs. Key improvements to your automation 1) Start the proactive window on both Running and heartbeat - Your current

after={"prefect.flow-run.heartbeat"}

pattern will catch runs that stop heartbeating after they’ve started heartbeating at least once. - It will not catch runs that enter Running but never send the first heartbeat (e.g., the process dies early). - Fix: include both events in

after

. 2) Expect all “healthy or terminal” events, not just heartbeats - You included terminal states, which is good. I recommend also including Suspended, TimedOut, and Retrying to avoid false positives when a run legitimately leaves Running and pauses/retries. 3) Make sure the flow has retries if you want an automatic re-run - Marking a run as CRASHED won’t re-run it unless the flow has

retries

configured. If you don’t want to add

retries

at the flow, add a second Automation that reacts to Crashed and “Run deployment” to start a new run. A tightened version of your automation

from datetime import timedelta
from prefect import get_client
from prefect.automations import AutomationCore
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.events.schemas.events import ResourceSpecification
from prefect.client.schemas.objects import StateType
from prefect.events.actions import ChangeFlowRunState

async def zombie_reaper() -> None:
    automation = AutomationCore(
        name="Crash zombie flows",
        description="Crash any flow run that stops heartbeating for more than 90 seconds.",
        enabled=True,
        trigger=EventTrigger(
            # Start the clock when a run starts OR we receive any heartbeat
            after={"prefect.flow-run.Running", "prefect.flow-run.heartbeat"},
            # Any of these events “satisfy” the expectation; if none occurs within 90s, trigger
            expect={
                "prefect.flow-run.heartbeat",   # subsequent heartbeats
                "prefect.flow-run.Completed",
                "prefect.flow-run.Failed",
                "prefect.flow-run.Cancelled",
                "prefect.flow-run.Crashed",
                "prefect.flow-run.Suspended",   # avoid false positives for intentional suspension
                "prefect.flow-run.TimedOut",    # explicit timeout
                "prefect.flow-run.Retrying",    # flow left Running to retry
            },
            match=ResourceSpecification({"prefect.resource.id": ["prefect.flow-run.*"]}),
            for_each={"prefect.resource.id"},
            posture=Posture.Proactive,
            threshold=1,
            within=timedelta(seconds=90),
        ),
        actions=[
            ChangeFlowRunState(
                name="zombie crash",
                state=StateType.CRASHED,
                message="Flow run marked as crashed due to missing heartbeats.",
            )
        ],
    )

    async with get_client() as client:
        existing = await client.read_automations_by_name(name=automation.name)
        if existing:
            for current in existing:
                await client.update_automation(current.id, automation)
            print(f"[prefect] Updated {len(existing)} automation(s) named {automation.name}.")
            return

        automation_id = await client.create_automation(automation)
        print(f"[prefect] Created automation {automation.name} with id {automation_id}")

Additional recommendations

- Add a flow-level timeout and retries to make this robust without depending exclusively on automations:

from prefect import flow

  @flow(timeout_seconds=3600, retries=3, retry_delay_seconds=300)
  def my_flow():
      ...

- If the pod hangs > 1h, Prefect will mark it Failed/TimedOut and retry. - Consider adding

activeDeadlineSeconds

on the Kubernetes Job as a hard kill for true zombies. This forces a Crashed state (and Prefect will retry if retries remain):

spec:
    backoffLimit: 0
    activeDeadlineSeconds: 3600
    ttlSecondsAfterFinished: 600  # optional cleanup; doesn’t affect retries
    template:
      spec:
        restartPolicy: Never

Reminders - Cancelled runs do not auto-retry. Use CRASHED/FAILED (with flow retries) or a second automation that reacts to Cancelled and “Run deployment”. - If you’re self-hosting Prefect Server, ensure the Automations service is running; proactive automations require the services stack. In Cloud, this is handled for you. Docs for reference - Automations: https://docs.prefect.io/latest/concepts/automations/ - Events: https://docs.prefect.io/latest/concepts/events/ - Retries: https://docs.prefect.io/latest/guides/retries/ - Prefect Kubernetes collection: https://prefect-python-sdk.github.io/prefect-kubernetes/ If you share whether you’re on Cloud or self-hosted (and your Prefect version), I can sanity-check that the automations evaluator is active and suggest heartbeat/timeout values aligned with your workload.