fix(downloads): don't let the supervisor kill a finishing child (review round 2)
- worker.py: the watchdog loop now treats only 'paused'/'cancelled' (or a vanished row) as a stop; a terminal status (done/error, or a re-queued contended asset) means the child already wrote its outcome and is exiting, so keep polling instead of killing it. Previously a proc.wait timeout landing between the child's terminal commit and its exit was misread as a route-cancel and SIGKILLed a successful download (dropping its completion notification). The stall/wall-clock/shutdown checks are now gated on status=='running' (#1, #5 test). - worker.py: guard the crash-path _job_liveness probe so a DB blip there can't propagate (the child is already dead; boot recovery is the backstop) (#2). - worker.py: drop _stop_child's redundant poll guard — escalate_kill no-ops on an exited proc and getpgid yields None for a reaped pid (#3). - worker.py: note the extraction heartbeat is per-attempt (a single over-timeout attempt is still killed, bounded by socket_timeout) (#4).
This commit is contained in:
+34
-24
@@ -279,9 +279,8 @@ def _signal_group(proc: subprocess.Popen, pgid: int | None, sig: int) -> None:
|
|||||||
def _stop_child(proc: subprocess.Popen, grace_s: float) -> None:
|
def _stop_child(proc: subprocess.Popen, grace_s: float) -> None:
|
||||||
"""SIGTERM the child's process GROUP, then SIGKILL if it outlives the grace period. Shares the
|
"""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
|
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."""
|
that we signal the whole group so ffmpeg grandchildren die too. `escalate_kill` no-ops if the
|
||||||
if proc.poll() is not None:
|
child already exited, and `getpgid` yields None for a reaped pid, so no separate poll guard."""
|
||||||
return
|
|
||||||
try:
|
try:
|
||||||
pgid: int | None = os.getpgid(proc.pid)
|
pgid: int | None = os.getpgid(proc.pid)
|
||||||
except OSError:
|
except OSError:
|
||||||
@@ -326,18 +325,23 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
|
|||||||
except subprocess.TimeoutExpired:
|
except subprocess.TimeoutExpired:
|
||||||
pass
|
pass
|
||||||
status, stale = _job_liveness(job_id)
|
status, stale = _job_liveness(job_id)
|
||||||
if status is None or status != "running":
|
if status == "running":
|
||||||
reason = "cancelled" # a route paused/cancelled it (or the row vanished)
|
# Only a genuinely-running child is subject to the stall / wall-clock / shutdown kill.
|
||||||
break
|
if stale is not None and stale > stall_s:
|
||||||
if stale is not None and stale > stall_s:
|
reason = "stalled"
|
||||||
reason = "stalled"
|
break
|
||||||
break
|
if time.monotonic() - started > wall_s:
|
||||||
if time.monotonic() - started > wall_s:
|
reason = "timeout"
|
||||||
reason = "timeout"
|
break
|
||||||
break
|
if _stop.is_set():
|
||||||
if _stop.is_set():
|
reason = "shutdown"
|
||||||
reason = "shutdown"
|
break
|
||||||
|
elif status is None or status in ("paused", "cancelled"):
|
||||||
|
reason = "cancelled" # a route stopped it (or the row vanished)
|
||||||
break
|
break
|
||||||
|
# else: a terminal status (done / error, or a re-queued contended asset) means the child
|
||||||
|
# already wrote its outcome and is exiting on its own — keep polling; the next proc.wait
|
||||||
|
# catches the exit. 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
|
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)
|
log.exception("supervisor loop failed for job %s; stopping the child", job_id)
|
||||||
_stop_child(proc, grace_s)
|
_stop_child(proc, grace_s)
|
||||||
@@ -351,17 +355,21 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
|
|||||||
|
|
||||||
if reason is None:
|
if reason is None:
|
||||||
# Child exited. If it died WITHOUT setting a terminal state (non-zero exit / crash / OOM),
|
# 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.
|
# the job is still 'running' — fail it here so nothing is left stuck. Guarded: the child is
|
||||||
|
# already dead (no orphan risk), so a DB blip in this probe must not propagate.
|
||||||
if proc.returncode != _CHILD_OK:
|
if proc.returncode != _CHILD_OK:
|
||||||
status, _ = _job_liveness(job_id)
|
try:
|
||||||
if status == "running":
|
status, _ = _job_liveness(job_id)
|
||||||
_reset_asset_pending(asset_id)
|
if status == "running":
|
||||||
_cleanup_staging(asset_id)
|
_reset_asset_pending(asset_id)
|
||||||
_set_job(
|
_cleanup_staging(asset_id)
|
||||||
job_id, status="error", error_code="unknown", phase=None,
|
_set_job(
|
||||||
error=f"The download process exited unexpectedly (code {proc.returncode}).",
|
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)
|
)
|
||||||
|
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 cleanup failed for job %s", job_id)
|
||||||
return
|
return
|
||||||
|
|
||||||
_stop_child(proc, grace_s)
|
_stop_child(proc, grace_s)
|
||||||
@@ -712,6 +720,8 @@ def _extract_with_fallback(job_id: int, asset_id: int, url: str, spec: dict, sta
|
|||||||
opts["postprocessor_hooks"] = [_make_pp_hook(job_id)]
|
opts["postprocessor_hooks"] = [_make_pp_hook(job_id)]
|
||||||
_touch_job(job_id) # stall-watchdog heartbeat: extraction writes no progress, so a slow /
|
_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.
|
# 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:
|
try:
|
||||||
with yt_dlp.YoutubeDL(opts) as ydl:
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
||||||
return ydl.extract_info(url, download=True)
|
return ydl.extract_info(url, download=True)
|
||||||
|
|||||||
@@ -157,6 +157,16 @@ def test_a_long_download_survives_while_progress_stays_fresh(monkeypatch, rec):
|
|||||||
assert _names(rec) == []
|
assert _names(rec) == []
|
||||||
|
|
||||||
|
|
||||||
|
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) == [] # never killed, asset/job untouched
|
||||||
|
|
||||||
|
|
||||||
def test_a_watchdog_error_still_stops_the_child(monkeypatch, rec):
|
def test_a_watchdog_error_still_stops_the_child(monkeypatch, rec):
|
||||||
# If the liveness probe raises (e.g. a transient DB error), the child must NOT be left running
|
# 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.
|
# unsupervised — the feature's whole point. It is killed and the job is failed.
|
||||||
|
|||||||
Reference in New Issue
Block a user