From ebae236307b21484c3d49c1ef811c8a38862ab70 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9A=B0=EC=A0=95?= Date: Mon, 13 Jul 2026 15:04:28 +0900 Subject: [PATCH] fix: re-check FreqBlog 'queued' (202) tracks once within the same run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A 202 from /lookup means an on-demand analysis was just triggered and is usually ready within ~15-60s. Previously the track only got retried on the next enrich run (next day's cron in production), so a playlist enriched interactively looked 'failed' even though the data was ready seconds later. enrich_pending now collects queued tracks and, after the main pass, waits QUEUED_RECHECK_DELAY (30s) and re-looks them up once: - ok → stored immediately (no second attempt bump for the same run) - rate_limited → stop, remaining tracks stay pending for the next run - still queued / not_found → left pending; next run retries them first (attempts DESC ordering) as before Each re-check spends a request the next-run retry would have spent anyway, so monthly quota use is net neutral. The re-check is skipped when the main pass already hit the rate limit. Fixes #2 Co-Authored-By: Claude Fable 5 --- src/mixmaster/enrich.py | 99 ++++++++++++++++++++++++++++++++--------- tests/test_enrich.py | 93 +++++++++++++++++++++++++++++++++++++- 2 files changed, 169 insertions(+), 23 deletions(-) diff --git a/src/mixmaster/enrich.py b/src/mixmaster/enrich.py index 30ed845..5090e95 100644 --- a/src/mixmaster/enrich.py +++ b/src/mixmaster/enrich.py @@ -2,8 +2,10 @@ State transitions: - ok → store enrichment, status='enriched', reset attempts -- queued → attempts+1 (waiting on on-demand analysis — re-checked on the - next run), marked 'failed' once MAX_ATTEMPTS is exceeded +- queued → attempts+1, then re-checked once more at the end of this + run (on-demand analysis is usually ready in ~15-60s); if + still queued, it stays pending for the next run and is + marked 'failed' once MAX_ATTEMPTS is exceeded - not_found → immediately 'failed' (the same query will never resolve) - rate_limited → end this run (progress so far is committed, resumes on the next run) @@ -14,6 +16,7 @@ from __future__ import annotations +import time from datetime import datetime, timezone from typing import Any @@ -24,14 +27,52 @@ MAX_ATTEMPTS = 10 COMMIT_EVERY = 20 +# FreqBlog's on-demand analysis after a 202 is usually ready in ~15-60s +# (docs/api-responses.md) — one in-run re-check turns "wait until +# tomorrow's cron" into "done this run" for freshly queued tracks. +QUEUED_RECHECK_DELAY = 30.0 + + +def _store_ok( + db: Database, + adapter: EnrichmentAdapter, + track: TrackToEnrich, + outcome: Any, + *, + bump_partial: bool = True, +) -> str: + """Store an 'ok' outcome; returns 'enriched' or 'partial'.""" + db.store_enrichment( + track.spotify_id, + adapter.name, + outcome.fields or {}, + raw=json.dumps(outcome.raw, ensure_ascii=False), + fetched_at=datetime.now(timezone.utc).isoformat(), + finalize=not outcome.partial, + ) + if outcome.partial: + # Data is stored; a later re-check (once backfill completes) will + # fill it in. If it's still incomplete by the attempt limit, we + # finalize the partial data as-is. + if bump_partial: + db.bump_enrich_attempt( + track.spotify_id, MAX_ATTEMPTS, terminal_status="enriched" + ) + return "partial" + return "enriched" def enrich_pending( - adapter: EnrichmentAdapter, db: Database, *, limit: int | None = None + adapter: EnrichmentAdapter, + db: Database, + *, + limit: int | None = None, + recheck_delay: float = QUEUED_RECHECK_DELAY, ) -> dict[str, Any]: pending = db.tracks_pending_enrichment(limit) counts = {"enriched": 0, "partial": 0, "queued": 0, "failed": 0} rate_limited = False + queued_tracks: list[TrackToEnrich] = [] try: for i, row in enumerate(pending, 1): if not row["artist"]: @@ -48,27 +89,11 @@ def enrich_pending( ) outcome = adapter.lookup(track) if outcome.status == "ok": - db.store_enrichment( - track.spotify_id, - adapter.name, - outcome.fields or {}, - raw=json.dumps(outcome.raw, ensure_ascii=False), - fetched_at=datetime.now(timezone.utc).isoformat(), - finalize=not outcome.partial, - ) - if outcome.partial: - # Data is stored; a later re-check (once backfill - # completes) will fill it in. If it's still incomplete by - # the attempt limit, we finalize the partial data as-is. - db.bump_enrich_attempt( - track.spotify_id, MAX_ATTEMPTS, terminal_status="enriched" - ) - counts["partial"] += 1 - else: - counts["enriched"] += 1 + counts[_store_ok(db, adapter, track, outcome)] += 1 elif outcome.status == "queued": db.bump_enrich_attempt(track.spotify_id, MAX_ATTEMPTS) counts["queued"] += 1 + queued_tracks.append(track) elif outcome.status == "not_found": db.mark_enrichment_failed(track.spotify_id) counts["failed"] += 1 @@ -82,6 +107,38 @@ def enrich_pending( if i % COMMIT_EVERY == 0: db.commit() print(f" enrichment progress: {i}/{len(pending)}") + # A 202 means the on-demand analysis was just triggered and is + # usually ready within ~15-60s — re-check once before giving up on + # this run, so freshly queued tracks don't sit pending until the + # next run (issue #2). Each re-check spends a request that the + # next-run retry would have spent anyway (net quota neutral). + if queued_tracks and not rate_limited and recheck_delay > 0: + db.commit() + print( + f" {len(queued_tracks)} tracks queued for on-demand analysis " + f"— re-checking in {recheck_delay:.0f}s..." + ) + time.sleep(recheck_delay) + for track in queued_tracks: + outcome = adapter.lookup(track) + if outcome.status == "ok": + # The attempt was already counted when it came back + # queued — don't bump again for the same run. + kind = _store_ok( + db, adapter, track, outcome, bump_partial=False + ) + counts[kind] += 1 + counts["queued"] -= 1 + elif outcome.status == "rate_limited": + rate_limited = True + print( + " rate limit during re-check — remaining queued tracks " + "will be collected on the next run." + ) + break + # still queued / not_found: leave it pending — the next run + # retries it first (attempts DESC ordering) without spending + # another attempt now. finally: # Every request consumes part of the monthly quota, so always commit # whatever results we already have, even on an exception or Ctrl-C. diff --git a/tests/test_enrich.py b/tests/test_enrich.py index 3929908..9b8f6dc 100644 --- a/tests/test_enrich.py +++ b/tests/test_enrich.py @@ -5,7 +5,7 @@ import pytest from mixmaster.db import LATEST_SCHEMA_VERSION, Database, _SCHEMA_V1 -from mixmaster.enrich import MAX_ATTEMPTS, enrich_pending +from mixmaster.enrich import MAX_ATTEMPTS, QUEUED_RECHECK_DELAY, enrich_pending from mixmaster.enrichment.base import ( EnrichmentAdapter, EnrichmentOutcome, @@ -21,6 +21,14 @@ def db(): d.close() +@pytest.fixture(autouse=True) +def fake_sleep(monkeypatch): + """No test should actually wait out the queued re-check delay.""" + calls = [] + monkeypatch.setattr("mixmaster.enrich.time.sleep", calls.append) + return calls + + def add_track(db, spotify_id, name="Song", artist_id="a1", isrc="KRA000000001"): db.upsert_track( spotify_id=spotify_id, name=name, album="Al", album_id="al1", @@ -35,12 +43,17 @@ class ScriptedAdapter(EnrichmentAdapter): name = "scripted" def __init__(self, outcomes): + # a value may be a list of outcomes, consumed one per lookup + # (the last one repeats) self.outcomes = outcomes self.calls = [] def lookup(self, track: TrackToEnrich) -> EnrichmentOutcome: self.calls.append(track.spotify_id) - return self.outcomes[track.spotify_id] + o = self.outcomes[track.spotify_id] + if isinstance(o, list): + return o.pop(0) if len(o) > 1 else o[0] + return o OK = EnrichmentOutcome( @@ -71,6 +84,82 @@ def test_queued_stays_pending_until_max_attempts(db): assert db.stats()["enrichment_failed"] == 1 +QUEUED = EnrichmentOutcome(status="queued") + + +def test_queued_recheck_recovers_within_run(db, fake_sleep): + # Issue #2: a 202 means the on-demand analysis is usually ready in + # ~15-60s — one in-run re-check should pick it up instead of leaving + # the track pending until the next run. + add_track(db, "t1") + adapter = ScriptedAdapter({"t1": [QUEUED, OK]}) + result = enrich_pending(adapter, db) + assert fake_sleep == [QUEUED_RECHECK_DELAY] + assert adapter.calls == ["t1", "t1"] + assert result["enriched"] == 1 + assert result["queued"] == 0 + assert db.stats()["enriched"] == 1 + assert db.stats()["enrichment_pending"] == 0 + + +def test_queued_recheck_still_queued_costs_one_attempt(db): + add_track(db, "t1") + result = enrich_pending(ScriptedAdapter({"t1": QUEUED}), db) + assert result["queued"] == 1 + assert db.stats()["enrichment_pending"] == 1 + attempts = db.conn.execute( + "SELECT enrich_attempts FROM tracks WHERE spotify_id='t1'" + ).fetchone()[0] + assert attempts == 1 # the re-check must not double-count the attempt + + +def test_queued_recheck_partial_stays_pending_without_extra_bump(db): + add_track(db, "t1") + result = enrich_pending(ScriptedAdapter({"t1": [QUEUED, PARTIAL]}), db) + assert result["partial"] == 1 + assert result["queued"] == 0 + row = db.conn.execute("SELECT bpm FROM enrichment WHERE track_id='t1'").fetchone() + assert row["bpm"] == 100.0 # partial data stored + assert db.stats()["enrichment_pending"] == 1 # awaiting backfill re-check + attempts = db.conn.execute( + "SELECT enrich_attempts FROM tracks WHERE spotify_id='t1'" + ).fetchone()[0] + assert attempts == 1 + + +def test_queued_recheck_skipped_when_rate_limited(db, fake_sleep): + add_track(db, "t1", isrc="KRA000000001") + add_track(db, "t2", isrc="KRA000000002") + # Force a deterministic processing order (added_at DESC) + db.conn.execute("UPDATE tracks SET added_at='2026-07-02T00:00:00Z' WHERE spotify_id='t1'") + db.commit() + adapter = ScriptedAdapter({ + "t1": QUEUED, + "t2": EnrichmentOutcome(status="rate_limited", detail="quota"), + }) + result = enrich_pending(adapter, db) + assert result["rate_limited"] is True + assert fake_sleep == [] # don't burn quota re-checking after a 429 + assert adapter.calls == ["t1", "t2"] + assert db.stats()["enrichment_pending"] == 2 + + +def test_queued_recheck_stops_on_rate_limit(db): + add_track(db, "t1", isrc="KRA000000001") + add_track(db, "t2", isrc="KRA000000002") + db.conn.execute("UPDATE tracks SET added_at='2026-07-02T00:00:00Z' WHERE spotify_id='t1'") + db.commit() + adapter = ScriptedAdapter({ + "t1": [QUEUED, EnrichmentOutcome(status="rate_limited", detail="quota")], + "t2": QUEUED, + }) + result = enrich_pending(adapter, db) + assert result["rate_limited"] is True + # t1 hit the rate limit on re-check → t2 is not re-checked this run + assert adapter.calls == ["t1", "t2", "t1"] + assert db.stats()["enrichment_pending"] == 2 + + def test_not_found_fails_immediately(db): add_track(db, "t1") result = enrich_pending(