feat(worker): separate download staging volume (STAGING_DIR)

Staging is no longer forced under the library. The worker reads STAGING_ROOT
(container path, default /music/.staging) and stages per-job downloads there;
docker-compose mounts ${STAGING_DIR:-${MUSIC_DIR}/.staging} at /staging with
STAGING_ROOT=/staging. Set STAGING_DIR to a fast local disk when the library
(MUSIC_DIR) is a network share so temp download I/O stays off the share.

Safe cross-volume: import_album already assembles into a temp dir on the
library volume and swaps atomically there, so partial downloads never touch
the library and the final swap stays atomic. clear_staging_root now clears the
root's contents (mount-safe) rather than removing the root. Documented in
.env.example; new tests cover the separate-root path.

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

.env.example

@@ -8,3 +8,9 @@ LYRA_SECRET_KEY=MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=
 
 # Host directory for the downloaded music library (bind-mounted into the worker at /music)
 MUSIC_DIR=./music
+
+# Host directory for per-job download staging (bind-mounted into the worker at /staging).
+# Defaults to MUSIC_DIR/.staging. Point this at a fast LOCAL disk when MUSIC_DIR is a network
+# share (SMB/NFS) so temporary downloads don't hammer the share; the final import copies the
+# finished album onto the library volume and swaps it in atomically.
+# STAGING_DIR=/srv/lyra-staging

docker-compose.yml

@@ -35,12 +35,16 @@ services:
     environment:
       DATABASE_URL: ${DATABASE_URL}
       LYRA_SECRET_KEY: ${LYRA_SECRET_KEY}
+      STAGING_ROOT: /staging
     # Qobuz's CDN also resolves to IPv6, but the container has no IPv6 route
     # ("Network is unreachable"); disable IPv6 so downloads use IPv4 only.
     sysctls:
       - net.ipv6.conf.all.disable_ipv6=1
     volumes:
       - ${MUSIC_DIR:-./music}:/music
+      # Per-job download staging on its own volume. Defaults to MUSIC_DIR/.staging (prior
+      # behavior); set STAGING_DIR to a fast local path when MUSIC_DIR is a network share.
+      - ${STAGING_DIR:-${MUSIC_DIR:-./music}/.staging}:/staging
     depends_on:
       db:
         condition: service_healthy

worker/lyra_worker/library.py

@@ -26,14 +26,28 @@ _COVER_STEMS = {"cover", "folder", "front"}
 _COVER_EXT = {".jpg", ".jpeg", ".png"}
 
 
-def staging_dir(dest_root: str, job_id: str) -> str:
-    """A per-job download staging directory, on the same filesystem as the library."""
-    return f"{dest_root}/.staging/{job_id}"
+def staging_dir(staging_root: str, job_id: str) -> str:
+    """A per-job download staging directory under `staging_root`. The staging root may be a
+    volume separate from the library (e.g. local disk while the library is an SMB share):
+    import_album assembles into a temp dir *on the library volume* and swaps atomically there,
+    so cross-volume staging stays safe."""
+    return f"{staging_root}/{job_id}"
 
 
-def clear_staging_root(dest_root: str) -> None:
-    """Remove every staging dir (called at worker startup to sweep crash orphans)."""
-    shutil.rmtree(f"{dest_root}/.staging", ignore_errors=True)
+def clear_staging_root(staging_root: str) -> None:
+    """Remove every per-job staging dir under `staging_root` (worker-startup crash-orphan sweep).
+    Clears the contents but not the root itself, so a mounted staging volume stays mounted."""
+    if not os.path.isdir(staging_root):
+        return
+    for name in os.listdir(staging_root):
+        path = os.path.join(staging_root, name)
+        if os.path.isdir(path):
+            shutil.rmtree(path, ignore_errors=True)
+        else:
+            try:
+                os.remove(path)
+            except OSError:
+                pass
 
 
 def _find_cover(staging: str) -> str | None:

worker/lyra_worker/main.py

@@ -1,3 +1,4 @@
+import os
 import time
 
 import psycopg
@@ -20,6 +21,10 @@ HEARTBEAT_SECONDS = 15.0  # how often to stamp worker.heartbeat (its updatedAt =
 MONITOR_TICK_SECONDS = 60.0
 DISCOVER_TICK_SECONDS = 300.0  # check the discovery interval every 5 min (interval itself is hours)
 DEST_ROOT = "/music"  # must match run_pipeline's default library root
+# Per-job download staging. Defaults inside the library (/music/.staging) to preserve prior
+# behavior; set STAGING_ROOT (mounted as a separate volume) to keep temp download I/O off a
+# network-share library.
+STAGING_ROOT = os.environ.get("STAGING_ROOT") or f"{DEST_ROOT}/.staging"
 DEFAULT_SCAN_CHUNK = 25  # albums resolved per loop iteration (~25-50s at MB ~1 req/s)
 _TRUE = {"1", "true", "yes", "on"}
 
@@ -132,7 +137,7 @@ def maybe_run_discovery(conn, sources, browser, config, dcfg, now, last_discover
 
 def run_forever() -> None:
     conn = wait_for_db()
-    clear_staging_root(DEST_ROOT)  # sweep any staging dirs orphaned by a prior crash
+    clear_staging_root(STAGING_ROOT)  # sweep any staging dirs orphaned by a prior crash
     adapters = build_adapters(get_config(conn))
     resolver = build_resolver()
     tagger = build_tagger()
@@ -184,7 +189,7 @@ def run_forever() -> None:
                 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)
+                                 staging_root=STAGING_ROOT, 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()

worker/lyra_worker/pipeline.py

@@ -142,6 +142,7 @@ def run_pipeline(
     tagger=None,
     min_confidence: float = 0.7,
     dest_root: str = "/music",
+    staging_root: str | None = None,
     upgrade_cutoff: int | None = None,
 ) -> None:
     """Real staged acquisition using source-agnostic adapters. Fakes in this plan."""
@@ -195,7 +196,7 @@ def run_pipeline(
     # 4. download (fall-through) into an isolated per-job staging dir
     _set_state(conn, job_id, "downloading", "download")
     by_source = {a.name: a for a in adapters}
-    staging = staging_dir(dest_root, job_id)
+    staging = staging_dir(staging_root or f"{dest_root}/.staging", job_id)
     winner = None
     result = None
     try:

worker/tests/test_staging.py

@@ -4,7 +4,9 @@ from lyra_worker.library import clear_staging_root, import_album, staging_dir
 
 
 def test_staging_dir_path():
-    assert staging_dir("/music", "job-1") == "/music/.staging/job-1"
+    # staging_dir now takes the staging root directly (the default root is <library>/.staging)
+    assert staging_dir("/music/.staging", "job-1") == "/music/.staging/job-1"
+    assert staging_dir("/staging", "job-1") == "/staging/job-1"  # separate-volume root
 
 
 def test_import_album_moves_audio_and_cover_into_fresh_final(tmp_path):
@@ -53,9 +55,16 @@ def test_import_album_leaves_no_temp_siblings(tmp_path):
     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)
-    (stg / "x").write_bytes(b"")
-    clear_staging_root(str(tmp_path))
-    assert not (tmp_path / ".staging").exists()
+def test_clear_staging_root_removes_jobs_but_keeps_root(tmp_path):
+    # Clears per-job dirs (and stray files) under the root, but leaves the root itself so a
+    # mounted staging volume stays mounted.
+    root = tmp_path / "staging"
+    (root / "j1").mkdir(parents=True)
+    (root / "j1" / "x").write_bytes(b"")
+    (root / "stray.txt").write_text("orphan")
+    clear_staging_root(str(root))
+    assert root.exists() and list(os.listdir(root)) == []
+
+
+def test_clear_staging_root_tolerates_missing(tmp_path):
+    clear_staging_root(str(tmp_path / "nope"))  # no-op, must not raise

worker/tests/test_staging_pipeline.py

@@ -37,6 +37,35 @@ class FlakyAdapter:
         return DownloadResult(ok=True, path=dest, track_count=3)
 
 
+class CompleteAdapter(FlakyAdapter):
+    """Always downloads the full 3-track set — for exercising a clean import path."""
+
+    def download(self, candidate, dest, on_progress):
+        os.makedirs(dest, exist_ok=True)
+        for i in range(3):
+            open(os.path.join(dest, f"{i} track.flac"), "wb").close()
+        return DownloadResult(ok=True, path=dest, track_count=3)
+
+
+def test_staging_on_a_separate_root_still_imports(conn, tmp_path):
+    # The SMB-share use case: staging on a different root than the library. Download stages
+    # under STAGING_ROOT, the album imports into the library, and the default library/.staging
+    # is never used.
+    lib = tmp_path / "library"
+    stg = tmp_path / "staging"
+    job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")
+    claim_next(conn)
+    run_pipeline(conn, job_id, [CompleteAdapter()], dest_root=str(lib), staging_root=str(stg))
+
+    with conn.cursor() as cur:
+        cur.execute('SELECT state FROM "Job" WHERE id = %s', (job_id,))
+        assert cur.fetchone()[0] == "imported"
+    assert sorted(os.listdir(lib / "Radiohead" / "In Rainbows")) == ["0 track.flac", "1 track.flac", "2 track.flac"]
+    assert stg.is_dir()                     # staging used the separate root...
+    assert not (lib / ".staging").exists()  # ...not the default library/.staging
+    assert list(stg.iterdir()) == []        # per-job staging cleaned up afterwards
+
+
 def test_partial_download_leaves_library_untouched(conn, tmp_path):
     adapter = FlakyAdapter()  # first call is partial
     job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")