Short answer: this isn’t a design defect, it’s a consequence of how timeouts and blocking calls work. In Prefect 3.x, if a synchronous task runs in a worker thread (the default ThreadPoolTaskRunner), Prefect’s

timeout_seconds

cannot interrupt a blocking system call like

subprocess.run()

. If

xtrabackup

hangs or stops making progress, the Python thread is still blocked in

subprocess.run()

so the UI continues to show the task as RUNNING until that call returns. The state only updates when the function returns or raises. Why you’re seeing “stalled file time but task is RUNNING” -

subprocess.run()

is a fully blocking call. In a worker thread, Prefect can’t preempt it to enforce

timeout_seconds

. - If

xtrabackup

hangs, exits but leaves a child process, or blocks on I/O, the task thread remains blocked so the UI never receives a state change. - If you’re on Windows, sync task timeouts can’t interrupt blocking calls at all; you must use native timeouts. How to make this reliable Pick one of these patterns to ensure timeouts actually terminate the process and the task state updates promptly. 1) Add a native subprocess timeout and kill the process group (most robust for CLI tools) - Works regardless of task runner or OS. - Ensures the entire

xtrabackup

tree is terminated if it stalls. ``` from prefect import task, get_run_logger import datetime as dt import os, signal, subprocess, shutil @task( task_run_name="gen_daily_incremental_backup", retries=1, retry_delay_seconds=120, timeout_seconds=21600, # Prefect-level safety net ) def gen_daily_incremental_backup(bdate: dt.datetime) -> str: logger = get_run_logger() bdate_str = bdate.strftime("%Y%m%d") inc_bak_dir = os.path.join(BACKUP_DIR, f"incbak_{bdate_str}") # Remove pre-existing directory safely if os.path.exists(inc_bak_dir): shutil.rmtree(inc_bak_dir, ignore_errors=True) base_dir = find_latest_backup_dir() os.makedirs(inc_bak_dir, exist_ok=True) if base_dir is None: msg = f"No previous backup found for {bdate_str}" logger.error(msg) send_slack_notification(f"{msg}.\nPlease check.") raise RuntimeError(msg) cmd = [ "xtrabackup", "--backup", "--compress", f"--host={MYSQL_CONFIG['host']}", f"--port={MYSQL_CONFIG['port']}", f"--user={MYSQL_CONFIG['user']}", f"--password={MYSQL_CONFIG['password']}", f"--target-dir={inc_bak_dir}", f"--incremental-basedir={base_dir}", "--parallel=12", "--rsync", ] # Choose a subprocess timeout that is <= Prefect timeout_seconds sb_timeout = 6 * 60 * 60 # 6h, adjust as needed # Start in a new process group/session so we can terminate the whole tree on timeout popen_kwargs = {} if os.name == "nt": popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP else: popen_kwargs["preexec_fn"] = os.setsid try: result = subprocess.run( cmd, stdout=subprocess.PIPE, # consider redirecting to a file to avoid huge memory stderr=subprocess.PIPE, text=True, check=False, timeout=sb_timeout, **popen_kwargs, ) except subprocess.TimeoutExpired as e: # Kill the entire process group if os.name == "nt": # Send CTRL_BREAK to the group; fall back to terminate/kill if needed try: os.kill(e.pid, signal.CTRL_BREAK_EVENT) # may require tracking Popen except Exception: pass else: try: # If we used setsid, kill the whole group os.killpg(os.getpgid(e.pid), signal.SIGTERM) # may require tracking Popen except Exception: pass msg = f"Incremental backup timed out for {bdate_str} after {sb_timeout}s" logger.error(msg)

send_slack_notification(f"{msg}") raise if result.returncode != 0: # capture last error line if present error_line = (result.stderr or "").splitlines()[-1] if (result.stderr or "").splitlines() else "unknown error" msg = f"Incremental backup failed for {bdate_str}: {error_line}" logger.error(msg) send_slack_notification(f"{msg}") raise RuntimeError(msg) logger.info(f"Incremental backup completed successfully for {bdate_str}") return inc_bak_dir

Notes:
- For strict group termination on timeout, you may prefer `Popen` + `communicate(timeout=...)` so you can reference `proc.pid` to send signals to the process group, then collect remaining output. Using `run()` hides the Popen instance; the pattern is similar but a few more lines.
- Consider writing output to files instead of capturing in-memory if `xtrabackup` is chatty.

2) Run the flow with a ProcessPoolTaskRunner
- Each task runs in its own process, so Prefect can reliably enforce `timeout_seconds` by terminating the task process.
- Minimal code change at the flow level:

from prefect import flow from prefect.task_runners import ProcessPoolTaskRunner @flow(name="backup-flow", task_runner=ProcessPoolTaskRunner()) def backup_flow(...): gen_daily_incremental_backup.submit(...)

3) Make the task async and use Prefect’s async process helpers
- Async cancellation is cooperative and reliably cleans up subprocesses.
- Example with `prefect.utilities.processutils.run_process`:

from prefect import task from prefect.utilities.processutils import run_process @task(timeout_seconds=21600) async def gen_daily_incremental_backup_async(...): cmd = [...] proc = await run_process(cmd, stream_output=True) return_code = await proc.wait() if return_code != 0: raise RuntimeError("xtrabackup failed") ``` What I’d like to confirm to tailor the fix - Which task runner are you using for this flow (default ThreadPoolTaskRunner vs ProcessPoolTaskRunner)? - OS/platform (Linux vs Windows)? - Are you running in Docker/Kubernetes? - Do you see a log like “Timeout configured but running in a worker thread…” when the task starts? Relevant docs - Task runners overview: https://docs.prefect.io/latest/concepts/task-runners/ - Orchestration basics and timeouts: https://docs.prefect.io/latest/ If you share the runner/platform and how you’re deploying, I can suggest the smallest change for your setup. In most cases, adding a native

timeout=

to

subprocess.run()

(and killing the process group on timeout) or switching to

ProcessPoolTaskRunner

resolves the “UI shows RUNNING while the process stalled/exited” symptom.

Got it — this sounds less like a timeout issue and more like the flow-run/infrastructure dying while the UI is still showing the task RUNNING. What’s going on - In Prefect 3.x, tasks report state from inside the flow-run process. If that process (the “infrastructure”: local subprocess, Docker container, or K8s pod) disappears abruptly (SIGKILL/OOM/restart), the engine never gets a chance to send final task states. The UI will keep showing RUNNING until something else marks the run as terminal. - There are two ways a run should get marked CRASHED: 1) Worker-side detection: the worker watches the infra process/container and, on a non-zero exit or abnormal stop, immediately proposes a CRASHED state. 2) Heartbeat staleness: if the flow-run heartbeats stop and the worker didn’t catch the exit (e.g., worker crashed/restarted), a later cleanup/automation can mark the run as CRASHED after a delay. - If the worker lost track (crashed, restarted, or lost network) before it could observe the infra exit, the flow/task can remain RUNNING in the UI until heartbeat-based cleanup or a manual action occurs. Common scenarios where you’ll see “infra gone but UI shows RUNNING” - Worker crash or restart after launching the infra: no one is left to report the abnormal exit. - Infra OOMKilled/SIGKILL: no graceful shutdown → no final state sent. - Network/API blip during state proposal: state transition never persisted. - Kubernetes/Docker workers lost watch permissions or were restarted mid-run: exit wasn’t observed. What to check next (most useful signals) - Which work pool/worker type are you using: Process, Docker, or Kubernetes? - Worker logs around that time. You should see lines like “Submitted infrastructure” followed by either a clean completion or messages about non-zero exit and “proposing CRASHED”. If you see neither, the worker likely died or lost its watch. - Infra native status: container/pod events (e.g., OOMKilled), or local subprocess exit code. - Flow run details: last heartbeat timestamp. If heartbeats stopped long ago and it still shows RUNNING, it points to missed crash detection. How to make this robust - Ensure your worker stays healthy and has continuous visibility into the infra (don’t restart/replace it mid-run without handoff). - For Docker/K8s, confirm the worker has permission to watch and read container/pod status and logs. - Add an Automation to flag/transition stale runs (missing heartbeat or long-running RUNNING state) so they don’t linger indefinitely. See Automations. - Keep Prefect and integrations up to date; we’ve shipped fixes over time improving crash detection in workers. See Workers. If you can share: - Work pool type and Prefect version - Worker logs around the incident - A flow run ID/time window …I can help pinpoint whether the worker missed the infra exit, a heartbeat-based cleanup didn’t trigger, or if there’s another edge case at play.

Thanks — looking just at the task code you shared, the most likely line(s)/patterns that can make the “infrastructure process disappear” (e.g., pod/container/process OOMKilled or terminated) while the UI still shows RUNNING are: Most likely culprit: capturing all xtrabackup output in memory - Line: the

subprocess.run(...)

invocation - Pattern:

stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True

- Why:

subprocess.run

will read the entire stdout/stderr into memory. xtrabackup can be very chatty over long runs. Hours of logs decoded to text can easily balloon memory and get your flow-run container/pod/process OOMKilled. When the infra dies abruptly, Prefect doesn’t get a final state and the UI lingers at RUNNING until a later cleanup. - What to change: - Don’t capture to memory; write to files or stream. - Example (write to files and keep a “last line” for error reporting):

out_path = os.path.join(inc_bak_dir, "xtrabackup.out")
    err_path = os.path.join(inc_bak_dir, "xtrabackup.err")
    with open(out_path, "w") as out, open(err_path, "w") as err:
        proc = subprocess.Popen(
            cmd,
            stdout=out,
            stderr=err,
            text=True,
            # optional: create a new process group so you can kill the whole tree if needed
            preexec_fn=os.setsid if os.name != "nt" else None,
            creationflags=(subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0),
        )
        ret = proc.wait()
    if ret != 0:
        # Read just the tail instead of whole file
        error_line = "see xtrabackup.err (tail)"
        raise RuntimeError(f"Incremental backup failed for {bdate_str}: {error_line}")

- Or stream+tail without holding everything:

proc = subprocess.Popen(
        cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1
    )
    last_line = ""
    for line in proc.stdout:
        last_line = line.rstrip()
        # optionally write to a rotating file here
    ret = proc.wait()
    if ret != 0:
        raise RuntimeError(f"Incremental backup failed for {bdate_str}: {last_line}")

High memory pressure from xtrabackup itself - Lines/flags:

--compress

with

--parallel=12

- Why: Compression + parallelism can use significant CPU and memory. In constrained containers/VMs this can trigger OOMKills independent of Python’s pipes. Combine this with in-memory log capture and it’s even riskier. - What to change: - Reduce

--parallel

or tune xtrabackup memory options. - Ensure the container/pod has sufficient memory limits/requests. Less likely but worth fixing - Using

rm -rf

via subprocess: - Line:

subprocess.run(["rm", "-rf", inc_bak_dir], check=False)

- Why: not the cause of infra disappearance, but brittle. Prefer

shutil.rmtree(inc_bak_dir, ignore_errors=True)

to avoid spawning extra processes and to keep behavior predictable. - Fragile stderr parsing: - Line:

error_line = result.stderr.splitlines()[-1]

- Why: if

stderr

is empty, this raises

IndexError

. That will fail the task (which is okay), but not cause the infra to vanish. Still, make it safe:

lines = (result.stderr or "").splitlines()
    error_line = lines[-1] if lines else "no stderr captured"

- Logger and Slack usage: - Use

get_run_logger()

instead of

get_logger(__name__)

for Prefect 3.x; not related to infra death but gets you consistent task logs. -

send_slack_notification

won’t “make infra disappear”; at worst it blocks if Slack is unreachable. How to confirm this is OOM/output-capture related - If you run in Kubernetes:

kubectl describe pod <pod>

and look for OOMKilled in the container’s last state. - If Docker:

docker inspect <container>

or check the daemon logs for OOM kill messages. - Local: check

dmesg

/ system logs for OOM killer entries around the run time.

- Quick experiment: remove

stdout=PIPE, stderr=PIPE, text=True

and redirect to files; re-run with the same data volume. If the “infra disappears” symptom goes away, you’ve found the cause. If you share where this runs (local/Docker/K8s) and memory limits, I can suggest exact redirection/streaming patterns for that environment (and how to tail errors safely without loading everything into memory).

Thanks for the logs — they pinpoint the root cause. What happened - Your worker crashed with a KeyError during worker teardown: -

KeyError: UUID('019c92ac-7d94-7785-899c-70888301db57')

inside

prefect.workers.base.__aexit__

- Followed by systemd:

prefect-worker.service: Main process exited, code=exited, status=1/FAILURE

- When the worker dies mid-run, the flow/task states are not updated and the UI can remain stuck in RUNNING until a later cleanup/automation marks them as CRASHED. This is not caused by your task code; it’s a known worker bug in certain Prefect 3.x versions - Very similar stack traces were reported here: Issue #17735 – “Exception occur in worker causing restarts” (Prefect 3.2.14) - It was fixed by: PR #19866 – “Fix KeyError crash caused by dead cancellation tracking code in worker” - Root cause: worker cancellation tracking tried to remove a flow run ID from an internal set that was never added, raising KeyError and crashing the worker. What to do next 1) Upgrade Prefect to the latest 3.x (the fix is included in newer releases beyond the versions shown in the issue/PR logs). After upgrading, restart your worker service. 2) Ensure your systemd unit restarts the worker on failure (Restart=always) so transient issues don’t leave runs orphaned. 3) Add an automation to fail stale RUNNING flow runs (missing heartbeats for N minutes) so the UI doesn’t linger if a worker does go away: Automations docs. 4) Optional hardening for your backup task: - Avoid

stdout=PIPE, stderr=PIPE

with very chatty commands like xtrabackup (can trigger high memory); write to files or stream to a rotating log. - Keep Prefect’s

timeout_seconds

as a safety net, but rely on native

subprocess

timeouts or process groups for termination. Can you share: - Your Prefect version (

prefect version

) and work pool/worker type? Once I have that, I can confirm whether your version contains the fix and suggest the exact upgrade path (e.g., image tag if Docker/K8s, or Poetry/pip version bump).

Great question. In Prefect 3.x, deployments and their schedules are managed by the server’s scheduler service. Here’s exactly what happens when you update or delete a deployment, and why you’re seeing “three old scheduled runs.” Key behaviors - Prefect pre-creates a small buffer of future runs for each active schedule - By default, the scheduler keeps at least 3 future runs on the books and at least 1 hour of runway. That’s why you often see “3” scheduled runs. - These defaults are controlled by server settings like: - PREFECT_API_SERVICES_SCHEDULER_MIN_RUNS (default 3) - PREFECT_API_SERVICES_SCHEDULER_MIN_SCHEDULED_TIME (default 1 hour) When you update a deployment - Future auto-scheduled runs are deleted, then immediately regenerated to match the new schedule. - Only auto-scheduled future runs are deleted. Manually created runs are not touched. - The scheduler then quickly re-fills the buffer (≥3 future runs by default), so you’ll see three upcoming runs again — they are new, not the original ones. - Past or already-running runs are unaffected. When you delete a deployment - All scheduled runs for that deployment are deleted (both auto-scheduled and manual). Why you might still see “old looking” scheduled runs after a change - Timing: The scheduler runs frequently and will repopulate at least 3 future runs right after an update. These are not the original runs; they are newly created to satisfy the “min 3 runs” rule. - Multiple schedules per deployment: If you have more than one active schedule, each one will maintain its own buffer of future runs. Practical implications - If you change the schedule (e.g., cron/interval), any future auto-scheduled runs will be dropped and replaced with runs that match the new expression. - If you pause a schedule, no new runs will be created, but already-created scheduled runs remain unless you delete them manually. - Deleting the deployment cleans up its scheduled runs entirely. If you want to change this behavior - Reduce or increase the pre-created buffer by adjusting server settings (e.g., PREFECT_API_SERVICES_SCHEDULER_MIN_RUNS), if you control the Prefect server. - Alternatively, pause schedules before updates if you want a completely “quiet” schedule window (no new runs generated) during a change, then unpause. If you share whether you’re on Prefect Cloud or self-hosted Orion Server, I can point you to where to adjust these settings (or confirm the current defaults for your environment).