diff --git a/app/api/stems.py b/app/api/stems.py index 89264add..b1ac7184 100644 --- a/app/api/stems.py +++ b/app/api/stems.py @@ -9,6 +9,7 @@ import re import subprocess import tempfile +import threading import uuid import zipfile from collections import deque @@ -1330,9 +1331,10 @@ async def _render_to_file( def _unlink_later(path: Path) -> None: - """Drop a rendered temp file once its response has been sent. + """Drop a rendered temp file once its response has been sent, or once its + render has failed. - Only ever attached to an uncached render. A cached one returns the cache + Only ever used on an uncached render or a zip. A cached one returns the cache entry itself, and deleting that would throw away the render the cache exists to keep.""" try: @@ -1341,6 +1343,26 @@ def _unlink_later(path: Path) -> None: logger.debug("could not remove rendered export %s", path, exc_info=True) +class _ZipTemp: + """The file a stems zip is built in. It goes once the request has given it + up (a failed build, or a cancel: the client or the server going away) and + the build thread has stopped writing it, whichever comes second. A cancel + does not stop the thread, so the request alone cannot tell when the file + is free, and on Windows removing it sooner fails.""" + + def __init__(self, path: Path) -> None: + self.path = path + self._lock = threading.Lock() + self._ended: set[str] = set() + + def end(self, side: str) -> None: + with self._lock: + self._ended.add(side) + both = self._ended == {"build", "request"} + if both: + _unlink_later(self.path) + + async def _ensure_cached_mp3(src: Path) -> Path: """Transcode `src` (a stem WAV) to a sibling `.mp3`, cached on disk. Re-encoding a full song on every request is the slow part of loading a track @@ -2277,7 +2299,9 @@ def _build_stems_zip( zf.write(p, arcname=_arcname(prefix, name, fmt)) return encode = _ENCODE_ARGS[fmt] - with tempfile.TemporaryDirectory() as td, zipfile.ZipFile(dest, "w", zipfile.ZIP_STORED) as zf: + # Beside the zip, so the startup sweep finds what a killed build left. + work = tempfile.TemporaryDirectory(prefix=".zipwork-", dir=dest.parent) + with work as td, zipfile.ZipFile(dest, "w", zipfile.ZIP_STORED) as zf: for name, p in sources: if p.suffix.lower() == f".{fmt}": zf.write(p, arcname=_arcname(prefix, name, fmt)) @@ -2412,7 +2436,7 @@ def _build_stems_zip_with_click( if click is None: _build_stems_zip(sources, fmt, dest, prefix) return - with tempfile.TemporaryDirectory() as td: + with tempfile.TemporaryDirectory(prefix=".zipwork-", dir=dest.parent) as td: path = _render_click_stem(click, sources[0][1], fmt, Path(td)) extra = [(CLICK_STEM_NAME, path)] if path is not None else [] _build_stems_zip([*sources, *extra], fmt, dest, prefix) @@ -2518,22 +2542,32 @@ async def get_all_stems_zip( click_sound, ) - fd, tmp = tempfile.mkstemp(prefix="stemdeck_zip_", suffix=".zip") + # In the export cache under a dot name, like a render: the cache prune + # leaves it alone, and the startup sweep removes it if the process dies. + _MIXDOWN_CACHE_DIR.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(prefix=".stemdeck_zip_", suffix=".zip", dir=_MIXDOWN_CACHE_DIR) os.close(fd) tmp_path = Path(tmp) + temp = _ZipTemp(tmp_path) + + def build() -> None: + try: + _build_stems_zip_with_click(sources, fmt, tmp_path, _title_slug(job.title), click_stem) + except Exception: + # Logged here: after a cancel nobody awaits the thread. + logger.exception("failed to build stems zip for job %s", job_id) + raise + finally: + temp.end("build") + try: - await asyncio.to_thread( - _build_stems_zip_with_click, - sources, - fmt, - tmp_path, - _title_slug(job.title), - click_stem, - ) + await asyncio.to_thread(build) except Exception: - tmp_path.unlink(missing_ok=True) - logger.exception("failed to build stems zip for job %s", job_id) + temp.end("request") raise HTTPException(status_code=500, detail="failed to build archive") from None + except BaseException: + temp.end("request") + raise filename = f"{_safe_title(job.title)}_stems.zip" return FileResponse( diff --git a/app/core/config.py b/app/core/config.py index ac9dbf57..5df3f621 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -318,6 +318,10 @@ def js_solver_available() -> bool: # Swept unconditionally -- even deployments with a persistent library must not # accumulate failure evidence forever. FAILED_TTL_SECONDS = max(3600, _env_int("STEMDECK_FAILED_TTL_SECONDS", 7 * 24 * 3600)) # 7 d +# How long an export cache temp must have gone unwritten before the startup +# sweep takes it. A second server that shares the cache folder (one started +# with only STEMDECK_JOBS_DIR changed) may be writing it; an orphan is older. +CACHE_TEMP_MIN_AGE_SECONDS = 600 # Depth of the import queue: jobs waiting for their turn, not counting the one # running. Counted separately by kind, because the two cost wildly different # things. A queued upload holds its source file on disk for the whole wait, so diff --git a/app/core/stem_files.py b/app/core/stem_files.py index 0b0b72e9..2ce92858 100644 --- a/app/core/stem_files.py +++ b/app/core/stem_files.py @@ -93,7 +93,9 @@ def is_stem_audio(path: Path) -> bool: @contextlib.contextmanager def decoded_wav(path: Path) -> Iterator[Path]: """`path` itself when it is a WAV, else a float32 WAV decoded from it by - ffmpeg in a temporary folder that is removed afterwards. + ffmpeg in a temporary folder beside it that is removed afterwards. Beside + it, not in the system temp folder, so the startup sweep + (collect.sweep_crash_leftovers) finds what a killed process left there. For a reader that hands the stem to a third-party library which may not decode MP3 the way ffmpeg does (the vocal split's audio-separator). Raises @@ -101,7 +103,7 @@ def decoded_wav(path: Path) -> Iterator[Path]: if path.suffix.lower() == ".wav": yield path return - folder = Path(tempfile.mkdtemp(prefix="stemdeck-decode-")) + folder = Path(tempfile.mkdtemp(prefix=".decode-", dir=path.parent)) try: out = folder / f"{path.stem}.wav" cmd = [ diff --git a/app/main.py b/app/main.py index a6203afa..7d1b7b06 100644 --- a/app/main.py +++ b/app/main.py @@ -116,7 +116,7 @@ ffmpeg_too_old, transpose_export, ) -from app.pipeline.collect import sweep_failed_jobs, sweep_old_jobs +from app.pipeline.collect import sweep_crash_leftovers, sweep_failed_jobs, sweep_old_jobs from app.pipeline.sections import sweep_orphaned_workspaces as sweep_orphaned_section_workspaces from app.pipeline.vocal_split_status import capability as vocal_split_capability @@ -250,6 +250,16 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: await asyncio.to_thread(sweep_render_temps, JOBS_DIR) except Exception: _log.exception("could not sweep stale render files") + # The same for a stem fill, a vocal split decode, a stem write, an export + # render and a stems zip. + try: + from app.api.stems import _CLICK_CACHE_DIR, _MIXDOWN_CACHE_DIR + + await asyncio.to_thread( + sweep_crash_leftovers, JOBS_DIR, (_MIXDOWN_CACHE_DIR, _CLICK_CACHE_DIR) + ) + except Exception: + _log.exception("could not sweep crash leftovers") # An FFmpeg older than 5.1 cannot run any export (app/pipeline/audio_encoders.py). # The server still starts, so imports and playback work and the page can diff --git a/app/pipeline/click_render.py b/app/pipeline/click_render.py index 38ffd987..45d90faa 100644 --- a/app/pipeline/click_render.py +++ b/app/pipeline/click_render.py @@ -17,6 +17,7 @@ import hashlib import logging +import uuid import wave from pathlib import Path @@ -425,13 +426,18 @@ def _render_events( pcm = (buf * 32767.0).astype(" None: _rmtree(d) except OSError: logger.warning("failed-quarantine sweep could not stat %s", d, exc_info=True) + + +# What a writer leaves when the process dies before its own cleanup runs: a +# stem fill's work folder (stem_fill), a decode for the vocal split +# (stem_files.decoded_wav), and a stem file or work folder written under a +# dot temp name beside its final one (stem_storage, stems._ensure_cached_mp3). +# Nothing a song keeps has one of these names. +_JOB_TEMP_DIR_RE = re.compile(r"\.fill-[0-9a-f]{8}\Z") +_STEMS_TEMP_DIR_RE = re.compile(r"\.decode-\w{8}\Z|\..+\.tmp\Z") +_STEMS_TEMP_FILE_RE = re.compile(r"\..+\.tmp(\.mp3)?\Z") +# The same in the export caches: a render or a click written beside its cache +# entry (stems._render_to_file, stems._stream_ffmpeg, click_render), a stems +# zip and the folder its stems are transcoded in (stems.get_all_stems_zip). +_CACHE_TEMP_DIR_RE = re.compile(r"\.zipwork-\w{8}\Z") +_CACHE_TEMP_FILE_RE = re.compile(r"\..+\.[0-9a-f]{32}\.tmp(\.\w+)?\Z|\.stemdeck_zip_\w{8}\.zip\Z") + + +def _matching(folder: Path, dir_re: re.Pattern[str], file_re: re.Pattern[str]) -> list[Path]: + if not folder.is_dir() or not is_real_entry(folder): + return [] + return [ + p + for p in folder.iterdir() + if (dir_re.match(p.name) and p.is_dir()) or (file_re.match(p.name) and p.is_file()) + ] + + +def sweep_crash_leftovers(jobs_dir: Path, cache_dirs: Iterable[Path]) -> int: + """Remove what a killed process left half-written in every song folder and + in the export caches (`cache_dirs`). Returns how many went. + + Call this at startup only, before the queue runs anything: no writer is + alive then, so every temp found is dead. Later it could take a file a + running job is writing. Only StemDeck's song folders are looked in, and no + link is followed or removed (registry.job_dirs, holds_job_files, + is_real_entry). A cache temp written in the last + CACHE_TEMP_MIN_AGE_SECONDS is left alone: it may be another server's.""" + found: list[Path] = [] + for job_dir in job_dirs(jobs_dir): + if not holds_job_files(job_dir): + continue + try: + found += [p for p in job_dir.iterdir() if _JOB_TEMP_DIR_RE.match(p.name) and p.is_dir()] + found += _matching(job_dir / "stems", _STEMS_TEMP_DIR_RE, _STEMS_TEMP_FILE_RE) + except OSError: + logger.warning("crash leftover sweep could not read %s", job_dir.name) + cutoff = time.time() - CACHE_TEMP_MIN_AGE_SECONDS + for cache_dir in cache_dirs: + try: + found += [ + p + for p in _matching(cache_dir, _CACHE_TEMP_DIR_RE, _CACHE_TEMP_FILE_RE) + if p.stat().st_mtime < cutoff + ] + except OSError: + logger.warning("crash leftover sweep could not read %s", cache_dir.name) + gone = 0 + for path in found: + if not is_real_entry(path): + continue + try: + if path.is_dir(): + shutil.rmtree(path) + else: + path.unlink() + gone += 1 + except OSError: + logger.warning("crash leftover not removed: %s", path.name) + return gone diff --git a/tests/conftest.py b/tests/conftest.py index 8eb03a8d..3173f803 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -100,6 +100,15 @@ def _isolate_backing_cache(tmp_path, monkeypatch): monkeypatch.setattr("app.pipeline.backing.CACHE_ROOT", tmp_path / "_backing_cache") +@pytest.fixture(autouse=True) +def _isolate_export_caches(tmp_path, monkeypatch): + """Exports render to a temporary cache, never the real one. The startup + sweep that every TestClient lifespan runs removes the half-written files + in these caches, which in the real one could be a running server's.""" + monkeypatch.setattr("app.api.stems._MIXDOWN_CACHE_DIR", tmp_path / "_cache" / "mixdown") + monkeypatch.setattr("app.api.stems._CLICK_CACHE_DIR", tmp_path / "_cache" / "click") + + @pytest.fixture(autouse=True) def _isolate_jobs_dir(tmp_path, monkeypatch): """Point every JOBS_DIR at a temp dir, for every test, no exceptions. diff --git a/tests/test_crash_leftovers.py b/tests/test_crash_leftovers.py new file mode 100644 index 00000000..d2181a9d --- /dev/null +++ b/tests/test_crash_leftovers.py @@ -0,0 +1,276 @@ +"""What a killed process leaves behind is removed at the next start, and only that. + +A stem fill, a decode for the vocal split, a stem written beside its final name, +an export render and an all-stems zip each write to a temp name first. A kill or +a power loss in the middle leaves that temp behind, and each can be hundreds of +MB. Every test here runs in the test's own throwaway folders. +""" + +from __future__ import annotations + +import asyncio +import os +import threading +import time +import types +import uuid +import wave +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from app.core import registry +from app.core.models import Job +from tests.test_jobs_folder_scope import _link + +HEX = uuid.uuid4().hex +DONE = "abcdefabcd01" +STOPPED = "abcdefabcd02" + + +@pytest.fixture(autouse=True) +def _isolate_registry(): + registry._jobs.clear() + yield + registry._jobs.clear() + + +@pytest.fixture +def caches(tmp_path: Path, monkeypatch) -> tuple[Path, Path]: + from app.api import stems as stems_mod + + mixdown = tmp_path / "cache" / "mixdown" + click = tmp_path / "cache" / "click" + mixdown.mkdir(parents=True) + click.mkdir(parents=True) + monkeypatch.setattr(stems_mod, "_MIXDOWN_CACHE_DIR", mixdown) + monkeypatch.setattr(stems_mod, "_CLICK_CACHE_DIR", click) + return mixdown, click + + +def _song(jobs: Path, job_id: str, status: str) -> list[Path]: + """A song with everything a user keeps, registered with `status`. Returns + every file it holds.""" + job = Job(id=job_id) + job.status = status + registry._jobs[job_id] = job + stems = jobs / job_id / "stems" + stems.mkdir(parents=True) + kept = [ + stems / "vocals.wav", + stems / "vocals.mp3", + stems / "drums.mp3", + stems / "peaks.json", + stems / "beats.json", + stems / ".DS_Store", + jobs / job_id / "source.mp3", + jobs / job_id / "metadata.json", + jobs / job_id / "takes" / "t0000aaaa" / "take.webm", + ] + for path in kept: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"keep") + return kept + + +def _leftovers(jobs: Path, job_id: str) -> list[Path]: + """What each writer leaves in a song's folder when the process dies mid-write, + named exactly as the writer names it.""" + job_dir = jobs / job_id + stems = job_dir / "stems" + fill = job_dir / f".fill-{HEX[:8]}" # stem_fill.fill_missing_stems + decode = stems / ".decode-ab_12cd3" # stem_files.decoded_wav + work = stems / f".vocal_split.{HEX}.tmp" # a stem writer's work folder + for folder in (fill, decode, work): + folder.mkdir() + (folder / "vocals.wav").write_bytes(b"x") + files = [ + stems / f".vocals.{HEX}.tmp.mp3", # stem_storage, WAV to MP3 + stems / f".vocals.mp3.{HEX}.tmp", # stems._ensure_cached_mp3 + stems / f".vocals.wav.{HEX}.tmp", # a stem written atomically + ] + for path in files: + path.write_bytes(b"x") + return [fill, decode, work, *files] + + +def test_startup_removes_crash_leftovers_and_keeps_every_song_file( + tmp_path: Path, caches: tuple[Path, Path] +): + from app.core.config import JOBS_DIR + from app.main import app + + mixdown, click = caches + kept = _song(JOBS_DIR, DONE, "done") + _song(JOBS_DIR, STOPPED, "cancelled") + gone = _leftovers(JOBS_DIR, DONE) + _leftovers(JOBS_DIR, STOPPED) + cache_kept = [ + mixdown / "0123abcd.wav", + click / "4567cdef.wav", + mixdown / ".DS_Store", + # Written a moment ago: another server sharing the cache may be writing it. + mixdown / f".render.{uuid.uuid4().hex}.tmp.wav", + ] + cache_gone = [ + mixdown / f".render.{HEX}.tmp.wav", # stems._render_to_file, no cache entry + mixdown / f".0123abcd.wav.{HEX}.tmp.wav", # stems._render_to_file into the cache + mixdown / f".0123abcd.mp3.{HEX}.tmp", # stems._stream_ffmpeg into the cache + mixdown / ".stemdeck_zip_k3j2h1g0.zip", # get_all_stems_zip + mixdown / ".zipwork-ab12cd34" / "vocals.mp3", # _build_stems_zip + click / f".4567cdef.wav.{HEX}.tmp", # click_render._render_events + ] + long_ago = time.time() - 3600 + for path in [*cache_kept, *cache_gone]: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"x") + for path in [*cache_kept[:3], *cache_gone]: + os.utime(path, (long_ago, long_ago)) + os.utime(mixdown / ".zipwork-ab12cd34", (long_ago, long_ago)) + # Not StemDeck's, in a shared jobs folder: a folder that is not a job, and + # one named like a job that holds nothing StemDeck writes. + foreign = [JOBS_DIR / name / f".fill-{HEX[:8]}" for name in ("photos", "0123456789ab")] + for folder in foreign: + folder.mkdir(parents=True) + + with TestClient(app): + pass + + assert [p for p in kept + cache_kept + foreign if not p.exists()] == [] + assert [p for p in gone + cache_gone if p.exists()] == [] + assert not (mixdown / ".zipwork-ab12cd34").exists() + + +def test_leftover_sweep_never_follows_a_link(tmp_path: Path, caches: tuple[Path, Path]): + from app.core.config import JOBS_DIR + from app.pipeline.collect import sweep_crash_leftovers + + outside = tmp_path / "outside" + (outside / "stems").mkdir(parents=True) + temp = outside / "stems" / f".vocals.{HEX}.tmp.mp3" + temp.write_bytes(b"x") + _link(JOBS_DIR / DONE, outside) + linked_cache = tmp_path / "elsewhere" + (linked_cache / "stems").mkdir(parents=True) + linked_temp = linked_cache / "stems" / f".drums.{HEX}.tmp.mp3" + linked_temp.write_bytes(b"x") + (JOBS_DIR / STOPPED).mkdir() + _link(JOBS_DIR / STOPPED / "stems", linked_cache / "stems") + + sweep_crash_leftovers(JOBS_DIR, caches) + + assert temp.is_file() + assert linked_temp.is_file() + + +# ─── The all-stems zip (F47) ────────────────────────────────────────────── + + +def test_zip_temp_is_removed_when_the_build_is_cancelled( + tmp_path: Path, caches: tuple[Path, Path], monkeypatch +): + """A request cancelled while its zip builds (the client or the server going + away) raises CancelledError, which is not an Exception, and the build + thread goes on writing: the temp zip was left behind. It now goes when the + build ends, and it is built in the export cache, where the startup sweep + finds it if the process dies instead.""" + from app.api import stems as stems_mod + + monkeypatch.setattr(stems_mod, "JOBS_DIR", tmp_path) + job = Job(id=DONE) + job.status = "done" + registry._jobs[DONE] = job + (tmp_path / DONE / "stems").mkdir(parents=True) + (tmp_path / DONE / "stems" / "vocals.wav").write_bytes(b"RIFF") + built: list[Path] = [] + started = threading.Event() + release = threading.Event() + + def slow_build(sources, fmt, dest, prefix, click): + built.append(dest) + with open(dest, "wb") as f: + f.write(b"PK partial") + started.set() + assert release.wait(timeout=10) + f.write(b" and the rest") + + monkeypatch.setattr(stems_mod, "_build_stems_zip_with_click", slow_build) + + async def cancel_mid_build() -> None: + request = asyncio.create_task( + stems_mod.get_all_stems_zip( + DONE, + fmt="wav", + stems=None, + click=False, + click_mult=1.0, + click_accent=-1, + click_gain=0.6, + click_groups="", + click_sound=stems_mod.DEFAULT_CLICK_SOUND, + ) + ) + assert await asyncio.to_thread(started.wait, 10) + request.cancel() + with pytest.raises(asyncio.CancelledError): + await request + # The build is still writing: the file must outlive the request. + assert built[0].exists() + release.set() + + try: + # asyncio.run returns once the build thread has ended. + asyncio.run(cancel_mid_build()) + assert not built[0].exists() + assert built[0].parent == caches[0] + finally: + for path in built: + path.unlink(missing_ok=True) + + +# ─── Click renders of one cache key at once (F39) ───────────────────────── + + +def test_two_renders_of_one_click_both_finish(tmp_path: Path, monkeypatch): + """Two exports rendering the same click at once wrote one shared temp name: + the first rename took the other's file away, and that export lost its + click. The first render here has written its file and waits to rename it + while the second runs from start to end.""" + from app.pipeline import click_render + + first_written = threading.Event() + second_done = threading.Event() + + class _Written: + def __init__(self, path: str, mode: str) -> None: + self._w = wave.open(path, mode) # noqa: SIM115 -- closed in __exit__ + + def __enter__(self): + return self._w + + def __exit__(self, *exc) -> None: + self._w.close() + if not first_written.is_set(): + first_written.set() + second_done.wait(timeout=10) + + monkeypatch.setattr(click_render, "wave", types.SimpleNamespace(open=_Written)) + dest = tmp_path / "click" / "samekey.wav" + errors: list[BaseException] = [] + + def render() -> None: + try: + click_render.render_click_wav(dest, [0.5, 1.0, 1.5, 2.0], [], 3.0) + except BaseException as exc: + errors.append(exc) + + first = threading.Thread(target=render) + first.start() + assert first_written.wait(timeout=10) + render() + second_done.set() + first.join(timeout=10) + + assert errors == [] + with wave.open(str(dest), "rb") as w: + assert w.getnframes() == 3 * 44100 + assert sorted(p.name for p in dest.parent.iterdir()) == ["samekey.wav"]