refactor(downloads): address /code-review on the download supervisor

- worker.py: wrap the watchdog loop so a failure in it (e.g. a transient DB error in
  _job_liveness) still hard-kills the child + fails the job, instead of leaving the child
  running unsupervised — the exact wedge the feature prevents (#1).
- worker.py: _touch_job heartbeat bumps updated_at at each extraction attempt, so a slow /
  multi-client-set fallback extraction (which writes no progress, and re-writes identical
  values so ORM onupdate does not fire) is not false-killed as "stalled" (#2).
- edit.py: extract escalate_kill (terminate->wait->kill escalation); _stop and worker's
  _stop_child (process-group variant) now share it (#3).
- worker.py: document the edit path's second, tighter run_ffmpeg watchdog and that the
  supervisor is POSIX/Linux-container-only (#4, #6).
- tests: a long download stays alive while progress is fresh, and a watchdog-loop error
  still stops the child (#1, #5).

backend/app/downloads/edit.py

@@ -288,20 +288,30 @@ def _drain_tail(stream, sink: list[str]) -> None:
         pass
 
 
-def _stop(proc: subprocess.Popen) -> None:
-    """Terminate, then kill — never wait unbounded on a process we've given up on. Safe to call on
-    a process that already exited (the single exit path below may reach it after a clean wait)."""
+def escalate_kill(proc: subprocess.Popen, terminate, kill, grace_s: float, label: str) -> None:
+    """terminate → wait(grace) → kill → wait(grace) escalation for a process we've given up on, so it
+    is never waited on unboundedly. `terminate`/`kill` deliver the stop to whatever scope the caller
+    owns — a single process (`proc.terminate`/`proc.kill`) or its whole group (`os.killpg`). No-op if
+    the process already exited."""
     if proc.poll() is not None:
         return
-    proc.terminate()
+    terminate()
     try:
-        proc.wait(timeout=5)
+        proc.wait(timeout=grace_s)
+        return
     except subprocess.TimeoutExpired:
-        proc.kill()
-        try:
-            proc.wait(timeout=5)
-        except subprocess.TimeoutExpired:
-            log.error("ffmpeg pid %s survived SIGKILL", proc.pid)
+        pass
+    kill()
+    try:
+        proc.wait(timeout=grace_s)
+    except subprocess.TimeoutExpired:
+        log.error("%s pid %s survived SIGKILL", label, proc.pid)
+
+
+def _stop(proc: subprocess.Popen) -> None:
+    """Terminate, then kill — never wait unbounded on a process we've given up on. Safe to call on
+    a process that already exited."""
+    escalate_kill(proc, proc.terminate, proc.kill, grace_s=5, label="ffmpeg")
 
 
 def run_ffmpeg(cmd, total_s, on_progress, should_cancel, stall_timeout_s=_FFMPEG_STALL_TIMEOUT_S) -> None:

backend/app/worker.py

@@ -96,6 +96,20 @@ def _job_status(job_id: int) -> str | None:
         ).scalar()
 
 
+def _touch_job(job_id: int) -> None:
+    """Bump a running job's `updated_at` without changing any value — a stall-watchdog heartbeat for
+    phases that legitimately write no progress. Extraction writes nothing, and a multi-client-set
+    fallback re-writes identical values (progress 0 / phase 'downloading'), so ORM `onupdate` would
+    NOT fire and two slow extraction attempts could otherwise cross the stall timeout and be killed
+    mid-extraction though they'd have succeeded."""
+    with SessionLocal() as db:
+        db.execute(
+            text("UPDATE download_jobs SET updated_at=now() WHERE id=:id AND status='running'"),
+            {"id": job_id},
+        )
+        db.commit()
+
+
 def _requeue_if_running(job_id: int, phase: str) -> None:
     """Put the job back to 'queued' for a later retry, but ONLY while it is still 'running'. The
     route flips status to paused/cancelled to stop a job; a blind `status='queued'` here would
@@ -214,6 +228,12 @@ def _claim_asset(asset_id: int) -> bool:
 # and one such thread starves the rest via the GIL — the incident that motivated this. The parent
 # thread supervises: it kills the child (and its process group, so ffmpeg grandchildren die too) on
 # a route pause/cancel, on a no-progress stall, or past an absolute wall-clock cap.
+#
+# POSIX-only: the kill path uses process groups (start_new_session / os.killpg / os.getpgid), so the
+# worker is Linux-container-only — never run `python -m app.worker` natively on Windows.
+# Edits ALSO carry a second, tighter watchdog inside run_ffmpeg (a per-ffmpeg stall timeout with a
+# cleaner "ffmpeg stalled" error); this process-level supervisor is the uniform backstop for both the
+# download and edit paths, and the last line of defence for anything run_ffmpeg can't see.
 
 _CHILD_OK = 0
 
@@ -257,25 +277,22 @@ 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 — the same
-    terminate→wait→kill escalation as run_ffmpeg, so a child that ignores SIGTERM is still reaped."""
+    """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
     try:
         pgid: int | None = os.getpgid(proc.pid)
     except OSError:
         pgid = None
-    _signal_group(proc, pgid, signal.SIGTERM)
-    try:
-        proc.wait(timeout=grace_s)
-        return
-    except subprocess.TimeoutExpired:
-        pass
-    _signal_group(proc, pgid, signal.SIGKILL)
-    try:
-        proc.wait(timeout=grace_s)
-    except subprocess.TimeoutExpired:
-        log.error("download child pid %s survived SIGKILL", proc.pid)
+    editmod.escalate_kill(
+        proc,
+        terminate=lambda: _signal_group(proc, pgid, signal.SIGTERM),
+        kill=lambda: _signal_group(proc, pgid, signal.SIGKILL),
+        grace_s=grace_s,
+        label="download child",
+    )
 
 
 def _run_supervised(job_id: int, asset_id: int) -> None:
@@ -301,25 +318,36 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
 
     started = time.monotonic()
     reason: str | None = None
-    while True:
-        try:
-            proc.wait(timeout=poll_s)
-            break  # child finished on its own
-        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"
-            break
+    try:
+        while True:
+            try:
+                proc.wait(timeout=poll_s)
+                break  # child finished on its own
+            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"
+                break
+    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.",
+        )
+        return
 
     if reason is None:
         # Child exited. If it died WITHOUT setting a terminal state (non-zero exit / crash / OOM),
@@ -682,6 +710,8 @@ def _extract_with_fallback(job_id: int, asset_id: int, url: str, spec: dict, sta
             settings.download_pot_base_url, clients,
         )
         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.
         try:
             with yt_dlp.YoutubeDL(opts) as ydl:
                 return ydl.extract_info(url, download=True)

backend/tests/test_worker_supervisor.py

@@ -146,6 +146,32 @@ def test_shutdown_kills_and_requeues_for_next_boot(monkeypatch, rec):
     assert "stop_child" in names and "requeue" in names and "set_job" not in names
 
 
+def test_a_long_download_survives_while_progress_stays_fresh(monkeypatch, rec):
+    # A download that outlives the stall timeout must NOT be killed as long as it keeps writing
+    # progress (updated_at stays fresh via the model's onupdate). Model: alive across several polls,
+    # liveness always fresh, then a clean exit — nothing killed, nothing failed.
+    monkeypatch.setattr(worker.settings, "download_stall_timeout_seconds", 10)
+    _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) == []
+
+
+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.
+    _spawn(monkeypatch, _FakeProc(["timeout"]))
+
+    def boom(job_id):
+        raise RuntimeError("db down")
+
+    monkeypatch.setattr(worker, "_job_liveness", boom)
+    worker._run_supervised(1, 99)
+    names = _names(rec)
+    assert "stop_child" in names and "reset_asset" in names
+    assert next(c for c in rec if c[0] == "set_job")[2]["status"] == "error"
+
+
 # --- _stop_child escalation -----------------------------------------------------------------
 
 @pytest.fixture