fix(adapters): shared retry helper + run_log failure_class + enable RSS (issues #1 #2 #9)

- adapters/__init__.py: add http_get() unified retry (429/5xx only, max 2 attempts, capped exp backoff) + AdapterHTTPError carrying failure_class; SourceAdapter.last_failure_class set on failure for pipeline capture. - arxiv/github/huggingface/hackernews/reddit: route HTTP through http_get. Preserves GitHub 403 rate-limit retry and Reddit 403/429 fast-bail. - schema.sql + pipeline.py: add run_log.failure_class column; rollup most- severe class across sources (5xx>4xx>429>error>zero_fetch>ok). - pipeline.py: ENABLE RSS in ENABLED_SOURCES (was registered, disabled). - RSS smoke test surfaced 3 broken feeds (anthropic 404, googleai 404, metaai 301) — left as-is, captured in feed_failures; URL fix is separate discovery task, not guessed. Verified: full dry-run fetches all 6 sources; github live fetch OK; Reddit 429 fast-bail preserved; no import/syntax errors.
2026-07-10 16:34:05 +00:00
parent 7ee1af3d7b
commit 23cce4d609
8 changed files with 209 additions and 137 deletions
@@ -1,11 +1,18 @@
 """Source adapters for AI Research Oracle."""

+import urllib.request
+import urllib.error
+import time
 from abc import ABC, abstractmethod


 class SourceAdapter(ABC):
    """Base class for all ingestion adapters."""

+    def __init__(self):
+        # Set by http_get when a request fails (issue #2 classification)
+        self.last_failure_class = None
+
    @abstractmethod
    def name(self) -> str:
        """Source name: 'github', 'arxiv', 'reddit'."""
@@ -15,3 +22,81 @@ class SourceAdapter(ABC):
    def fetch(self, query: str = "", limit: int = 20) -> list[dict]:
        """Return entries matching DB schema fields."""
        pass
+
+
+class AdapterHTTPError(Exception):
+    """Raised by http_get on non-retryable or exhausted HTTP failures.
+
+    failure_class is one of: '429', '5xx', '4xx', 'error'.
+    Consumed by pipeline.py to populate run_log.failure_class (issue #2).
+    """
+
+    def __init__(self, status, failure_class, message=""):
+        super().__init__(message)
+        self.status = status
+        self.failure_class = failure_class
+
+
+def classify_http_status(code):
+    """Map an HTTP status code to a run_log failure_class."""
+    if code == 429:
+        return "429"
+    if 500 <= code < 600:
+        return "5xx"
+    return "4xx"  # 401/403/404 etc = client/config problem, not retried
+
+
+def http_get(url, headers=None, timeout=15, max_retries=2,
+             backoff_base=2, retry_403_ratelimit=False, return_headers=False,
+             owner=None):
+    """GET with a unified retry policy shared by all adapters (issue #1).
+
+    Retries ONLY on 429 and 5xx (transient). 4xx other than 429 are NOT
+    retried (config/client problems). Optional GitHub-style 403 rate-limit
+    handling via retry_403_ratelimit (waits for X-RateLimit-Reset header).
+
+    Returns response body bytes (or (body, headers) tuple if return_headers).
+    Raises AdapterHTTPError on non-retryable or exhausted failures, carrying
+    .failure_class for run_log classification. If owner is provided, sets
+    owner.last_failure_class so the pipeline can record it (issue #2).
+    """
+    def _fail(code, fclass, msg):
+        if owner is not None:
+            owner.last_failure_class = fclass
+        raise AdapterHTTPError(code, fclass, msg)
+
+    req = urllib.request.Request(url, headers=headers or {})
+    last_err = None
+    for attempt in range(max_retries + 1):
+        try:
+            with urllib.request.urlopen(req, timeout=timeout) as resp:
+                body = resp.read()
+                if return_headers:
+                    return body, resp.headers
+                return body
+        except urllib.error.HTTPError as e:
+            last_err = e
+            code = e.code
+            if code == 429 or (code == 403 and retry_403_ratelimit):
+                if attempt < max_retries:
+                    if code == 403:
+                        reset = int(e.headers.get("X-RateLimit-Reset", 0))
+                        wait = (max(reset - int(time.time()), 0) + 1
+                                if reset else backoff_base * (attempt + 1) * 15)
+                        time.sleep(min(wait, 300))
+                    else:
+                        time.sleep(min(backoff_base * (attempt + 1), 30))
+                    continue
+                fclass = "429" if code == 429 else "4xx"
+                _fail(code, fclass, f"HTTP {code} for {url} (exhausted)")
+            if 500 <= code < 600:
+                if attempt < max_retries:
+                    time.sleep(min(backoff_base * (attempt + 1), 30))
+                    continue
+                _fail(code, "5xx", f"HTTP {code} for {url} (exhausted)")
+            # other 4xx (401/403 w/o flag/404) — not retried
+            _fail(code, classify_http_status(code), f"HTTP {code} for {url}")
+        except Exception as e:  # non-HTTP (timeout, DNS, conn reset)
+            _fail(None, "error", f"Request error: {e}")
+    # Defensive: loop should always raise or return
+    _fail(getattr(last_err, "code", None), "error", str(last_err))
@@ -32,7 +32,7 @@ import xml.etree.ElementTree as ET
 from datetime import datetime, timedelta, timezone
 from html import unescape

-from adapters import SourceAdapter
+from adapters import SourceAdapter, http_get, AdapterHTTPError

 # arXiv API
 ARXIV_API = "http://export.arxiv.org/api/query"
@@ -151,14 +151,12 @@ class ArxivAdapter(SourceAdapter):
            f"&max_results={max_results}"
        )

-        req = urllib.request.Request(url, headers={"User-Agent": "ai-oracle/0.1"})
-
        try:
-            with urllib.request.urlopen(req, timeout=30) as resp:
-                xml_data = resp.read().decode("utf-8")
-            return self._parse_atom(xml_data)
-        except urllib.error.HTTPError as e:
-            print(f"    HTTP {e.code} for arXiv query")
+            raw = http_get(url, headers={"User-Agent": "ai-oracle/0.1"},
+                           timeout=30, max_retries=2, owner=self)
+            return self._parse_atom(raw.decode("utf-8"))
+        except AdapterHTTPError as e:
+            print(f"    {e.failure_class}: arXiv query ({e})")
            return []
        except Exception as e:
            print(f"    arXiv request error: {e}")
@@ -17,7 +17,7 @@ import urllib.error
 import urllib.parse
 from datetime import datetime, timedelta, timezone

-from adapters import SourceAdapter
+from adapters import SourceAdapter, http_get, AdapterHTTPError


 class GitHubAdapter(SourceAdapter):
@@ -43,40 +43,27 @@ class GitHubAdapter(SourceAdapter):
        return headers

    def _request(self, url: str, max_retries: int = 2) -> dict | list | None:
-        """Make a GET request with retry on 403 (rate limit)."""
-        req = urllib.request.Request(url, headers=self._headers())
-
-        for attempt in range(max_retries + 1):
-            try:
-                with urllib.request.urlopen(req, timeout=15) as resp:
-                    data = json.loads(resp.read().decode("utf-8"))
-
-                # Check rate limit headers
-                remaining = int(resp.headers.get("X-RateLimit-Remaining", 0))
-                if remaining <= 5:
-                    print(f"    ⚠ Rate limit low ({remaining} remaining), stopping")
-                    break
-
-                return data
-            except urllib.error.HTTPError as e:
-                if e.code == 403:
-                    # Rate limited — reset time is in headers
-                    reset = int(e.headers.get("X-RateLimit-Reset", 0))
-                    if reset:
-                        wait = max(reset - int(time.time()), 0) + 1
-                        print(f"    ⚠ Rate limited, wait {wait}s")
-                    else:
-                        wait = 30 * (attempt + 1)
-                        print(f"    403 on attempt {attempt + 1}, retry in {wait}s")
-                    time.sleep(min(wait, 300))  # cap at 5 min
-                    continue
-                print(f"    HTTP {e.code} for {url}")
-                return None
-            except Exception as e:
-                print(f"    Request error: {e}")
-                return None
-
-        return None
+        """GET via shared retry helper; 403 rate-limit handled as transient."""
+        try:
+            raw, headers = http_get(
+                url, headers=self._headers(), timeout=15,
+                max_retries=max_retries, retry_403_ratelimit=True,
+                return_headers=True, owner=self)
+        except AdapterHTTPError as e:
+            print(f"    {e.failure_class}: GitHub {url}")
+            return None
+        # Informational: flag if we're close to the unauth rate ceiling
+        try:
+            remaining = int(headers.get("X-RateLimit-Remaining", 0))
+            if remaining <= 5:
+                print(f"    ⚠ Rate limit low ({remaining} remaining)")
+        except Exception:
+            pass
+        try:
+            return json.loads(raw.decode("utf-8"))
+        except Exception as e:
+            print(f"    GitHub decode error: {e}")
+            return None

    def _search_repos(self, query: str, sort: str = "stars", order: str = "desc", per_page: int = 30) -> list:
        """Search repositories via GitHub API."""
@@ -20,7 +20,7 @@ import urllib.request
 import urllib.error
 from datetime import datetime, timezone

-from adapters import SourceAdapter
+from adapters import SourceAdapter, http_get, AdapterHTTPError


 class HackerNewsAdapter(SourceAdapter):
@@ -54,24 +54,19 @@ class HackerNewsAdapter(SourceAdapter):
        return "hackernews"

    def _request(self, path: str, max_retries: int = 2) -> dict | list | None:
-        """Make a GET request to the HN Firebase API."""
+        """GET via shared retry helper (retries 429/5xx)."""
        url = f"{self.BASE}{path}"
-        req = urllib.request.Request(url, headers={"User-Agent": self.user_agent})
-
-        for attempt in range(max_retries + 1):
-            try:
-                with urllib.request.urlopen(req, timeout=15) as resp:
-                    return json.loads(resp.read().decode("utf-8"))
-            except (urllib.error.HTTPError, urllib.error.URLError) as e:
-                if attempt < max_retries:
-                    time.sleep(3 * (attempt + 1))
-                    continue
-                print(f"    HTTP error: {e}")
-                return None
-            except Exception as e:
-                print(f"    Request error: {e}")
-                return None
-        return None
+        try:
+            raw = http_get(url, headers={"User-Agent": self.user_agent},
+                           timeout=15, max_retries=max_retries, owner=self)
+        except AdapterHTTPError as e:
+            print(f"    {e.failure_class}: HN {path}")
+            return None
+        try:
+            return json.loads(raw.decode("utf-8"))
+        except Exception as e:
+            print(f"    HN decode error: {e}")
+            return None

    def _is_ai_relevant(self, title: str) -> bool:
        """Check if a story title is AI/ML relevant.
@@ -34,7 +34,7 @@ import urllib.request
 import urllib.error
 from datetime import datetime, timedelta, timezone

-from adapters import SourceAdapter
+from adapters import SourceAdapter, http_get, AdapterHTTPError


 class HuggingFaceAdapter(SourceAdapter):
@@ -84,24 +84,19 @@ class HuggingFaceAdapter(SourceAdapter):
        return headers

    def _request(self, path: str, max_retries: int = 2) -> list | dict | None:
-        """Make a GET request to the HF API."""
+        """GET via shared retry helper (retries 429/5xx)."""
        url = f"{self.BASE}{path}"
-        req = urllib.request.Request(url, headers=self._headers())
-
-        for attempt in range(max_retries + 1):
-            try:
-                with urllib.request.urlopen(req, timeout=20) as resp:
-                    return json.loads(resp.read().decode("utf-8"))
-            except (urllib.error.HTTPError, urllib.error.URLError) as e:
-                if attempt < max_retries:
-                    time.sleep(3 * (attempt + 1))
-                    continue
-                print(f"    HF API error: {e}")
-                return None
-            except Exception as e:
-                print(f"    Request error: {e}")
-                return None
-        return None
+        try:
+            raw = http_get(url, headers=self._headers(), timeout=20,
+                           max_retries=max_retries, owner=self)
+        except AdapterHTTPError as e:
+            print(f"    {e.failure_class}: HF {path}")
+            return None
+        try:
+            return json.loads(raw.decode("utf-8"))
+        except Exception as e:
+            print(f"    HF decode error: {e}")
+            return None

    def _is_ai_relevant(self, model: dict) -> bool:
        """Check if a model/dataset is AI/ML relevant.
@@ -24,7 +24,7 @@ import xml.etree.ElementTree as ET
 from datetime import datetime, timezone
 from html import unescape

-from adapters import SourceAdapter
+from adapters import SourceAdapter, http_get, AdapterHTTPError


 class RedditAdapter(SourceAdapter):
@@ -101,33 +101,27 @@ class RedditAdapter(SourceAdapter):
        return False

    def _fetch_rss(self, subreddit: str) -> list[dict]:
-        """Fetch RSS feed for a subreddit."""
-        url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50"
-        req = urllib.request.Request(url, headers={"User-Agent": self.user_agent})
+        """Fetch RSS feed for a subreddit (shared retry helper).

-        for attempt in range(2):  # max 2 attempts, fail fast
-            try:
-                with urllib.request.urlopen(req, timeout=10) as resp:
-                    xml_data = resp.read().decode("utf-8")
-                break
-            except urllib.error.HTTPError as e:
-                if e.code in (403,):
-                    print(f"    RSS blocked (HTTP {e.code}) for r/{subreddit}")
-                    return []
-                if e.code == 429:
-                    if attempt == 0:
-                        time.sleep(2)  # single retry with short backoff
-                        continue
-                    print(f"    RSS rate-limited for r/{subreddit}, skip")
-                    return []
-                print(f"    RSS HTTP {e.code} for r/{subreddit}")
-                return []
-            except Exception as e:
-                print(f"    RSS error r/{subreddit}: {e}")
-                return []
-        else:
-            print(f"    r/{subreddit}: still rate limited, skip")
+        Preserves prior fast-bail: 403 -> immediate []; 429 -> single 2s
+        retry then []; 5xx -> helper retry then []. Not slower than before.
+        """
+        url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50"
+        try:
+            raw = http_get(url, headers={"User-Agent": self.user_agent},
+                           timeout=10, max_retries=1, backoff_base=2, owner=self)
+        except AdapterHTTPError as e:
+            if e.status == 403:
+                print(f"    RSS blocked (HTTP 403) for r/{subreddit}")
+            elif e.status == 429:
+                print(f"    RSS rate-limited for r/{subreddit}, skip")
+            else:
+                print(f"    RSS {e.failure_class} for r/{subreddit}")
            return []
+        except Exception as e:
+            print(f"    RSS error r/{subreddit}: {e}")
+            return []
+        xml_data = raw.decode("utf-8")

        # Parse Atom XML
        entries = []
@@ -303,33 +297,27 @@ class RedditAdapter(SourceAdapter):
        return tags

    def _fetch_rss(self, subreddit: str) -> list[dict]:
-        """Fetch RSS feed for a subreddit."""
-        url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50"
-        req = urllib.request.Request(url, headers={"User-Agent": self.user_agent})
+        """Fetch RSS feed for a subreddit (shared retry helper).

-        for attempt in range(2):  # max 2 attempts, fail fast
-            try:
-                with urllib.request.urlopen(req, timeout=10) as resp:
-                    xml_data = resp.read().decode("utf-8")
-                break
-            except urllib.error.HTTPError as e:
-                if e.code in (403,):
-                    print(f"    RSS blocked (HTTP {e.code}) for r/{subreddit}")
-                    return []
-                if e.code == 429:
-                    if attempt == 0:
-                        time.sleep(2)  # single retry with short backoff
-                        continue
-                    print(f"    RSS rate-limited for r/{subreddit}, skip")
-                    return []
-                print(f"    RSS HTTP {e.code} for r/{subreddit}")
-                return []
-            except Exception as e:
-                print(f"    RSS error r/{subreddit}: {e}")
-                return []
-        else:
-            print(f"    r/{subreddit}: still rate limited, skip")
+        Preserves prior fast-bail: 403 -> immediate []; 429 -> single 2s
+        retry then []; 5xx -> helper retry then []. Not slower than before.
+        """
+        url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50"
+        try:
+            raw = http_get(url, headers={"User-Agent": self.user_agent},
+                           timeout=10, max_retries=1, backoff_base=2, owner=self)
+        except AdapterHTTPError as e:
+            if e.status == 403:
+                print(f"    RSS blocked (HTTP 403) for r/{subreddit}")
+            elif e.status == 429:
+                print(f"    RSS rate-limited for r/{subreddit}, skip")
+            else:
+                print(f"    RSS {e.failure_class} for r/{subreddit}")
            return []
+        except Exception as e:
+            print(f"    RSS error r/{subreddit}: {e}")
+            return []
+        xml_data = raw.decode("utf-8")

        # Parse Atom XML
        entries = []