diff --git a/core/amazon_client.py b/core/amazon_client.py index 2414051e..5d61daf2 100644 --- a/core/amazon_client.py +++ b/core/amazon_client.py @@ -70,7 +70,16 @@ _meta_cache_lock = threading.Lock() class AmazonClientError(RuntimeError): - """Raised on unrecoverable T2Tunes API errors.""" + """Raised on unrecoverable T2Tunes API errors. + + Carries the HTTP ``status_code`` when the failure was an HTTP error, so + callers (the worker's outage detection) can tell a source outage (5xx) from + a per-item miss without parsing the message. + """ + + def __init__(self, *args, status_code=None): + super().__init__(*args) + self.status_code = status_code # --------------------------------------------------------------------------- @@ -703,7 +712,8 @@ class AmazonClient: ) continue raise AmazonClientError( - f"HTTP {exc.response.status_code} for {url} — body: {body!r}" + f"HTTP {exc.response.status_code} for {url} — body: {body!r}", + status_code=exc.response.status_code, ) from exc except requests.RequestException as exc: raise AmazonClientError(f"Request failed for {url}: {exc}") from exc diff --git a/core/amazon_outage.py b/core/amazon_outage.py new file mode 100644 index 00000000..b8205464 --- /dev/null +++ b/core/amazon_outage.py @@ -0,0 +1,61 @@ +"""Amazon enrichment outage detection + back-off — pure, importable, testable. + +The Amazon worker enriches via a public T2Tunes proxy instance. When that +instance is down (HTTP 5xx, "Amazon Music API is not initialized", or an +unreachable host), the worker must NOT treat every album as an individual +failure: doing so floods the logs with an error per item, churns network + DB +continuously, and permanently marks the whole library ``error`` (which the +retry tiers never re-attempt) for what is really a transient outage. + +Instead it recognizes "the whole source is down", leaves the item untouched so +it's retried once the instance recovers, and backs off hard. These two pure +helpers carry that logic so it can be unit-tested without the worker, the DB, +or the network. +""" + +from __future__ import annotations + +import re + +# HTTP statuses that mean "the source/proxy is unhealthy", not "no match". +_OUTAGE_STATUS = {500, 502, 503, 504} + +# Substrings (lower-cased) in an error message that indicate a source outage +# rather than a per-item miss: proxy not ready, gateway errors, the host being +# unreachable, or an error page returned instead of JSON. +_OUTAGE_PHRASES = ( + "not initialized", "not configured", "service unavailable", + "bad gateway", "gateway time", "request failed", "response not json", + "max retries", "connection", "timed out", "temporarily unavailable", +) + +# Back-off schedule while the source is down. +_NORMAL_DELAY = 2 # seconds between items when healthy +_OUTAGE_BASE = 30 # first back-off step +_OUTAGE_CAP = 1800 # 30 minutes max + + +def is_source_outage(exc: Exception) -> bool: + """True when ``exc`` indicates the Amazon source/proxy is down (transient, + whole-source), as opposed to a normal per-item error. + + Robust to how the error is surfaced: an explicit ``status_code`` attribute, + an ``HTTP `` prefix in the message, or an outage phrase (covers + connection failures and non-JSON error pages that carry no status code).""" + code = getattr(exc, "status_code", None) + if isinstance(code, int) and code in _OUTAGE_STATUS: + return True + msg = str(exc).lower() + m = re.search(r"http\s+(\d{3})", msg) + if m and int(m.group(1)) in _OUTAGE_STATUS: + return True + return any(p in msg for p in _OUTAGE_PHRASES) + + +def next_poll_delay_seconds(outage_streak: int) -> int: + """Seconds to wait before the next item. Normal cadence when healthy; + escalating back-off (30s, 60s, 120s, … capped at 30 min) the longer the + source has been down, so a dead instance can't flood logs/CPU/DB.""" + if outage_streak <= 0: + return _NORMAL_DELAY + return min(_OUTAGE_BASE * (2 ** min(outage_streak - 1, 6)), _OUTAGE_CAP) diff --git a/core/amazon_worker.py b/core/amazon_worker.py index 2ac47ee4..c4dbc0c7 100644 --- a/core/amazon_worker.py +++ b/core/amazon_worker.py @@ -9,6 +9,7 @@ from database.music_database import MusicDatabase from core.amazon_client import AmazonClient from core.worker_utils import interruptible_sleep, set_album_api_track_count from core.enrichment.manual_match_honoring import honor_stored_match +from core.amazon_outage import is_source_outage, next_poll_delay_seconds logger = get_logger("amazon_worker") @@ -39,6 +40,11 @@ class AmazonWorker: self.retry_days = 30 self.name_similarity_threshold = 0.80 + # Source-outage circuit breaker: counts consecutive whole-source + # failures (proxy down / "not initialized" / unreachable) so the loop + # backs off instead of grinding the whole library item-by-item. + self._outage_streak = 0 + logger.info("Amazon background worker initialized") def _ensure_amazon_schema(self, cursor) -> None: @@ -151,7 +157,9 @@ class AmazonWorker: continue self._process_item(item) - interruptible_sleep(self._stop_event, 2) + # Normal 2s cadence when healthy; escalating back-off (up to + # 30 min) while the source is in an outage streak. + interruptible_sleep(self._stop_event, next_poll_delay_seconds(self._outage_streak)) except Exception as e: logger.error(f"Error in worker loop: {e}") @@ -275,7 +283,32 @@ class AmazonWorker: elif item_type == 'track': self._process_track(item_id, item_name, item.get('artist', ''), item) + # The source answered (match or not_found) — clear any outage streak. + if self._outage_streak: + logger.info("Amazon source recovered after %d outage(s), resuming", + self._outage_streak) + self._outage_streak = 0 + except Exception as e: + if is_source_outage(e): + # The whole source is down (proxy 5xx / "not initialized" / + # unreachable). Do NOT mark the item 'error' — that would burn + # the entire library to a state the retry tiers never re-attempt + # for a transient outage. Leave it untouched so it's retried once + # the instance recovers, and let the loop back off. Log once per + # streak to avoid flooding. + self._outage_streak += 1 + if self._outage_streak == 1: + logger.warning("Amazon source unavailable — pausing enrichment " + "until it recovers: %s", e) + else: + logger.debug("Amazon source still unavailable (streak=%d): %s", + self._outage_streak, e) + return + # A non-outage error means the source actually answered (e.g. a + # 404/parse error on a real response), so the outage is over — + # clear the streak and handle this as a normal per-item error. + self._outage_streak = 0 logger.error(f"Error processing {item['type']} #{item['id']}: {e}") self.stats['errors'] += 1 try: diff --git a/tests/test_amazon_outage.py b/tests/test_amazon_outage.py new file mode 100644 index 00000000..9a3b8357 --- /dev/null +++ b/tests/test_amazon_outage.py @@ -0,0 +1,86 @@ +"""Tests for Amazon enrichment outage detection + back-off (core/amazon_outage.py). + +Pins the contract behind issue #759: when the public T2Tunes proxy is down +(503 "Amazon Music API is not initialized", 5xx, or unreachable) the worker must +recognize a *source outage* — not treat every album as a per-item error and +grind/flood the whole library. +""" + +from __future__ import annotations + +from core.amazon_client import AmazonClientError +from core.amazon_outage import is_source_outage, next_poll_delay_seconds + + +# --- is_source_outage: the reported case + the surfaces it can arrive in ----- + + +def test_503_not_initialized_is_outage_via_status_code(): + exc = AmazonClientError("HTTP 503 for .../search — body: '...not initialized...'", + status_code=503) + assert is_source_outage(exc) is True + + +def test_503_is_outage_even_without_status_code_attr(): + # Message-only (e.g. re-raised/wrapped) — parsed from the "HTTP 503" prefix. + assert is_source_outage(Exception("HTTP 503 for https://t2tunes.site/...")) is True + + +def test_not_initialized_phrase_is_outage_without_code(): + assert is_source_outage(Exception("Amazon Music API is not initialized")) is True + + +def test_gateway_5xx_are_outages(): + for code in (500, 502, 504): + assert is_source_outage(AmazonClientError("x", status_code=code)) is True + + +def test_connection_failure_is_outage(): + assert is_source_outage(AmazonClientError( + "Request failed for https://t2tunes.site/...: Connection refused")) is True + assert is_source_outage(Exception("Max retries exceeded ... Connection timed out")) is True + + +def test_non_json_error_page_is_outage(): + assert is_source_outage(AmazonClientError( + "Response not JSON for ...: '503 Service Unavailable'")) is True + + +# --- NOT outages: real per-item misses / client errors ----------------------- + + +def test_404_is_not_outage(): + assert is_source_outage(AmazonClientError("HTTP 404 for ...", status_code=404)) is False + + +def test_transient_400_failed_to_search_is_not_outage(): + # A per-query Amazon-side hiccup, not the whole source being down. + assert is_source_outage(AmazonClientError( + "HTTP 400 for ... — body: 'Failed to search'", status_code=400)) is False + + +def test_generic_error_is_not_outage(): + assert is_source_outage(ValueError("something unrelated")) is False + + +# --- next_poll_delay_seconds: normal cadence vs escalating back-off ---------- + + +def test_healthy_uses_normal_cadence(): + assert next_poll_delay_seconds(0) == 2 + assert next_poll_delay_seconds(-1) == 2 + + +def test_backoff_escalates_then_caps(): + assert next_poll_delay_seconds(1) == 30 + assert next_poll_delay_seconds(2) == 60 + assert next_poll_delay_seconds(3) == 120 + assert next_poll_delay_seconds(4) == 240 + # Escalation is capped at 30 minutes so it never grows unbounded. + assert next_poll_delay_seconds(50) == 1800 + + +def test_backoff_is_monotonic_nondecreasing(): + delays = [next_poll_delay_seconds(s) for s in range(1, 20)] + assert delays == sorted(delays) + assert max(delays) == 1800 diff --git a/web_server.py b/web_server.py index d7719c50..9572dfe8 100644 --- a/web_server.py +++ b/web_server.py @@ -32899,9 +32899,13 @@ try: amazon_db = MusicDatabase() amazon_worker = AmazonWorker(database=amazon_db) amazon_worker.start() - if config_manager.get('amazon_enrichment_paused', False): + # Opt-in by default: Amazon enrichment depends on an external public proxy + # (T2Tunes) that can be down, so it stays paused unless the user has + # explicitly enabled it (amazon_enrichment_paused=False). This stops an + # instance outage from grinding/log-flooding installs that never opted in. + if config_manager.get('amazon_enrichment_paused', True): amazon_worker.pause() - logger.info("Amazon enrichment worker initialized (paused — restored from config)") + logger.info("Amazon enrichment worker initialized (paused — enable it in Settings)") else: logger.info("Amazon enrichment worker initialized and started") except Exception as e: