From 9ba587d13664524c3988d2be3be9ddeb4a06aade Mon Sep 17 00:00:00 2001 From: Jonathan Date: Sun, 12 Jul 2026 10:42:48 +0200 Subject: [PATCH] fix(worker): run discovery at most once per loop iteration The scheduled-tick branch and the discover.requested one-shot branch both read the same stale per-iteration config snapshot, so an enabled+due sweep coinciding with a pending request ran discovery twice back-to-back (idempotent but wasteful). Extract maybe_run_discovery, which unifies both triggers into a single run and is now unit-tested. Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/main.py | 38 +++++++++++++--------- worker/tests/test_discovery_trigger.py | 45 +++++++++++++++++++++++++- 2 files changed, 67 insertions(+), 16 deletions(-) diff --git a/worker/lyra_worker/main.py b/worker/lyra_worker/main.py index bdf827f..fc228b4 100644 --- a/worker/lyra_worker/main.py +++ b/worker/lyra_worker/main.py @@ -45,6 +45,26 @@ def _run_discovery(conn, sources, browser, config=None) -> None: print(f"worker: discovery done — {result.artists} artists, {result.albums} albums", flush=True) +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 + (enabled + interval elapsed) or a one-shot ``discover.requested`` flag. A single run + satisfies both, so the two triggers must not fire it twice off the same stale config + snapshot. Returns the (possibly advanced) last_discover_tick.""" + requested = str(config.get("discover.requested", "")).strip().lower() in _TRUE + due = dcfg.enabled and now - last_discover_tick >= tick_seconds + if not (requested or due): + return last_discover_tick + try: + _run_discovery(conn, sources, browser, config) # also clears discover.requested + except Exception as e: # a discovery error must never kill the worker + print(f"worker: discovery run failed: {e}", flush=True) + conn.rollback() + if requested: + _set_config(conn, "discover.requested", "false") + return now if due else last_discover_tick + + def run_forever() -> None: conn = wait_for_db() clear_staging_root(DEST_ROOT) # sweep any staging dirs orphaned by a prior crash @@ -72,13 +92,9 @@ def run_forever() -> None: last_tick = now dcfg = DiscoveryConfig.from_config(config) - if dcfg.enabled and now - last_discover_tick >= DISCOVER_TICK_SECONDS: - try: - _run_discovery(conn, similarity_sources, browser, config) - except Exception as e: # a discovery error must never kill the worker - print(f"worker: discovery sweep failed: {e}", flush=True) - conn.rollback() - last_discover_tick = now + last_discover_tick = maybe_run_discovery( + conn, similarity_sources, browser, config, dcfg, now, last_discover_tick + ) if str(config.get("scan.requested", "")).strip().lower() in _TRUE: try: @@ -88,14 +104,6 @@ def run_forever() -> None: conn.rollback() _set_config(conn, "scan.requested", "false") - if str(config.get("discover.requested", "")).strip().lower() in _TRUE: - try: - _run_discovery(conn, similarity_sources, browser, config) - except Exception as e: # a discovery error must never kill the worker - print(f"worker: discovery run failed: {e}", flush=True) - conn.rollback() - _set_config(conn, "discover.requested", "false") - job_id = claim_next(conn) if job_id is None: time.sleep(IDLE_SLEEP) diff --git a/worker/tests/test_discovery_trigger.py b/worker/tests/test_discovery_trigger.py index c0c7c9e..5b15d6a 100644 --- a/worker/tests/test_discovery_trigger.py +++ b/worker/tests/test_discovery_trigger.py @@ -1,5 +1,7 @@ +from lyra_worker import main from lyra_worker.adapters.fakes import FakeMbBrowser, FakeSimilaritySource -from lyra_worker.main import _run_discovery +from lyra_worker.discovery import DiscoveryConfig +from lyra_worker.main import _run_discovery, maybe_run_discovery from lyra_worker.similarity.base import SimilarArtist from tests.conftest import insert_watched_artist @@ -33,3 +35,44 @@ def test_run_discovery_writes_result_and_clears_flag(conn): with conn.cursor() as cur: cur.execute('SELECT count(*) FROM "DiscoverySuggestion" WHERE kind = \'artist\'') assert cur.fetchone()[0] == 1 + + +def test_maybe_run_discovery_runs_once_when_due_and_requested(conn, monkeypatch): + # A scheduled tick that is due AND a pending one-shot request in the same loop + # iteration must trigger discovery exactly once, not twice. + calls = [] + monkeypatch.setattr(main, "_run_discovery", lambda *a, **k: calls.append(1)) + config = {"discover.enabled": "true", "discover.requested": "true"} + dcfg = DiscoveryConfig.from_config(config) + new_tick = maybe_run_discovery( + conn, [], FakeMbBrowser(), config, dcfg, now=10_000.0, last_discover_tick=0.0 + ) + assert calls == [1] # exactly one run + assert new_tick == 10_000.0 # schedule tick advanced + + +def test_maybe_run_discovery_requested_when_disabled_does_not_advance_tick(conn, monkeypatch): + # A one-shot request runs even when discovery is disabled, but must not advance the + # schedule tick (there is no active schedule). + calls = [] + monkeypatch.setattr(main, "_run_discovery", lambda *a, **k: calls.append(1)) + config = {"discover.enabled": "false", "discover.requested": "true"} + dcfg = DiscoveryConfig.from_config(config) + new_tick = maybe_run_discovery( + conn, [], FakeMbBrowser(), config, dcfg, now=10_000.0, last_discover_tick=42.0 + ) + assert calls == [1] + assert new_tick == 42.0 # unchanged + + +def test_maybe_run_discovery_idle_does_nothing(conn, monkeypatch): + calls = [] + monkeypatch.setattr(main, "_run_discovery", lambda *a, **k: calls.append(1)) + config = {"discover.enabled": "true", "discover.requested": "false"} + dcfg = DiscoveryConfig.from_config(config) + # enabled but not yet due (tick was recent relative to now) + new_tick = maybe_run_discovery( + conn, [], FakeMbBrowser(), config, dcfg, now=100.0, last_discover_tick=99.0 + ) + assert calls == [] + assert new_tick == 99.0