This usually happens because the Ray future never resolves after the worker/node crashes, so your Prefect task stays “Running” waiting on it. Here are reliable ways to prevent indefinite runs and recover cleanly (assuming Prefect 3.x + Ray): 1) Add a hard cap with task timeouts - Set a timeout on the task itself so Prefect will fail it if it doesn’t finish in time.

from prefect import task

@task(timeout_seconds=60, retries=2, retry_delay_seconds=10)
def do_work(...):
    ...

- This is the simplest way to ensure a stuck Ray task doesn’t run forever. See “Tasks > Timeouts” in the docs: Prefect docs. 2) If you’re using prefect-ray, time-bound result collection and cancel the Ray task - Install and use the Ray task runner:

pip install "prefect[ray]"  # or: pip install prefect-ray

- In your flow, collect results with a timeout and explicitly cancel the Ray task on timeout/crash:

import ray
from prefect import flow, task
from prefect_ray import RayTaskRunner

@task(timeout_seconds=60, retries=2, retry_delay_seconds=10)
def do_work(x):
    # your Ray-executed work
    return x * 2

@flow(task_runner=RayTaskRunner())
def my_flow(items: list[int]):
    futs = [do_work.submit(i) for i in items]
    results = []
    for f in futs:
        try:
            # Don’t block forever on a crashed worker
            res = f.result(timeout=75, raise_on_failure=True)
            results.append(res)
        except TimeoutError:
            # Stop waiting and request cancellation on the Ray side
            try:
                ray.cancel(getattr(f, "wrapped_future", None), force=True)
            except Exception:
                pass
            results.append(None)
        except Exception:
            # Ray worker/node failures surface as exceptions; you can choose to retry
            results.append(None)
    return results

Notes: -

f.result(timeout=...)

prevents your flow from waiting forever. -

ray.cancel(...)

requests cancellation of the underlying Ray task; it’s not automatic, so call it if you want to stop any lingering work on the cluster. - In prefect-ray, the underlying Ray ObjectRef is typically available via

future.wrapped_future

. 3) Retry only on Ray-crash exceptions (optional) - You can make retries conditional on Ray-specific failures:

from prefect import task

def retry_on_ray_crash(exc: BaseException) -> bool:
    try:
        import ray
        crash_types = (
            ray.exceptions.WorkerCrashedError,
            ray.exceptions.RayActorError,
            getattr(ray.exceptions, "ObjectRefLostError", Exception),
        )
        return isinstance(exc, crash_types)
    except Exception:
        return False

@task(retries=3, retry_delay_seconds=10, retry_condition_fn=retry_on_ray_crash)
def do_work(...):
    ...

4) Defense in depth: use library-native timeouts inside the task - For blocking I/O, set timeouts on the libraries you call (e.g., requests/httpx DB clients) so the Ray-side work can exit quickly even if the worker is under stress. References - Ray integration package: prefect-ray - Main Prefect docs: Prefect docs (see Integrations > Ray and Tasks > Timeouts) If you can share: - Your Prefect and Ray versions - Whether you’re using `prefect-ray`’s

RayTaskRunner

- A minimal snippet of how you submit and wait for the task …I can tailor the example exactly to your setup.

Sorry, I encountered an error while processing your request. Please try again.

Thanks for the clear postmortem — that lines up with what we see when a flow pod (driver) gets SIGTERM: the Ray job is torn down and any in-flight Ray tasks are force-killed. The “RUNNING forever” task is just an orphaned record because the driver died before it could report a terminal state. Below is a practical recipe to make your task submissions robust in the face of driver SIGTERM/resubmission, spot evictions, and Ray worker crashes. It’s focused on at-least-once execution with safe deduping so a resubmitted flow only picks up missing work. TL;DR (actionable checklist) - Make each unit of work idempotent and verifiable (per-clip, per-frame-window). - Persist results and/or “done” markers to S3 with deterministic keys. - Use Prefect task-level caching and result persistence to dedupe re-submissions. - Add task timeouts + retries with jitter; optionally retry only on Ray/system errors. - Bound waits for results; cancel Ray futures on timeout. - Use concurrency limits to prevent duplicate “same-clip-window” submissions. - Harden your K8s driver pod against evictions (grace period, preStop, PDB). - Add a small cleanup job to mark orphaned RUNNING tasks as CRASHED after a TTL. 1) Idempotency + S3 checkpointing/done-markers Have each task write an atomic “done” marker keyed by clip + frame window. On resubmission, list S3 and only submit missing windows.

from prefect import flow, task
import boto3, json, os

S3_BUCKET = os.getenv("S3_BUCKET", "my-bucket")
S3_PREFIX = "clips"  # e.g., clips/{clip_id}/{start}-{end}/done.json

s3 = boto3.client("s3")

def done_key(clip_id: str, start: int, end: int) -> str:
    return f"{S3_PREFIX}/{clip_id}/{start}-{end}/done.json"

def is_done(clip_id: str, start: int, end: int) -> bool:
    try:
        s3.head_object(Bucket=S3_BUCKET, Key=done_key(clip_id, start, end))
        return True
    except s3.exceptions.ClientError:
        return False

def write_done(clip_id: str, start: int, end: int, meta: dict):
    # Atomic-ish: write to tmp then copy to final key (S3 has no real rename)
    tmp_key = f"{S3_PREFIX}/{clip_id}/{start}-{end}/.tmp-{os.urandom(4).hex()}.json"
    body = json.dumps({"status": "done", "meta": meta}).encode()
    s3.put_object(Bucket=S3_BUCKET, Key=tmp_key, Body=body)
    s3.copy_object(Bucket=S3_BUCKET, CopySource={"Bucket": S3_BUCKET, "Key": tmp_key}, Key=done_key(clip_id, start, end))
    s3.delete_object(Bucket=S3_BUCKET, Key=tmp_key)

@task(
    # Cache + persistence make retries/resubmissions a no-op if already done
    persist_result=True,
    cache_expiration=None,
    cache_policy="INPUTS",  # cache by parameters
    result_storage_key="clip-{parameters[clip_id]}-{parameters[start]}-{parameters[end]}"
)
def process_window(clip_id: str, start: int, end: int) -> dict:
    if is_done(clip_id, start, end):
        return {"skipped": True, "reason": "already-done-marker"}

    # Do the work (Ray, GPU, etc.)
    result_meta = run_heavy_clip_op(clip_id, start, end)

    # Write outputs to S3 first, then mark done
    write_outputs_to_s3(clip_id, start, end, result_meta)
    write_done(clip_id, start, end, result_meta)
    return {"skipped": False, "meta": result_meta}

@flow
def clip_job(clip_id: str, windows: list[tuple[int, int]]):
    to_run = [(s, e) for (s, e) in windows if not is_done(clip_id, s, e)]
    futs = [process_window.submit(clip_id, s, e) for (s, e) in to_run]
    return [f.result() for f in futs]

2) Prefect caching + result persistence (defense in depth) - Ensure

persist_result=True

and deterministic

result_storage_key

so resubmitted flows reuse results. - For global defaults:

prefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true
# Optionally: set a default storage block for all results
# prefect config set PREFECT_DEFAULT_RESULT_STORAGE_BLOCK="s3-bucket/<your-block>"

3) Timeouts, retries, and Ray-aware retry conditions - Add a hard cap so a hung submission never runs forever.

- Use jittered backoff and retry only on Ray/system failures (not on your business exceptions).

from prefect import task
from prefect.tasks import exponential_backoff

def retry_on_system_or_ray(exc: BaseException) -> bool:
    msg = f"{type(exc).__name__}: {exc}"
    ray_signals = ("ray.exceptions", "RayTaskError", "WorkerCrashed", "ObjectRefLost", "Task was cancelled")
    kube_signals = ("SIGTERM", "Evicted", "PodUnschedulable", "DeadlineExceeded")
    return any(s in msg for s in (*ray_signals, *kube_signals))

@task(
    timeout_seconds=1800,               # 30m cap per window
    retries=5,
    retry_delay_seconds=exponential_backoff(2),  # 2,4,8,16,32s (jitter below)
    retry_jitter_factor=0.5,
    retry_condition_fn=retry_on_system_or_ray,
)
def process_window(...):
    ...

4) Bound result waits and cancel Ray work on timeout If you’re using prefect-ray’s RayTaskRunner or managing Ray futures yourself, don’t block forever and cancel on timeout:

import ray
from prefect import flow, task
from prefect_ray import RayTaskRunner

@task(timeout_seconds=1800)
def process_window(...):
    # Example if you’re calling ray directly:
    @ray.remote
    def do_work(...): ...
    obj_ref = do_work.remote(...)

    try:
        return ray.get(obj_ref, timeout=1850)  # Slightly > task timeout
    except ray.exceptions.GetTimeoutError:
        # Optional: request cancellation if you don't want the orphan to continue
        try:
            ray.cancel(obj_ref, force=True)
        except Exception:
            pass
        raise  # triggers Prefect retry per above

If you’re calling

future.result()

on Prefect futures from RayTaskRunner, use the timeout there too:

res = f.result(timeout=120, raise_on_failure=True)

5) Concurrency guardrails to prevent duplicate submissions Tag your tasks and enforce a limit so that, even on a bursty resubmission, you never double-run the same clip/window at the same time. - Add a tag to the task:

@task(tags=["clip-window"])
def process_window(...): ...

- Create a concurrency limit:

prefect concurrency-limit create clip-window --limit 50

6) Harden the Kubernetes driver pod - Give it time to exit cleanly and flush progress: - terminationGracePeriodSeconds: 30+ - preStop hook to stop Ray cleanly (if you init Ray in the pod): - command: sleep 10 && ray stop --force - Add a PodDisruptionBudget so drains don’t evict all drivers: - minAvailable: 1 (or suitable for your fleet) - Prefer nodes without frequent preemption for driver pods (avoid spot for the driver; run workers on spot if desired). 7) Automatic cleanup of orphaned RUNNING tasks You did the right manual fix. To automate it, run a small periodic flow/cron that finds old RUNNING task runs and marks them CRASHED so dashboards and SLAs stay clean.

import asyncio
from datetime import datetime, timedelta, timezone
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.states import Crashed

@flow
async def mark_orphans_crashed(max_age_minutes: int = 30):
    cutoff = datetime.now(timezone.utc) - timedelta(minutes=max_age_minutes)
    async with get_client() as client:
        # Narrow this to the specific flow or tag if desired
        task_runs = await client.read_task_runs(limit=1000)
        to_mark = [tr for tr in task_runs if tr.state.type.value == "RUNNING" and (tr.start_time or cutoff) < cutoff]
        for tr in to_mark:
            await client.set_task_run_state(
                tr.id,
                Crashed(message="Orphaned after driver SIGTERM/resubmission"),
                force=True,
            )
    return len(to_mark)

8) Consider decoupling the compute from the driver If you can, submit work to a persistent Ray cluster (Ray Jobs/Serve) rather than tying the Ray job’s lifetime to the flow pod. Then a driver restart won’t kill compute. The tradeoff: you need explicit idempotency and bookkeeping since the compute outlives the flow pod.

Where to focus first for your pipeline - Keep your current S3 check for “17/18 files done,” but make it formal: one done marker per (clip_id, start, end). Only submit missing windows. - Add

timeout_seconds

+ jittered retries; bound any `future.result()`/`ray.get()` waits and cancel on timeout. - Tag + add a concurrency limit for clip-window tasks to prevent accidental double-submissions under resubmission. - Add a small periodic cleanup flow to mark stale RUNNING tasks as CRASHED. References - Prefect docs (3.x): Prefect docs - Ray integration: prefect-ray If you can share a snippet of how you submit “clip frames 50000–100000” (with/without prefect-ray), I can tailor the code to your exact pattern (including the S3 key scheme and the best cache policy for your parameters).