fix(pipeline): let cancellation reach the processes it is meant to stop

app/api/jobs.py

@@ -326,6 +326,16 @@ def cancel_job(job_id: str) -> dict:
     if job is None:
         raise HTTPException(status_code=404, detail="job not found")
     if job.status in ("done", "error", "cancelled"):
+        # A vocal split only ever runs on a done job, so this early return made
+        # it uncancellable by construction: the flag was never even set, while
+        # the split held _pipeline_lock and stalled the whole import queue for
+        # its full duration (#519). Terminating the worker is enough -- the
+        # split's own error path marks it failed and releases the lock.
+        if job.vocal_split == "running":
+            job.cancel_requested = True
+            proc = registry_get_proc(job_id)
+            if proc is not None and proc.poll() is None:
+                proc.terminate()
         return job.to_state()
     job.cancel_requested = True
 

app/core/process.py

@@ -38,3 +38,53 @@ def process_exists(pid: int) -> bool:
         kernel32.CloseHandle(handle)
         return True
     return ctypes.get_last_error() != ERROR_INVALID_PARAMETER
+
+
+_PARENT_POLL_SECONDS = 1.0
+
+
+def _watch_parent(parent_pid: int) -> None:
+    """Exit as soon as the process that spawned us is gone.
+
+    A worker's stdin EOF only covers a parent that exits between jobs.
+    Mid-inference the worker reads nothing, and the parent may have been killed
+    in a way that ran no cleanup at all (SIGKILL, Force Quit, Task Manager, a
+    crash). Without this the worker keeps running -- holding a GPU, in the
+    demucs and vocal-split cases -- with nobody left to collect the result.
+
+    os._exit rather than sys.exit: this runs on a daemon thread, and raising
+    SystemExit there would not interrupt inference running in C code. Nothing
+    here needs flushing.
+    """
+    import sys
+    import time
+
+    while True:
+        if not process_exists(parent_pid):
+            sys.stderr.write("@@ERROR@@parent process exited\n")
+            sys.stderr.flush()
+            os._exit(1)
+        time.sleep(_PARENT_POLL_SECONDS)
+
+
+def arm_parent_watchdog() -> None:
+    """Start the parent-death watchdog if the parent asked for one.
+
+    Shared by every long-running worker. It lived in demucs_worker.py, which is
+    why vocal_split_worker and section_worker never had it: a Force-Quit during
+    a vocal split orphaned an onnxruntime process holding the GPU, and a
+    section pass outlived the parent whose TIMEOUT_SECTIONS was its only bound
+    (#519).
+    """
+    import threading
+
+    raw = os.environ.get("STEMDECK_PARENT_PID", "").strip()
+    if not raw:
+        return
+    try:
+        parent_pid = int(raw)
+    except ValueError:
+        return
+    if parent_pid <= 0 or parent_pid == os.getpid():
+        return
+    threading.Thread(target=_watch_parent, args=(parent_pid,), daemon=True).start()

app/pipeline/demucs_worker.py

@@ -35,14 +35,11 @@
 from __future__ import annotations
 
 import json
-import os
 import sys
-import threading
-import time
 from pathlib import Path
 
 from app.core.config import DEMUCS_MODEL
-from app.core.process import process_exists
+from app.core.process import arm_parent_watchdog
 
 
 def _run_one_job(model, device: str, req: dict) -> None:
@@ -89,47 +86,9 @@ def _run_one_job(model, device: str, req: dict) -> None:
         )
 
 
-_PARENT_POLL_SECONDS = 1.0
-
-
-def _watch_parent(parent_pid: int) -> None:
-    """Exit as soon as the process that spawned us is gone.
-
-    The stdin EOF in the loop below only covers a parent that exits between
-    jobs. Mid-separation the worker is inside torch and reads nothing, and the
-    parent may have been killed in a way that ran no cleanup at all (SIGKILL,
-    Force Quit, Task Manager, a crash). Without this, the worker would keep a
-    GPU busy with nobody left to collect the result.
-
-    os._exit rather than sys.exit: this runs on a daemon thread, and raising
-    SystemExit there would not interrupt inference running in C code. Nothing
-    here needs flushing -- a half-written model directory is cleared before the
-    job is retried.
-    """
-    while True:
-        if not process_exists(parent_pid):
-            sys.stderr.write("@@ERROR@@parent process exited\n")
-            sys.stderr.flush()
-            os._exit(1)
-        time.sleep(_PARENT_POLL_SECONDS)
-
-
-def _arm_parent_watchdog() -> None:
-    raw = os.environ.get("STEMDECK_PARENT_PID", "").strip()
-    if not raw:
-        return
-    try:
-        parent_pid = int(raw)
-    except ValueError:
-        return
-    if parent_pid <= 0 or parent_pid == os.getpid():
-        return
-    threading.Thread(target=_watch_parent, args=(parent_pid,), daemon=True).start()
-
-
 def main() -> None:
     device = sys.argv[1] if len(sys.argv) > 1 else "cpu"
-    _arm_parent_watchdog()
+    arm_parent_watchdog()
 
     from demucs.pretrained import get_model
 

app/pipeline/runner.py

@@ -14,6 +14,7 @@
 from app.core.models import Job, JobCancelled, _set
 from app.core.redact import redact
 from app.core.registry import persist as persist_registry
+from app.core.registry import set_proc
 from app.pipeline.analyze import analyze
 from app.pipeline.beatgrid import compute_beat_grid
 from app.pipeline.collect import (
@@ -49,6 +50,30 @@ def _check_cancel(job: Job) -> None:
         raise JobCancelled()
 
 
+def _run_registered_ffmpeg(job: Job, cmd: list[str], timeout: int) -> tuple[int, bytes]:
+    """Run ffmpeg with the process registered, so cancel can reach it.
+
+    subprocess.run() cannot be interrupted: POST /cancel sets the flag, but
+    nothing looks at it until the call returns, so a cancel during a large
+    upload's transcode was a no-op for up to TIMEOUT_FFMPEG per call -- twice
+    over on the .mp4 path, which runs both this and the video extract (#519).
+
+    Mirrors collect._run_ffmpeg, which registers for exactly this reason.
+    """
+    proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
+    set_proc(job.id, proc)
+    try:
+        try:
+            _, stderr = proc.communicate(timeout=timeout)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            proc.communicate()
+            raise
+        return proc.returncode, stderr or b""
+    finally:
+        set_proc(job.id, None)
+
+
 def _extract_video_track(job: Job, source: Path, job_dir: Path) -> None:
     """For an .mp4 upload, preserve a silent video-only track at
     video.mp4 so the studio can later mux it with a custom stem mix
@@ -76,15 +101,15 @@ def _extract_video_track(job: Job, source: Path, job_dir: Path) -> None:
         str(dest),
     ]
     try:
-        result = subprocess.run(cmd, capture_output=True, timeout=TIMEOUT_FFMPEG)
+        returncode, _ = _run_registered_ffmpeg(job, cmd, TIMEOUT_FFMPEG)
     except (OSError, subprocess.SubprocessError) as e:
         # ffmpeg missing or timed out. Distinct from an .mp4 that simply has no
         # video stream, and the only one of the two worth surfacing (#436).
         dest.unlink(missing_ok=True)
         job.video_status = "failed"
         logger.warning("video extract failed for job %s: %s", job.id, e)
         return
-    if result.returncode != 0 or not dest.is_file() or dest.stat().st_size == 0:
+    if returncode != 0 or not dest.is_file() or dest.stat().st_size == 0:
         dest.unlink(missing_ok=True)
         job.video_status = "unavailable"
         logger.info("no video track preserved for job %s (source has no video stream?)", job.id)
@@ -127,10 +152,10 @@ def _prepare_local_source(job: Job, source: Path, job_dir: Path) -> Path:
         "-y",
         str(dest),
     ]
-    result = subprocess.run(cmd, capture_output=True, timeout=TIMEOUT_FFMPEG)
-    if result.returncode != 0:
+    returncode, stderr = _run_registered_ffmpeg(job, cmd, TIMEOUT_FFMPEG)
+    if returncode != 0:
         raise RuntimeError(
-            "ffmpeg transcode failed: " + result.stderr.decode("utf-8", errors="replace").strip()
+            "ffmpeg transcode failed: " + stderr.decode("utf-8", errors="replace").strip()
         )
     source.unlink(missing_ok=True)
     return dest

app/pipeline/section_worker.py

@@ -10,6 +10,7 @@
 import threading
 from pathlib import Path
 
+from app.core.process import arm_parent_watchdog
 from app.pipeline.section_refine import refine_segments
 
 _HEARTBEAT_SECONDS = 10
@@ -66,6 +67,9 @@ def _load_beat_grid(path: Path | None) -> object | None:
 
 
 def main(argv: list[str] | None = None) -> int:
+    # Without this a CPU inference pass outlives the parent whose
+    # TIMEOUT_SECTIONS was its only bound (#519).
+    arm_parent_watchdog()
     args = _parser().parse_args(argv)
     for path in (args.stems_dir / f"{name}.wav" for name in ("bass", "drums", "other", "vocals")):
         if not path.is_file():

app/pipeline/sections.py

@@ -214,6 +214,10 @@ def _terminate(proc: subprocess.Popen) -> None:
 def _run_registered_process(job: Job, cmd: list[str]) -> tuple[int, list[str], list[str]]:
     """Run a child with cancellation, total timeout, and output-stall detection."""
     env = os.environ.copy()
+    # The worker arms a watchdog on this and hard-exits when we disappear, so a
+    # kill that runs no cleanup (SIGKILL, Force Quit, Task Manager, a crash)
+    # cannot leave it running with nobody to collect the result (#519).
+    env["STEMDECK_PARENT_PID"] = str(os.getpid())
     env["PYTHONIOENCODING"] = "utf-8:replace"
     proc = subprocess.Popen(
         cmd,

app/pipeline/vocal_split.py

@@ -61,6 +61,10 @@ def split_vocals(job: Job, stems_dir: Path) -> list[str]:
     # the mismatch simply moves rather than being fixed. Demucs and audio-
     # separator both emit progress bars and can echo track metadata, neither of
     # which is guaranteed to be cp1252-safe.
+    # The worker arms a watchdog on this and hard-exits when we disappear, so a
+    # kill that runs no cleanup (SIGKILL, Force Quit, Task Manager, a crash)
+    # cannot leave it running with nobody to collect the result (#519).
+    env["STEMDECK_PARENT_PID"] = str(os.getpid())
     env["PYTHONIOENCODING"] = "utf-8:replace"
     try:
         import certifi

app/pipeline/vocal_split_worker.py

@@ -24,6 +24,7 @@
 import sys
 
 from app.core.config import VOCAL_SPLIT_MODEL
+from app.core.process import arm_parent_watchdog
 
 
 def _run(device: str, vocals_path: str, out_dir: str) -> None:
@@ -57,6 +58,10 @@ def _run(device: str, vocals_path: str, out_dir: str) -> None:
 
 
 def main() -> None:
+    # A Force-Quit of the app otherwise orphans this process holding the GPU:
+    # onnxruntime reads nothing from stdin mid-inference, so EOF never arrives
+    # and nothing else bounds it (#519).
+    arm_parent_watchdog()
     if len(sys.argv) < 4:
         sys.stderr.write("@@ERROR@@usage: vocal_split_worker <device> <vocals_wav> <out_dir>\n")
         sys.stderr.flush()