diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index 17c4bba..adeb2fa 100644 --- a/worker/lyra_worker/pipeline.py +++ b/worker/lyra_worker/pipeline.py @@ -126,7 +126,12 @@ def run_pipeline( _set_state(conn, job_id, "matching", "match") found: list[Candidate] = [] for adapter in adapters: - for c in adapter.search(target): + try: + results = adapter.search(target) + except Exception as e: # a down source contributes no candidates, never crashes the job + print(f"pipeline: adapter {adapter.name} search failed: {e}", flush=True) + continue + for c in results: found.append(replace(c, source_tier=adapter.tier, confidence=score_confidence(target, c))) _persist_candidates(conn, job_id, found) diff --git a/worker/tests/test_pipeline_hardening.py b/worker/tests/test_pipeline_hardening.py new file mode 100644 index 0000000..195f563 --- /dev/null +++ b/worker/tests/test_pipeline_hardening.py @@ -0,0 +1,35 @@ +from lyra_worker.adapters.youtube import YouTubeAdapter +from lyra_worker.claim import claim_next +from lyra_worker.pipeline import run_pipeline +from tests.conftest import insert_request +from tests.test_youtube_adapter import FakeYtClient + + +class ExplodingAdapter: + name = "boom" + tier = 0 + + def health(self): + return True + + def search(self, target): + raise RuntimeError("search backend down") + + def download(self, candidate, dest, on_progress): + raise AssertionError("should never be called") + + +def test_one_failing_source_does_not_crash_the_job(conn): + job_id = insert_request(conn, artist="John Mayer", album="Continuum") + claim_next(conn) + # track_count must be 10 to match FakeYtClient.download's hardcoded return value, + # otherwise the pipeline's tag-stage integrity check fails independently of hardening. + yt = [{"source_ref": "yt1", "title": "Continuum", "artist": "John Mayer", "track_count": 10}] + # The exploding adapter must not abort the pipeline; YouTube should still win. + run_pipeline(conn, job_id, [ExplodingAdapter(), YouTubeAdapter(FakeYtClient(results=yt))], dest_root="/tmp/lib") + + with conn.cursor() as cur: + cur.execute('SELECT state FROM "Job" WHERE id = %s', (job_id,)) + assert cur.fetchone()[0] == "imported" + cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,)) + assert cur.fetchone()[0] == "youtube"