fix(downloads): supervisor edge-cases from review round 3

- 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

@@ -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,43 +342,42 @@ 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:
+        # 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,
@@ -369,26 +385,29 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
                     )
                     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)
+            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:

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):