From d47ed12e8ec4849840e093bdcbee54d7a9de1f9d Mon Sep 17 00:00:00 2001 From: Jonathan Date: Wed, 15 Jul 2026 13:41:04 +0200 Subject: [PATCH] feat(worker): fall through to the next source on an incomplete download MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completeness/duration verification was applied only to the first candidate that downloaded successfully; if that copy was incomplete or too short the whole job failed, even when a later-ranked source had a complete copy (e.g. John Mayer 'The Search for Everything' — Qobuz reliably fails one hi-res track, so it never fell back to the deluxe edition or Soulseek). Extract the checks into a pure _download_problem() and run it INSIDE the download loop: a candidate that downloads but fails verification now falls through to the next-ranked source, and the job only fails (needs_attention) when every candidate is incomplete — reporting the last shortfall reason. Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/pipeline.py | 88 +++++++++++--------- worker/tests/test_incomplete_fallthrough.py | 91 +++++++++++++++++++++ 2 files changed, 139 insertions(+), 40 deletions(-) create mode 100644 worker/tests/test_incomplete_fallthrough.py diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index 91fe58f..a1d0532 100644 --- a/worker/lyra_worker/pipeline.py +++ b/worker/lyra_worker/pipeline.py @@ -70,6 +70,40 @@ def _count_staged_audio(staging: str) -> int: return n +def _download_problem(result, staging: str, target, winner, measure_duration) -> str | None: + """Why a completed download is unusable — ``"incomplete download"`` / ``"download too short + (...)"`` — or None if it's good. Completeness is judged against the CHOSEN SOURCE's own track + count (``winner.track_count``), NOT MusicBrainz's canonical release: MB picks an arbitrary + releases[0] that is often a deluxe/expanded edition with more tracks (and runtime) than the + standard album a source legitimately delivers, so gating on ``target.track_count`` falsely + rejects complete standard-edition downloads (The Script "No Sound Without Silence" 11 vs MB 12, + Skillet "Unleashed" 12 vs MB's 20-track deluxe). Pure — no DB I/O — so the download loop can + call it per candidate and fall through to the next source on a problem.""" + expected = winner.track_count + # (a) Truncated: fewer files than the source promised (a track failed mid-download). Count the + # NON-EMPTY audio on disk too: a silently-skipped track leaves a 0-byte placeholder the adapter + # still counts. (`staged and ...` keeps fake adapters that stage nothing on the count path.) + staged = _count_staged_audio(staging) + truncated = result.track_count < expected or bool(staged and staged < expected) + # (b) Implausibly small vs MB's album — a lone-track "full album" video / wrong match, as + # distinct from a legitimately smaller edition. A real edition keeps more than half MB's tracks. + too_small = target.track_count is not None and expected * 2 <= target.track_count + if truncated or too_small: + return "incomplete download" + # Duration: a truncated file or short preview substituted for a real track shows up as playtime + # well under the expected total. Scale MB's total to the DELIVERED edition's size so a smaller + # edition isn't judged against a larger one's runtime. Only when the total is known (measured is + # None for the offline fakes). + if target.total_duration_s: + expected_total = target.total_duration_s + if target.track_count: + expected_total *= winner.track_count / target.track_count + measured = measure_duration(staging) + if measured is not None and measured < expected_total * _DURATION_MIN_RATIO: + return f"download too short ({measured / 60:.0f} of ~{expected_total / 60:.0f} min)" + return None + + def _poll_download_progress(job_id: str, staging: str, expected: int, stop: "threading.Event", reports: "threading.Event") -> None: """Until `stop`, periodically write Job.downloadProgress = files-in-staging / expected @@ -343,7 +377,7 @@ def run_pipeline( by_source = {a.name: a for a in adapters} staging = staging_dir(staging_root or f"{dest_root}/.staging", job_id) winner = None - result = None + last_problem = None # why the most recent downloaded-but-rejected candidate was unusable _set_download_progress(conn, job_id, 0.0) # reset for this run expected = target.track_count or (ranked[0].track_count if ranked else 0) _stop = threading.Event() @@ -362,54 +396,28 @@ def run_pipeline( shutil.rmtree(staging, ignore_errors=True) # clean slate per attempt _reports.clear() # this attempt hasn't reported yet → poller estimates until it does result = adapter.download(candidate, staging, _make_on_progress(conn, job_id, _reports)) - if result.ok: + if not result.ok: + continue + # Verify completeness on the staging copy. A download that lands incomplete or too + # short falls through to the NEXT-ranked source (e.g. Qobuz repeatedly fails one + # track → try the deluxe edition or Soulseek) rather than failing the whole job on + # the first source's shortfall. + last_problem = _download_problem(result, staging, target, candidate, measure_duration) + if last_problem is None: winner = candidate break finally: _stop.set() _poller.join(timeout=3) - if winner is None or result is None or not result.ok: - _fail(conn, job_id, "all downloads failed") + if winner is None: + # No source produced a complete album: surface why the last one was rejected (incomplete + # / too short), or "all downloads failed" if none even downloaded. + _fail(conn, job_id, last_problem or "all downloads failed") return _set_download_progress(conn, job_id, 1.0) # download done — UI shows 100% into Finishing - # 5. verify completeness on the staging copy, then promote + tag + # 5. promote + tag the verified winner _set_state(conn, job_id, "tagging", "tag") - # Judge completeness against what the CHOSEN SOURCE offered, NOT MusicBrainz's canonical - # release: MB picks an arbitrary releases[0] that is often a deluxe/expanded edition with - # more tracks than the standard album a source legitimately delivers, so gating purely on - # target.track_count falsely rejects complete standard-edition downloads (e.g. The Script - # "No Sound Without Silence" 11 vs MB 12, Skillet "Unleashed" 12 vs MB's 20-track deluxe). - expected = winner.track_count - # (a) Truncated: fewer files than the source promised (a track failed mid-download). - # Count the NON-EMPTY audio actually on disk too: a silently-skipped track leaves a 0-byte - # placeholder the adapter still counts, so trust the files when any are staged. (`staged - # and ...` keeps fake adapters that stage nothing on the count path.) - staged = _count_staged_audio(staging) - truncated = result.track_count < expected or bool(staged and staged < expected) - # (b) Implausibly small vs MB's album — a lone-track "full album" video or a wrong match, - # as distinct from a legitimately smaller edition (standard vs deluxe). A real edition - # keeps more than half of MB's tracks; a 1-of-12 match does not. - too_small = target.track_count is not None and expected * 2 <= target.track_count - if truncated or too_small: - _fail(conn, job_id, "incomplete download") - return - - # Duration check: even with the right track count, a truncated file or a short preview - # substituted for a real track shows up as a total playtime well under the expected total. - # Scale MB's total to the DELIVERED edition's size (source vs MB track count) so a smaller - # standard edition isn't measured against a larger deluxe edition's runtime. Only applies - # when the total is known (measured is None for the offline fakes). - if target.total_duration_s: - expected_total = target.total_duration_s - if target.track_count: - expected_total *= winner.track_count / target.track_count - measured = measure_duration(staging) - if measured is not None and measured < expected_total * _DURATION_MIN_RATIO: - _fail(conn, job_id, - f"download too short ({measured / 60:.0f} of ~{expected_total / 60:.0f} min)") - return - final = album_dir(dest_root, target) existing_q = _library_quality(conn, target) new_q = quality_class(winner.quality) diff --git a/worker/tests/test_incomplete_fallthrough.py b/worker/tests/test_incomplete_fallthrough.py new file mode 100644 index 0000000..aea3523 --- /dev/null +++ b/worker/tests/test_incomplete_fallthrough.py @@ -0,0 +1,91 @@ +"""When the top-ranked source downloads but is incomplete (e.g. Qobuz consistently fails one +track), the pipeline must fall through to the next-ranked source instead of failing the whole +job — only giving up (needs_attention) when every candidate is incomplete.""" +import os + +from lyra_worker.claim import claim_next +from lyra_worker.pipeline import run_pipeline +from lyra_worker.types import Candidate, DownloadResult, Quality +from tests.conftest import insert_request + +_HIRES = Quality(fmt="FLAC", lossless=True, bit_depth=24, sample_rate=96000) +_CD = Quality(fmt="FLAC", lossless=True, bit_depth=16, sample_rate=44100) + + +def _stage(dest, n): + os.makedirs(dest, exist_ok=True) + for i in range(n): + with open(os.path.join(dest, f"{i + 1:02d}.flac"), "wb") as fh: + fh.write(b"audio") + + +class _Qobuz: + """Higher-quality (ranks first) Qobuz that promises 3 tracks but delivers `deliver`.""" + name = "qobuz" + tier = 0 + + def __init__(self, deliver): + self._deliver = deliver + + def health(self): + return True + + def search(self, target): + return [Candidate(source="qobuz", source_ref="q", matched_artist=target.artist, + matched_album=target.album, quality=_HIRES, track_count=3, source_tier=0)] + + def download(self, candidate, dest, on_progress): + _stage(dest, self._deliver) + return DownloadResult(ok=True, path=dest, track_count=self._deliver) + + +class _Soulseek: + """Lower-quality (ranks second) Soulseek that promises 3 tracks and delivers `deliver`.""" + name = "soulseek" + tier = 1 + + def __init__(self, deliver): + self._deliver = deliver + + def health(self): + return True + + def search(self, target): + return [Candidate(source="soulseek", source_ref="s", matched_artist=target.artist, + matched_album=target.album, quality=_CD, track_count=3, source_tier=1)] + + def download(self, candidate, dest, on_progress): + _stage(dest, self._deliver) + return DownloadResult(ok=True, path=dest, track_count=self._deliver) + + +def _state_error(conn, job_id): + with conn.cursor() as cur: + cur.execute('SELECT state, error FROM "Job" WHERE id = %s', (job_id,)) + return cur.fetchone() + + +def _chosen_source(conn, job_id): + with conn.cursor() as cur: + cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,)) + row = cur.fetchone() + return row[0] if row else None + + +def test_incomplete_top_source_falls_through_to_complete_next(conn, tmp_path): + # Qobuz (ranks first on quality) delivers only 2 of its promised 3 tracks -> incomplete. + # The pipeline must try the next source (Soulseek, complete) rather than failing the job. + job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") + claim_next(conn) + run_pipeline(conn, job_id, [_Qobuz(deliver=2), _Soulseek(deliver=3)], dest_root=str(tmp_path)) + assert _state_error(conn, job_id) == ("imported", None) + assert _chosen_source(conn, job_id) == "soulseek" + + +def test_all_sources_incomplete_still_needs_attention(conn, tmp_path): + # Every candidate is incomplete -> no fallback succeeds -> needs_attention (incomplete). + job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") + claim_next(conn) + run_pipeline(conn, job_id, [_Qobuz(deliver=2), _Soulseek(deliver=1)], dest_root=str(tmp_path)) + assert _state_error(conn, job_id) == ("needs_attention", "incomplete download") + assert not (tmp_path / "Radiohead" / "In Rainbows").exists() # nothing imported