What you’re describing does sound like a real worker/observer failure mode, not just a sizing issue. Short version: - The root operational trigger is the large burst of K8s Job failures caused by pending backlog / ICE /

activeDeadlineSeconds

. - But the bad Prefect behavior is that the Kubernetes Jobs observer appears to die when crash-marking a run hits

Abort: already in terminal state ...

, which then leaves later failed Jobs unobserved and flow runs stuck

Running

. - The

httpx.PoolTimeout

you’re seeing is also consistent with observer/API pressure during that burst and likely makes the recovery path worse. I couldn’t fully inspect source / issues from tools right now because GitHub access and the deeper research agent both failed in this session, so I don’t want to overclaim a specific fixed version. But based on the symptoms, I’d recommend treating this as two problems: burst control and observer robustness. Recommendations 1. Upgrade off 3.6.22 if you can You’re on an older 3.x patch line. Even though I couldn’t confirm the exact PR/release for this watcher behavior, moving to the latest 3.x patch is the first thing I’d try. Current latest release notes available to me show active bugfix work continuing in 3.7.x: Prefect releases If you’re self-hosting workers plus server, I’d upgrade both Prefect core and any `prefect-kubernetes`/worker image components together. 2. Reduce the fan-out burst at the orchestration layer

max_concurrent=1000

against scarce GPU capacity plus EC2 ICE is effectively creating a synchronized failure wave. Even if Prefect handled terminal-state races perfectly, that’s still an ugly load shape for: - K8s API - Prefect API - observer reconciliation - worker HTTP client pools I’d strongly consider: - lowering

max_concurrent

- adding queue/work-pool concurrency limits - batching subflow submission - staggering launches instead of a single 1000-run burst In practice, keeping the number of simultaneously submitted infrastructure runs closer to actual schedulable GPU capacity + some headroom is usually much more stable than letting 1000 sit Pending. 3. Revisit

activeDeadlineSeconds=3600

If jobs can sit Pending for a long time before ever getting a GPU node,

activeDeadlineSeconds

can create exactly the synchronized

DeadlineExceeded

burst you’re seeing. If the deadline is meant to bound runtime rather than queue wait, consider: - increasing it substantially, or - handling timeout inside the flow code instead of only with K8s Job deadline, or - using a submission pattern that avoids long Pending queues in the first place 4. Treat observer restartability as critical Since the watcher dying is what creates zombie

Running

UI state, make sure the worker pods: - auto-restart on failure - have liveness/readiness probes that catch a dead/stuck observer process if possible - emit logs/alerts specifically on kopf watcher exits Even if the underlying bug remains, rapid restart reduces the zombie window. 5. Investigate / tune API client pressure Your

httpx.PoolTimeout

with pool max 16 strongly suggests the worker/observer is saturating its API connection pool during the failure burst. I wasn’t able to confirm an exposed Prefect setting for that pool size from tools here, so I don’t want to invent one. But operationally: - reduce concurrent state-update pressure first - check whether you’ve customized worker polling / prefetch aggressively - inspect server latency / DB saturation during the burst - if you have ingress / LB connection bottlenecks, check those too The pool timeout may be a symptom, but it can absolutely amplify the watcher failure. 6. Backfill zombie runs For runs already stuck

Running

after the Job is gone, you’ll likely need a cleanup process:

- query

Running

flow runs from the affected deployment/work pool and age window - reconcile against K8s Job existence/final status - mark missing/failed ones as

Crashed

or

Failed

If this is recurring, a temporary reconciliation script is worth having until you upgrade/stabilize. What this most likely is This smells like a race where: - K8s observer receives terminal Job events in a burst - multiple worker replicas / observer paths attempt to mark the same flow run terminal - Prefect correctly rejects duplicate terminal transition with

Abort

- but kopf/observer incorrectly treats that exception as fatal for the watch loop That should be handled as an expected/idempotent condition, not as an unrecoverable watcher error. What I’d do next, in order 1. Upgrade from

3.6.22

to latest stable 3.x. 2. Cut

max_concurrent

sharply and/or enforce work-pool concurrency aligned to GPU capacity. 3. Increase or rethink

activeDeadlineSeconds

. 4. Add alerting on worker observer restarts / kopf handler crashes. 5. Add a reconciliation script for orphaned

Running

runs. If you want, I can help with either of these next: - a concrete mitigation plan for EKS + Kubernetes work pool at this scale - a zombie-run reconciler script for Prefect 3.x self-hosted - a GitHub issue writeup template with the exact logs/details maintainers would want If you can paste one stack trace from the kopf watcher death and one

httpx.PoolTimeout

trace, I can help narrow whether this is purely duplicate-terminal-state handling or also a separate client-pool bottleneck.

Thanks — that stack trace is very helpful. This confirms the main bug shape: - the K8s observer handler

prefect_kubernetes.observer._mark_flow_run_as_crashed

- calls

propose_state(...)

- Prefect returns

Abort

because the run is already terminal (

COMPLETED

here) - that exception bubbles out of the handler -

kopf

treats it as unrecoverable and stops the Jobs watcher entirely So yes: this is an observer robustness bug / idempotency bug, not just “too much load”. What that means A duplicate terminal-state race is expected under bursty conditions: - Job event arrives late - another path already marked the run

COMPLETED

or

CRASHED

- observer tries to set

Crashed

- API rejects it with

Abort

That part is fine. The bug is that the observer does not swallow

Abort

as an expected no-op. My recommendation 1. Upgrade first I’d move off

3.6.22

to the latest 3.x available before doing anything else: Prefect releases I can’t confirm from tools here which exact release fixes this specific path, but this is exactly the kind of bug that may have been fixed in later patch lines. 2. Lower concurrency / smooth launch rate immediately Even if you upgrade,

1000

concurrent GPU subflows with frequent Pending/ICE means you’ll keep producing the race window and a lot of API churn. I’d reduce: - deployment concurrency - work-pool / work-queue concurrency - parent flow fan-out rate 3. Increase or remove the synchronized deadline burst

activeDeadlineSeconds=3600

is creating a huge coordinated failure event. If runs can spend most of that time Pending, that deadline is effectively a “mass-fail timer”. 4. Run multiple workers, but assume observer can still fail You already have 3 replicas, which is good for availability, but if each replica runs its own watcher and each can hit the same unhandled

Abort

, that alone doesn’t solve it. Short-term mitigation ideas Best mitigation without code changes: - reduce fan-out - reduce long Pending backlog - avoid synchronized

DeadlineExceeded

- upgrade Prefect If you are comfortable patching your worker image temporarily: wrap the observer crash-marking call so

prefect.exceptions.Abort

is treated as expected and logged at warning/debug, not raised to kopf. Conceptually, the bad section is effectively:

await propose_state(client=client, flow_run_id=flow_run_id, state=Crashed(...))

and the safer behavior would be more like:

try:
    await propose_state(client=client, flow_run_id=flow_run_id, state=Crashed(...))
except prefect.exceptions.Abort:
    # run already terminal; ignore
    return

If you maintain a custom worker image, this is probably the most direct protection until you confirm an upstream fix. Why you’re getting zombie

Running

runs Once kopf stops watching Jobs: - later failed Jobs are no longer reconciled back into Prefect - no one marks the corresponding flow runs terminal - the flow runs remain

Running

in Prefect even though the K8s Job is already

Failed

or deleted That matches your symptom exactly. About the

httpx.PoolTimeout

I think it’s likely secondary but important: - during the failure burst, observers/workers are all trying to update states - the client connection pool saturates - some updates get delayed or dropped - the system becomes more race-prone and less able to recover So I’d treat it as a scaling signal, but the watcher death is still the bigger correctness issue. Practical next steps I’d use - Today: reduce

max_concurrent

substantially - Today: increase

activeDeadlineSeconds

or stop using it as the primary timeout mechanism - Next maintenance window: upgrade to latest 3.x worker/server images - If issue persists after upgrade: patch

prefect_kubernetes.observer

locally to swallow

Abort

- Add cleanup: periodic reconciler for stale

Running

flow runs older than N minutes where the K8s Job no longer exists If you want, I can help with one of these right now: 1. a monkey patch / custom image patch for the Kubernetes observer to ignore

Abort

2. a reconciler script to clean up zombie

Running

runs 3. a fan-out redesign for launching GPU subflows without creating the Pending/deadline storm

Here’s a practical Prefect 3.x reconciler script you can run out-of-band to find stale

Running

flow runs, check whether their corresponding K8s Job still exists, and mark missing/failed ones as

Crashed

. Assumptions - Self-hosted Prefect 3.x - Kubernetes work pool creates Jobs whose names include the flow run ID - You can reach: - Prefect API via

PREFECT_API_URL

- Kubernetes API via in-cluster config or local kubeconfig - You want a safe script with

--dry-run

first What it does - reads

Running

flow runs older than some age - optionally filters by deployment / work pool name - tries to map each flow run to a K8s Job by flow run ID - if Job is: - Failed → mark run

Crashed

- Missing → optionally mark run

Crashed

- Active / Pending / Running → leave alone Because Prefect/K8s naming can vary a bit, I made the K8s lookup use the flow run ID substring in Job name. That’s usually the most robust recovery approach. ``` import argparse import asyncio import os from datetime import datetime, timedelta, timezone from typing import Optional from kubernetes import client as k8s_client from kubernetes import config as k8s_config from kubernetes.client import BatchV1Api from kubernetes.client.rest import ApiException from prefect.client.orchestration import get_client from prefect.client.schemas.filters import ( FlowRunFilter, FlowRunFilterState, FlowRunFilterStateName, FlowRunFilterStartTime, DeploymentFilter, DeploymentFilterName, WorkPoolFilter, WorkPoolFilterName, ) from prefect.client.schemas.sorting import FlowRunSort from prefect.states import Crashed from prefect.exceptions import Abort def utcnow() -> datetime: return datetime.now(timezone.utc) def state_name(state) -> str: if not state: return "UNKNOWN" return getattr(state, "name", None) or getattr(state, "type", None) or "UNKNOWN" def load_kube(): try: k8s_config.load_incluster_config() print("Loaded in-cluster Kubernetes config") except Exception: k8s_config.load_kube_config() print("Loaded local kubeconfig") def job_status_summary(job) -> str: status = job.status if status is None: return "unknown" if getattr(status, "succeeded", 0): return "succeeded" if getattr(status, "failed", 0): return "failed" if getattr(status, "active", 0): return "active" conditions = getattr(status, "conditions", None) or [] for cond in conditions: if cond.type == "Failed" and cond.status == "True": return "failed" if cond.type == "Complete" and cond.status == "True": return "succeeded" return "unknown" def is_job_failed(job) -> bool: status = job.status if status is None: return False if getattr(status, "failed", 0): return True conditions = getattr(status, "conditions", None) or [] for cond in conditions: if cond.type == "Failed" and cond.status == "True": return True return False def is_job_active(job) -> bool: status = job.status if status is None: return False return bool(getattr(status, "active", 0)) def is_job_succeeded(job) -> bool: status = job.status if status is None: return False if getattr(status, "succeeded", 0): return True conditions = getattr(status, "conditions", None) or [] for cond in conditions: if cond.type == "Complete" and cond.status == "True": return True return False def find_jobs_for_flow_run(batch_api: BatchV1Api, namespace: str, flow_run_id: str): jobs = [] cont = None while True: resp = batch_api.list_namespaced_job( namespace=namespace, limit=200, _continue=cont, ) for job in resp.items: name = job.metadata.name or "" if flow_run_id in name: jobs.append(job)

cont = resp.metadata._continue if not cont: break return jobs def newest_job(jobs): if not jobs: return None return sorted( jobs, key=lambda j: j.metadata.creation_timestamp or datetime.min.replace(tzinfo=timezone.utc), reverse=True, )[0] async def reconcile( namespace: str, min_running_minutes: int, limit: int, dry_run: bool, mark_missing_as_crashed: bool, deployment_name: Optional[str], work_pool_name: Optional[str], ): load_kube() batch_api = k8s_client.BatchV1Api() cutoff = utcnow() - timedelta(minutes=min_running_minutes) flow_run_filter = FlowRunFilter( state=FlowRunFilterState( name=FlowRunFilterStateName(any_=["Running"]) ), start_time=FlowRunFilterStartTime(before_=cutoff), ) deployment_filter = None if deployment_name: deployment_filter = DeploymentFilter( name=DeploymentFilterName(any_=[deployment_name]) ) work_pool_filter = None if work_pool_name: work_pool_filter = WorkPoolFilter( name=WorkPoolFilterName(any_=[work_pool_name]) ) async with get_client() as client: flow_runs = await client.read_flow_runs( flow_run_filter=flow_run_filter, deployment_filter=deployment_filter, work_pool_filter=work_pool_filter, sort=FlowRunSort.START_TIME_DESC, limit=limit, ) print(f"Found {len(flow_runs)} candidate running flow runs older than {min_running_minutes} minutes") for flow_run in flow_runs: flow_run_id = str(flow_run.id) run_name = flow_run.name start_time = flow_run.start_time current_state = state_name(flow_run.state) print(f"\nFlow run: {run_name} ({flow_run_id})") print(f" state: {current_state}") print(f" start_time: {start_time}") jobs = find_jobs_for_flow_run(batch_api, namespace, flow_run_id) job = newest_job(jobs) if not job: print(" k8s job: MISSING") if mark_missing_as_crashed: msg = ( f"Kubernetes reconciliation: no Job found in namespace '{namespace}' " f"for running flow run {flow_run_id}. " "This usually indicates the observer missed a terminal event or the Job was removed." ) if dry_run: print(f" DRY RUN: would mark as Crashed: {msg}") else: try: await client.set_flow_run_state( flow_run.id, Crashed(message=msg).to_state_create(), force=True, ) print(" marked as Crashed") except Abort as exc: print(f" skipped: terminal-state race: {exc}") except Exception as exc: print(f" ERROR setting crashed state: {exc}") continue job_name = job.metadata.name summary = job_status_summary(job) print(f" k8s job: {job_name} ({summary})") if is_job_active(job): print(" action: leave alone (job still active)") continue if is_job_succeeded(job): print(" action: leave alone (job succeeded)") continue if is_job_failed(job): conditions = getattr(job.status, "conditions", None) or [] reason = None message = None for cond in conditions: if cond.type == "Failed" and cond.status == "True":

reason = cond.reason message = cond.message break msg = ( f"Kubernetes reconciliation: Job '{job_name}' failed" + (f" with reason '{reason}'" if reason else "") + (f": {message}" if message else "") ) if dry_run: print(f" DRY RUN: would mark as Crashed: {msg}") else: try: await client.set_flow_run_state( flow_run.id, Crashed(message=msg).to_state_create(), force=True, ) print(" marked as Crashed") except Abort as exc: print(f" skipped: terminal-state race: {exc}") except Exception as exc: print(f" ERROR setting crashed state: {exc}") continue print(" action: unknown job state, left unchanged") def parse_args(): parser = argparse.ArgumentParser(description="Reconcile stale Prefect running flow runs against Kubernetes Jobs") parser.add_argument("--namespace", required=True, help="Kubernetes namespace where worker Jobs run") parser.add_argument( "--min-running-minutes", type=int, default=30, help="Only inspect flow runs that have been running at least this long", ) parser.add_argument( "--limit", type=int, default=500, help="Maximum number of Prefect flow runs to inspect", ) parser.add_argument( "--deployment-name", default=None, help="Optional Prefect deployment name filter", ) parser.add_argument( "--work-pool-name", default=None, help="Optional Prefect work pool name filter", ) parser.add_argument( "--mark-missing-as-crashed", action="store_true", help="Mark runs as crashed when no matching Kubernetes Job is found", ) parser.add_argument( "--apply", action="store_true", help="Actually update Prefect states. Without this flag, dry-run only.", ) return parser.parse_args() if name == "__main__": args = parse_args() if not os.getenv("PREFECT_API_URL"): raise RuntimeError("PREFECT_API_URL must be set") asyncio.run( reconcile( namespace=args.namespace, min_running_minutes=args.min_running_minutes, limit=args.limit, dry_run=not args.apply, mark_missing_as_crashed=args.mark_missing_as_crashed, deployment_name=args.deployment_name, work_pool_name=args.work_pool_name, ) )

*Install deps*

pip install "prefect==3.6.22" kubernetes

*Example usage*

*Dry run first:*

export PREFECT_API_URL="http://your-prefect-api:4200/api" python reconcile_prefect_k8s_runs.py \ --namespace your-jobs-namespace \ --min-running-minutes 90 \ --work-pool-name kubernetes-pool \ --mark-missing-as-crashed

*Actually apply:*

export PREFECT_API_URL="http://your-prefect-api:4200/api" python reconcile_prefect_k8s_runs.py \ --namespace your-jobs-namespace \ --min-running-minutes 90 \ --work-pool-name kubernetes-pool \ --mark-missing-as-crashed \ --apply ``` Notes / caveats 1. Force mode I used

force=True

when setting the state so the reconciler can clean up runs even if orchestration would otherwise reject the transition. 2. Missing Job behavior

--mark-missing-as-crashed

is useful in your scenario because the Job may already be deleted after failure. But I recommend: - use it only with a sufficiently high

--min-running-minutes

- and ideally filter to the affected deployment/work pool 3. Name matching

This script assumes the Job name contains the flow run ID. If your generated Job names don’t, we should switch the lookup to labels/annotations instead. 4. Large namespaces

find_jobs_for_flow_run

currently scans Jobs in the namespace. That’s okay for incident cleanup, but not ideal for very large namespaces. If needed, I can give you a more efficient version that: - narrows by label selector, or - uses Job annotations if Prefect sets the flow run ID there in your setup 5. Failed vs Crashed I used

Crashed

because this is infrastructure-observer reconciliation, not application exception reporting. That usually fits best. If you want, I can provide a v2 of this script that: - matches Jobs via labels/annotations instead of name substring - deletes stale K8s Jobs too - emits a CSV/JSON report - runs as a CronJob in-cluster on EKS

Yes — for self-hosted Prefect 3.x, you can create the equivalent automation yourself. The doc section you linked is specifically about the managed Cloud automation, but the same pattern is supported in self-hosted as a normal automation: Detect zombie flows Important caveat for your case: This heartbeat-based automation helps with true zombie runs where the flow run remains

Running

but stops heartbeating. It is useful, but it is not a perfect replacement for the Kubernetes Jobs observer: - If the flow process never starts heartbeating, this automation won’t fire. - If the observer dies after the run has been heartbeating, this automation can clean up the stuck

Running

run later. So in your situation, I’d view it as a good safety net, not the primary fix. What to create You can create an automation that: - watches for

prefect.flow-run.heartbeat

- expects continued

prefect.flow-run.*

events for each flow run - if no heartbeat arrives within a window, marks the flow run

CRASHED

Example:

from datetime import timedelta

from prefect.automations import Automation
from prefect.client.schemas.objects import StateType
from prefect.events.actions import ChangeFlowRunState
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.events.schemas.events import ResourceSpecification

automation = Automation(
    name="Crash zombie flow runs",
    trigger=EventTrigger(
        after={"prefect.flow-run.heartbeat"},
        expect={"prefect.flow-run.*"},
        match=ResourceSpecification(
            {"prefect.resource.id": ["prefect.flow-run.*"]}
        ),
        for_each={"prefect.resource.id"},
        posture=Posture.Proactive,
        threshold=1,
        within=timedelta(minutes=9),
    ),
    actions=[
        ChangeFlowRunState(
            state=StateType.CRASHED,
            message="Flow run marked as crashed due to missing heartbeats.",
        )
    ],
)

if __name__ == "__main__":
    automation.create()

How to use it 1. Ensure your self-hosted server is running automation services. 2. Run the script in an environment pointed at your Prefect API:

export PREFECT_API_URL="<http://your-prefect-api:4200/api>"
python create_zombie_automation.py

Why

within=9 minutes

? Prefect flow heartbeats are typically every 180 seconds, so 9 minutes gives tolerance for about 3 missed heartbeats. That’s the same general pattern described in the docs. Docs - Zombie flow detection: Detect zombie flows - Automations: Automations For self-hosted: confirm services are running Your server needs the background services that evaluate automations. If you split server components, make sure

prefect server services start

is running somewhere. If you want, I can also give you: 1. a YAML automation definition instead of Python, or 2. a version tuned for your EKS workload with a longer timeout like 15–20 minutes.

but if you have a specific repro for the zombies, happy to look into that

You’re probably not doing anything “wrong” conceptually — you’re likely hitting a known behavior/bug in proactive automations around heartbeat-based zombie detection. Why it fires after a successful

Completed

event Your trigger is:

after: ["prefect.flow-run.heartbeat"]
expect: ["prefect.flow-run.*"]
posture: "Proactive"
threshold: 1
within: 600
for_each: ["prefect.resource.id"]

That means: - a heartbeat starts the evaluation window - Prefect then expects at least 1 matching event within 600s - because posture is Proactive, it fires when not enough expected events are seen In theory,

prefect.flow-run.Completed

should count as an expected event because it matches

prefect.flow-run.*

. So if you still see the automation fire ~8–10 minutes later after completion, that matches a bug that has affected this pattern. The practical symptom is exactly what you saw: a run completes successfully, then the zombie automation still triggers later. Recommended fixes 1. Upgrade Prefect if possible You’re on

3.6.22

, and newer releases include automation fixes. I’d strongly recommend testing on the latest 3.x first: Prefect releases 2. Use explicit terminal events instead of only the wildcard Even though the wildcard should work, I’d change the automation to explicitly include terminal flow-run events:

{
  "type": "event",
  "match": {
    "prefect.resource.id": "prefect.flow-run.*"
  },
  "match_related": {
    "prefect.resource.id": [
      "prefect.flow.5dec07f4-f468-4cdf-99a3-5016aa05c383",
      "prefect.flow.acdba04f-fc39-4e67-990b-b9ff25947e24",
      "prefect.flow.baebc35a-bcaf-47ab-9819-a80f873d88b5",
      "prefect.flow.400db30b-8136-48b0-93bf-f391bded4377",
      "prefect.flow.d0b6d613-82d4-42a4-b2c0-113b186d817b",
      "prefect.flow.ee31395e-3769-4db5-a72c-6fe6a0470e1a",
      "prefect.flow.50bc3efe-2c90-4d06-b82b-08db0faf1a8b",
      "prefect.flow.b992f437-56c4-4644-90a8-4ef6305993c4",
      "prefect.flow.293b63ca-651d-4e8a-b518-9c63d124bf5a"
    ],
    "prefect.resource.role": "flow"
  },
  "after": [
    "prefect.flow-run.heartbeat"
  ],
  "expect": [
    "prefect.flow-run.heartbeat",
    "prefect.flow-run.Completed",
    "prefect.flow-run.Failed",
    "prefect.flow-run.Cancelled",
    "prefect.flow-run.Crashed"
  ],
  "for_each": [
    "prefect.resource.id"
  ],
  "posture": "Proactive",
  "threshold": 1,
  "within": 600
}

That makes the intent much clearer: - heartbeat starts the timer - another heartbeat or a terminal state satisfies it - only true silence should trigger 3. Consider a slightly shorter or longer window based on heartbeat frequency If your heartbeat frequency is 180s, then: -

within: 600

is okay -

within: 540

is a clean “3 missed heartbeats” rule Important nuance A proactive trigger does not automatically “shut off” just because a run completed. It only knows whether the bucket of expected events was satisfied according to trigger logic. That’s why explicit terminal events are helpful, and why this bug is so noticeable with heartbeat-based detection. Docs - Zombie flow detection: Detect zombie flows - Automations concepts: Automations My concrete recommendation for you - Update the automation to list terminal events explicitly - If possible, upgrade Prefect to latest 3.x - Test with one short-lived flow and verify that no automation fires ~10 minutes after

Completed

If you want, I can also rewrite your JSON into: - a full automation payload for API/CLI use, or - a Python

Automation(...)

definition you can apply directly.

That sounds like a bug on Prefect Cloud. Who should I contact @Marvin?

If you think this is a Prefect Cloud bug, the best path is to contact Prefect here: prefect.io/contact I’d include: - your automation definition - the flow run ID(s) - timestamps for the

Completed

event and the later automation firing - your Prefect version / Cloud workspace context - a note that this appears related to proactive zombie-flow automation firing after terminal completion If you want, I can help you turn what you’ve gathered into a concise bug report.