release: 0.63.0 (improvement/download-subprocess-watchdog)
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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 <job_id>", 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()
|
||||
+225
-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
|
||||
@@ -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)
|
||||
|
||||
@@ -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 == []
|
||||
@@ -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}
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user