_cancel_run skips kill_infrastructure for running flow runs, leaving zombie pods indefinitely

[chuqCTC]

Bug summary

When a flow run is set to Cancelling (via UI, API, or automations), BaseWorker._cancel_run silently skips killing infrastructure if the flow has already started (start_time is not None). This assumes the flow engine inside the pod will detect the Cancelling state and self-terminate. If the flow is hung for any reason — swallowed exception, stuck future, blocked I/O — the pod lives forever with no mechanism to clean it up.

All major worker integrations implement kill_infrastructure (Kubernetes deletes the Job, ECS stops the task, Docker stops the container, etc.), but it is never called for running flows. The only call site is _cancel_run, which bails out early:

https://github.com/PrefectHQ/prefect/blob/3d18732034/src/prefect/workers/base.py#L1767-L1769

# Only cancel if the flow run was pending (never started)
if flow_run.start_time is not None:
    return

There is no other actor in the system that will kill the infrastructure:

• The Kopf observer only fires when the K8s Job exceeds backoffLimit — a hung pod is "healthy" from Kubernetes' perspective
• The foreman service only monitors worker heartbeats, not flow run liveness
• Flow run heartbeats continue even when hung (they run on a separate OS thread)

Minimal Reproducible Example

import concurrent.futures
from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner

@task
def bad_task():
    return "result"

@flow(task_runner=ProcessPoolTaskRunner())
def my_flow():
    # If deserialization of the task result fails silently
    # (e.g. due to a code/image version mismatch), the future
    # is never resolved and the flow hangs forever.
    future = bad_task.submit()
    return future.result()

To reproduce the zombie pod scenario:

1. Deploy my_flow to a Kubernetes work pool
2. Trigger a run — it enters Running state and a pod is created
3. Simulate a hang: the flow blocks on future.result() indefinitely (in the real case, this was caused by a cloudpickle deserialization error silently swallowed by concurrent.futures.Future._invoke_callbacks)
4. Cancel the flow run from the UI or API
5. Expected: The worker kills the Kubernetes Job and marks the run as Cancelled
6. Actual: _cancel_run returns immediately without calling kill_infrastructure because start_time is not None. The pod continues running forever, consuming cluster resources.

In a parent/subflow scenario, this compounds: the parent retries and spawns new subflows, each of which also hangs, accumulating hundreds of zombie pods.

Proposed Fix

Remove the start_time guard in _cancel_run:

     async def _cancel_run(self, flow_run_id: UUID) -> None:
         """
-        Cancel a flow run by killing its infrastructure and marking it cancelled.
-
-        Only cancels flow runs that were pending (not yet started).
+        Cancel a flow run by killing its infrastructure and marking it cancelled.
         """
         try:
             flow_run = await self.client.read_flow_run(flow_run_id)
@@ -1764,10 +1762,6 @@

         run_logger = self.get_flow_run_logger(flow_run)

-        # Only cancel if the flow run was pending (never started)
-        if flow_run.start_time is not None:
-            return
-
         # No infrastructure to kill if no pid
         if not flow_run.infrastructure_pid:

All worker integrations (Kubernetes, ECS, Docker, Azure Container Instances, Cloud Run, Vertex AI) already implement kill_infrastructure and handle the InfrastructureNotFound case for infrastructure that has already exited. The change is safe for flows that have already completed naturally — kill_infrastructure will raise InfrastructureNotFound, which _cancel_run already catches and handles gracefully.

Additional Context

• Related: Fix _UnpicklingFuture.add_done_callback swallowing deserialization errors #21612 fixed a specific instance of this (silently swallowed deserialization errors in _UnpicklingFuture), but the infrastructure cleanup gap is a separate systemic issue affecting any hung flow regardless of cause.
• The ProcessWorker is less affected because it spawns a local subprocess and waits for it — the parent process can be killed. But for Kubernetes, ECS, and other remote infrastructure workers where run() returns immediately after creating the job, there is no parent process managing the lifecycle.

Version info

Version:              3.6.23
API version:          0.8.4
Python version:       3.11.15
Git commit:           db5763e0
Built:                Thu, Mar 19, 2026 08:56 PM
OS/Arch:              linux/x86_64
Profile:              ephemeral
Server type:          server
Pydantic version:     2.12.5
Server:
  Database:           sqlite
  SQLite version:     3.53.0
Integrations:
  prefect-docker:     0.0.1.dev32+g450c9c392.d20260116
  prefect-kubernetes: 0.0.1.dev1+gd9623cb12.d20260406
  prefect-ray:        0.4.5
  prefect-redis:      0.2.8
  prefect-azure:      0.0.1.dev23+g3724d73b2.d20260117

Additional context

Related to #21616 which causes hanging runs

[desertaxle]

Thanks for the detailed report @chuqCTC!

BaseWorker._cancel_run is intentionally only the pre-start cancellation path. For work-pool runs using the standard prefect flow-run execute entrypoint, once the run has started, responsibility for cancellation shifts to a process running within the infrastructure itself. Because of that ownership split, _cancel_run is not the only cancellation path for already-
running runs, and we do not think removing the start_time guard is the right fix here.

That said, we do agree there is an abandoned-infrastructure cleanup gap when that in-infrastructure cancellation path cannot make progress. In those cases, infrastructure can remain live longer than it should.

We plan on addressing that more directly with a narrow server-to-worker cleanup channel for best-effort infrastructure teardown after the server has already decided that a run should no longer own its infrastructure. That is the direction we expect to take for this class of cleanup problem, rather than changing _cancel_run to unconditionally kill started runs.