From 88ef4149d958dd89d02d99cbd7f3a9b66cd11ab6 Mon Sep 17 00:00:00 2001 From: Jonathan Date: Sat, 11 Jul 2026 15:27:48 +0200 Subject: [PATCH] fix: atomic import swap, disk/DB replace consistency, guarded pipeline loop Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/library.py | 38 +++++++++++++-------- worker/lyra_worker/main.py | 8 +++-- worker/lyra_worker/pipeline.py | 38 +++++++++++++++------ worker/tests/test_staging.py | 12 +++++++ worker/tests/test_staging_pipeline.py | 49 +++++++++++++++++++++++++++ 5 files changed, 119 insertions(+), 26 deletions(-) diff --git a/worker/lyra_worker/library.py b/worker/lyra_worker/library.py index b9817e6..94595da 100644 --- a/worker/lyra_worker/library.py +++ b/worker/lyra_worker/library.py @@ -46,19 +46,29 @@ def _find_cover(staging: str) -> str | None: def import_album(staging: str, final: str) -> str: - """Move the album's audio files (+ one cover image) from `staging` into a fresh `final`, - replacing any existing folder there. Non-audio artifacts and nested subfolders are left - behind (the caller removes `staging`). Tolerates a missing `staging`. Returns `final`.""" + """Atomically place the album's audio (+ one cover) at `final`, replacing any existing folder. + Assembles into a sibling temp dir, then swaps via rename so a mid-assembly failure never + destroys the existing copy. Non-audio artifacts/subfolders in `staging` are left behind. + Tolerates a missing `staging`. Returns `final`.""" + os.makedirs(os.path.dirname(final), exist_ok=True) + tmp = final + ".importing" + shutil.rmtree(tmp, ignore_errors=True) + os.makedirs(tmp) + if os.path.isdir(staging): + for name in sorted(os.listdir(staging)): + src = os.path.join(staging, name) + if os.path.isfile(src) and os.path.splitext(name)[1].lower() in _AUDIO_EXT: + shutil.move(src, os.path.join(tmp, name)) + cover = _find_cover(staging) + if cover is not None: + shutil.move(cover, os.path.join(tmp, "cover.jpg")) + # swap: keep the old copy until the new one is assembled, then rename into place if os.path.isdir(final): - shutil.rmtree(final) - os.makedirs(final, exist_ok=True) - if not os.path.isdir(staging): - return final - for name in sorted(os.listdir(staging)): - src = os.path.join(staging, name) - if os.path.isfile(src) and os.path.splitext(name)[1].lower() in _AUDIO_EXT: - shutil.move(src, os.path.join(final, name)) - cover = _find_cover(staging) - if cover is not None: - shutil.move(cover, os.path.join(final, "cover.jpg")) + old = final + ".old" + shutil.rmtree(old, ignore_errors=True) + os.rename(final, old) + os.rename(tmp, final) + shutil.rmtree(old, ignore_errors=True) + else: + os.rename(tmp, final) return final diff --git a/worker/lyra_worker/main.py b/worker/lyra_worker/main.py index de1dba3..e7eb2b6 100644 --- a/worker/lyra_worker/main.py +++ b/worker/lyra_worker/main.py @@ -40,8 +40,12 @@ def run_forever() -> None: time.sleep(IDLE_SLEEP) continue print(f"worker: claimed job {job_id}", flush=True) - run_pipeline(conn, job_id, adapters, resolver=resolver, tagger=tagger, - upgrade_cutoff=mcfg.quality_cutoff) + try: + run_pipeline(conn, job_id, adapters, resolver=resolver, tagger=tagger, + upgrade_cutoff=mcfg.quality_cutoff) + except Exception as e: # one bad job must not take down the worker loop + print(f"worker: pipeline failed for job {job_id}: {e}", flush=True) + conn.rollback() if mcfg.enabled: try: reconcile(conn, job_id, mcfg) diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index 25b6ef1..ae7b02c 100644 --- a/worker/lyra_worker/pipeline.py +++ b/worker/lyra_worker/pipeline.py @@ -80,6 +80,16 @@ def _already_in_library_at_cutoff(conn: psycopg.Connection, target: MBTarget, cu return cur.fetchone() is not None +def _library_quality(conn: psycopg.Connection, target: MBTarget) -> int | None: + with conn.cursor() as cur: + cur.execute( + 'SELECT "qualityClass" FROM "LibraryItem" WHERE artist = %s AND album = %s', + (target.artist, target.album), + ) + row = cur.fetchone() + return row[0] if row else None + + def _persist_candidates(conn: psycopg.Connection, job_id: str, candidates: list[Candidate]) -> None: with conn.cursor() as cur: for c in candidates: @@ -211,16 +221,24 @@ def run_pipeline( return final = album_dir(dest_root, target) - import_album(staging, final) # clean audio(+cover) move into the library, replacing any prior copy - - if tagger is not None: - try: - tagger.tag_album(final, target) - except Exception as e: # a tagging failure must not discard a good download - print(f"pipeline: tagging failed for job {job_id}: {e}", flush=True) - - # 6. import (DB) + existing_q = _library_quality(conn, target) + new_q = quality_class(winner.quality) _set_state(conn, job_id, "imported", "import") - _import(conn, job_id, target, winner, final) + if existing_q is None or new_q > existing_q: + import_album(staging, final) # clean audio(+cover) move, atomically replacing any prior copy + if tagger is not None: + try: + tagger.tag_album(final, target) + except Exception as e: # a tagging failure must not discard a good download + print(f"pipeline: tagging failed for job {job_id}: {e}", flush=True) + _import(conn, job_id, target, winner, final) + else: + # an existing copy is already at >= this quality — keep it untouched, just complete the request + with conn.cursor() as cur: + cur.execute( + "UPDATE \"Request\" SET status = 'completed' WHERE id = %s", + (_request_id(conn, job_id),), + ) + conn.commit() finally: shutil.rmtree(staging, ignore_errors=True) # a partial/failed download never persists diff --git a/worker/tests/test_staging.py b/worker/tests/test_staging.py index efba482..a9be861 100644 --- a/worker/tests/test_staging.py +++ b/worker/tests/test_staging.py @@ -41,6 +41,18 @@ def test_import_album_tolerates_missing_staging(tmp_path): assert os.path.isdir(final) and os.listdir(final) == [] +def test_import_album_leaves_no_temp_siblings(tmp_path): + final = tmp_path / "Artist" / "Album" + final.mkdir(parents=True) + (final / "old.flac").write_bytes(b"old") + staging = tmp_path / "stg" + staging.mkdir() + (staging / "01 New.flac").write_bytes(b"new") + import_album(str(staging), str(final)) + assert sorted(os.listdir(final)) == ["01 New.flac"] # replaced + assert os.listdir(tmp_path / "Artist") == ["Album"] # no .importing / .old leftovers + + def test_clear_staging_root_removes_staging(tmp_path): stg = tmp_path / ".staging" / "j1" stg.mkdir(parents=True) diff --git a/worker/tests/test_staging_pipeline.py b/worker/tests/test_staging_pipeline.py index 089f95e..60a316d 100644 --- a/worker/tests/test_staging_pipeline.py +++ b/worker/tests/test_staging_pipeline.py @@ -68,3 +68,52 @@ def test_retry_imports_clean_with_no_orphans(conn, tmp_path): final = tmp_path / "Radiohead" / "In Rainbows" # exactly the 3 real tracks — the attempt-1 "orphan-a.flac" is NOT present assert sorted(os.listdir(final)) == ["0 track.flac", "1 track.flac", "2 track.flac"] + + +def test_existing_equal_quality_copy_is_kept_not_replaced(conn, tmp_path): + # An existing lossy (q1) copy on disk + DB. A monitored upgrade job whose only candidate is also + # q1 (FakeYouTube) must NOT replace the files or the DB row. + import os as _os + from lyra_worker.adapters.fakes import FakeYouTube + from lyra_worker.library import album_dir + from lyra_worker.types import MBTarget + + target = MBTarget(artist="A", album="B") + final = album_dir(str(tmp_path), target) + _os.makedirs(final, exist_ok=True) + open(_os.path.join(final, "existing.opus"), "wb").close() + with conn.cursor() as cur: + cur.execute( + "INSERT INTO \"MonitoredRelease\" (id, \"artistMbid\", \"artistName\", \"rgMbid\", album, " + "\"secondaryTypes\", monitored, state, \"currentQualityClass\", \"createdAt\") " + "VALUES (gen_random_uuid()::text, 'a1', 'A', 'rg1', 'B', '{}', true, 'grabbed', 1, now()) RETURNING id" + ) + rel_id = cur.fetchone()[0] + cur.execute( + "INSERT INTO \"Request\" (id, artist, album, status, \"monitoredReleaseId\", \"createdAt\") " + "VALUES (gen_random_uuid()::text, 'A', 'B', 'pending', %s, now()) RETURNING id", + (rel_id,), + ) + req = cur.fetchone()[0] + cur.execute( + "INSERT INTO \"LibraryItem\" (id, \"requestId\", artist, album, path, source, format, " + "\"qualityClass\", \"importedAt\") " + "VALUES (gen_random_uuid()::text, %s, 'A', 'B', %s, 'youtube', 'OPUS', 1, now())", + (req, final), + ) + cur.execute( + "INSERT INTO \"Job\" (id, \"requestId\", state, \"currentStage\", attempts, \"createdAt\", \"updatedAt\") " + "VALUES (gen_random_uuid()::text, %s, 'requested', 'intake', 0, now(), now()) RETURNING id", + (req,), + ) + job_id = cur.fetchone()[0] + conn.commit() + + run_pipeline(conn, job_id, [FakeYouTube()], dest_root=str(tmp_path), upgrade_cutoff=2) + + with conn.cursor() as cur: + cur.execute('SELECT state FROM "Job" WHERE id = %s', (job_id,)) + assert cur.fetchone()[0] == "imported" + cur.execute('SELECT count(*) FROM "LibraryItem" WHERE artist=%s AND album=%s AND source=%s', ("A", "B", "youtube")) + assert cur.fetchone()[0] == 1 # the original row is untouched + assert _os.path.isfile(_os.path.join(final, "existing.opus")) # the existing files were kept