feat(downloads): run each download/edit in a supervised child process with a hard-kill watchdog
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).
This commit is contained in:
+166
-8
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user