From ecdc03aaf4f5d3202e10e8164be864e043232cac Mon Sep 17 00:00:00 2001 From: Thales <> Date: Tue, 3 Nov 2026 07:26:23 +0000 Subject: [PATCH] fix(registry): rebuild a song as done only when its separation finished restore() rebuilt a "done" song from any folder holding one stem, so a process killed during collect(), before original.wav or mix.wav, or in the middle of MP3 storage came back as a finished song with files missing. A kill between the two vocal-split renames brought back a lead vocal without its backing. The last write of a separation is now stems/.complete, after every stem file and the folder are fsynced, and recovery rebuilds a folder as done only when it has the marker. A folder without it goes back in the queue, or comes back as a stopped upload, and its files are kept. Songs from before the marker get it at startup while the registry holds them as done; one whose record was already lost is still rebuilt by its metadata.json. A rebuilt song counts as split only when both halves of the pair are there. --- app/core/registry.py | 91 ++++++-- app/pipeline/runner.py | 15 +- app/pipeline/vocal_split.py | 4 +- tests/test_deleted_jobs_stay_deleted.py | 1 + tests/test_registry_persistence.py | 11 +- tests/test_registry_restore_edges.py | 2 + tests/test_stem_completeness.py | 263 ++++++++++++++++++++++++ tests/test_takes_api.py | 3 +- 8 files changed, 361 insertions(+), 29 deletions(-) create mode 100644 tests/test_stem_completeness.py diff --git a/app/core/registry.py b/app/core/registry.py index 19cc3cf5..07cacb66 100644 --- a/app/core/registry.py +++ b/app/core/registry.py @@ -37,7 +37,7 @@ clean_work, ) from app.core.settings import fsync_dir, quarantine_corrupt -from app.core.stem_files import stem_exists, stem_url +from app.core.stem_files import stem_exists, stem_names_on_disk, stem_path, stem_url from app.core.takes import reconcile_tracks, recover_takes from app.core.versions import clean_stems @@ -540,6 +540,8 @@ def restore(jobs_dir: Path) -> None: for job in to_add.values(): if _sync_tracks(job, jobs_dir / job.id): changed = True + if job.status in _TERMINAL: + _backfill_stems_complete(jobs_dir / job.id) if copy_saved: # Songs finished before metadata.json kept a copy get one now, # or a registry lost later would still reset all of them. A @@ -668,13 +670,13 @@ def _copy_saved_fields(job: Job, jobs_dir: Path, held: set[str]) -> bool: def _resume_or_recover(job: Job, job_dir: Path) -> Job | None: """Decide what a job that was in flight when the process died becomes. - Three states of the same directory need telling apart. If the stems are all - there, the crash landed between the last stem being written and the "done" - persist, so the work is finished and re-running it would duplicate the - library entry. Otherwise the job goes back in the queue from the top, after - the partial demucs output is cleared so collect() cannot mistake it for - results. A job that has already burned its resume is failed loudly rather - than retried forever. + Three states of the same directory need telling apart. If the stems are + marked complete, the crash landed between the last stem being written and + the "done" persist, so the work is finished and re-running it would + duplicate the library entry. Otherwise the job goes back in the queue from + the top, after the partial demucs output is cleared so collect() cannot + mistake it for results. A job that has already burned its resume is failed + loudly rather than retried forever. A job the user trashed before the restart is not resumed (#748). It came back as a Paused row in the queue, offered to Start a song that was in the @@ -789,6 +791,40 @@ def _stopped_from_source(job_dir: Path) -> Job | None: ) +# The last file a separation writes into stems/ (mark_stems_complete): a folder +# without it was cut off part way. +STEMS_COMPLETE = ".complete" + + +def mark_stems_complete(job_dir: Path) -> None: + """Record that a separation wrote every stem file it makes in stems/. + Called last, so a kill at any point before leaves a folder without the + marker. Every stem file and the folder are on the disk before the marker + is: a power cut could otherwise keep the marker and lose a stem's bytes. + Raises OSError.""" + stems_dir = job_dir / "stems" + for name in stem_names_on_disk(stems_dir): + # Opened for writing: Windows only flushes a handle that can write. + with open(stem_path(stems_dir, name), "r+b") as fh: + os.fsync(fh.fileno()) + fsync_dir(stems_dir) + _write_atomic(stems_dir / STEMS_COMPLETE, "") + + +def _backfill_stems_complete(job_dir: Path) -> None: + """The marker for a song made before it, which the registry holds as done, + so a registry lost later still brings it back (songs from before 0.2 have + no metadata.json either). Its stems are long on the disk: only the marker + is written. Never fatal: restore() runs at import time.""" + stems_dir = job_dir / "stems" + try: + if (stems_dir / STEMS_COMPLETE).is_file() or not stem_names_on_disk(stems_dir): + return + _write_atomic(stems_dir / STEMS_COMPLETE, "") + except OSError: + logger.warning("could not mark the stems of %s complete", job_dir.name, exc_info=True) + + def _recover_done_job(job_dir: Path) -> Job | None: found = _rebuild_from_folder(job_dir) return found[0] if found is not None else None @@ -796,29 +832,40 @@ def _recover_done_job(job_dir: Path) -> Job | None: def _rebuild_from_folder(job_dir: Path) -> tuple[Job, bool] | None: """A finished song rebuilt from its folder, and whether its metadata.json - held the SAVED_FIELDS (a folder written before they were kept has none).""" + held the SAVED_FIELDS (a folder written before they were kept has none). + None for a folder whose separation did not finish.""" stems_dir = job_dir / "stems" if not stems_dir.is_dir(): return None + # A restart between the split finishing and its next registry persist + # would otherwise report the job as never split -- derive from disk the + # same way `stems` below does, so the recovered library entry doesn't + # regress (#275). Both or neither: the pair moves in with two renames + # (vocal_split._move_in), and a kill between them leaves a lead alone. + has_split = all(stem_exists(stems_dir, name) for name in EXTRA_STEM_NAMES) + vocal_split = "done" if has_split else "none" # WAV or MP3, whichever the song was stored as (#783). stems = [ {"name": name, "url": stem_url(job_dir.name, stems_dir, name)} - for name in ("original", *STEM_NAMES, *EXTRA_STEM_NAMES) + for name in ("original", *STEM_NAMES, *(EXTRA_STEM_NAMES if has_split else ())) if stem_exists(stems_dir, name) ] if not stems: return None + meta_path = job_dir / "metadata.json" + # A song made before the marker gets it while the registry holds it + # (_backfill_stems_complete). One whose record was already gone has + # metadata.json, written once a song is done, and now only after the marker. + if not (stems_dir / STEMS_COMPLETE).is_file() and not meta_path.is_file(): + logger.warning( + "%s holds stems of a separation that did not finish; not rebuilt as done", + job_dir.name, + ) + return None mix_url = None if stem_exists(stems_dir, "mix"): mix_url = stem_url(job_dir.name, stems_dir, "mix") selected = [stem["name"] for stem in stems if stem["name"] in STEM_NAMES] or list(STEM_NAMES) - # A restart between the split finishing and its next registry persist - # would otherwise report the job as never split -- derive from disk the - # same way `stems` above does, so the recovered library entry doesn't - # regress (#275). - has_split = all(stem_exists(stems_dir, name) for name in EXTRA_STEM_NAMES) - vocal_split = "done" if has_split else "none" - meta_path = job_dir / "metadata.json" meta: dict = {} if meta_path.is_file(): try: @@ -828,11 +875,11 @@ def _rebuild_from_folder(job_dir: Path) -> tuple[Job, bool] | None: if not isinstance(meta, dict): meta = {} # read as none, so the folders after this one still rebuild else: - # Crash window (#284): the process died between status=done and the - # metadata write, leaving a complete stems dir that used to be - # unrecoverable. Recover with a placeholder title and write a minimal - # metadata.json immediately, so the NEXT restart takes the normal - # path -- self-healing, not a permanent special case. + # Crash window (#284): the process died between the stems being + # marked complete and the metadata write, leaving a complete stems dir + # that used to be unrecoverable. Recover with a placeholder title and + # write a minimal metadata.json immediately, so the NEXT restart takes + # the normal path -- self-healing, not a permanent special case. meta = {"title": f"Recovered track {job_dir.name[:6]}"} try: meta_path.write_text(json.dumps(meta, indent=2) + "\n", encoding="utf-8") diff --git a/app/pipeline/runner.py b/app/pipeline/runner.py index 2b1199b7..408c3f62 100644 --- a/app/pipeline/runner.py +++ b/app/pipeline/runner.py @@ -19,7 +19,7 @@ ) from app.core.models import Job, JobCancelled, _set, settle_cancelled from app.core.redact import redact -from app.core.registry import is_upload, saved_fields +from app.core.registry import is_upload, mark_stems_complete, saved_fields from app.core.registry import persist as persist_registry from app.pipeline import jobqueue from app.pipeline.analyze import analyze, refine_key_from_stems @@ -342,6 +342,17 @@ def _store_stems(job: Job, job_dir: Path) -> None: _lap(job, "store", mark) +def _mark_stems_complete(job: Job, job_dir: Path) -> None: + """Last: nothing writes into stems/ after it. A failure costs the marker, + not the song, whose stems are whole: failing the job would delete them. + A song without the marker is rebuilt by the metadata.json it gets once + done, as a song made before the marker is.""" + try: + mark_stems_complete(job_dir) + except OSError: + logger.exception("[%s] could not mark the stems complete", job.id) + + def _run_blocking(job: Job, url: str, job_dir: Path) -> None: _check_cancel(job) mark = time.monotonic() @@ -349,6 +360,7 @@ def _run_blocking(job: Job, url: str, job_dir: Path) -> None: _lap(job, "download", mark) _run_with_band_lookup(job, source, job_dir) _store_stems(job, job_dir) + _mark_stems_complete(job, job_dir) def _run_local_blocking(job: Job, source_path: Path, job_dir: Path) -> None: @@ -358,6 +370,7 @@ def _run_local_blocking(job: Job, source_path: Path, job_dir: Path) -> None: _lap(job, "prepare", mark) _run_with_band_lookup(job, source, job_dir) _store_stems(job, job_dir) + _mark_stems_complete(job, job_dir) def _write_metadata(job: Job, job_dir: Path) -> None: diff --git a/app/pipeline/vocal_split.py b/app/pipeline/vocal_split.py index e58c955f..8cb93ea3 100644 --- a/app/pipeline/vocal_split.py +++ b/app/pipeline/vocal_split.py @@ -93,7 +93,9 @@ def split_vocals(job: Job, stems_dir: Path) -> list[str]: def _move_in(work: Path, stems_dir: Path, names: list[str]) -> list[str]: """Move the pair from `work` into the stems. A move that fails takes back - the one already made; a kill between the two renames is not covered.""" + the one already made. A kill between the two renames leaves a lead alone, + which a song rebuilt from its folder does not count as split + (registry._rebuild_from_folder).""" placed: list[Path] = [] try: for name in names: diff --git a/tests/test_deleted_jobs_stay_deleted.py b/tests/test_deleted_jobs_stay_deleted.py index b9342a09..e0fdea7a 100644 --- a/tests/test_deleted_jobs_stay_deleted.py +++ b/tests/test_deleted_jobs_stay_deleted.py @@ -47,6 +47,7 @@ def _done_job_dir(jobs_dir, job_id="a1b2c3d4e5f6"): stems = jobs_dir / job_id / "stems" stems.mkdir(parents=True) (stems / "vocals.wav").write_bytes(b"RIFF") + _registry.mark_stems_complete(jobs_dir / job_id) return jobs_dir / job_id diff --git a/tests/test_registry_persistence.py b/tests/test_registry_persistence.py index dd59f184..ed2c0c92 100644 --- a/tests/test_registry_persistence.py +++ b/tests/test_registry_persistence.py @@ -30,6 +30,8 @@ def _stems_dir(tmp_path: Path, job_id: str, names=("vocals", "drums")) -> None: d.mkdir(parents=True, exist_ok=True) for n in names: (d / f"{n}.wav").write_bytes(b"RIFF") + # As a separation that finished leaves them. + _registry.mark_stems_complete(tmp_path / job_id) def test_a_queued_job_survives_a_restart(tmp_path: Path): @@ -279,14 +281,15 @@ def test_restore_recovers_automatic_sections_from_metadata(tmp_path: Path): def test_restore_recovers_orphan_without_metadata(tmp_path: Path): - """#284: a crash between status=done and the metadata write used to leave - a complete stems dir permanently unrecoverable. Now it comes back with a - placeholder title, and a minimal metadata.json is written so the next - restart takes the normal recovery path (self-healing).""" + """#284: a crash between the stems being marked complete and the metadata + write used to leave a complete stems dir permanently unrecoverable. Now it + comes back with a placeholder title, and a minimal metadata.json is written + so the next restart takes the normal recovery path (self-healing).""" job_dir = tmp_path / "abcdefabcde0" stems_dir = job_dir / "stems" stems_dir.mkdir(parents=True) (stems_dir / "vocals.wav").write_bytes(b"RIFF") + _registry.mark_stems_complete(job_dir) restore_registry(tmp_path) diff --git a/tests/test_registry_restore_edges.py b/tests/test_registry_restore_edges.py index 8de1b3cb..6261066d 100644 --- a/tests/test_registry_restore_edges.py +++ b/tests/test_registry_restore_edges.py @@ -41,6 +41,8 @@ def _stems(jobs_dir: Path, job_id: str, *names: str) -> Path: stems.mkdir(parents=True, exist_ok=True) for name in names or ("vocals",): (stems / f"{name}.wav").write_bytes(b"RIFF") + # As a separation that finished leaves them. + _registry.mark_stems_complete(jobs_dir / job_id) return jobs_dir / job_id diff --git a/tests/test_stem_completeness.py b/tests/test_stem_completeness.py new file mode 100644 index 00000000..74a1201e --- /dev/null +++ b/tests/test_stem_completeness.py @@ -0,0 +1,263 @@ +"""A song folder is rebuilt as "done" only when its separation finished. + +restore() rebuilds a song from its folder when the registry says it was still +running, or has no entry for it. It used to take any folder holding one stem +for a finished song, so a process killed part way through the pipeline came +back as a "done" song with stems, original.wav or mix.wav missing. The last +thing a separation writes is now a marker in stems/, after every stem file is +on the disk, and only a folder that has it, or a song made before the marker +existed (it has metadata.json, written once a song was done), is rebuilt. +""" + +from __future__ import annotations + +import contextlib +import json +import os +import shutil +from pathlib import Path +from unittest.mock import patch + +import pytest + +from app.core import registry as _registry +from app.core.config import STEM_NAMES +from app.core.models import Job +from app.pipeline import runner, stem_storage +from app.pipeline.vocal_split import _move_in + +SONG = "c0ffee5e1ec7" + + +class _Killed(BaseException): + """The process dying: no `except Exception` on the way out sees it.""" + + +@pytest.fixture(autouse=True) +def _empty_registry(): + _registry._jobs.clear() + _registry._pending_resume.clear() + yield + _registry._jobs.clear() + _registry._pending_resume.clear() + + +def _fake_separate(job: Job, source: Path, job_dir: Path) -> Path: + out = job_dir / "htdemucs_6s" / "source" + out.mkdir(parents=True) + for name in STEM_NAMES: + (out / f"{name}.wav").write_bytes(b"RIFF") + return out + + +def _fake_ffmpeg(job: Job, cmd: list[str]) -> bool: + Path(cmd[-1]).write_bytes(b"RIFF") + return True + + +def _kill(*args, **kwargs): + raise _Killed + + +def _dies_on_second_call(target: str, real): + calls: list[tuple] = [] + + def call(*args, **kwargs): + if calls: + raise _Killed + calls.append(args) + return real(*args, **kwargs) + + return patch(target, call) + + +def _mp3_storage_dies_on_second_call(target: str, real) -> contextlib.ExitStack: + """The real MP3 storage step, with `target` dying on its second call + inside it. Encoding and its check are stubbed: what is under test is the + order the files move in.""" + + def encode(job, wav, tmp, kbps): + tmp.write_bytes(b"ID3") + + def store(job, job_dir): + with _dies_on_second_call(target, real): + stem_storage.store_stems(job, job_dir) + + stack = contextlib.ExitStack() + for p in ( + patch.object(runner, "store_stems", store), + patch.object(stem_storage, "get_stem_format", return_value="mp3"), + patch.object(stem_storage, "get_stem_mp3_kbps", return_value=128), + patch.object(stem_storage, "_encode", encode), + patch.object(stem_storage, "verify_mp3"), + ): + stack.enter_context(p) + return stack + + +def _kill_after(target: str): + real = getattr(runner, target) + + def run_then_die(*args, **kwargs): + real(*args, **kwargs) + raise _Killed + + return patch.object(runner, target, run_then_die) + + +# Each step of the pipeline that writes into stems/, and the gap after the last. +KILLS = { + "collect, one stem moved": lambda: _dies_on_second_call( + "app.pipeline.collect.shutil.move", shutil.move + ), + "before original.wav": lambda: patch.object(runner, "make_original_track", _kill), + "before mix.wav": lambda: patch.object(runner, "make_selected_mix", _kill), + "after mix.wav": lambda: patch.object(runner, "compute_stem_peaks", _kill), + "before stem storage": lambda: patch.object(runner, "store_stems", _kill), + "stem storage, one MP3 moved in": lambda: _mp3_storage_dies_on_second_call( + "app.pipeline.stem_storage.os.replace", os.replace + ), + "stem storage, one WAV deleted": lambda: _mp3_storage_dies_on_second_call( + "pathlib.Path.unlink", Path.unlink + ), + "after stem storage": lambda: _kill_after("_store_stems"), +} + + +def _run(jobs_dir: Path, kill=None) -> Job: + """An upload run through the real pipeline steps that write stems/, with + the registry saved as the app saves it when the job starts.""" + job = Job( + id=SONG, + status="separating", + title="Song", + source_url="local:Song", + selected_stems=["vocals", "drums"], + auto_lyrics=False, + ) + _registry._jobs[job.id] = job + _registry.persist(jobs_dir) + job_dir = jobs_dir / SONG + job_dir.mkdir() + source = job_dir / "source.wav" + source.write_bytes(b"RIFF") + with ( + patch.object(runner, "_prepare_local_source", return_value=source), + patch.object(runner.IdentifyLookup, "start", return_value=None), + patch.object(runner, "analyze"), + patch.object(runner, "separate", _fake_separate), + patch.object(runner, "refine_key_from_stems"), + patch("app.pipeline.collect._run_ffmpeg", _fake_ffmpeg), + patch.object(runner, "compute_stem_peaks", return_value={}), + patch.object(runner, "compute_beat_grid"), + patch.object(runner, "store_stems"), + ): + if kill is None: + runner._run_local_blocking(job, source, job_dir) + else: + with kill(), pytest.raises(_Killed): + runner._run_local_blocking(job, source, job_dir) + return job + + +def _restart(jobs_dir: Path, *, registry_lost: bool) -> Job | None: + _registry._jobs.clear() + _registry._pending_resume.clear() + if registry_lost: + _registry.registry_path(jobs_dir).unlink() + _registry.restore(jobs_dir) + return _registry.get(SONG) + + +@pytest.mark.parametrize("registry_lost", [False, True], ids=["registry kept", "registry lost"]) +@pytest.mark.parametrize("step", list(KILLS)) +def test_a_song_killed_part_way_is_never_rebuilt_as_done(tmp_path: Path, step, registry_lost): + _run(tmp_path, KILLS[step]) + assert (tmp_path / SONG / "stems").is_dir(), "the kill left stems behind" + + job = _restart(tmp_path, registry_lost=registry_lost) + + assert job is not None and job.status != "done" + if registry_lost: + # Kept as the upload it is, so Extract can run it again. + assert job.status == "stopped" + else: + assert job.status == "queued" + assert _registry.take_pending_resume() == [SONG] + assert (tmp_path / SONG / "source.wav").is_file() + + +@pytest.mark.parametrize("registry_lost", [False, True], ids=["registry kept", "registry lost"]) +def test_a_song_whose_separation_finished_is_rebuilt_as_done(tmp_path: Path, registry_lost): + """The process died after the last write, before "done" was saved.""" + _run(tmp_path) + + job = _restart(tmp_path, registry_lost=registry_lost) + + assert job is not None and job.status == "done" + assert {s["name"] for s in job.stems} == {"original", *STEM_NAMES} + assert job.mix_url == f"/api/jobs/{SONG}/stems/mix.wav" + assert _registry.take_pending_resume() == [] + + +def test_a_marker_that_cannot_be_written_does_not_fail_the_song(tmp_path: Path, caplog): + """Failing the job would delete stems that are whole. The song is then + rebuilt by its metadata.json, as a song made before the marker is.""" + with patch.object(runner, "mark_stems_complete", side_effect=OSError("disk full")): + job = _run(tmp_path) + assert "could not mark the stems complete" in caplog.text + assert not (tmp_path / SONG / "stems" / _registry.STEMS_COMPLETE).exists() + runner._write_metadata(job, tmp_path / SONG) # what the runner does once done + + rebuilt = _restart(tmp_path, registry_lost=True) + + assert rebuilt is not None and rebuilt.status == "done" + assert rebuilt.title == "Song" + + +def test_a_kill_between_the_two_split_renames_leaves_the_song_unsplit(tmp_path: Path): + """The lead/backing pair moves in with two renames. A song rebuilt with + only one of them would list a lead without its backing.""" + _run(tmp_path) + stems_dir = tmp_path / SONG / "stems" + (tmp_path / SONG / "metadata.json").write_text(json.dumps({"title": "Song"}), "utf-8") + work = stems_dir / ".vocal_split.0123.tmp" + work.mkdir() + for name in ("lead_vocals", "backing_vocals"): + (work / f"{name}.wav").write_bytes(b"RIFF") + real = os.replace + renames: list[str] = [] + + def replace(src, dst): + if renames: + raise _Killed + renames.append(str(dst)) + return real(src, dst) + + with patch("app.pipeline.vocal_split.os.replace", replace), pytest.raises(_Killed): + _move_in(work, stems_dir, ["lead_vocals", "backing_vocals"]) + assert (stems_dir / "lead_vocals.wav").is_file() + + job = _restart(tmp_path, registry_lost=True) + + assert job is not None and job.status == "done" + assert job.vocal_split == "none" + assert not {"lead_vocals", "backing_vocals"} & {s["name"] for s in job.stems} + + +def test_a_song_made_before_the_marker_gets_it_while_the_registry_holds_it(tmp_path: Path): + """Songs from before 0.2 have no metadata.json: the registry is the only + record that they finished, so the marker comes from it.""" + stems_dir = tmp_path / SONG / "stems" + stems_dir.mkdir(parents=True) + for name in STEM_NAMES: + (stems_dir / f"{name}.wav").write_bytes(b"RIFF") + _registry._jobs[SONG] = Job(id=SONG, status="done", title="Old Song") + _registry.persist(tmp_path) + + _restart(tmp_path, registry_lost=False) + assert (stems_dir / _registry.STEMS_COMPLETE).is_file() + job = _restart(tmp_path, registry_lost=True) + + assert job is not None and job.status == "done" + assert {s["name"] for s in job.stems} == set(STEM_NAMES) diff --git a/tests/test_takes_api.py b/tests/test_takes_api.py index 48c5e597..b7d7bd46 100644 --- a/tests/test_takes_api.py +++ b/tests/test_takes_api.py @@ -512,9 +512,10 @@ async def test_takes_survive_a_registry_rebuild_from_disk(jobs_dir, client): """_recover_done_job rebuilds a job from its folder; the takes are files first, so they come back with it.""" skip_without_ffmpeg() - from app.core.registry import _recover_done_job + from app.core.registry import _recover_done_job, mark_stems_complete _make_job(jobs_dir) + mark_stems_complete(jobs_dir / JOB) async with client as c: tid = await _finished_take(c, jobs_dir) await _create(c) # unfinished: not a lane