diff --git a/.gitignore b/.gitignore index d86469d3..731ba93d 100644 --- a/.gitignore +++ b/.gitignore @@ -39,6 +39,7 @@ htmlcov/ # Runtime job artifacts and local build scratch data/ jobs/ +cache/ settings.json .run/ .build diff --git a/app/api/stems.py b/app/api/stems.py index 4523a520..8f1351da 100644 --- a/app/api/stems.py +++ b/app/api/stems.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import hashlib import logging import os import re @@ -15,7 +16,14 @@ from fastapi.responses import FileResponse, Response, StreamingResponse from starlette.background import BackgroundTask -from app.core.config import JOB_ID_RE, JOBS_DIR, STEM_NAMES, TIMEOUT_FFMPEG, ffmpeg_executable +from app.core.config import ( + CACHE_DIR, + JOB_ID_RE, + JOBS_DIR, + STEM_NAMES, + TIMEOUT_FFMPEG, + ffmpeg_executable, +) from app.core.registry import get as registry_get from app.core.settings import get_export_sample_rate @@ -46,6 +54,67 @@ MIXDOWN_CODECS = {ext: [*args, "-f", ext] for ext, args in _ENCODE_ARGS.items()} MIXDOWN_MEDIA_TYPES = {"wav": "audio/wav", "mp3": "audio/mpeg", "flac": "audio/flac"} +# Mixdown render cache (#290): identical render params re-run the full ffmpeg +# graph on every request today. On a shared server, repeat downloads of the +# same mix (a common case -- re-downloading, or several listeners pulling the +# same export) burn CPU for no reason since the output is a pure function of +# the inputs. Bounded so a churny cache (many one-off region trims) can't +# grow unboundedly on a long-running server. +_MIXDOWN_CACHE_DIR = CACHE_DIR / "mixdown" +_MIXDOWN_CACHE_MAX_FILES = 20 +_MIXDOWN_CACHE_MAX_BYTES = 500 * 1024 * 1024 # 500 MB + + +def _mixdown_cache_key( + job_id: str, + ext: str, + names: list[str], + gains: list[float], + start: float | None, + end: float | None, +) -> str: + """Every render input is in the key, so a different mixer state, region, + or a Settings change (export sample rate) misses cleanly -- never a stale + hit. Stems are immutable once a job reaches "done", so job_id alone pins + the underlying audio; gains are formatted to match the %.6f precision + already used in the ffmpeg volume filter, so cosmetically-different but + ffmpeg-equivalent gain strings (e.g. "1" vs "1.0") share a cache entry.""" + raw = "|".join( + [ + job_id, + ext, + ",".join(names), + ",".join(f"{g:.6f}" for g in gains), + "" if start is None else f"{start:.6f}", + "" if end is None else f"{end:.6f}", + str(get_export_sample_rate()), + ] + ) + # Cache key, not a security context -- usedforsecurity=False documents + # that for both readers and static analysis (bandit flags bare sha1()). + return hashlib.sha1(raw.encode("utf-8"), usedforsecurity=False).hexdigest() + + +def _prune_mixdown_cache(cache_dir: Path) -> None: + """Evict oldest-first once either the file-count or total-size budget is + exceeded. Best-effort: a failed prune just means the cache grows past + budget until the next successful render, not a broken export.""" + try: + entries = sorted( + (p for p in cache_dir.iterdir() if p.is_file() and not p.name.startswith(".")), + key=lambda p: p.stat().st_mtime, + ) + total = sum(p.stat().st_size for p in entries) + except OSError: + return + while entries and (len(entries) > _MIXDOWN_CACHE_MAX_FILES or total > _MIXDOWN_CACHE_MAX_BYTES): + oldest = entries.pop(0) + try: + total -= oldest.stat().st_size + oldest.unlink() + except OSError: + logger.debug("mixdown cache prune: could not remove %s", oldest, exc_info=True) + def _validate_stem_path(job_id: str, name: str): """Shared guard: validate job_id, name, job state, and path. Returns resolved Path.""" @@ -93,14 +162,21 @@ async def _drain_stderr(stream: asyncio.StreamReader, sink: deque[str]) -> None: sink.append(line.decode("utf-8", "replace").rstrip()) -async def _stream_ffmpeg(cmd: list[str], context: str = ""): +async def _stream_ffmpeg(cmd: list[str], context: str = "", cache_path: Path | None = None): """Yield ffmpeg stdout in 64 KB chunks; kill process on client disconnect. stderr is captured (bounded tail) and logged at WARNING when ffmpeg exits non-zero (#280): the HTTP status is already committed mid-stream, so a failed render reaches the client as a truncated file -- the log entry is the only place the failure can surface. Kills we initiated (client - disconnect) are expected and not logged as failures.""" + disconnect) are expected and not logged as failures. + + When `cache_path` is given, every chunk is also teed to a per-request + temp file alongside it (#290). A clean finish atomically renames the temp + file into place as the cache entry and prunes the cache to its budget; + any failure or client disconnect removes the temp file instead -- a + render the client didn't get in full never becomes a cache hit for the + next request.""" proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, @@ -113,14 +189,25 @@ async def _stream_ffmpeg(cmd: list[str], context: str = ""): # what distinguishes "ffmpeg finished on its own" from "client # disconnected mid-stream and we killed it". finished = False + + tmp_path: Path | None = None + tmp_file = None + if cache_path is not None: + cache_path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = cache_path.with_name(f".{cache_path.name}.{uuid.uuid4().hex}.tmp") + tmp_file = open(tmp_path, "wb") # noqa: SIM115 -- closed in finally, not a context manager here try: while True: chunk = await proc.stdout.read(65536) if not chunk: finished = True break + if tmp_file is not None: + tmp_file.write(chunk) yield chunk finally: + if tmp_file is not None: + tmp_file.close() if not finished and proc.returncode is None: proc.kill() await proc.wait() @@ -135,6 +222,12 @@ async def _stream_ffmpeg(cmd: list[str], context: str = ""): context, " | ".join(list(stderr_tail)[-8:]) or "(no stderr)", ) + if tmp_path is not None: + if finished and proc.returncode == 0: + os.replace(tmp_path, cache_path) + _prune_mixdown_cache(cache_path.parent) + else: + tmp_path.unlink(missing_ok=True) async def _ensure_cached_mp3(src: Path) -> Path: @@ -308,12 +401,16 @@ async def get_mixdown( gains: str = Query(..., description="Comma-separated linear gains, parallel to stems"), start: float | None = Query(default=None, ge=0, description="Trim start in seconds"), end: float | None = Query(default=None, gt=0, description="Trim end in seconds"), -) -> StreamingResponse: - """Render a fresh mixdown of the given lanes at the given gains, streamed as - WAV or MP3. Mirrors the studio mixer (per-stem volume, mute, solo) so the +) -> FileResponse | StreamingResponse: + """Render a mixdown of the given lanes at the given gains, streamed as WAV, + MP3, or FLAC. Mirrors the studio mixer (per-stem volume, mute, solo) so the exported file matches what is heard. The master fader is intentionally not applied -- it is a monitoring level, not part of the mix. Optional ?start=&end= - trims to a loop region.""" + trims to a loop region. + + Identical params (including start/end and the current export sample rate) + hit a render cache instead of re-running ffmpeg (#290) -- a cheap win on a + shared server where the same export gets re-downloaded.""" if ext not in ("wav", "mp3", "flac"): raise HTTPException(status_code=404, detail="not found") @@ -324,9 +421,22 @@ async def get_mixdown( detail="start and end are both required and start must be less than end", ) - # Validates job_id (404), job done (404), and path traversal (404) per stem. + # Validates job_id (404), job done (404), and path traversal (404) per + # stem -- deliberately before the cache lookup below, so a deleted or + # not-yet-done job 404s the same way it always has instead of serving a + # stale cache entry from before the job was removed. paths = [_validate_stem_path(job_id, name) for name in names] + media_type = MIXDOWN_MEDIA_TYPES[ext] + cache_key = _mixdown_cache_key(job_id, ext, names, parsed_gains, start, end) + cache_path = _MIXDOWN_CACHE_DIR / f"{cache_key}.{ext}" + if cache_path.is_file(): + return FileResponse( + cache_path, + media_type=media_type, + headers={"Content-Disposition": f'attachment; filename="mixdown.{ext}"'}, + ) + pre_seek = ["-ss", str(start)] if start is not None else [] post_seek = ["-t", str(end - start)] if start is not None else [] @@ -359,9 +469,10 @@ async def get_mixdown( "pipe:1", ] - media_type = MIXDOWN_MEDIA_TYPES[ext] return StreamingResponse( - _stream_ffmpeg(cmd, context=f"mixdown job={job_id} ext={ext} stems={stems}"), + _stream_ffmpeg( + cmd, context=f"mixdown job={job_id} ext={ext} stems={stems}", cache_path=cache_path + ), media_type=media_type, headers={"Content-Disposition": f'attachment; filename="mixdown.{ext}"'}, ) diff --git a/tests/test_stems_api.py b/tests/test_stems_api.py index da174d24..decc73a7 100644 --- a/tests/test_stems_api.py +++ b/tests/test_stems_api.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import os import pytest from fastapi.testclient import TestClient @@ -18,9 +19,12 @@ def _isolate_registry(): @pytest.fixture def client(tmp_path, monkeypatch): - import app.api.stems as stems_mod + from app.api import stems as stems_mod monkeypatch.setattr(stems_mod, "JOBS_DIR", tmp_path) + # Isolate the mixdown render cache to a per-test dir -- these tests must + # never read or pollute the developer's real cache. + monkeypatch.setattr(stems_mod, "_MIXDOWN_CACHE_DIR", tmp_path / "cache" / "mixdown") from app.main import app return TestClient(app) @@ -379,6 +383,107 @@ def test_mixdown_honors_export_sample_rate(client, tmp_path, monkeypatch): assert int.from_bytes(r.content[24:28], "little") == 48000 +# ─── #290: mixdown render cache ──────────────────────────────────────────── + + +def test_mixdown_cache_hit_skips_second_render(client, tmp_path, monkeypatch): + _skip_without_ffmpeg() + from app.api import stems as stems_mod + + calls = {"n": 0} + original = stems_mod._stream_ffmpeg + + def counting_stream_ffmpeg(*args, **kwargs): + calls["n"] += 1 + return original(*args, **kwargs) + + monkeypatch.setattr(stems_mod, "_stream_ffmpeg", counting_stream_ffmpeg) + + job = _done_job_with_stems(tmp_path, "abcdef000020", ["vocals", "drums"]) + url = f"/api/jobs/{job.id}/mixdown.wav?stems=vocals,drums&gains=1,0.5" + + r1 = client.get(url) + assert r1.status_code == 200 + assert calls["n"] == 1 + cache_files = list((tmp_path / "cache" / "mixdown").glob("*.wav")) + assert len(cache_files) == 1 + + r2 = client.get(url) + assert r2.status_code == 200 + assert calls["n"] == 1 # no second render -- served from cache + assert r2.content == r1.content + + +def test_mixdown_cache_key_varies_with_params(client, tmp_path): + _skip_without_ffmpeg() + job = _done_job_with_stems(tmp_path, "abcdef000021", ["vocals", "drums"]) + base = f"/api/jobs/{job.id}/mixdown.wav?stems=vocals,drums" + + urls = [ + f"{base}&gains=1,0.5", + f"{base}&gains=1,0.9", # different gains + f"{base}&gains=1,0.5&start=0.05&end=0.15", # region trim + ] + for url in urls: + r = client.get(url) + assert r.status_code == 200, url + + cache_files = list((tmp_path / "cache" / "mixdown").glob("*.wav")) + assert len(cache_files) == 3 # three distinct cache entries, no collisions + + +def test_mixdown_failed_render_leaves_no_cache_entry(client, tmp_path): + _skip_without_ffmpeg() + job = _done_job_with_stems(tmp_path, "abcdef000022", ["vocals"]) + # Not a real WAV -- ffmpeg will exit non-zero trying to decode it. + (tmp_path / job.id / "stems" / "vocals.wav").write_bytes(b"not audio data at all") + + r = client.get(f"/api/jobs/{job.id}/mixdown.wav?stems=vocals&gains=1") + assert r.status_code == 200 # HTTP status is already committed mid-stream (#280) + assert r.content == b"" # ffmpeg produced nothing before failing + + cache_dir = tmp_path / "cache" / "mixdown" + leftover = list(cache_dir.glob("*")) if cache_dir.is_dir() else [] + assert leftover == [], "a failed render must not leave a cache entry or a stray temp file" + + +def test_prune_mixdown_cache_bounds_file_count(tmp_path, monkeypatch): + from app.api import stems as stems_mod + + monkeypatch.setattr(stems_mod, "_MIXDOWN_CACHE_MAX_FILES", 3) + monkeypatch.setattr(stems_mod, "_MIXDOWN_CACHE_MAX_BYTES", 10**9) + cache_dir = tmp_path / "mixdown" + cache_dir.mkdir() + for i in range(5): + p = cache_dir / f"entry{i}.wav" + p.write_bytes(b"x") + os.utime(p, (i, i)) # oldest first: entry0 is oldest + + stems_mod._prune_mixdown_cache(cache_dir) + + remaining = {p.name for p in cache_dir.iterdir()} + assert len(remaining) == 3 + assert remaining == {"entry2.wav", "entry3.wav", "entry4.wav"} # newest 3 survive + + +def test_prune_mixdown_cache_bounds_total_size(tmp_path, monkeypatch): + from app.api import stems as stems_mod + + monkeypatch.setattr(stems_mod, "_MIXDOWN_CACHE_MAX_FILES", 100) + monkeypatch.setattr(stems_mod, "_MIXDOWN_CACHE_MAX_BYTES", 25) + cache_dir = tmp_path / "mixdown" + cache_dir.mkdir() + for i in range(5): + p = cache_dir / f"entry{i}.wav" + p.write_bytes(b"x" * 10) # 5 * 10 = 50 bytes total, budget is 25 + os.utime(p, (i, i)) + + stems_mod._prune_mixdown_cache(cache_dir) + + remaining = list(cache_dir.iterdir()) + assert sum(p.stat().st_size for p in remaining) <= 25 + + @pytest.mark.asyncio async def test_stream_ffmpeg_logs_stderr_on_failure(caplog): """#280: a mid-stream ffmpeg failure can't change the HTTP status, so the