diff --git a/app/api/stems.py b/app/api/stems.py index 974a83e3..dc2aa947 100644 --- a/app/api/stems.py +++ b/app/api/stems.py @@ -65,6 +65,21 @@ "ogg": ["-c:a", "libvorbis", "-q:a", "6"], } MIXDOWN_CODECS = {ext: [*args, "-f", ext] for ext, args in _ENCODE_ARGS.items()} + +# Containers whose muxer finishes the file by seeking back to a header it wrote +# earlier, which is impossible on a pipe (#458): +# +# wav RIFF and data chunk sizes, left as the 0xFFFFFFFF placeholder, so the +# file claims ~4 GB of audio and strict hardware refuses it +# flac STREAMINFO total samples and the MD5 signature, left as zeros +# mp3 the Xing/Info frame, which ffmpeg simply omits rather than writing a +# wrong one -- and these are VBR (-q:a 2), so without it there is no +# duration and no seek table +# +# ogg is absent on purpose: a granule position rides on every page, so nothing +# is patched afterwards. mp4 is handled separately by get_video_mixdown, which +# already muxes fragmented (frag_keyframe+empty_moov) for exactly this reason. +_SEEKABLE_OUTPUT_EXTS = frozenset({"wav", "flac", "mp3"}) MIXDOWN_MEDIA_TYPES = { "wav": "audio/wav", "mp3": "audio/mpeg", @@ -84,6 +99,18 @@ _MIXDOWN_CACHE_MAX_BYTES = 500 * 1024 * 1024 # 500 MB +# Folded into every cache key. Bump it whenever a render's *output* changes for +# inputs that are otherwise identical, so entries written by an older build can +# never be served by a newer one. +# +# "2": everything cached before #458 was streamed through a pipe and carries an +# unpatched container header. Those files are wrong, the key that produced them +# is still reachable, and _prune_mixdown_cache evicts by age rather than +# validity -- so without this a user who exported before upgrading would be +# handed the same broken file back for the same parameters indefinitely. +_RENDER_CACHE_VERSION = "2" + + def _mixdown_cache_key( job_id: str, ext: str, @@ -101,6 +128,7 @@ def _mixdown_cache_key( ffmpeg-equivalent gain strings (e.g. "1" vs "1.0") share a cache entry.""" raw = "|".join( [ + _RENDER_CACHE_VERSION, job_id, ext, ",".join(names), @@ -397,6 +425,96 @@ async def _stream_ffmpeg(cmd: list[str], context: str = "", cache_path: Path | N tmp_path.unlink(missing_ok=True) +async def _render_to_file( + cmd: list[str], suffix: str, context: str = "", cache_path: Path | None = None +) -> Path: + """Run ffmpeg to completion against a real file and return the path. + + The sibling of _stream_ffmpeg, for the containers in _SEEKABLE_OUTPUT_EXTS + whose muxer finishes the file by seeking back to a header it wrote earlier + (#458). A pipe cannot seek, so those headers were never patched and every + exported WAV claimed roughly 4 GB of audio. + + Returning a path rather than yielding chunks is what lets the caller answer + with FileResponse, and that is worth more than the extra buffering costs. + A streamed render commits HTTP 200 before ffmpeg has exited, so a failure + reaches the client as a truncated file that only the server log records; + here the exit code is known while the response is still ours to choose, so + a failed render is an honest 500. It also brings Content-Length and range + requests, which chunked encoding cannot offer at all. + + `cmd` must not carry an output path -- this appends one. With `cache_path` + the temp file is created beside it and a clean render is renamed into place + as the cache entry (#290); the caller then serves the cache entry and must + not delete it. Without one the caller owns the returned temp file and is + responsible for removing it once the response has been sent. + """ + 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{suffix}") + else: + _MIXDOWN_CACHE_DIR.mkdir(parents=True, exist_ok=True) + tmp_path = _MIXDOWN_CACHE_DIR / f".render.{uuid.uuid4().hex}.tmp{suffix}" + # Both names start with a dot, which is what keeps an in-flight render out + # of _prune_mixdown_cache's eviction list and out of the cache-hit path. + + proc = await asyncio.create_subprocess_exec( + *cmd, + "-y", + str(tmp_path), + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.PIPE, + ) + stderr_tail: deque[str] = deque(maxlen=30) + drain_task = asyncio.create_task(_drain_stderr(proc.stderr, stderr_tail)) + ok = False + try: + try: + await asyncio.wait_for(proc.wait(), timeout=TIMEOUT_FFMPEG) + except (TimeoutError, asyncio.TimeoutError): + proc.kill() + await proc.wait() + try: + await asyncio.wait_for(drain_task, timeout=5) + except (TimeoutError, asyncio.TimeoutError): + drain_task.cancel() + if proc.returncode != 0: + logger.warning( + "render ffmpeg exit %s [%s]: %s", + proc.returncode, + context, + " | ".join(list(stderr_tail)[-8:]) or "(no stderr)", + ) + raise HTTPException(status_code=500, detail="export failed") + ok = True + finally: + # A cancelled request (client gone) unwinds through here too, so the + # process is never left running and the temp file is never orphaned. + if proc.returncode is None: + proc.kill() + await proc.wait() + if not ok: + tmp_path.unlink(missing_ok=True) + + if cache_path is not None: + os.replace(tmp_path, cache_path) + _prune_mixdown_cache(cache_path.parent) + return cache_path + return tmp_path + + +def _unlink_later(path: Path) -> None: + """Drop a rendered temp file once its response has been sent. + + Only ever attached to an uncached render. A cached one returns the cache + entry itself, and deleting that would throw away the render the cache + exists to keep.""" + try: + path.unlink(missing_ok=True) + except OSError: + logger.debug("could not remove rendered export %s", path, exc_info=True) + + 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 @@ -533,16 +651,17 @@ async def get_stem( "pcm_s16le", "-f", "wav", - "pipe:1", ] - return StreamingResponse( - _stream_ffmpeg(cmd, context=f"stem-region job={job_id} stem={name}"), + rendered = await _render_to_file(cmd, ".wav", context=f"stem-region job={job_id} stem={name}") + return FileResponse( + rendered, media_type="audio/wav", headers={ "Content-Disposition": ( f'attachment; filename="{_stem_download_name(job_id, name, "wav", "_region")}"' ) }, + background=BackgroundTask(_unlink_later, rendered), ) @@ -596,14 +715,18 @@ async def get_stem_mp3( "2", # VBR ~190 kbps "-f", "mp3", - "pipe:1", ] # Only the trimmed branch reaches here; the untrimmed one returned above. filename = _stem_download_name(job_id, name, "mp3", "_region") - return StreamingResponse( - _stream_ffmpeg(cmd, context=f"stem-mp3 job={job_id} stem={name}"), + # Rendered rather than streamed for the same reason as the WAV region + # above: on a pipe ffmpeg omits the Xing frame entirely, and these are VBR, + # so the result has no duration and no seek table (#458). + rendered = await _render_to_file(cmd, ".mp3", context=f"stem-mp3 job={job_id} stem={name}") + return FileResponse( + rendered, media_type="audio/mpeg", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, + background=BackgroundTask(_unlink_later, rendered), ) @@ -739,19 +862,27 @@ async def get_mixdown( *post_seek, *codec, *rate, - "pipe:1", ] + context = f"mixdown job={job_id} ext={ext} stems={stems}" + headers = { + "Content-Disposition": ( + f'attachment; filename="{_mixdown_download_name(job_id, ext, start is not None)}"' + ) + } + if ext in _SEEKABLE_OUTPUT_EXTS: + # Rendered to a file so the muxer can seek back and finish its header + # (#458). It lands in the cache, so nothing is deleted afterwards -- + # this is the same file a later identical request will be served from + # by the cache-hit branch above. + rendered = await _render_to_file(cmd, f".{ext}", context=context, cache_path=cache_path) + return FileResponse(rendered, media_type=media_type, headers=headers) + # ogg needs no back-patching, so it keeps streaming and the client sees + # bytes as soon as ffmpeg produces them. return StreamingResponse( - _stream_ffmpeg( - cmd, context=f"mixdown job={job_id} ext={ext} stems={stems}", cache_path=cache_path - ), + _stream_ffmpeg([*cmd, "pipe:1"], context=context, cache_path=cache_path), media_type=media_type, - headers={ - "Content-Disposition": ( - f'attachment; filename="{_mixdown_download_name(job_id, ext, start is not None)}"' - ) - }, + headers=headers, ) diff --git a/tests/test_stems_api.py b/tests/test_stems_api.py index dbdc0024..df9b074e 100644 --- a/tests/test_stems_api.py +++ b/tests/test_stems_api.py @@ -604,13 +604,13 @@ def test_mixdown_cache_hit_skips_second_render(client, tmp_path, monkeypatch): from app.api import stems as stems_mod calls = {"n": 0} - original = stems_mod._stream_ffmpeg + original = stems_mod._render_to_file # WAV renders via the seekable-output path - def counting_stream_ffmpeg(*args, **kwargs): + def counting_render(*args, **kwargs): calls["n"] += 1 return original(*args, **kwargs) - monkeypatch.setattr(stems_mod, "_stream_ffmpeg", counting_stream_ffmpeg) + monkeypatch.setattr(stems_mod, "_render_to_file", counting_render) 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" @@ -652,8 +652,11 @@ def test_mixdown_failed_render_leaves_no_cache_entry(client, tmp_path): (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 + # A rendered format knows ffmpeg's exit code before a byte is sent, so the + # failure is an honest status rather than the empty 200 a streamed render + # was stuck with once its headers had gone out (#458, and the limitation + # #280 documented). + assert r.status_code == 500 cache_dir = tmp_path / "cache" / "mixdown" leftover = list(cache_dir.glob("*")) if cache_dir.is_dir() else [] @@ -826,3 +829,147 @@ def test_video_mux_happy(client, tmp_path): assert r.headers["content-type"] == "video/mp4" # ISO-BMFF: bytes 4-8 of the first box are the "ftyp" type. assert r.content[4:8] == b"ftyp" + + +# ─── WAV/FLAC headers must be finalised (seekable output, not pipe:1) ───── + + +def _assert_wav_header_sizes_match(content: bytes) -> None: + """ffmpeg's WAV muxer can only patch the RIFF and data sizes on a seekable + output; streamed via pipe:1 both stay 0xFFFFFFFF and strict hardware + players (samplers, drum machines) reject the file.""" + assert content[:4] == b"RIFF" and content[8:12] == b"WAVE" + riff_size = int.from_bytes(content[4:8], "little") + assert riff_size == len(content) - 8, f"RIFF size {riff_size:#x} vs file {len(content)}" + pos = 12 + while pos + 8 <= len(content): + cid = content[pos : pos + 4] + clen = int.from_bytes(content[pos + 4 : pos + 8], "little") + if cid == b"data": + assert clen != 0xFFFFFFFF, "data chunk size left as placeholder" + assert pos + 8 + clen == len(content), "data chunk does not end at EOF" + return + pos += 8 + clen + (clen & 1) + raise AssertionError("no data chunk found") + + +def test_mixdown_wav_header_sizes_are_finalised(client, tmp_path): + _skip_without_ffmpeg() + job = _done_job_with_stems(tmp_path, "abcdef000030", ["vocals", "drums"]) + r = client.get(f"/api/jobs/{job.id}/mixdown.wav?stems=vocals,drums&gains=1,0.5") + assert r.status_code == 200 + _assert_wav_header_sizes_match(r.content) + # The cached copy is the same finalised file. + (cached,) = (tmp_path / "cache" / "mixdown").glob("*.wav") + assert cached.read_bytes() == r.content + r2 = client.get(f"/api/jobs/{job.id}/mixdown.wav?stems=vocals,drums&gains=1,0.5") + _assert_wav_header_sizes_match(r2.content) + + +def test_stem_region_wav_header_sizes_are_finalised(client, tmp_path): + _skip_without_ffmpeg() + job = _done_job_with_stems(tmp_path, "abcdef000031", ["vocals"]) + r = client.get(f"/api/jobs/{job.id}/stems/vocals.wav?start=0&end=0.05") + assert r.status_code == 200 + _assert_wav_header_sizes_match(r.content) + # No temp file left behind for the uncached region render. + cache_dir = tmp_path / "cache" / "mixdown" + assert not [p for p in cache_dir.glob(".*") if cache_dir.is_dir()] + + +def test_mixdown_flac_streaminfo_has_total_samples(client, tmp_path): + _skip_without_ffmpeg() + job = _done_job_with_stems(tmp_path, "abcdef000032", ["vocals"]) + r = client.get(f"/api/jobs/{job.id}/mixdown.flac?stems=vocals&gains=1") + assert r.status_code == 200 + assert r.content[:4] == b"fLaC" + # STREAMINFO block: 4-byte header, then 34 bytes; total_samples is the low + # 36 bits of bytes 13..21 of the block body. + body = r.content[8 : 8 + 34] + total_samples = int.from_bytes(body[13:21], "big") & ((1 << 36) - 1) + assert total_samples > 0, "STREAMINFO total_samples left at 0 (unseekable output)" + + +def _assert_has_xing_header(content: bytes) -> None: + """A VBR MP3 carries its duration and seek table in a Xing (or Info) frame + near the start. ffmpeg writes that frame by seeking back once the encode + ends, and on a pipe it simply omits it rather than writing a wrong one -- + so the file plays but reports an estimated duration and seeks badly.""" + assert content[:3] == b"ID3" or content[:2] == b"\xff\xfb", "not an MP3 stream" + head = content[:8192] + assert b"Xing" in head or b"Info" in head, "no Xing/Info frame (unseekable output)" + + +def test_mixdown_mp3_has_a_xing_header(client, tmp_path): + _skip_without_ffmpeg() + job = _done_job_with_stems(tmp_path, "abcdef000033", ["vocals"]) + r = client.get(f"/api/jobs/{job.id}/mixdown.mp3?stems=vocals&gains=1") + assert r.status_code == 200 + _assert_has_xing_header(r.content) + + +def test_stem_region_mp3_has_a_xing_header(client, tmp_path): + """The MP3 region export goes through its own endpoint, which the original + fix left on pipe:1 while the WAV one beside it was corrected.""" + _skip_without_ffmpeg() + job = _done_job_with_stems(tmp_path, "abcdef000034", ["vocals"]) + r = client.get(f"/api/jobs/{job.id}/stems/vocals.mp3?start=0&end=0.05") + assert r.status_code == 200 + _assert_has_xing_header(r.content) + # Uncached render: the temp file must not survive the response. + cache_dir = tmp_path / "cache" / "mixdown" + assert not (list(cache_dir.glob(".*")) if cache_dir.is_dir() else []) + + +def test_rendered_exports_declare_their_length(client, tmp_path): + """The point of answering with a file rather than a chunked stream: the + client is told how big the download is, so a progress bar and a range + request are both possible.""" + _skip_without_ffmpeg() + job = _done_job_with_stems(tmp_path, "abcdef000035", ["vocals"]) + for url in ( + f"/api/jobs/{job.id}/mixdown.wav?stems=vocals&gains=1", + f"/api/jobs/{job.id}/mixdown.flac?stems=vocals&gains=1", + f"/api/jobs/{job.id}/mixdown.mp3?stems=vocals&gains=1", + f"/api/jobs/{job.id}/stems/vocals.wav?start=0&end=0.05", + ): + r = client.get(url) + assert r.status_code == 200, url + assert int(r.headers["content-length"]) == len(r.content), url + + +def test_ogg_is_still_streamed(client, tmp_path): + """Ogg keeps a granule position on every page and patches nothing + afterwards, so it has no reason to pay for a full render first.""" + _skip_without_ffmpeg() + job = _done_job_with_stems(tmp_path, "abcdef000036", ["vocals"]) + r = client.get(f"/api/jobs/{job.id}/mixdown.ogg?stems=vocals&gains=1") + assert r.status_code == 200 + assert r.content[:4] == b"OggS" + assert "content-length" not in r.headers, "ogg should still be chunked" + + +def test_bumping_the_render_version_invalidates_the_cache(monkeypatch): + """Every entry written before #458 holds an unpatched container header, and + the key that produced it is still reachable. Without a version in the key + those files stay serveable forever, because the cache evicts by age rather + than by validity.""" + from app.api import stems as stems_mod + + args = ("abcdef000037", "wav", ["vocals"], [1.0], None, None, None) + before = stems_mod._mixdown_cache_key(*args) + monkeypatch.setattr(stems_mod, "_RENDER_CACHE_VERSION", "999") + assert stems_mod._mixdown_cache_key(*args) != before + + +def test_cached_render_survives_its_response(client, tmp_path): + """The cleanup task is only ever attached to an uncached render. Attaching + it to a cached one would delete the entry the cache exists to keep, turning + every request into a fresh render.""" + _skip_without_ffmpeg() + job = _done_job_with_stems(tmp_path, "abcdef000038", ["vocals"]) + url = f"/api/jobs/{job.id}/mixdown.wav?stems=vocals&gains=1" + assert client.get(url).status_code == 200 + (cached,) = (tmp_path / "cache" / "mixdown").glob("*.wav") + assert cached.is_file(), "the cache entry was deleted with the response" + assert client.get(url).content == cached.read_bytes()