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) <noreply@anthropic.com>
This commit is contained in:
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user