diff --git a/.env.example b/.env.example index 2af306c..84f83cd 100644 --- a/.env.example +++ b/.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 diff --git a/docker-compose.yml b/docker-compose.yml index 0054b0c..fe9b2fb 100644 --- a/docker-compose.yml +++ b/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 diff --git a/worker/lyra_worker/library.py b/worker/lyra_worker/library.py index 94595da..7d288dd 100644 --- a/worker/lyra_worker/library.py +++ b/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: diff --git a/worker/lyra_worker/main.py b/worker/lyra_worker/main.py index 70a2da4..acdd8e3 100644 --- a/worker/lyra_worker/main.py +++ b/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() diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index ae7b02c..7dfbbe4 100644 --- a/worker/lyra_worker/pipeline.py +++ b/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: diff --git a/worker/tests/test_staging.py b/worker/tests/test_staging.py index a9be861..b1bb320 100644 --- a/worker/tests/test_staging.py +++ b/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 /.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 diff --git a/worker/tests/test_staging_pipeline.py b/worker/tests/test_staging_pipeline.py index 60a316d..eca7448 100644 --- a/worker/tests/test_staging_pipeline.py +++ b/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")