From cce862cde72127ecfae40a9710ff80c1c61b7151 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Wed, 12 Aug 2026 23:38:37 +0200 Subject: [PATCH] fix(downloads): supervisor edge-cases from review round 3 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - worker.py: keep an absolute wall-clock backstop for ANY still-alive child. A child that reached a terminal state (done/error) but wedges in teardown is now killed to free the worker slot ("abandon"), with its DB status left intact; only a still-'running' over-cap job is failed. Previously gating the wall-clock on status=='running' let a wedged finisher loop forever (#1). - worker.py: reconcile the claimed asset on ANY clean child exit (not only non-zero), so a child that exits without finalizing it — e.g. an early return on a job DELETEd mid-flight — no longer leaves the asset stuck 'downloading' until boot recovery (#2). - worker.py: extract _stop_and_free + _fail_after_kill; the exception / stall / timeout / cancel / shutdown / abandon paths now share one kill+free (+fail) tail instead of three copies (#3). - tests: a wedged-finished child is killed by the wall-clock without a status change; clean exits now assert the reconciliation no-op (#1, #4). --- backend/app/worker.py | 93 +++++++++++++++---------- backend/tests/test_worker_supervisor.py | 22 ++++-- 2 files changed, 73 insertions(+), 42 deletions(-) diff --git a/backend/app/worker.py b/backend/app/worker.py index 3cb9397d..7bd3f764 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -294,6 +294,23 @@ def _stop_child(proc: subprocess.Popen, grace_s: float) -> None: ) +def _stop_and_free(job_id: int, asset_id: int, proc: subprocess.Popen, grace_s: float) -> None: + """Kill the child, free its claimed asset, and clear its staging — the common tail of every path + that stops a child (a route pause/cancel, a shutdown, an over-cap abandon).""" + _stop_child(proc, grace_s) + _reset_asset_pending(asset_id) + _cleanup_staging(asset_id) + + +def _fail_after_kill( + job_id: int, asset_id: int, proc: subprocess.Popen, grace_s: float, error: str, error_code: str +) -> None: + """`_stop_and_free` + mark the job errored — for the watchdog's stall/timeout kills and the + supervisor-error fallback.""" + _stop_and_free(job_id, asset_id, proc, grace_s) + _set_job(job_id, status="error", error=error, error_code=error_code, phase=None) + + def _run_supervised(job_id: int, asset_id: int) -> None: """Run the heavy download/edit for an already-claimed asset in a child process and watchdog it. @@ -325,70 +342,72 @@ 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 in ("paused", "cancelled"): + reason = "cancelled" # a route stopped it (or the row vanished) + break 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) + # Absolute wall-clock cap frees the worker slot even for a FINISHED child wedged in + # teardown. A still-running job over the cap is failed ("timeout"); a terminal one (already + # done/error) just gets its lingering child killed, its DB status left intact ("abandon"). + if time.monotonic() - started > wall_s: + reason = "timeout" if status == "running" else "abandon" 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). + # else: a terminal status under the cap (done/error, or a re-queued contended asset) means + # the child is exiting on its own — keep polling; 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) - _reset_asset_pending(asset_id) - _cleanup_staging(asset_id) - _set_job( - job_id, status="error", error_code="unknown", phase=None, - error="The download was stopped after a worker error.", + _fail_after_kill( + job_id, asset_id, proc, grace_s, "The download was stopped after a worker error.", "unknown" ) return 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. Guarded: the child is - # already dead (no orphan risk), so a DB blip in this probe must not propagate. - if proc.returncode != _CHILD_OK: - try: + # Child exited. Reconcile the asset we claimed: free it if the child left it non-finalized — + # a crash/OOM, or an early return on a job deleted mid-flight — a no-op when it finalized + # normally. Then, if it died non-zero while the job is still 'running', fail the job so it is + # not left stuck. Guarded: the child is already dead, so a DB blip here must not propagate. + try: + _reset_asset_pending(asset_id) + 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) - 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) + except Exception: # noqa: BLE001 — boot recovery is the backstop for a stuck 'running' job + log.exception("post-exit reconciliation failed for job %s", job_id) return - _stop_child(proc, grace_s) - _reset_asset_pending(asset_id) - _cleanup_staging(asset_id) - if reason == "cancelled": - # The route already set paused/cancelled (or the job is gone); just free the asset. - log.info("job %s: child stopped (route pause/cancel)", job_id) + if reason == "stalled": + _fail_after_kill( + job_id, asset_id, proc, grace_s, + f"The download stalled (no progress for {int(stall_s)}s) and was stopped.", "network", + ) return + if reason == "timeout": + _fail_after_kill( + job_id, asset_id, proc, grace_s, + f"The download exceeded the {int(wall_s)}s time limit and was stopped.", "network", + ) + return + # cancelled / abandon / shutdown: stop + free the asset, but leave the job's own DB status. + _stop_and_free(job_id, asset_id, proc, grace_s) if reason == "shutdown": _requeue_if_running(job_id, None) # resume on the next boot - return - msg = ( - f"The download stalled (no progress for {int(stall_s)}s) and was stopped." - if reason == "stalled" - else f"The download exceeded the {int(wall_s)}s time limit and was stopped." - ) - _set_job(job_id, status="error", error=msg, error_code="network", phase=None) - log.warning("job %s: stopped by watchdog (%s)", job_id, reason) + elif reason == "abandon": + log.warning("job %s: killed a child wedged past the wall-clock; DB status left intact", job_id) + else: # cancelled + log.info("job %s: child stopped (route pause/cancel)", job_id) def _cleanup_staging(asset_id: int) -> None: diff --git a/backend/tests/test_worker_supervisor.py b/backend/tests/test_worker_supervisor.py index 3a17d5a6..e4b66006 100644 --- a/backend/tests/test_worker_supervisor.py +++ b/backend/tests/test_worker_supervisor.py @@ -69,11 +69,11 @@ def _liveness(monkeypatch, value): def test_clean_child_exit_leaves_the_outcome_alone(monkeypatch, rec): - # Child finished on its own (exit 0) — it wrote the terminal state; the supervisor must not kill, - # reset, or re-fail anything. + # Child finished on its own (exit 0) — it wrote the terminal state. The supervisor only reconciles + # the asset (a no-op when the child finalized it) and must not kill or re-fail anything. _spawn(monkeypatch, _FakeProc([("exit", 0)])) worker._run_supervised(1, 99) - assert _names(rec) == [] + assert _names(rec) == ["reset_asset"] def test_no_progress_stall_kills_and_fails_the_job(monkeypatch, rec): @@ -154,7 +154,7 @@ def test_a_long_download_survives_while_progress_stays_fresh(monkeypatch, rec): _spawn(monkeypatch, _FakeProc(["timeout", "timeout", "timeout", ("exit", 0)])) _liveness(monkeypatch, ("running", 2.0)) # always well under the 10s stall worker._run_supervised(1, 99) - assert _names(rec) == [] + assert _names(rec) == ["reset_asset"] # only the clean-exit reconciliation (a no-op), no kill def test_a_finishing_child_is_not_mistaken_for_a_cancel(monkeypatch, rec): @@ -164,7 +164,19 @@ def test_a_finishing_child_is_not_mistaken_for_a_cancel(monkeypatch, rec): _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 + assert _names(rec) == ["reset_asset"] # not killed; only the clean-exit reconciliation (no-op) + + +def test_a_wedged_finished_child_is_killed_by_the_wall_clock(monkeypatch, rec): + # A child that reached a terminal state (done) but never exits (e.g. wedged in teardown on a stuck + # mount) must still be killed by the absolute wall-clock cap to free the worker slot — WITHOUT + # overwriting its DB status (it already succeeded). + monkeypatch.setattr(worker.settings, "download_wall_clock_max_seconds", 0) + _spawn(monkeypatch, _FakeProc(["timeout"])) # never exits within the test + _liveness(monkeypatch, ("done", 1.0)) # finished, but the process lingers + worker._run_supervised(1, 99) + names = _names(rec) + assert "stop_child" in names and "set_job" not in names # killed to free the slot, status kept def test_a_watchdog_error_still_stops_the_child(monkeypatch, rec):