From b42b340594f81e10ab4a08b1010dd5667c29fee5 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Wed, 12 Aug 2026 23:15:41 +0200 Subject: [PATCH] fix(downloads): don't let the supervisor kill a finishing child (review round 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - worker.py: the watchdog loop now treats only 'paused'/'cancelled' (or a vanished row) as a stop; a terminal status (done/error, or a re-queued contended asset) means the child already wrote its outcome and is exiting, so keep polling instead of killing it. Previously a proc.wait timeout landing between the child's terminal commit and its exit was misread as a route-cancel and SIGKILLed a successful download (dropping its completion notification). The stall/wall-clock/shutdown checks are now gated on status=='running' (#1, #5 test). - worker.py: guard the crash-path _job_liveness probe so a DB blip there can't propagate (the child is already dead; boot recovery is the backstop) (#2). - worker.py: drop _stop_child's redundant poll guard — escalate_kill no-ops on an exited proc and getpgid yields None for a reaped pid (#3). - worker.py: note the extraction heartbeat is per-attempt (a single over-timeout attempt is still killed, bounded by socket_timeout) (#4). --- backend/app/worker.py | 58 +++++++++++++++---------- backend/tests/test_worker_supervisor.py | 10 +++++ 2 files changed, 44 insertions(+), 24 deletions(-) diff --git a/backend/app/worker.py b/backend/app/worker.py index 4d3cef68..3cb9397d 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -279,9 +279,8 @@ def _signal_group(proc: subprocess.Popen, pgid: int | None, sig: int) -> None: def _stop_child(proc: subprocess.Popen, grace_s: float) -> None: """SIGTERM the child's process GROUP, then SIGKILL if it outlives the grace period. Shares the terminate→wait→kill escalation with run_ffmpeg (`editmod.escalate_kill`); the only difference is - that we signal the whole group so ffmpeg grandchildren die too.""" - if proc.poll() is not None: - return + that we signal the whole group so ffmpeg grandchildren die too. `escalate_kill` no-ops if the + child already exited, and `getpgid` yields None for a reaped pid, so no separate poll guard.""" try: pgid: int | None = os.getpgid(proc.pid) except OSError: @@ -326,18 +325,23 @@ def _run_supervised(job_id: int, asset_id: int) -> None: except subprocess.TimeoutExpired: pass status, stale = _job_liveness(job_id) - if status is None or status != "running": - reason = "cancelled" # a route paused/cancelled it (or the row vanished) - break - if stale is not None and stale > stall_s: - reason = "stalled" - break - if time.monotonic() - started > wall_s: - reason = "timeout" - break - if _stop.is_set(): - reason = "shutdown" + if status == "running": + # Only a genuinely-running child is subject to the stall / wall-clock / shutdown kill. + if stale is not None and stale > stall_s: + reason = "stalled" + break + if time.monotonic() - started > wall_s: + reason = "timeout" + break + if _stop.is_set(): + reason = "shutdown" + break + elif status is None or status in ("paused", "cancelled"): + reason = "cancelled" # a route stopped it (or the row vanished) break + # else: a terminal status (done / error, or a re-queued contended asset) means the child + # already wrote its outcome and is exiting on its own — keep polling; the next proc.wait + # catches the exit. NEVER kill a finishing child (that could drop its completion notice). except Exception: # noqa: BLE001 — a watchdog error must NEVER leave the child running unsupervised log.exception("supervisor loop failed for job %s; stopping the child", job_id) _stop_child(proc, grace_s) @@ -351,17 +355,21 @@ def _run_supervised(job_id: int, asset_id: int) -> None: if reason is None: # Child exited. If it died WITHOUT setting a terminal state (non-zero exit / crash / OOM), - # the job is still 'running' — fail it here so nothing is left stuck. + # the job is still 'running' — fail it here so nothing is left stuck. Guarded: the child is + # already dead (no orphan risk), so a DB blip in this probe must not propagate. if proc.returncode != _CHILD_OK: - status, _ = _job_liveness(job_id) - if status == "running": - _reset_asset_pending(asset_id) - _cleanup_staging(asset_id) - _set_job( - job_id, status="error", error_code="unknown", phase=None, - error=f"The download process exited unexpectedly (code {proc.returncode}).", - ) - log.warning("job %s: child exited %s without finishing", job_id, proc.returncode) + try: + status, _ = _job_liveness(job_id) + if status == "running": + _reset_asset_pending(asset_id) + _cleanup_staging(asset_id) + _set_job( + job_id, status="error", error_code="unknown", phase=None, + error=f"The download process exited unexpectedly (code {proc.returncode}).", + ) + log.warning("job %s: child exited %s without finishing", job_id, proc.returncode) + except Exception: # noqa: BLE001 — boot recovery is the backstop for a stuck 'running' job + log.exception("post-exit cleanup failed for job %s", job_id) return _stop_child(proc, grace_s) @@ -712,6 +720,8 @@ def _extract_with_fallback(job_id: int, asset_id: int, url: str, spec: dict, sta opts["postprocessor_hooks"] = [_make_pp_hook(job_id)] _touch_job(job_id) # stall-watchdog heartbeat: extraction writes no progress, so a slow / # multi-client-set fallback would otherwise cross the stall timeout with a stale updated_at. + # (Per-ATTEMPT only; a single extract_info that itself runs past the stall timeout is still + # killed — acceptable, as socket_timeout + extractor_retries bound one attempt well under it.) try: with yt_dlp.YoutubeDL(opts) as ydl: return ydl.extract_info(url, download=True) diff --git a/backend/tests/test_worker_supervisor.py b/backend/tests/test_worker_supervisor.py index ca80d82b..3a17d5a6 100644 --- a/backend/tests/test_worker_supervisor.py +++ b/backend/tests/test_worker_supervisor.py @@ -157,6 +157,16 @@ def test_a_long_download_survives_while_progress_stays_fresh(monkeypatch, rec): assert _names(rec) == [] +def test_a_finishing_child_is_not_mistaken_for_a_cancel(monkeypatch, rec): + # The child committed its terminal state (done/error) but hasn't exited yet (e.g. slow staging + # cleanup outlasting a poll). A non-'running' status must NOT be treated as a route-cancel and + # kill the finisher (which could drop its completion notification) — let it exit on its own. + _spawn(monkeypatch, _FakeProc(["timeout", ("exit", 0)])) # alive one poll, then a clean exit + _liveness(monkeypatch, ("done", 1.0)) + worker._run_supervised(1, 99) + assert _names(rec) == [] # never killed, asset/job untouched + + def test_a_watchdog_error_still_stops_the_child(monkeypatch, rec): # If the liveness probe raises (e.g. a transient DB error), the child must NOT be left running # unsupervised — the feature's whole point. It is killed and the job is failed.