fix: atomic import swap, disk/DB replace consistency, guarded pipeline loop

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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`,
+    """Atomically place the album's audio (+ one cover) at `final`, replacing any existing folder.
-    replacing any existing folder there. Non-audio artifacts and nested subfolders are left
+    Assembles into a sibling temp dir, then swaps via rename so a mid-assembly failure never
-    behind (the caller removes `staging`). Tolerates a missing `staging`. Returns `final`."""
+    destroys the existing copy. Non-audio artifacts/subfolders in `staging` are left behind.
-    if os.path.isdir(final):
+    Tolerates a missing `staging`. Returns `final`."""
-        shutil.rmtree(final)
+    os.makedirs(os.path.dirname(final), exist_ok=True)
-    os.makedirs(final, exist_ok=True)
+    tmp = final + ".importing"
-    if not os.path.isdir(staging):
+    shutil.rmtree(tmp, ignore_errors=True)
-        return final
+    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(final, name))
+                shutil.move(src, os.path.join(tmp, name))
         cover = _find_cover(staging)
         if cover is not None:
-        shutil.move(cover, os.path.join(final, "cover.jpg"))
+            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):
+        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

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)
+            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)

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
+        existing_q = _library_quality(conn, target)
-
+        new_q = quality_class(winner.quality)
+        _set_state(conn, job_id, "imported", "import")
+        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)
-
-        # 6. import (DB)
-        _set_state(conn, job_id, "imported", "import")
             _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

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)

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