diff --git a/backend/app/downloads/edit.py b/backend/app/downloads/edit.py index ed671a47..45337dad 100644 --- a/backend/app/downloads/edit.py +++ b/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: diff --git a/backend/app/worker.py b/backend/app/worker.py index 57dce801..4d3cef68 100644 --- a/backend/app/worker.py +++ b/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) diff --git a/backend/tests/test_worker_supervisor.py b/backend/tests/test_worker_supervisor.py index ce8373c6..ca80d82b 100644 --- a/backend/tests/test_worker_supervisor.py +++ b/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