From 6f325d30b9e96b5af09e67039a2c4d7ff47e2922 Mon Sep 17 00:00:00 2001 From: Jonathan Date: Mon, 13 Jul 2026 15:39:38 +0200 Subject: [PATCH] fix(worker): evict removed-seed contributions + clear scan worklist on abandon Two row-leak bugs found in whole-branch review: - discovery.run_discovery now prunes DiscoverySeedContribution rows whose seedMbid is no longer a WatchedArtist (unfollow/removal has no FK cascade) and recomputes the touched candidates' scores this sweep, instead of letting a dead seed's contribution inflate the aggregate forever. - maybe_run_scan's exception handler around scan_chunk now calls clear_worklist and clears scan.id, matching the drain path. Previously an exception mid-chunk left the already-done rows plus the remaining worklist for that scan_id orphaned forever, since the next scan mints a fresh uuid. Also clear scan.id in the build_worklist failure handler for consistency. Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/discovery.py | 12 ++++++++++- worker/lyra_worker/main.py | 5 ++++- worker/tests/test_discovery.py | 34 +++++++++++++++++++++++++++++++ worker/tests/test_scan_trigger.py | 20 ++++++++++++++++++ 4 files changed, 69 insertions(+), 2 deletions(-) diff --git a/worker/lyra_worker/discovery.py b/worker/lyra_worker/discovery.py index 62d4a4b..53fcb50 100644 --- a/worker/lyra_worker/discovery.py +++ b/worker/lyra_worker/discovery.py @@ -225,7 +225,17 @@ def run_discovery(conn: psycopg.Connection, sources: list[SimilaritySource], followed = _followed_mbids(conn) live = _healthy_sources(sources) - affected: set[str] = set() + # Evict contributions from seeds no longer followed — unfollow/removal has no + # FK cascade, so without this a removed seed's score would linger and inflate + # the aggregate forever. Seed `affected` with the touched candidates so their + # suggestion scores are recomputed from the remaining live seeds this sweep. + with conn.cursor() as cur: + cur.execute('SELECT DISTINCT "candidateMbid" FROM "DiscoverySeedContribution" ' + 'WHERE "seedMbid" NOT IN (SELECT mbid FROM "WatchedArtist")') + affected = {r[0] for r in cur.fetchall()} + cur.execute('DELETE FROM "DiscoverySeedContribution" ' + 'WHERE "seedMbid" NOT IN (SELECT mbid FROM "WatchedArtist")') + for seed_mbid, _seed_name in seeds: affected |= _record_contributions(conn, seed_mbid, live, followed, cfg) diff --git a/worker/lyra_worker/main.py b/worker/lyra_worker/main.py index 69ed3f1..819f258 100644 --- a/worker/lyra_worker/main.py +++ b/worker/lyra_worker/main.py @@ -76,6 +76,7 @@ def maybe_run_scan(conn, resolver, probe, browser, config, dest_root: str = DEST except Exception as e: # a scan error must never kill the worker print(f"worker: library scan worklist build failed: {e}", flush=True) conn.rollback() + _set_config(conn, "scan.id", "") _set_config(conn, "scan.inProgress", "false") _set_config(conn, "scan.requested", "false") return @@ -91,7 +92,9 @@ def maybe_run_scan(conn, resolver, probe, browser, config, dest_root: str = DEST except Exception as e: # a scan error must never kill the worker print(f"worker: library scan chunk failed: {e}", flush=True) conn.rollback() - _set_config(conn, "scan.inProgress", "false") + clear_worklist(conn, scan_id) + _set_config(conn, "scan.id", "") + _set_config(conn, "scan.inProgress", "false") # abandon the scan; a new request restarts it _set_config(conn, "scan.requested", "false") return imported_total = imported_so_far + imp diff --git a/worker/tests/test_discovery.py b/worker/tests/test_discovery.py index 5425d84..82b36f9 100644 --- a/worker/tests/test_discovery.py +++ b/worker/tests/test_discovery.py @@ -165,6 +165,40 @@ def test_seed_dropoff_lowers_seedcount(conn): assert _artist_score(conn, "c1") == (0.5, 1) # only s2 remains +def test_evicts_contributions_from_removed_seed(conn): + # Two live seeds contribute to c1 -> 1.1/2. Then s1 is unfollowed (deleted from + # WatchedArtist, simulating DELETE /api/artists/[id]) without any FK cascade on + # DiscoverySeedContribution. The next sweep must evict s1's orphaned contribution + # rather than leaving it to inflate c1's score forever. + insert_watched_artist(conn, mbid="s1", name="Seed One", + last_discovered_sql="now() - interval '200 hours'") + insert_watched_artist(conn, mbid="s2", name="Seed Two", + last_discovered_sql="now() - interval '200 hours'") + src = FakeSimilaritySource(similar={ + "s1": [SimilarArtist("c1", "Shared", 0.6)], + "s2": [SimilarArtist("c1", "Shared", 0.5)], + }) + run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig()) + assert _artist_score(conn, "c1") == (1.1, 2) + + # Unfollow s1 (no FK cascade on DiscoverySeedContribution). + with conn.cursor() as cur: + cur.execute('DELETE FROM "WatchedArtist" WHERE mbid = %s', ("s1",)) + conn.commit() + + # Make s2 due again and sweep; s1 can no longer be a seed since it's gone. + with conn.cursor() as cur: + cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() - interval \'200 hours\' WHERE mbid = %s', ("s2",)) + conn.commit() + src2 = FakeSimilaritySource(similar={"s2": [SimilarArtist("c1", "Shared", 0.5)]}) + run_discovery(conn, [src2], FakeMbBrowser(), DiscoveryConfig()) + + assert _artist_score(conn, "c1") == (0.5, 1) # s1's stale 0.6 evicted, not summed + with conn.cursor() as cur: + cur.execute('SELECT count(*) FROM "DiscoverySeedContribution" WHERE "seedMbid" = %s', ("s1",)) + assert cur.fetchone()[0] == 0 + + def test_run_discovery_processes_at_most_chunk_size_seeds(conn): from lyra_worker.discovery import DiscoveryConfig, run_discovery from lyra_worker.adapters.fakes import FakeMbBrowser, FakeSimilaritySource diff --git a/worker/tests/test_scan_trigger.py b/worker/tests/test_scan_trigger.py index e13361c..99623e4 100644 --- a/worker/tests/test_scan_trigger.py +++ b/worker/tests/test_scan_trigger.py @@ -14,6 +14,11 @@ class FakeResolver: return MBTarget(artist=artist, album=album, rg_mbid=f"rg-{album}", artist_mbid="a1") +class RaisingProbe: + def probe(self, path): + raise RuntimeError("probe boom") + + def _set(conn, key, value): with conn.cursor() as cur: cur.execute( @@ -78,6 +83,21 @@ def test_scan_chunks_across_iterations_and_totals_correctly(conn, tmp_path): assert cur.fetchone()[0] == 3 +def test_scan_chunk_exception_clears_worklist_not_just_flags(conn, tmp_path): + # scan_chunk commits per item, so if it throws mid-chunk the except block must + # still clear the worklist — otherwise the done + not-yet-processed ScanWorkItem + # rows for this scan_id are orphaned forever (the next scan mints a fresh uuid). + _album(tmp_path, "Artist", "Album") + _set(conn, "scan.requested", "true") + + maybe_run_scan(conn, FakeResolver(), RaisingProbe(), FakeMbBrowser(), _cfg(conn), dest_root=str(tmp_path)) + + assert _get(conn, "scan.inProgress") == "false" + with conn.cursor() as cur: + cur.execute('SELECT count(*) FROM "ScanWorkItem"') + assert cur.fetchone()[0] == 0 + + def test_maybe_run_scan_idle_when_not_requested_or_in_progress(conn, tmp_path): # no flags set -> no-op, no result written maybe_run_scan(conn, FakeResolver(), FakeProbe(), FakeMbBrowser(), _cfg(conn), dest_root=str(tmp_path))