From 034ae400848c887dee9d14f720a711274ed3c88a Mon Sep 17 00:00:00 2001 From: Jonathan Date: Tue, 14 Jul 2026 11:17:51 +0200 Subject: [PATCH] feat(worker): share the MusicBrainz cache from scans/discovery The worker's MB browser (discography + artist search, used by monitor sweeps, library scans, and Last.fm discovery) re-fetched every time. Route it through the same ApiCache table + logical keys the web app uses, so a discography one process fetched warms the other, and repeat lookups stop hitting MB. - apicache.py cached_api(): read-through, serve-stale-on-outage, never caches null, best-effort (a cache-DB error degrades to a live fetch, never breaks scan/sweep/discovery). Its own DEDICATED autocommit connection so it can't commit the worker's in-flight transaction, and autocommit avoids freezing Postgres now() (which would break TTL math). - CachingMbBrowser decorates MusicBrainzBrowser, serializing to the SAME camelCase JSON shape the web writes (round-trip tested). registry.build_browser wraps it when a DSN is present; the Last.fm source's browser rides it too. - Hourly prune drops ApiCache rows > 90d so the table stays bounded. - worker 213 tests / 7-skip. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 1 + worker/lyra_worker/_mbbrowser.py | 66 ++++++++++++++++++++ worker/lyra_worker/apicache.py | 82 ++++++++++++++++++++++++ worker/lyra_worker/main.py | 27 +++++++- worker/lyra_worker/registry.py | 18 +++--- worker/tests/conftest.py | 2 + worker/tests/test_apicache.py | 104 +++++++++++++++++++++++++++++++ 7 files changed, 291 insertions(+), 9 deletions(-) create mode 100644 worker/lyra_worker/apicache.py create mode 100644 worker/tests/test_apicache.py diff --git a/.gitignore b/.gitignore index f94ff6f..255a2c2 100644 --- a/.gitignore +++ b/.gitignore @@ -42,3 +42,4 @@ Thumbs.db *.swp slskd-downloads/ backups/ +.playwright-mcp/ diff --git a/worker/lyra_worker/_mbbrowser.py b/worker/lyra_worker/_mbbrowser.py index 4253e89..8d6b544 100644 --- a/worker/lyra_worker/_mbbrowser.py +++ b/worker/lyra_worker/_mbbrowser.py @@ -1,6 +1,72 @@ +from lyra_worker.apicache import DAY, MONTH, cached_api, norm_key, open_cache_conn from lyra_worker.browser import ArtistHit, ReleaseGroupInfo +# Serialize to the SAME camelCase JSON shape the web app stores (web/src/lib/musicbrainz.ts), +# so a discography one process fetched warms the cache for the other. +def _artist_hit_to_json(a: ArtistHit) -> dict: + return {"mbid": a.mbid, "name": a.name, "disambiguation": a.disambiguation} + + +def _artist_hit_from_json(d: dict) -> ArtistHit: + return ArtistHit(mbid=d["mbid"], name=d.get("name", ""), disambiguation=d.get("disambiguation", "")) + + +def _rg_to_json(r: ReleaseGroupInfo) -> dict: + return { + "rgMbid": r.rg_mbid, + "title": r.title, + "primaryType": r.primary_type or None, + "secondaryTypes": list(r.secondary_types), + "firstReleaseDate": r.first_release_date or None, + } + + +def _rg_from_json(d: dict) -> ReleaseGroupInfo: + return ReleaseGroupInfo( + rg_mbid=d["rgMbid"], + title=d.get("title", "") or "", + primary_type=d.get("primaryType") or "", + secondary_types=tuple(d.get("secondaryTypes") or ()), + first_release_date=d.get("firstReleaseDate") or "", + ) + + +class CachingMbBrowser: + """Wraps an MbBrowser with the shared ApiCache read-through, keyed on the same logical + strings the web side uses (`artist-search:{q}`, `artist-rgs:{mbid}`). Uses a dedicated, + lazily (re)opened cache connection; if it can't connect, it just calls through uncached.""" + + def __init__(self, inner, dsn: str): + self._inner = inner + self._dsn = dsn + self._conn = None + + def _cache_conn(self): + if self._conn is not None and not self._conn.closed: + return self._conn + self._conn = open_cache_conn(self._dsn) + return self._conn + + def search_artist(self, name: str) -> list[ArtistHit]: + conn = self._cache_conn() + if conn is None: + return self._inner.search_artist(name) + key = f"artist-search:{norm_key(name)}" + rows = cached_api(conn, key, DAY, + lambda: [_artist_hit_to_json(a) for a in self._inner.search_artist(name)]) + return [_artist_hit_from_json(d) for d in rows] + + def browse_release_groups(self, artist_mbid: str) -> list[ReleaseGroupInfo]: + conn = self._cache_conn() + if conn is None: + return self._inner.browse_release_groups(artist_mbid) + key = f"artist-rgs:{artist_mbid}" + rows = cached_api(conn, key, MONTH, + lambda: [_rg_to_json(r) for r in self._inner.browse_release_groups(artist_mbid)]) + return [_rg_from_json(d) for d in rows] + + class MusicBrainzBrowser: """Real MbBrowser via musicbrainzngs (imported lazily; not unit-tested offline).""" diff --git a/worker/lyra_worker/apicache.py b/worker/lyra_worker/apicache.py new file mode 100644 index 0000000..1437e8b --- /dev/null +++ b/worker/lyra_worker/apicache.py @@ -0,0 +1,82 @@ +"""Read-through cache over the shared ApiCache table (same table + logical keys the web app +uses, so web and worker share entries). TTL applied here from fetchedAt; serves a stale row +if the fresh fetch fails; never caches None. + +Best-effort: all DB access is guarded so a cache-layer problem NEVER breaks the caller +(scan/sweep/discovery) — on a DB error it just degrades to a live fetch. It is given a +DEDICATED connection (not the worker's main conn) so its commits can't commit the worker's +in-flight transaction.""" +from typing import Callable, TypeVar + +import psycopg +from psycopg.types.json import Jsonb + +T = TypeVar("T") + +DAY = 86_400 +WEEK = 7 * 86_400 +MONTH = 30 * 86_400 + + +def _read(conn, key: str): + try: + with conn.cursor() as cur: + cur.execute( + 'SELECT json, extract(epoch from (now() - "fetchedAt")) FROM "ApiCache" WHERE key = %s', + (key,), + ) + return cur.fetchone() + except Exception: + try: + conn.rollback() + except Exception: + pass + return None + + +def _write(conn, key: str, value) -> None: + try: + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "ApiCache" (key, json, "fetchedAt") VALUES (%s, %s, now()) ' + 'ON CONFLICT (key) DO UPDATE SET json = EXCLUDED.json, "fetchedAt" = now()', + (key, Jsonb(value)), + ) + conn.commit() + except Exception: + try: + conn.rollback() + except Exception: + pass + + +def cached_api(conn, key: str, ttl_sec: int, fetch_fresh: Callable[[], T]) -> T: + row = _read(conn, key) + if row is not None and row[1] < ttl_sec: + return row[0] + try: + fresh = fetch_fresh() + except Exception: + if row is not None: + return row[0] # serve stale on an upstream outage + raise + if fresh is not None: + _write(conn, key, fresh) + return fresh + + +def norm_key(s: str) -> str: + """Match the web side's normKey (trim + lowercase) so search keys don't fragment.""" + return s.strip().lower() + + +def open_cache_conn(dsn: str): + """A dedicated AUTOCOMMIT connection for cache I/O, or None if it can't be opened (the + cache then simply no-ops). Kept separate from the worker's main connection. Autocommit is + essential: a read that hits the TTL returns without committing, and a non-autocommit conn + would leave that transaction open — freezing Postgres `now()` at the transaction start + (now() = transaction timestamp), which would break every subsequent TTL comparison.""" + try: + return psycopg.connect(dsn, autocommit=True) + except Exception: + return None diff --git a/worker/lyra_worker/main.py b/worker/lyra_worker/main.py index 0ca0d5d..ca261e1 100644 --- a/worker/lyra_worker/main.py +++ b/worker/lyra_worker/main.py @@ -23,6 +23,7 @@ IDLE_SLEEP = 2.0 HEARTBEAT_SECONDS = 15.0 # how often to stamp worker.heartbeat (its updatedAt = liveness) MONITOR_TICK_SECONDS = 60.0 DISCOVER_TICK_SECONDS = 300.0 # check the discovery interval every 5 min (interval itself is hours) +CACHE_PRUNE_SECONDS = 3600.0 # prune the shared ApiCache at most hourly (drops rows > 90d old) DEST_ROOT = "/music" # must match run_pipeline's default library root # Per-job download staging. Defaults inside the library (/music/.staging) to preserve prior # behavior; set STAGING_ROOT (mounted as a separate volume) to keep temp download I/O off a @@ -167,6 +168,23 @@ def maybe_run_discovery(conn, sources, browser, config, dcfg, now, last_discover return now if due else last_discover_tick +def prune_api_cache(conn) -> None: + """Drop ApiCache rows older than 90 days so the shared cache stays bounded. Best-effort: + a failure here (e.g. a transient error) must never take down the worker loop.""" + try: + with conn.cursor() as cur: + cur.execute('DELETE FROM "ApiCache" WHERE "fetchedAt" < now() - interval \'90 days\'') + conn.commit() + except psycopg.OperationalError: + raise # a real disconnect — let the loop's reconnect handler deal with it + except Exception as e: + print(f"worker: ApiCache prune failed: {e}", flush=True) + try: + conn.rollback() + except Exception: + pass + + def _require_secret_key() -> None: """Fail fast (or warn loudly) on a missing/placeholder/public LYRA_SECRET_KEY before the worker touches any encrypted credential.""" @@ -192,9 +210,10 @@ def run_forever() -> None: adapters = build_adapters(get_config(conn)) resolver = build_resolver() tagger = build_tagger() - browser = build_browser() + _dsn = os.environ.get("DATABASE_URL") + browser = build_browser(_dsn) # shared ApiCache read-through when a DSN is present probe = build_probe() - similarity_sources = build_similarity_sources(get_config(conn)) + similarity_sources = build_similarity_sources(get_config(conn), _dsn) # Backfill: releases pressed while the monitor was off never got reconciled, so they linger # on the wanted list. Fix any already-owned ones once at startup. fixed = fulfill_owned_releases(conn, MonitorConfig.from_config(get_config(conn))) @@ -205,6 +224,7 @@ def run_forever() -> None: last_tick = 0.0 last_discover_tick = 0.0 last_heartbeat = 0.0 + last_prune = 0.0 try: while True: # A transient DB connection loss (e.g. AdminShutdown when Postgres restarts) raises @@ -218,6 +238,9 @@ def run_forever() -> None: if now - last_heartbeat >= HEARTBEAT_SECONDS: _set_config(conn, "worker.heartbeat", "1") # value irrelevant; updatedAt is the signal last_heartbeat = now + if now - last_prune >= CACHE_PRUNE_SECONDS: + prune_api_cache(conn) # keep the shared ApiCache bounded + last_prune = now if mcfg.enabled and now - last_tick >= MONITOR_TICK_SECONDS: try: sweep(conn, browser, mcfg) diff --git a/worker/lyra_worker/registry.py b/worker/lyra_worker/registry.py index 352a983..6a1f3cc 100644 --- a/worker/lyra_worker/registry.py +++ b/worker/lyra_worker/registry.py @@ -1,4 +1,4 @@ -from lyra_worker._mbbrowser import MusicBrainzBrowser +from lyra_worker._mbbrowser import CachingMbBrowser, MusicBrainzBrowser from lyra_worker._musicbrainz import MusicBrainzResolver from lyra_worker._mutagen import MutagenTagger from lyra_worker._probe import MutagenProbe @@ -42,9 +42,12 @@ def build_tagger() -> Tagger: return MutagenTagger() -def build_browser() -> MbBrowser: - """The MusicBrainz browser used by the background monitor.""" - return MusicBrainzBrowser() +def build_browser(dsn: str | None = None) -> MbBrowser: + """The MusicBrainz browser used by the background monitor. When a DSN is given, wrap it + in the shared ApiCache read-through so scans/discovery stop re-fetching discographies + (and warm the same cache the web modals read).""" + inner = MusicBrainzBrowser() + return CachingMbBrowser(inner, dsn) if dsn else inner def build_probe() -> AudioProbe: @@ -52,14 +55,15 @@ def build_probe() -> AudioProbe: return MutagenProbe() -def build_similarity_sources(config: dict) -> list[SimilaritySource]: +def build_similarity_sources(config: dict, dsn: str | None = None) -> list[SimilaritySource]: """Similarity sources for discovery. ListenBrainz needs no credentials, so it is always constructed; per-run reachability is checked via each source's health(). - Last.fm is added only when an API key is configured.""" + Last.fm is added only when an API key is configured. Its MB name→MBID lookups ride the + shared ApiCache when a DSN is given.""" sources: list[SimilaritySource] = [ ListenBrainzSource(base_url=config.get("discover.listenBrainzUrl") or "") ] key = config.get("lastfm.api_key") if key: - sources.append(LastfmSource(api_key=key, browser=MusicBrainzBrowser())) + sources.append(LastfmSource(api_key=key, browser=build_browser(dsn))) return sources diff --git a/worker/tests/conftest.py b/worker/tests/conftest.py index 68e6b98..75daeba 100644 --- a/worker/tests/conftest.py +++ b/worker/tests/conftest.py @@ -30,6 +30,7 @@ def conn(): cur.execute('DELETE FROM "DiscoverySuggestion"') cur.execute('DELETE FROM "DiscoverySeedContribution"') cur.execute('DELETE FROM "ScanWorkItem"') + cur.execute('DELETE FROM "ApiCache"') cur.execute('DELETE FROM "Config"') connection.commit() yield connection @@ -47,6 +48,7 @@ def conn(): cur.execute('DELETE FROM "DiscoverySuggestion"') cur.execute('DELETE FROM "DiscoverySeedContribution"') cur.execute('DELETE FROM "ScanWorkItem"') + cur.execute('DELETE FROM "ApiCache"') cur.execute('DELETE FROM "Config"') connection.commit() connection.close() diff --git a/worker/tests/test_apicache.py b/worker/tests/test_apicache.py new file mode 100644 index 0000000..8507fae --- /dev/null +++ b/worker/tests/test_apicache.py @@ -0,0 +1,104 @@ +"""Read-through + shared-shape tests for the worker cache. cached_api uses the real +lyra_test DB (via the conn fixture); MB is never hit (fetch_fresh is a fake).""" +from lyra_worker.apicache import cached_api, norm_key +from lyra_worker._mbbrowser import ( + _artist_hit_from_json, _artist_hit_to_json, _rg_from_json, _rg_to_json, +) +from lyra_worker.browser import ArtistHit, ReleaseGroupInfo + + +def _backdate(conn, key, seconds): + with conn.cursor() as cur: + cur.execute( + 'UPDATE "ApiCache" SET "fetchedAt" = now() - make_interval(secs => %s) WHERE key = %s', + (seconds, key), + ) + conn.commit() + + +def test_read_through_fetches_once(conn): + calls = [] + + def fresh(): + calls.append(1) + return {"v": 1} + + assert cached_api(conn, "k1", 3600, fresh) == {"v": 1} + assert cached_api(conn, "k1", 3600, fresh) == {"v": 1} + assert len(calls) == 1 # second served from cache + + +def test_refetches_after_ttl(conn): + calls = [] + + def fresh(): + calls.append(1) + return {"v": len(calls)} + + cached_api(conn, "k2", 3600, fresh) + _backdate(conn, "k2", 7200) # older than the ttl + cached_api(conn, "k2", 3600, fresh) + assert len(calls) == 2 + + +def test_serves_stale_on_fetch_error(conn): + cached_api(conn, "k3", 3600, lambda: {"v": "old"}) + _backdate(conn, "k3", 7200) + + def boom(): + raise RuntimeError("mb down") + + assert cached_api(conn, "k3", 3600, boom) == {"v": "old"} + + +def test_reraises_when_no_cached_row(conn): + def boom(): + raise RuntimeError("mb down") + + try: + cached_api(conn, "k4-missing", 3600, boom) + except RuntimeError as e: + assert "mb down" in str(e) + else: + raise AssertionError("expected the fetch error to propagate") + + +def test_does_not_cache_none(conn): + calls = [] + + def fresh(): + calls.append(1) + return None + + cached_api(conn, "k5", 3600, fresh) + cached_api(conn, "k5", 3600, fresh) + assert len(calls) == 2 # None never cached + with conn.cursor() as cur: + cur.execute('SELECT 1 FROM "ApiCache" WHERE key = %s', ("k5",)) + assert cur.fetchone() is None + + +def test_norm_key(): + assert norm_key(" Radiohead ") == "radiohead" + + +def test_release_group_json_roundtrip(): + r = ReleaseGroupInfo(rg_mbid="rg1", title="OK Computer", primary_type="Album", + secondary_types=("Live",), first_release_date="1997-05-21") + assert _rg_from_json(_rg_to_json(r)) == r + # a bare group (no type/date) round-trips to the worker's empty-string defaults + bare = ReleaseGroupInfo(rg_mbid="rg2", title="X") + assert _rg_from_json(_rg_to_json(bare)) == bare + + +def test_artist_hit_json_roundtrip(): + a = ArtistHit(mbid="a1", name="Radiohead", disambiguation="UK band") + assert _artist_hit_from_json(_artist_hit_to_json(a)) == a + + +def test_worker_shape_matches_web_camelcase(): + # the stored dict MUST use the same camelCase keys the web app writes/reads + d = _rg_to_json(ReleaseGroupInfo(rg_mbid="rg", title="T", primary_type="Album", + secondary_types=(), first_release_date="2000")) + assert set(d) == {"rgMbid", "title", "primaryType", "secondaryTypes", "firstReleaseDate"} + assert set(_artist_hit_to_json(ArtistHit(mbid="a", name="N"))) == {"mbid", "name", "disambiguation"}