diff --git a/app/core/models.py b/app/core/models.py index ca2e1b6..973f5df 100644 --- a/app/core/models.py +++ b/app/core/models.py @@ -63,6 +63,10 @@ class Job: # Device the separation actually ran on ("cuda" / "mps" / "cpu"), recorded # per job for diagnostics -- settings may change between jobs. compute_device: str | None = None + # True when a GPU separation attempt failed and the job completed on the + # CPU fallback (#276) -- kept loud in state/metadata so the fallback is + # never silent (the #247 lesson). + gpu_fallback: bool = False # Wall-clock seconds per pipeline stage ({"download": 12.3, ...}); written # to metadata.json and the one-line completion summary in the log. stage_timings: dict[str, float] | None = None @@ -101,6 +105,7 @@ def to_state(self) -> dict[str, Any]: "error": self.error, "error_detail": self.error_detail, "compute_device": self.compute_device, + "gpu_fallback": self.gpu_fallback, "stage_timings": self.stage_timings, "created_at": self.created_at, } diff --git a/app/pipeline/runner.py b/app/pipeline/runner.py index b32ff91..0a7ad81 100644 --- a/app/pipeline/runner.py +++ b/app/pipeline/runner.py @@ -208,6 +208,7 @@ def _write_metadata(job: Job, job_dir: Path) -> None: "tags": job.tags, "has_video": job.has_video, "compute_device": job.compute_device, + "gpu_fallback": job.gpu_fallback, "stage_timings": job.stage_timings, } try: diff --git a/app/pipeline/separate.py b/app/pipeline/separate.py index dc059d0..58864b6 100644 --- a/app/pipeline/separate.py +++ b/app/pipeline/separate.py @@ -3,6 +3,7 @@ import logging import os import re +import shutil import subprocess import sys import threading @@ -13,7 +14,7 @@ from app.core.models import Job, JobCancelled, _set from app.core.registry import set_proc from app.core.settings import get_demucs_device -from app.pipeline.errors import SeparationError +from app.pipeline.errors import SeparationError, classify_failure logger = logging.getLogger("stemdeck.pipeline") @@ -23,16 +24,10 @@ # while still catching genuine hangs (GPU deadlock, OOM stall, etc.). -def separate(job: Job, source: Path, job_dir: Path) -> Path: - _set(job, status="separating", progress=0.0, stage="Separating stems...") - - # Read the device fresh per job (not a frozen import) so a Settings change - # applies to the next separation without a restart. Recorded on the job for - # the completion summary / metadata / failure quarantine. - device = get_demucs_device() - job.compute_device = device - logger.info("[%s] separating on device=%s", job.id, device) - cmd = [ +def _demucs_cmd(device: str, source: Path, job_dir: Path) -> list[str]: + """Build the demucs CLI invocation. Module-level seam so tests can swap + in a stub executable without touching the process-management machinery.""" + return [ sys.executable, "-m", "demucs", @@ -44,6 +39,13 @@ def separate(job: Job, source: Path, job_dir: Path) -> Path: str(job_dir), str(source), ] + + +def _run_demucs(job: Job, source: Path, job_dir: Path, device: str) -> tuple[int, list[str]]: + """One demucs attempt on `device`: spawn, stream progress, watchdog stalls. + + Returns (returncode, stderr_tail). Raises JobCancelled when the exit was + caused by POST /cancel. The retry policy lives in separate().""" env = os.environ.copy() try: import certifi @@ -54,7 +56,7 @@ def separate(job: Job, source: Path, job_dir: Path) -> Path: pass proc = subprocess.Popen( - cmd, + _demucs_cmd(device, source, job_dir), stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, @@ -123,15 +125,69 @@ def _watchdog() -> None: # that into JobCancelled before the generic "demucs failed" path. if job.cancel_requested: raise JobCancelled() - if proc.returncode != 0: + return proc.returncode, tail + + +def separate(job: Job, source: Path, job_dir: Path) -> Path: + """Run demucs on the configured device, falling back to CPU once when a + GPU attempt fails (#276). + + The fallback is deliberately loud, never silent (the #247 lesson): the + stage line says so while it runs, the WARNING log carries the full stderr + tail, and gpu_fallback/compute_device persist to job state and metadata. + It applies even when the user forced cuda/mps in Settings -- a dead job + with no diagnostics is strictly worse for them than a slow one that + explains itself.""" + _set(job, status="separating", progress=0.0, stage="Separating stems...") + + # Read the device fresh per job (not a frozen import) so a Settings change + # applies to the next separation without a restart. Recorded on the job for + # the completion summary / metadata / failure quarantine. + device = get_demucs_device() + job.compute_device = device + logger.info("[%s] separating on device=%s", job.id, device) + + rc, tail = _run_demucs(job, source, job_dir, device) + + if rc != 0 and device != "cpu": + cause = classify_failure("\n".join(tail)) + logger.warning( + "[%s] demucs failed on %s (exit %s, cause=%s); retrying on CPU. tail:\n%s", + job.id, + device, + rc, + cause, + "\n".join(tail[-15:]) or "(no stderr captured)", + ) + # Partial output from the failed attempt must not be mistaken for + # results by collect(); CPU restarts from scratch, so does progress. + shutil.rmtree(job_dir / DEMUCS_MODEL, ignore_errors=True) + _set(job, progress=0.0, stage="GPU failed — retrying on CPU (slower)...") + job.gpu_fallback = True + job.compute_device = f"cpu (fallback from {device})" + first_tail = tail + rc, tail = _run_demucs(job, source, job_dir, "cpu") + if rc != 0: + combined = [ + f"--- attempt on {device} ---", + *first_tail[-20:], + "--- cpu fallback attempt ---", + *tail[-20:], + ] + last = tail[-1] if tail else f"exit status {rc}" + logger.error("[%s] cpu fallback also failed (exit %s)", job.id, rc) + raise SeparationError( + f"demucs failed: {last}", tail=combined, device=f"{device}, then cpu" + ) + elif rc != 0: detail = "\n".join(tail[-15:]) if tail else "(no stderr captured)" - logger.error("[%s] demucs exited %s; tail:\n%s", job.id, proc.returncode, detail) - last = tail[-1] if tail else f"exit status {proc.returncode}" + logger.error("[%s] demucs exited %s; tail:\n%s", job.id, rc, detail) + last = tail[-1] if tail else f"exit status {rc}" # SeparationError carries the stderr tail + device so the runner's # failure quarantine can preserve the evidence (#277). raise SeparationError(f"demucs failed: {last}", tail=tail[-40:], device=device) stems_root = job_dir / DEMUCS_MODEL / source.stem if not stems_root.is_dir(): - raise SeparationError(f"demucs output not found at {stems_root}", device=device) + raise SeparationError(f"demucs output not found at {stems_root}", device=job.compute_device) return stems_root diff --git a/tests/test_separate_fallback.py b/tests/test_separate_fallback.py new file mode 100644 index 0000000..b90348b --- /dev/null +++ b/tests/test_separate_fallback.py @@ -0,0 +1,159 @@ +"""Tests for the GPU->CPU separation fallback (#276). + +The demucs invocation is swapped for stub Python one-liners via the +_demucs_cmd seam, so the real process machinery (Popen, stderr streaming, +watchdog, cancel translation) runs end-to-end without demucs or a GPU. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +from app.core.models import Job, JobCancelled +from app.pipeline import separate as sep_mod +from app.pipeline.errors import SeparationError + + +def _stub_cmds(fail_devices: set[str], calls: list[str]): + """A _demucs_cmd replacement: fails with a CUDA-OOM message on the given + devices, succeeds (writing a stem WAV where demucs would) elsewhere.""" + + def fake_cmd(device: str, source: Path, job_dir: Path) -> list[str]: + calls.append(device) + if device in fail_devices: + code = ( + "import sys;" + " sys.stderr.write('RuntimeError: CUDA out of memory. Tried 2 GiB\\n');" + " sys.exit(1)" + ) + else: + out_dir = job_dir / sep_mod.DEMUCS_MODEL / source.stem + code = ( + "import os, sys;" + f" d = {str(out_dir)!r};" + " os.makedirs(d, exist_ok=True);" + " open(os.path.join(d, 'vocals.wav'), 'wb').write(b'RIFF');" + " sys.stderr.write('100%|separated\\n')" + ) + return [sys.executable, "-c", code] + + return fake_cmd + + +@pytest.fixture() +def job(tmp_path: Path): + j = Job(id="abcdefabc276") + (tmp_path / "source.wav").write_bytes(b"RIFF") + return j + + +def test_gpu_failure_falls_back_to_cpu(job, tmp_path, monkeypatch, caplog): + import logging + + calls: list[str] = [] + monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "cuda") + monkeypatch.setattr(sep_mod, "_demucs_cmd", _stub_cmds({"cuda"}, calls)) + + with caplog.at_level(logging.WARNING, logger="stemdeck.pipeline"): + stems_root = sep_mod.separate(job, tmp_path / "source.wav", tmp_path) + + assert calls == ["cuda", "cpu"] + assert (stems_root / "vocals.wav").is_file() + assert job.gpu_fallback is True + assert job.compute_device == "cpu (fallback from cuda)" + # Loud, never silent: the warning names device, cause, and stderr. + warning = next(r.message for r in caplog.records if "retrying on CPU" in r.message) + assert "cause=out-of-memory" in warning + assert "CUDA out of memory" in warning + + +def test_gpu_success_needs_no_fallback(job, tmp_path, monkeypatch): + calls: list[str] = [] + monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "cuda") + monkeypatch.setattr(sep_mod, "_demucs_cmd", _stub_cmds(set(), calls)) + + stems_root = sep_mod.separate(job, tmp_path / "source.wav", tmp_path) + + assert calls == ["cuda"] + assert stems_root.is_dir() + assert job.gpu_fallback is False + assert job.compute_device == "cuda" + + +def test_cpu_failure_does_not_retry(job, tmp_path, monkeypatch): + calls: list[str] = [] + monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "cpu") + monkeypatch.setattr(sep_mod, "_demucs_cmd", _stub_cmds({"cpu"}, calls)) + + with pytest.raises(SeparationError) as exc_info: + sep_mod.separate(job, tmp_path / "source.wav", tmp_path) + + assert calls == ["cpu"] # exactly one attempt + assert job.gpu_fallback is False + assert exc_info.value.device == "cpu" + + +def test_both_attempts_failing_raises_with_both_tails(job, tmp_path, monkeypatch): + calls: list[str] = [] + monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "mps") + monkeypatch.setattr(sep_mod, "_demucs_cmd", _stub_cmds({"mps", "cpu"}, calls)) + + with pytest.raises(SeparationError) as exc_info: + sep_mod.separate(job, tmp_path / "source.wav", tmp_path) + + assert calls == ["mps", "cpu"] + err = exc_info.value + assert err.device == "mps, then cpu" + # The quarantine's error.txt gets both attempts' evidence. + joined = "\n".join(err.tail) + assert "--- attempt on mps ---" in joined + assert "--- cpu fallback attempt ---" in joined + + +def test_cancel_during_gpu_attempt_skips_fallback(job, tmp_path, monkeypatch): + calls: list[str] = [] + monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "cuda") + monkeypatch.setattr(sep_mod, "_demucs_cmd", _stub_cmds({"cuda"}, calls)) + job.cancel_requested = True # POST /cancel arrived before/mid attempt + + with pytest.raises(JobCancelled): + sep_mod.separate(job, tmp_path / "source.wav", tmp_path) + + assert calls == ["cuda"] # no CPU retry after a cancel + + +def test_partial_gpu_output_cleared_before_retry(job, tmp_path, monkeypatch): + """A failed GPU attempt's partial stems must not leak into the CPU run.""" + calls: list[str] = [] + marker = tmp_path / sep_mod.DEMUCS_MODEL / "partial-garbage.wav" + + def fake_cmd(device: str, source: Path, job_dir: Path) -> list[str]: + calls.append(device) + if device == "cuda": + # Simulate demucs dying after writing partial output. + code = ( + "import os, sys;" + f" os.makedirs(os.path.dirname({str(marker)!r}), exist_ok=True);" + f" open({str(marker)!r}, 'wb').write(b'junk');" + " sys.stderr.write('RuntimeError: CUDA error\\n');" + " sys.exit(1)" + ) + return [sys.executable, "-c", code] + out_dir = tmp_path / sep_mod.DEMUCS_MODEL / source.stem + code = ( + "import os;" + f" os.makedirs({str(out_dir)!r}, exist_ok=True);" + f" open(os.path.join({str(out_dir)!r}, 'vocals.wav'), 'wb').write(b'RIFF')" + ) + return [sys.executable, "-c", code] + + monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "cuda") + monkeypatch.setattr(sep_mod, "_demucs_cmd", fake_cmd) + + sep_mod.separate(job, tmp_path / "source.wav", tmp_path) + + assert calls == ["cuda", "cpu"] + assert not marker.exists(), "partial GPU output must be cleared before the CPU retry"