diff --git a/VERSION b/VERSION index 7c6425a5..c149bbff 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.62.2 \ No newline at end of file +0.63.0 \ No newline at end of file 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/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/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..7bd3f764 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 @@ -93,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 @@ -205,19 +222,217 @@ 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. +# +# 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 + + +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. 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. `escalate_kill` no-ops if the + child already exited, and `getpgid` yields None for a reaped pid, so no separate poll guard.""" + try: + pgid: int | None = os.getpgid(proc.pid) + except OSError: + pgid = None + 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 _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. + + 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 + 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 in ("paused", "cancelled"): + reason = "cancelled" # a route stopped it (or the row vanished) + break + if status == "running": + if stale is not None and stale > stall_s: + reason = "stalled" + break + if _stop.is_set(): + reason = "shutdown" + break + # 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 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) + _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. 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": + _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 reconciliation failed for job %s", job_id) + return + + 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 + 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: + """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 +442,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 +456,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: @@ -524,6 +737,10 @@ 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. + # (Per-ATTEMPT only; a single extract_info that itself runs past the stall timeout is still + # killed — acceptable, as socket_timeout + extractor_retries bound one attempt well under it.) 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 new file mode 100644 index 00000000..e4b66006 --- /dev/null +++ b/backend/tests/test_worker_supervisor.py @@ -0,0 +1,223 @@ +"""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 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) == ["reset_asset"] + + +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 + + +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) == ["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): + # The child committed its terminal state (done/error) but hasn't exited yet (e.g. slow staging + # cleanup outlasting a poll). A non-'running' status must NOT be treated as a route-cancel and + # kill the finisher (which could drop its completion notification) — let it exit on its own. + _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) == ["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): + # 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 +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} diff --git a/frontend/src/lib/releaseNotes.ts b/frontend/src/lib/releaseNotes.ts index 4578aacc..3a6673fc 100644 --- a/frontend/src/lib/releaseNotes.ts +++ b/frontend/src/lib/releaseNotes.ts @@ -14,6 +14,13 @@ export interface ReleaseEntry { } export const RELEASE_NOTES: ReleaseEntry[] = [ + { + version: "0.63.0", + date: "2026-08-12", + chores: [ + "Hardened the download worker: each download or clip now runs in its own supervised process, so a stuck or misbehaving one can be stopped cleanly and can never freeze the rest of the download queue. No change to how downloads look or work.", + ], + }, { version: "0.62.2", date: "2026-08-12",