Merge pull request #532 from BigBodyCobain/fix/xquik-failure-backoff

fix(xquik): respect outage backoff and credential rotation

backend/services/fetchers/xquik_news.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import logging
 import os
 import re
@@ -26,7 +27,8 @@ _MAX_TEXT_LENGTH = 500
 _cache_lock = threading.Lock()
 _cache_signature: tuple[str, int] | None = None
 _cache_entries: list[dict[str, Any]] = []
-_cache_fetched_at = 0.0
+_attempt_signature: tuple[str, int, str] | None = None
+_last_attempt_at: float | None = None
 
 
 def xquik_fetch_enabled() -> bool:
@@ -63,13 +65,22 @@ def _normalize_tweet(tweet: object) -> dict[str, Any] | None:
     if not isinstance(tweet, dict):
         return None
     tweet_id = str(tweet.get("id") or "").strip()
-    text = " ".join(str(tweet.get("text") or "").split())
+    raw_text = tweet.get("text")
     author = tweet.get("author")
     username = str(author.get("username") or "").strip() if isinstance(author, dict) else ""
-    if not _TWEET_ID_RE.fullmatch(tweet_id) or not _USERNAME_RE.fullmatch(username) or not text:
+    created_at = tweet.get("createdAt")
+    if (
+        not _TWEET_ID_RE.fullmatch(tweet_id)
+        or not _USERNAME_RE.fullmatch(username)
+        or not isinstance(raw_text, str)
+        or not isinstance(created_at, str)
+    ):
         return None
 
-    created_at = tweet.get("createdAt")
+    text = " ".join(raw_text.split())
+    created_at = created_at.strip()
+    if not text:
+        return None
     published_parts = _published_parts(created_at)
     if published_parts is None:
         return None
@@ -121,6 +132,11 @@ def _request_entries(api_key: str, query: str, limit: int) -> list[dict[str, Any
     return entries
 
 
+def _credential_fingerprint(api_key: str) -> str:
+    """Return a non-secret cache key so credential rotation retries immediately."""
+    return hashlib.sha256(api_key.encode("utf-8")).hexdigest()[:16]
+
+
 def fetch_xquik_entries() -> list[dict[str, Any]]:
     """Return cached, normalized X posts for the configured search query."""
     if not xquik_fetch_enabled():
@@ -134,30 +150,42 @@ def fetch_xquik_entries() -> list[dict[str, Any]]:
 
     limit = _bounded_int("XQUIK_SEARCH_LIMIT", 20, 1, _MAX_RESULTS)
     interval_seconds = _bounded_int("XQUIK_SEARCH_INTERVAL_MINUTES", 30, 5, 1440) * 60
-    signature = (query, limit)
+    cache_signature = (query, limit)
+    attempt_signature = (query, limit, _credential_fingerprint(api_key))
 
-    global _cache_entries, _cache_fetched_at, _cache_signature
+    global _cache_entries, _cache_signature
+    global _attempt_signature, _last_attempt_at
     with _cache_lock:
+        now = time.monotonic()
         if (
-            _cache_signature == signature
-            and _cache_fetched_at
-            and time.monotonic() - _cache_fetched_at < interval_seconds
+            _attempt_signature == attempt_signature
+            and _last_attempt_at is not None
+            and now - _last_attempt_at < interval_seconds
         ):
-            return [dict(entry) for entry in _cache_entries]
+            if _cache_signature == cache_signature:
+                return [dict(entry) for entry in _cache_entries]
+            return []
+
+        # Record every outbound attempt, not only successes. This keeps an
+        # unhealthy provider from being retried by the 5-minute news scheduler
+        # more frequently than the operator-configured Xquik interval.
+        _attempt_signature = attempt_signature
+        _last_attempt_at = now
 
         entries = _request_entries(api_key, query, limit)
         if entries is not None:
-            _cache_signature = signature
+            _cache_signature = cache_signature
             _cache_entries = entries
-            _cache_fetched_at = time.monotonic()
-        if _cache_signature == signature:
+        if _cache_signature == cache_signature:
             return [dict(entry) for entry in _cache_entries]
         return []
 
 
 def _reset_cache_for_tests() -> None:
-    global _cache_entries, _cache_fetched_at, _cache_signature
+    global _cache_entries, _cache_signature
+    global _attempt_signature, _last_attempt_at
     with _cache_lock:
         _cache_signature = None
         _cache_entries = []
-        _cache_fetched_at = 0.0
+        _attempt_signature = None
+        _last_attempt_at = None

backend/tests/test_xquik_news.py

@@ -93,6 +93,8 @@ def test_request_is_bounded_and_normalizes_untrusted_posts(monkeypatch) -> None:
                     _tweet(tweet_id="9876543210", username="invalid-name"),
                     {**_tweet(tweet_id="111"), "createdAt": "not-a-date"},
                     {"id": "42", "text": "", "author": {"username": "empty"}},
+                    {**_tweet(tweet_id="222"), "text": {"unexpected": "shape"}},
+                    {**_tweet(tweet_id="333"), "createdAt": 1234567890},
                     "not-an-object",
                 ]
             }
@@ -139,7 +141,7 @@ def test_successful_results_are_cached_and_copied(monkeypatch) -> None:
     assert second[0]["title"] == "Missile strike reported in Kyiv"
 
 
-def test_http_failure_keeps_cached_results_without_retry(monkeypatch) -> None:
+def test_http_failure_keeps_cached_results_and_respects_poll_interval(monkeypatch) -> None:
     _enable(monkeypatch)
     responses = [_Response({"tweets": [_tweet()]}), _Response({}, status_code=429)]
     calls = 0
@@ -150,11 +152,48 @@ def test_http_failure_keeps_cached_results_without_retry(monkeypatch) -> None:
         return responses.pop(0)
 
     monkeypatch.setattr(xquik_news.requests, "get", fake_get)
-    clock = iter((1000.0, 2801.0))
+    clock = iter((1000.0, 2801.0, 2802.0))
     monkeypatch.setattr(xquik_news.time, "monotonic", lambda: next(clock))
-    expected = xquik_news.fetch_xquik_entries()
 
+    expected = xquik_news.fetch_xquik_entries()
     assert xquik_news.fetch_xquik_entries() == expected
+    assert xquik_news.fetch_xquik_entries() == expected
+    assert calls == 2
+
+
+def test_failure_without_cache_is_throttled(monkeypatch) -> None:
+    _enable(monkeypatch)
+    calls = 0
+
+    def fake_get(*args, **kwargs):
+        nonlocal calls
+        calls += 1
+        return _Response({}, status_code=503)
+
+    monkeypatch.setattr(xquik_news.requests, "get", fake_get)
+    clock = iter((1000.0, 1001.0))
+    monkeypatch.setattr(xquik_news.time, "monotonic", lambda: next(clock))
+
+    assert xquik_news.fetch_xquik_entries() == []
+    assert xquik_news.fetch_xquik_entries() == []
+    assert calls == 1
+
+
+def test_api_key_rotation_bypasses_failed_attempt_backoff(monkeypatch) -> None:
+    _enable(monkeypatch)
+    responses = [_Response({}, status_code=401), _Response({"tweets": [_tweet()]})]
+    calls = 0
+
+    def fake_get(*args, **kwargs):
+        nonlocal calls
+        calls += 1
+        return responses.pop(0)
+
+    monkeypatch.setattr(xquik_news.requests, "get", fake_get)
+
+    assert xquik_news.fetch_xquik_entries() == []
+    monkeypatch.setenv("XQUIK_API_KEY", "rotated-unit-test-key")
+    assert len(xquik_news.fetch_xquik_entries()) == 1
     assert calls == 2
 
 
@@ -162,7 +201,9 @@ def test_failed_query_change_does_not_reuse_another_query(monkeypatch) -> None:
     _enable(monkeypatch)
     responses = [_Response({"tweets": [_tweet()]}), _Response({}, status_code=503)]
     monkeypatch.setattr(
-        xquik_news.requests, "get", lambda *args, **kwargs: responses.pop(0)
+        xquik_news.requests,
+        "get",
+        lambda *args, **kwargs: responses.pop(0),
     )
     assert len(xquik_news.fetch_xquik_entries()) == 1
 
@@ -176,6 +217,7 @@ def test_redirect_and_invalid_json_fail_closed(monkeypatch) -> None:
 
     monkeypatch.setattr(xquik_news.requests, "get", lambda *args, **kwargs: responses.pop(0))
     assert xquik_news.fetch_xquik_entries() == []
+    monkeypatch.setenv("XQUIK_SEARCH_QUERY", "different region")
     assert xquik_news.fetch_xquik_entries() == []