From 082365005c01d424518e5f967845f5855d1fa6ee Mon Sep 17 00:00:00 2001 From: Jonathan Date: Sun, 12 Jul 2026 15:08:24 +0200 Subject: [PATCH] fix(worker): reconcile every job + backfill owned releases Wanted releases pressed while the monitor is off never got reconciled, so they lingered on the wanted list. Extract finish_job() and call it for every finished job (not just when monitor.enabled) so the MonitoredRelease transitions to fulfilled. Add fulfill_owned_releases(), run once at startup, to backfill releases already owned at/above the quality cutoff (self-heals items stuck from before this fix). Both idempotent + tested. Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/main.py | 25 ++++++++++++----- worker/lyra_worker/monitor.py | 20 +++++++++++++ worker/tests/test_monitor_reconcile.py | 39 +++++++++++++++++++++++++- 3 files changed, 76 insertions(+), 8 deletions(-) diff --git a/worker/lyra_worker/main.py b/worker/lyra_worker/main.py index 64b45c8..a4df1d4 100644 --- a/worker/lyra_worker/main.py +++ b/worker/lyra_worker/main.py @@ -7,7 +7,7 @@ from lyra_worker.config import get_config from lyra_worker.db import wait_for_db from lyra_worker.discovery import DiscoveryConfig, run_discovery from lyra_worker.library import clear_staging_root -from lyra_worker.monitor import MonitorConfig, reconcile, sweep +from lyra_worker.monitor import MonitorConfig, fulfill_owned_releases, reconcile, sweep from lyra_worker.pipeline import run_pipeline from lyra_worker.registry import ( build_adapters, build_browser, build_probe, build_resolver, @@ -98,6 +98,17 @@ def _run_discovery(conn, sources, browser, config=None) -> None: print(f"worker: discovery done — {result.artists} artists, {result.albums} albums", flush=True) +def finish_job(conn, job_id, mcfg) -> None: + """Feed a finished job's outcome back into its MonitoredRelease. Runs for EVERY job, + monitor enabled or not: a manually-wanted release must still transition to fulfilled so + it drops off the Wanted list. A no-op for plain (unlinked) requests.""" + try: + reconcile(conn, job_id, mcfg) + except Exception as e: # reconcile must never take down the worker loop + print(f"worker: reconcile failed for job {job_id}: {e}", flush=True) + conn.rollback() + + def maybe_run_discovery(conn, sources, browser, config, dcfg, now, last_discover_tick, tick_seconds: float = DISCOVER_TICK_SECONDS) -> float: """Run discovery at most once per loop iteration, whether triggered by the schedule @@ -127,6 +138,11 @@ def run_forever() -> None: browser = build_browser() probe = build_probe() similarity_sources = build_similarity_sources(get_config(conn)) + # 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))) + if fixed: + print(f"worker: reconciled {fixed} already-owned release(s) to fulfilled", flush=True) print(f"worker: {len(adapters)} adapter(s) enabled: {[a.name for a in adapters]}", flush=True) print("worker: waiting for jobs", flush=True) last_tick = 0.0 @@ -167,12 +183,7 @@ def run_forever() -> None: except Exception as e: # one bad job must not take down the worker loop print(f"worker: pipeline failed for job {job_id}: {e}", flush=True) conn.rollback() - if mcfg.enabled: - try: - reconcile(conn, job_id, mcfg) - except Exception as e: - print(f"worker: reconcile failed for job {job_id}: {e}", flush=True) - conn.rollback() + finish_job(conn, job_id, mcfg) print(f"worker: finished job {job_id}", flush=True) except psycopg.OperationalError as e: print(f"worker: db connection lost ({e}); reconnecting...", flush=True) diff --git a/worker/lyra_worker/monitor.py b/worker/lyra_worker/monitor.py index 15f9df7..a4e2f59 100644 --- a/worker/lyra_worker/monitor.py +++ b/worker/lyra_worker/monitor.py @@ -146,6 +146,26 @@ def enqueue_due(conn: psycopg.Connection, cfg: MonitorConfig) -> int: return enqueued +def fulfill_owned_releases(conn: psycopg.Connection, cfg: MonitorConfig) -> int: + """Mark fulfilled any monitored, not-yet-fulfilled release already owned at/above the + quality cutoff. Run once at startup so releases pressed while the monitor was off (and + therefore never reconciled) drop off the wanted list. Idempotent. Returns #rows fixed.""" + with conn.cursor() as cur: + cur.execute( + 'UPDATE "MonitoredRelease" mr SET state = \'fulfilled\', ' + ' "currentQualityClass" = li.q, ' + ' "firstGrabbedAt" = COALESCE(mr."firstGrabbedAt", now()) ' + 'FROM (SELECT artist, album, max("qualityClass") AS q FROM "LibraryItem" ' + ' GROUP BY artist, album) li ' + 'WHERE li.artist = mr."artistName" AND li.album = mr.album ' + " AND mr.monitored = true AND mr.state <> 'fulfilled' AND li.q >= %s", + (cfg.quality_cutoff,), + ) + fixed = cur.rowcount + conn.commit() + return fixed + + def reconcile(conn: psycopg.Connection, job_id: str, cfg: MonitorConfig) -> None: """Feed a finished job's outcome back into its MonitoredRelease (if any).""" with conn.cursor() as cur: diff --git a/worker/tests/test_monitor_reconcile.py b/worker/tests/test_monitor_reconcile.py index 23a3946..42f7426 100644 --- a/worker/tests/test_monitor_reconcile.py +++ b/worker/tests/test_monitor_reconcile.py @@ -1,4 +1,4 @@ -from lyra_worker.monitor import MonitorConfig, reconcile +from lyra_worker.monitor import MonitorConfig, fulfill_owned_releases, reconcile from tests.conftest import insert_monitored_release CFG = MonitorConfig(enabled=True, quality_cutoff=2) @@ -69,6 +69,43 @@ def test_needs_attention_leaves_wanted(conn): assert _release(conn, rel) == ("wanted", None, False) +def test_finish_job_reconciles_even_when_monitor_disabled(conn): + # Regression: a manually-wanted release that gets pressed must transition to fulfilled + # (and drop off the Wanted list) even with the monitor off — reconcile is core + # correctness, not a monitor-sweep feature. + from lyra_worker.main import finish_job + + rel = insert_monitored_release(conn, artist_name="A", album="B", rg_mbid="rg-1", + monitored=True, state="wanted") + req, job = _make_job(conn, rel, "A", "B", "imported") + _add_library(conn, req, "A", "B", 3) + finish_job(conn, job, MonitorConfig(enabled=False, quality_cutoff=2)) + assert _release(conn, rel) == ("fulfilled", 3, True) + + +def test_fulfill_owned_releases_backfills_stuck_wanted(conn): + # A release pressed while the monitor was off: owned in the library at cutoff, but its + # MonitoredRelease is still 'wanted' (never reconciled). Startup backfill fixes it. + rel = insert_monitored_release(conn, artist_name="A", album="B", rg_mbid="rg-1", + monitored=True, state="wanted") + other = insert_monitored_release(conn, artist_name="C", album="D", rg_mbid="rg-2", + monitored=True, state="wanted") # not owned → stays wanted + _add_library(conn, None, "A", "B", 3) + cfg = MonitorConfig(quality_cutoff=2) + assert fulfill_owned_releases(conn, cfg) == 1 + assert _release(conn, rel) == ("fulfilled", 3, True) + assert _release(conn, other)[0] == "wanted" + assert fulfill_owned_releases(conn, cfg) == 0 # idempotent + + +def test_fulfill_owned_releases_ignores_below_cutoff(conn): + rel = insert_monitored_release(conn, artist_name="A", album="B", rg_mbid="rg-1", + monitored=True, state="wanted") + _add_library(conn, None, "A", "B", 1) # owned but below cutoff → keep wanting (upgrade) + assert fulfill_owned_releases(conn, MonitorConfig(quality_cutoff=2)) == 0 + assert _release(conn, rel)[0] == "wanted" + + def test_unlinked_job_is_ignored(conn): with conn.cursor() as cur: cur.execute(