From 0de57a3ee286cbcd5c2026b16eea5b6c916617b1 Mon Sep 17 00:00:00 2001 From: Thales <> Date: Tue, 3 Nov 2026 06:43:43 +0000 Subject: [PATCH] fix: sweep crash leftovers at startup, free a cancelled zip, give each click render its own temp A stem fill, a vocal split decode, 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 left that temp on disk for good, and each can be hundreds of MB. The startup sweep now removes them: only in StemDeck's own song folders, never through a link or junction, never a file a song keeps, and in the export caches only by the exact temp names, and only when nothing has written them for ten minutes (a second server may share the cache). The decode and the zip, with its transcode folder, moved out of the system temp folder (beside the stem, and into the export cache) so the sweep never has to delete in a folder other programs share. A cancelled zip request (the server stopping) skipped the cleanup, because CancelledError is not an Exception, and the build thread went on writing. The zip now goes once the request has given it up and the thread has ended. Two exports rendering the same click at once shared one temp name, so one could rename the other's file away and lose its click. Each render now has its own uuid dot name, removed in a finally. --- app/api/stems.py | 64 ++++++-- app/core/config.py | 4 + app/core/stem_files.py | 6 +- app/main.py | 12 +- app/pipeline/click_render.py | 20 ++- app/pipeline/collect.py | 74 ++++++++- tests/conftest.py | 9 ++ tests/test_crash_leftovers.py | 276 ++++++++++++++++++++++++++++++++++ 8 files changed, 439 insertions(+), 26 deletions(-) create mode 100644 tests/test_crash_leftovers.py 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"]