Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ htmlcov/
# Runtime job artifacts and local build scratch
data/
jobs/
cache/
settings.json
.run/
.build
Expand Down
131 changes: 121 additions & 10 deletions app/api/stems.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from __future__ import annotations

import asyncio
import hashlib
import logging
import os
import re
Expand All @@ -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

Expand Down Expand Up @@ -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:
Comment thread
github-code-quality[bot] marked this conversation as resolved.
Fixed
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."""
Expand Down Expand Up @@ -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,
Expand All @@ -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()
Expand All @@ -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:
Expand Down Expand Up @@ -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")

Expand All @@ -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 []

Expand Down Expand Up @@ -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}"'},
)
Expand Down
107 changes: 106 additions & 1 deletion tests/test_stems_api.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from __future__ import annotations

import json
import os

import pytest
from fastapi.testclient import TestClient
Expand All @@ -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)
Expand Down Expand Up @@ -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
Expand Down