From 57f97e9838ce02ec887f43ec586ea80aa7338ab8 Mon Sep 17 00:00:00 2001 From: Jonathan Date: Wed, 15 Jul 2026 19:15:43 +0200 Subject: [PATCH] feat(worker): defer Qobuz upgrades when it's gated instead of re-grabbing same quality MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With qualityCutoff=3 (want hi-res), a Soulseek-grabbed CD copy stays below cutoff and is re-attempted for upgrade. Without a guard, when Qobuz is gated by pacing those attempts would (a) re-search every owned album each monitor cycle and (b) re-download a same-quality Soulseek copy that keep-if-better discards — pure churn. Add an upgrade guard: a >= CD-lossless copy can only be improved by hi-res Qobuz, so when Qobuz is gated, skip the job WITHOUT searching (pre-search skip); and as a safety net, skip the download when the best-ranked candidate can't beat the existing copy. Force jobs always proceed. Extracted the shared keep-existing path into a helper. Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/pipeline.py | 45 ++++++++-- worker/tests/test_upgrade_guard.py | 133 +++++++++++++++++++++++++++++ 2 files changed, 171 insertions(+), 7 deletions(-) create mode 100644 worker/tests/test_upgrade_guard.py diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index 8a6363d..a62a585 100644 --- a/worker/lyra_worker/pipeline.py +++ b/worker/lyra_worker/pipeline.py @@ -29,6 +29,10 @@ _DURATION_MIN_RATIO = 0.85 # album returns ~200 Soulseek candidates), each a slow attempt that also leaves an abandoned slskd # transfer. The best candidates rank first, so 6 attempts is plenty. _MAX_DOWNLOAD_ATTEMPTS = 6 +# The best quality any non-Qobuz source offers (Soulseek FLAC = CD-lossless). A library copy at or +# above this can only be improved by hi-res Qobuz, so when Qobuz is gated by pacing an upgrade of +# such a copy is deferred without even searching. +_CD_LOSSLESS_CLASS = 2 def _measure_staged_duration_s(staging: str) -> float | None: @@ -266,6 +270,17 @@ def _library_quality(conn: psycopg.Connection, target: MBTarget) -> int | None: return row[0] if row else None +def _keep_existing_and_complete(conn: psycopg.Connection, job_id: str) -> None: + """Finish a job without importing — keep the existing library copy — and mark its request + completed. Used when a (deferred) upgrade has nothing better to offer right now; the release + stays below the cutoff and is re-attempted later.""" + _set_state(conn, job_id, "imported", "import") + with conn.cursor() as cur: + cur.execute("UPDATE \"Request\" SET status = 'completed' WHERE id = %s", + (_request_id(conn, job_id),)) + conn.commit() + + def _persist_candidates(conn: psycopg.Connection, job_id: str, candidates: list[Candidate]) -> None: with conn.cursor() as cur: for c in candidates: @@ -358,6 +373,16 @@ def run_pipeline( conn.commit() return + # Pre-search upgrade skip: a >= CD-lossless copy can only be improved by hi-res Qobuz. If Qobuz + # is gated by pacing, don't even search (no other source can beat it) — defer to when Qobuz + # frees up. This avoids re-searching every already-owned album (~2 min Soulseek search each) on + # every monitor cycle. A user-forced re-acquire always proceeds. + _existing_q = _library_quality(conn, target) + if _existing_q is not None and _existing_q >= _CD_LOSSLESS_CLASS \ + and not _is_force_job(conn, job_id) and not gate_open(conn): + _keep_existing_and_complete(conn, job_id) + return + # 2. match — search all adapters CONCURRENTLY (Soulseek's search polls up to ~2 min; run it # alongside Qobuz's instant search so total match time = the slowest source, not their sum). # Only Qobuz touches streamrip's shared asyncio loop and there's a single Qobuz adapter, so no @@ -386,6 +411,17 @@ def run_pipeline( _fail(conn, job_id, "no candidate above confidence threshold") return + # Upgrade guard: for an album already in the library, skip the download when no currently- + # available source beats the copy we have (ranked[0] is the highest-quality candidate). This + # happens when Qobuz is gated by pacing and only a same-quality Soulseek copy is left — the + # download would be discarded by keep-if-better, so defer it. The release stays below the cutoff + # and is re-attempted when a better source frees up. A user-forced re-acquire always proceeds. + existing_q = _library_quality(conn, target) + if existing_q is not None and not _is_force_job(conn, job_id) \ + and quality_class(ranked[0].quality) <= existing_q: + _keep_existing_and_complete(conn, job_id) # keep the existing copy; nothing better available + return + # 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} @@ -455,12 +491,7 @@ def run_pipeline( 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() + # an existing copy is already at >= this quality — keep it untouched, just complete + _keep_existing_and_complete(conn, job_id) finally: shutil.rmtree(staging, ignore_errors=True) # a partial/failed download never persists diff --git a/worker/tests/test_upgrade_guard.py b/worker/tests/test_upgrade_guard.py new file mode 100644 index 0000000..58c2f85 --- /dev/null +++ b/worker/tests/test_upgrade_guard.py @@ -0,0 +1,133 @@ +"""Upgrade guard: when an album is already in the library and no currently-available source can +beat it (e.g. Qobuz is gated by pacing and only a same-quality Soulseek copy is left), the pipeline +must NOT waste a download — keep-if-better would discard it anyway. It should skip and let the +release be re-attempted later (when Qobuz frees up). A genuinely better source still upgrades.""" +import os + +from lyra_worker.claim import claim_next +from lyra_worker.pipeline import run_pipeline +from lyra_worker.types import Candidate, DownloadResult, Quality + +_CD = Quality(fmt="FLAC", lossless=True, bit_depth=16, sample_rate=44100) # class 2 +_HIRES = Quality(fmt="FLAC", lossless=True, bit_depth=24, sample_rate=96000) # class 3 + + +def _cfg(conn, key, value): + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "Config"(key,value,secret,"updatedAt") VALUES (%s,%s,false,now()) ' + 'ON CONFLICT (key) DO UPDATE SET value=EXCLUDED.value, "updatedAt"=now()', + (key, value), + ) + conn.commit() + + +class _Counting: + def __init__(self, name, tier, quality): + self.name, self.tier, self._q = name, tier, quality + self.calls = 0 + self.searches = 0 + + def health(self): + return True + + def search(self, target): + self.searches += 1 + return [Candidate(source=self.name, source_ref=self.name, matched_artist=target.artist, + matched_album=target.album, quality=self._q, track_count=3, source_tier=self.tier)] + + def download(self, candidate, dest, on_progress): + self.calls += 1 + os.makedirs(dest, exist_ok=True) + for i in range(3): + with open(os.path.join(dest, f"0{i}.flac"), "wb") as fh: + fh.write(b"a") + return DownloadResult(ok=True, path=dest, track_count=3) + + +def _seed_library(conn, artist, album, quality_class, path): + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "LibraryItem" (id, artist, album, path, source, format, "qualityClass", "importedAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, %s, 'soulseek', 'FLAC', %s, now())", + (artist, album, path, quality_class), + ) + conn.commit() + + +def _insert_upgrade_job(conn, artist, album): + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "MonitoredRelease" (id, "artistMbid", "artistName", "rgMbid", album, ' + 'monitored, state, "secondaryTypes", "createdAt") ' + "VALUES (gen_random_uuid()::text, 'm', %s, 'rg', %s, true, 'grabbed', '{}', now()) RETURNING id", + (artist, album), + ) + rel_id = cur.fetchone()[0] + cur.execute( + 'INSERT INTO "Request" (id, artist, album, status, "monitoredReleaseId", "createdAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, 'pending', %s, now()) RETURNING id", + (artist, album, rel_id), + ) + req = cur.fetchone()[0] + 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 = cur.fetchone()[0] + conn.commit() + return job + + +def _lib(conn, artist, album): + with conn.cursor() as cur: + cur.execute('SELECT path, "qualityClass" FROM "LibraryItem" WHERE artist=%s AND album=%s', (artist, album)) + return cur.fetchone() + + +def _state(conn, job_id): + with conn.cursor() as cur: + cur.execute('SELECT state FROM "Job" WHERE id=%s', (job_id,)) + return cur.fetchone()[0] + + +def test_upgrade_skipped_pre_search_when_qobuz_gated(conn): + # Qobuz gated (empty active window) + existing >= CD copy: only hi-res Qobuz could improve it, + # so defer WITHOUT even searching — avoids re-searching every owned album each monitor cycle. + _cfg(conn, "qobuz.pacing.activeStartHour", "5") + _cfg(conn, "qobuz.pacing.activeEndHour", "5") # empty window → gate closed + _seed_library(conn, "A", "B", 2, "/m/existing") + job_id = _insert_upgrade_job(conn, "A", "B") + claim_next(conn) + soulseek = _Counting("soulseek", 1, _CD) + run_pipeline(conn, job_id, [soulseek], dest_root="/tmp/lib-ug", upgrade_cutoff=3) + assert soulseek.searches == 0 # no wasted search + assert soulseek.calls == 0 # no wasted download + assert _lib(conn, "A", "B") == ("/m/existing", 2) + assert _state(conn, job_id) == "imported" + + +def test_upgrade_skipped_post_search_when_only_same_quality_found(conn): + # Gate OPEN, so it searches, but the only source found is same-quality → skip the download. + _cfg(conn, "qobuz.pacing.activeStartHour", "0") + _cfg(conn, "qobuz.pacing.activeEndHour", "24") # gate open + _seed_library(conn, "A", "B", 2, "/m/existing") + job_id = _insert_upgrade_job(conn, "A", "B") + claim_next(conn) + soulseek = _Counting("soulseek", 1, _CD) + run_pipeline(conn, job_id, [soulseek], dest_root="/tmp/lib-ug", upgrade_cutoff=3) + assert soulseek.searches == 1 # it did search (gate open) + assert soulseek.calls == 0 # but skipped the useless download + assert _lib(conn, "A", "B") == ("/m/existing", 2) + + +def test_upgrade_proceeds_when_a_better_source_is_available(conn): + # Existing class-2 copy; a class-3 (hi-res) source IS available → download + upgrade. + _seed_library(conn, "A", "B", 2, "/m/existing") + job_id = _insert_upgrade_job(conn, "A", "B") + claim_next(conn) + qobuz = _Counting("qobuz", 0, _HIRES) + run_pipeline(conn, job_id, [qobuz], dest_root="/tmp/lib-ug2", upgrade_cutoff=3) + assert qobuz.calls == 1 # downloaded the better copy + assert _lib(conn, "A", "B")[1] == 3 # upgraded to hi-res