From 0ed82fe6b6226cf715628ebbef714515bdab5ee7 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Wed, 12 Aug 2026 22:39:17 +0200 Subject: [PATCH] feat(downloads): run each download/edit in a supervised child process with a hard-kill watchdog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Defense-in-depth after the ReDoS incident. The heavy work (yt-dlp download / ffmpeg edit) now runs in a child process (app.downloads.runner) supervised by the worker thread, not in the thread itself. A thread can't be killed and a C-level spin can't even be cooperatively cancelled, so one wedged download used to pin a core, hold the GIL and freeze the whole queue; a child process can be SIGTERM/SIGKILLed. The supervisor kills the child (and its process group, so ffmpeg grandchildren die too) when: a route pauses/cancels the job, it makes no progress for download_stall_timeout_seconds (150s), it runs past download_wall_clock_max_seconds (2h), the worker is shutting down, or the child crashes without setting state. On a kill it frees the asset and fails the job with a retryable network error (stalls are not auto-requeued). A clean exit is left alone — the child already wrote the terminal state, including a cooperative pause via the progress hook. - config.py: stall / wall-clock / supervisor-poll / kill-grace settings. - worker.py: _run_supervised + _spawn_download_child / _job_liveness / _stop_child / _cleanup_staging; _process_job claims the asset, then hands the heavy work to the child. - app/downloads/runner.py: the child entrypoint. - docker-compose.localdev.yml: init:true on the worker so tini (PID 1) reaps orphaned ffmpeg grandchildren left by a hard-kill (the worker reaps its own direct children). - tests: watchdog decision logic + SIGTERM->SIGKILL escalation (fake proc, no real subprocess). --- backend/app/config.py | 12 ++ backend/app/downloads/runner.py | 51 +++++++ backend/app/worker.py | 174 +++++++++++++++++++++-- backend/tests/test_worker_supervisor.py | 175 ++++++++++++++++++++++++ docker-compose.localdev.yml | 3 + 5 files changed, 407 insertions(+), 8 deletions(-) create mode 100644 backend/app/downloads/runner.py create mode 100644 backend/tests/test_worker_supervisor.py diff --git a/backend/app/config.py b/backend/app/config.py index 78bb9a6f..e0f2707f 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -173,6 +173,18 @@ class Settings(BaseSettings): download_worker_concurrency: int = 2 # How often (empty pending queue) the worker polls for a new job, in seconds. download_poll_seconds: int = 5 + # --- download-subprocess watchdog (each download/edit runs in a supervised child process so a + # wedged yt-dlp/ffmpeg can be hard-killed; a thread cannot) --- + # Kill a running download if its job row hasn't advanced (no progress written) for this long. + # Generous: a live download writes progress ~1/s, so only a genuine 0-progress stall trips it. + download_stall_timeout_seconds: int = 150 + # Absolute ceiling on a single download/edit regardless of progress (a legitimately huge file + # still finishes well under this; a runaway is capped). + download_wall_clock_max_seconds: int = 7200 # 2 hours + # How often the supervisor checks the child's liveness + the job row. + download_supervisor_poll_seconds: int = 3 + # Grace between SIGTERM and SIGKILL when stopping a child (matches run_ffmpeg's escalation). + download_kill_grace_seconds: int = 5 # Total cache size cap across ALL users (bytes). When >0 and exceeded, the GC evicts the # least-recently-accessed ready assets until under it. 0 = unlimited (disk is the limit). download_total_max_bytes: int = 0 diff --git a/backend/app/downloads/runner.py b/backend/app/downloads/runner.py new file mode 100644 index 00000000..44e9f64d --- /dev/null +++ b/backend/app/downloads/runner.py @@ -0,0 +1,51 @@ +"""Download/edit child process — runs ONE job's heavy work (yt-dlp download or ffmpeg edit) in its +own process, launched and supervised by the worker (`app.worker._run_supervised`). + +Why a separate process: a worker THREAD can't be killed, a C-level spin (a catastrophic regex, a +tight yt-dlp retry loop) can't even be cooperatively cancelled, and one such thread starves the +others via the GIL. Running the work in a child means the supervisor can SIGTERM/SIGKILL it (and its +process group, so ffmpeg grandchildren die too) on a stall / pause / wall-clock breach. + +Contract: the asset is already claimed ('downloading') by the parent. This process does the work and +writes the terminal job/asset state itself (done / error, or a cooperative pause via the in-flight +progress hook), then exits 0. An unexpected crash exits non-zero and the parent fails the job. +""" +import logging +import sys + +from app.db import SessionLocal +from app.models import DownloadJob + +log = logging.getLogger("siftlode.worker.runner") + + +def main() -> None: + logging.basicConfig(level=logging.INFO) + if len(sys.argv) != 2: + print("usage: python -m app.downloads.runner ", file=sys.stderr) + raise SystemExit(2) + job_id = int(sys.argv[1]) + + with SessionLocal() as db: + job = db.get(DownloadJob, job_id) + if job is None: + return + asset_id = job.asset_id + job_kind = job.job_kind + spec = dict(job.profile_snapshot or {}) + source_kind, source_ref = job.source_kind, job.source_ref + if asset_id is None: + return + + # Imported here (not at module top) so the child pays the yt-dlp import cost only when it has + # real work, and to keep the import graph obvious: runner is a thin entry over worker's helpers. + from app.worker import _download, _process_edit + + if job_kind == "edit": + _process_edit(job_id, asset_id) + else: + _download(job_id, asset_id, source_kind, source_ref, spec) + + +if __name__ == "__main__": + main() diff --git a/backend/app/worker.py b/backend/app/worker.py index 0364703e..57dce801 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -16,9 +16,12 @@ Design: * Crash-safe: on startup, orphaned `running` jobs are requeued and `downloading` assets reset. """ import logging +import os import re import shutil import signal +import subprocess +import sys import threading import time from datetime import date, datetime, timedelta, timezone @@ -205,19 +208,176 @@ def _claim_asset(asset_id: int) -> bool: return row is not None +# --- supervised child execution + watchdog -------------------------------------------------- +# The heavy work runs in a child process so a wedged yt-dlp/ffmpeg is HARD-killable. A worker thread +# can't be killed, a C-level spin (e.g. a catastrophic regex) can't even be cooperatively cancelled, +# 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. + +_CHILD_OK = 0 + + +def _spawn_download_child(job_id: int) -> subprocess.Popen: + """Start the child that does the actual download/edit for `job_id`, in its OWN session/process + group so stopping it also stops any ffmpeg / helper processes yt-dlp spawned.""" + return subprocess.Popen( + [sys.executable, "-m", "app.downloads.runner", str(job_id)], + start_new_session=True, + ) + + +def _job_liveness(job_id: int) -> tuple[str | None, float | None]: + """(status, seconds-since-last-row-update) for the job, computed with the DB clock so there is + no parent↔DB skew. A live download rewrites its row ~1/s; a stalled one never does. Returns + (None, None) if the row is gone.""" + with SessionLocal() as db: + row = db.execute( + text( + "SELECT status, EXTRACT(EPOCH FROM (now() - updated_at)) " + "FROM download_jobs WHERE id = :id" + ), + {"id": job_id}, + ).one_or_none() + if row is None: + return None, None + return row[0], (float(row[1]) if row[1] is not None else None) + + +def _signal_group(proc: subprocess.Popen, pgid: int | None, sig: int) -> None: + """Signal the child's whole process group (falling back to the child alone), swallowing the race + where it has already exited.""" + try: + if pgid is not None: + os.killpg(pgid, sig) + else: + proc.send_signal(sig) + except (ProcessLookupError, OSError): + pass + + +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.""" + 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) + + +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. + + A clean child exit means the child already wrote the terminal job/asset state (done/error, or a + cooperative pause via the in-child progress hook) — we leave it. We step in only when the child + can't finish or self-abort: a route paused/cancelled it (kill, free the asset, keep the route's + status), it made no progress for `stall_timeout` or ran past `wall_clock_max` (kill, fail the + job), the worker is shutting down (kill, requeue), or it crashed without setting state.""" + poll_s = max(1, settings.download_supervisor_poll_seconds) + stall_s = settings.download_stall_timeout_seconds + wall_s = settings.download_wall_clock_max_seconds + grace_s = settings.download_kill_grace_seconds + + try: + proc = _spawn_download_child(job_id) + except Exception: # noqa: BLE001 — spawn failed; don't leave the asset stuck 'downloading' + log.exception("failed to spawn download child for job %s", job_id) + _reset_asset_pending(asset_id) + _requeue_if_running(job_id, None) + return + + 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 + + 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. + 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) + 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) + return + 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) + + +def _cleanup_staging(asset_id: int) -> None: + """Best-effort removal of a killed child's leftover staging dirs (its own `finally` never ran). + The next attempt on this asset also wipes staging on start, so this is just tidiness/disk.""" + for path in (_staging_dir(asset_id), _edit_staging_dir(asset_id)): + shutil.rmtree(path, ignore_errors=True) + + # --- processing ----------------------------------------------------------------------------- def _process_job(job_id: int) -> None: + """Dispatch a claimed job. The trivial outcomes (cache hit, dead/contended asset) are handled + inline and cheaply; the actual heavy work (yt-dlp download / ffmpeg edit) is claimed here and + then run in a supervised CHILD PROCESS (`_run_supervised`) so a wedged yt-dlp/ffmpeg can be + hard-killed — a worker THREAD cannot be killed, nor (when it spins in a C call) even + cooperatively cancelled, and one such thread starves the others via the GIL.""" with SessionLocal() as db: job = db.get(DownloadJob, job_id) if job is None: return asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None - spec = dict(job.profile_snapshot or {}) - source_kind, source_ref = job.source_kind, job.source_ref - job_kind = job.job_kind asset_id = asset.id if asset else None asset_status = asset.status if asset else None + asset_error = asset.error if asset else None if asset is None: _set_job(job_id, status="error", error="No media asset", phase=None) @@ -227,7 +387,7 @@ def _process_job(job_id: int) -> None: _finish_from_cache(job_id, asset_id) return if asset_status == "error": - _set_job(job_id, status="error", error=asset.error or "Download failed", phase=None) + _set_job(job_id, status="error", error=asset_error or "Download failed", phase=None) return if asset_status == "downloading": # Another worker owns this asset; wait and let the job be reclaimed later. @@ -241,10 +401,8 @@ def _process_job(job_id: int) -> None: time.sleep(2) return - if job_kind == "edit": - _process_edit(job_id, asset_id) - else: - _download(job_id, asset_id, source_kind, source_ref, spec) + # The asset is now ours ('downloading'): run the heavy work in a supervised child process. + _run_supervised(job_id, asset_id) def _finish_from_cache(job_id: int, asset_id: int) -> None: diff --git a/backend/tests/test_worker_supervisor.py b/backend/tests/test_worker_supervisor.py new file mode 100644 index 00000000..ce8373c6 --- /dev/null +++ b/backend/tests/test_worker_supervisor.py @@ -0,0 +1,175 @@ +"""The download supervisor/watchdog: `_run_supervised` decides when to hard-kill a child, and +`_stop_child` escalates SIGTERM→SIGKILL. No real subprocess or DB here — a fake proc stands in and +the DB-liveness lookup + side effects are monkeypatched, so the decision logic is exercised on any +machine. These are regression tests for the class of bug that motivated the whole change: a wedged +download that a thread could neither kill nor cancel. +""" +import signal +import subprocess + +import pytest + +from app import worker + + +class _FakeProc: + """Stands in for the child Popen. `waits` scripts each wait() call: "timeout" = still alive, + ("exit", rc) = the child finished with that code.""" + + def __init__(self, waits, pid=4242): + self._waits = list(waits) + self.pid = pid + self.returncode = None + self.signals = [] + + def wait(self, timeout=None): + step = self._waits.pop(0) if self._waits else "timeout" + if step == "timeout": + raise subprocess.TimeoutExpired("child", timeout or 0) + self.returncode = step[1] + return step[1] + + def poll(self): + return self.returncode + + def send_signal(self, sig): + self.signals.append(sig) + + +@pytest.fixture +def rec(monkeypatch): + """Record supervisor side effects and stub out the real spawn/kill/liveness/staging so a test + drives them. `calls` collects (name, kwargs) so assertions read the decisions, not the plumbing.""" + calls = [] + + def record(name): + def _f(*args, **kwargs): + calls.append((name, args, kwargs)) + return _f + + monkeypatch.setattr(worker, "_reset_asset_pending", record("reset_asset")) + monkeypatch.setattr(worker, "_requeue_if_running", record("requeue")) + monkeypatch.setattr(worker, "_set_job", record("set_job")) + monkeypatch.setattr(worker, "_cleanup_staging", record("cleanup")) + monkeypatch.setattr(worker, "_stop_child", record("stop_child")) + monkeypatch.setattr(worker._stop, "is_set", lambda: False) + return calls + + +def _names(calls): + return [c[0] for c in calls] + + +def _spawn(monkeypatch, proc): + monkeypatch.setattr(worker, "_spawn_download_child", lambda job_id: proc) + + +def _liveness(monkeypatch, value): + monkeypatch.setattr(worker, "_job_liveness", lambda job_id: 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. + _spawn(monkeypatch, _FakeProc([("exit", 0)])) + worker._run_supervised(1, 99) + assert _names(rec) == [] + + +def test_no_progress_stall_kills_and_fails_the_job(monkeypatch, rec): + _spawn(monkeypatch, _FakeProc(["timeout"])) + _liveness(monkeypatch, ("running", 9999.0)) # far past the stall timeout + worker._run_supervised(1, 99) + names = _names(rec) + assert "stop_child" in names and "reset_asset" in names + set_job = next(c for c in rec if c[0] == "set_job") + assert set_job[2]["status"] == "error" and set_job[2]["error_code"] == "network" + assert "stalled" in set_job[2]["error"].lower() + + +def test_route_pause_kills_but_keeps_the_route_status(monkeypatch, rec): + # A route flipped the job to 'paused'/'cancelled'. Kill + free the asset, but do NOT overwrite the + # status with an error and do NOT requeue. + _spawn(monkeypatch, _FakeProc(["timeout"])) + _liveness(monkeypatch, ("paused", 1.0)) + worker._run_supervised(1, 99) + names = _names(rec) + assert "stop_child" in names and "reset_asset" in names + assert "set_job" not in names and "requeue" not in names + + +def test_wall_clock_cap_kills_even_while_progressing(monkeypatch, rec): + monkeypatch.setattr(worker.settings, "download_wall_clock_max_seconds", 0) + _spawn(monkeypatch, _FakeProc(["timeout"])) + _liveness(monkeypatch, ("running", 1.0)) # fresh progress, but over the absolute cap + worker._run_supervised(1, 99) + set_job = next(c for c in rec if c[0] == "set_job") + assert set_job[2]["status"] == "error" + assert "time limit" in set_job[2]["error"].lower() + + +def test_child_crash_without_finishing_is_failed(monkeypatch, rec): + # Non-zero exit and the job is still 'running' → the child died (crash/OOM) before writing state. + _spawn(monkeypatch, _FakeProc([("exit", 1)])) + _liveness(monkeypatch, ("running", 1.0)) + worker._run_supervised(1, 99) + names = _names(rec) + assert "reset_asset" in names + set_job = next(c for c in rec if c[0] == "set_job") + assert set_job[2]["status"] == "error" and "unexpectedly" in set_job[2]["error"].lower() + + +def test_child_nonzero_exit_that_already_set_state_is_left_alone(monkeypatch, rec): + # Child exited non-zero but the job is no longer 'running' (it recorded its own error) → no double. + _spawn(monkeypatch, _FakeProc([("exit", 1)])) + _liveness(monkeypatch, ("error", None)) + worker._run_supervised(1, 99) + assert "set_job" not in _names(rec) + + +def test_spawn_failure_frees_the_asset_and_requeues(monkeypatch, rec): + def boom(job_id): + raise OSError("no fork") + + monkeypatch.setattr(worker, "_spawn_download_child", boom) + worker._run_supervised(1, 99) + names = _names(rec) + assert "reset_asset" in names and "requeue" in names + + +def test_shutdown_kills_and_requeues_for_next_boot(monkeypatch, rec): + monkeypatch.setattr(worker._stop, "is_set", lambda: True) + _spawn(monkeypatch, _FakeProc(["timeout"])) + _liveness(monkeypatch, ("running", 1.0)) + worker._run_supervised(1, 99) + names = _names(rec) + assert "stop_child" in names and "requeue" in names and "set_job" not in names + + +# --- _stop_child escalation ----------------------------------------------------------------- + +@pytest.fixture +def killpg_rec(monkeypatch): + sent = [] + monkeypatch.setattr(worker.os, "getpgid", lambda pid: pid) + monkeypatch.setattr(worker.os, "killpg", lambda pgid, sig: sent.append(sig)) + return sent + + +def test_stop_child_sigterm_is_enough(killpg_rec): + proc = _FakeProc([("exit", 0)]) # dies on the SIGTERM wait + worker._stop_child(proc, grace_s=0.01) + assert killpg_rec == [signal.SIGTERM] + + +def test_stop_child_escalates_to_sigkill(killpg_rec): + proc = _FakeProc(["timeout", ("exit", -9)]) # survives SIGTERM, dies on SIGKILL + worker._stop_child(proc, grace_s=0.01) + assert killpg_rec == [signal.SIGTERM, signal.SIGKILL] + + +def test_stop_child_noop_if_already_exited(killpg_rec): + proc = _FakeProc([]) + proc.returncode = 0 # already gone + worker._stop_child(proc, grace_s=0.01) + assert killpg_rec == [] diff --git a/docker-compose.localdev.yml b/docker-compose.localdev.yml index 988afd6f..48560639 100644 --- a/docker-compose.localdev.yml +++ b/docker-compose.localdev.yml @@ -76,6 +76,9 @@ services: GIT_SHA: ${GIT_SHA:-unknown} BUILD_DATE: ${BUILD_DATE:-} command: ["python", "-m", "app.worker"] + # tini as PID 1: reaps orphaned ffmpeg/yt-dlp grandchildren when the watchdog hard-kills a + # download child's process group (the worker reaps its own direct children via subprocess.wait). + init: true env_file: .env environment: DATABASE_URL: postgresql+psycopg://${POSTGRES_USER:-siftlode}:${POSTGRES_PASSWORD:-siftlode}@db:5432/${POSTGRES_DB:-siftlode}