fix: sweep crash leftovers at startup, free a cancelled zip, give each click render its own temp

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 `<name>.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(

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

app/core/stem_files.py

@@ -93,15 +93,17 @@ 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
     RuntimeError when the decode fails."""
     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 = [

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

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("<i2")
 
     dest.parent.mkdir(parents=True, exist_ok=True)
-    tmp = dest.with_suffix(".wav.tmp")
-    with wave.open(str(tmp), "wb") as w:
-        w.setnchannels(1)
-        w.setsampwidth(2)
-        w.setframerate(sample_rate)
-        w.writeframes(pcm.tobytes())
-    tmp.replace(dest)
+    # A unique dot name: two exports may render the same click at once, and
+    # the export cache's prune and the startup sweep both go by the dot.
+    tmp = dest.with_name(f".{dest.name}.{uuid.uuid4().hex}.tmp")
+    try:
+        with wave.open(str(tmp), "wb") as w:
+            w.setnchannels(1)
+            w.setsampwidth(2)
+            w.setframerate(sample_rate)
+            w.writeframes(pcm.tobytes())
+        tmp.replace(dest)
+    finally:
+        tmp.unlink(missing_ok=True)
     return dest
 
 

app/pipeline/collect.py

@@ -3,14 +3,17 @@
 import json
 import logging
 import os
+import re
 import shutil
 import subprocess
 import threading
 import time
 import uuid
+from collections.abc import Iterable
 from pathlib import Path
 
 from app.core.config import (
+    CACHE_TEMP_MIN_AGE_SECONDS,
     DEMUCS_MODEL,
     FAILED_TTL_SECONDS,
     JOB_TTL_SECONDS,
@@ -23,7 +26,7 @@
 from app.core.models import FINISHED_STATUSES, Job
 from app.core.registry import all_jobs as registry_all
 from app.core.registry import deleted_ids as registry_deleted_ids
-from app.core.registry import holds_job_files, job_dirs, quarantine_dirs, set_proc
+from app.core.registry import holds_job_files, is_real_entry, job_dirs, quarantine_dirs, set_proc
 from app.core.registry import persist as registry_persist
 from app.core.registry import remove as registry_remove
 from app.core.stem_files import is_stem_audio, stem_names_on_disk, stem_path
@@ -563,3 +566,72 @@ def sweep_failed_jobs(jobs_dir: Path) -> 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

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.