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) <noreply@anthropic.com>
This commit is contained in:
Jonathan
2026-07-13 15:39:38 +02:00
parent 0115b82867
commit 6f325d30b9
4 changed files with 69 additions and 2 deletions
+11 -1
View File
@@ -225,7 +225,17 @@ def run_discovery(conn: psycopg.Connection, sources: list[SimilaritySource],
followed = _followed_mbids(conn) followed = _followed_mbids(conn)
live = _healthy_sources(sources) 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: for seed_mbid, _seed_name in seeds:
affected |= _record_contributions(conn, seed_mbid, live, followed, cfg) affected |= _record_contributions(conn, seed_mbid, live, followed, cfg)
+4 -1
View File
@@ -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 except Exception as e: # a scan error must never kill the worker
print(f"worker: library scan worklist build failed: {e}", flush=True) print(f"worker: library scan worklist build failed: {e}", flush=True)
conn.rollback() conn.rollback()
_set_config(conn, "scan.id", "")
_set_config(conn, "scan.inProgress", "false") _set_config(conn, "scan.inProgress", "false")
_set_config(conn, "scan.requested", "false") _set_config(conn, "scan.requested", "false")
return 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 except Exception as e: # a scan error must never kill the worker
print(f"worker: library scan chunk failed: {e}", flush=True) print(f"worker: library scan chunk failed: {e}", flush=True)
conn.rollback() 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") _set_config(conn, "scan.requested", "false")
return return
imported_total = imported_so_far + imp imported_total = imported_so_far + imp
+34
View File
@@ -165,6 +165,40 @@ def test_seed_dropoff_lowers_seedcount(conn):
assert _artist_score(conn, "c1") == (0.5, 1) # only s2 remains 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): def test_run_discovery_processes_at_most_chunk_size_seeds(conn):
from lyra_worker.discovery import DiscoveryConfig, run_discovery from lyra_worker.discovery import DiscoveryConfig, run_discovery
from lyra_worker.adapters.fakes import FakeMbBrowser, FakeSimilaritySource from lyra_worker.adapters.fakes import FakeMbBrowser, FakeSimilaritySource
+20
View File
@@ -14,6 +14,11 @@ class FakeResolver:
return MBTarget(artist=artist, album=album, rg_mbid=f"rg-{album}", artist_mbid="a1") 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): def _set(conn, key, value):
with conn.cursor() as cur: with conn.cursor() as cur:
cur.execute( cur.execute(
@@ -78,6 +83,21 @@ def test_scan_chunks_across_iterations_and_totals_correctly(conn, tmp_path):
assert cur.fetchone()[0] == 3 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): def test_maybe_run_scan_idle_when_not_requested_or_in_progress(conn, tmp_path):
# no flags set -> no-op, no result written # no flags set -> no-op, no result written
maybe_run_scan(conn, FakeResolver(), FakeProbe(), FakeMbBrowser(), _cfg(conn), dest_root=str(tmp_path)) maybe_run_scan(conn, FakeResolver(), FakeProbe(), FakeMbBrowser(), _cfg(conn), dest_root=str(tmp_path))