From feba2fe7b7a6746cc3c47aa3c7c932a734660a06 Mon Sep 17 00:00:00 2001 From: Epictetus Date: Sun, 12 Jul 2026 04:51:24 +0000 Subject: [PATCH] feat: apply fix/adapter-health-1-2-9 changes (unified retry, failure_class, RSS enabled) - adapters/__init__.py: add http_get() unified retry helper + AdapterHTTPError + failure_class classification (429/5xx/4xx/error/zero_fetch/ok) + last_failure_class on SourceAdapter for pipeline capture - pipeline.py: ENABLED_SOURCES now includes RSS + failure_class rollup in run_log (most severe across all adapters) + per-source failure_class in source_stats - schema.sql: add failure_class column to run_log table Backport of fix/adapter-health-1-2-9 branch (issues #1, #2, #9). --- adapters/__init__.py | 85 ++++++++++++++++++++++++++++++++++++++++++++ pipeline.py | 36 +++++++++++++++---- schema.sql | 3 ++ 3 files changed, 117 insertions(+), 7 deletions(-) diff --git a/adapters/__init__.py b/adapters/__init__.py index 317cdec..83ab6f3 100644 --- a/adapters/__init__.py +++ b/adapters/__init__.py @@ -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)) diff --git a/pipeline.py b/pipeline.py index c7bdd5f..16b723f 100644 --- a/pipeline.py +++ b/pipeline.py @@ -37,7 +37,7 @@ ADAPTERS = { } # Default enabled sources -ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews", "huggingface"] +ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews", "huggingface", "rss"] def init_db(db_path: str, schema_path: str) -> sqlite3.Connection: @@ -252,9 +252,14 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo entries = adapter.fetch(limit=limit) except Exception as e: print(f" ✗ {source_name} failed: {e}") - source_stats[source_name] = {"fetched": 0, "stored": 0, "error": str(e)} + source_stats[source_name] = {"fetched": 0, "stored": 0, + "error": str(e), + "failure_class": "error"} continue + # Capture classification from the adapter (set by http_get on failure) + fc = getattr(adapter, "last_failure_class", None) + # Add adapter_version to metadata for entry in entries: meta = json.loads(entry["raw_metadata"]) if isinstance(entry["raw_metadata"], str) else entry["raw_metadata"] @@ -262,7 +267,8 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo entry["raw_metadata"] = json.dumps(meta) all_entries.extend(entries) - source_stats[source_name] = {"fetched": len(entries), "stored": 0} + source_stats[source_name] = {"fetched": len(entries), "stored": 0, + "failure_class": fc or "ok"} print(f" Fetched: {len(entries)} entries") # Small spacing between sources @@ -294,16 +300,32 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo # Zero-fetch (e.g. Reddit fully rate-limited) raises no exception but # is still a degraded run — record it so run_log can tell # "intermittent vs consistently-broken" apart over time. - zero = [s for s, st in source_stats.items() if st.get("fetched", 0) == 0 and not st.get("error")] + zero = [s for s, st in source_stats.items() + if st.get("fetched", 0) == 0 and not st.get("error")] notes_parts = [f"{s}: {st['error']}" for s, st in source_stats.items() if st.get("error")] if zero: notes_parts.append(f"no-fetch (degraded): {', '.join(zero)}") notes = "; ".join(notes_parts) or "all sources ok" + + # Rollup failure_class (issue #2): most severe across sources. + # Priority: 5xx > 4xx > 429 > error > zero_fetch > ok + rank = {"5xx": 5, "4xx": 4, "429": 3, "error": 2, "zero_fetch": 1, "ok": 0} + classes = [st.get("failure_class", "ok") for st in source_stats.values()] + if any(c in ("5xx", "4xx", "429", "error") for c in classes): + run_fc = max((c for c in classes if c in rank), + key=lambda c: rank[c]) + elif zero: + run_fc = "zero_fetch" + else: + run_fc = "ok" + try: conn.execute(""" - INSERT INTO run_log (total_fetched, total_stored, sources_ok, sources_failed, notes) - VALUES (?, ?, ?, ?, ?) - """, (len(all_entries), stored, json.dumps(ok), json.dumps(failed), notes)) + INSERT INTO run_log (total_fetched, total_stored, sources_ok, + sources_failed, failure_class, notes) + VALUES (?, ?, ?, ?, ?, ?) + """, (len(all_entries), stored, json.dumps(ok), json.dumps(failed), + run_fc, notes)) conn.commit() except Exception as e: print(f" ⚠ run_log write failed: {e}") diff --git a/schema.sql b/schema.sql index 623ec96..700db4d 100644 --- a/schema.sql +++ b/schema.sql @@ -22,6 +22,8 @@ CREATE INDEX IF NOT EXISTS idx_entries_category ON entries(category_tags); -- Partial failures (e.g. Reddit rate-limited) are detectable here, not hidden -- as a "complete" run. Also enables future pruning decisions (entries older -- than N days with no re-fetch can be archived). +-- failure_class (issue #2): one of 4xx / 5xx / 429 / zero_fetch / ok, derived +-- from the real HTTP response via adapters.http_get, not guessed after the fact. CREATE TABLE IF NOT EXISTS run_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_time TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), @@ -29,6 +31,7 @@ CREATE TABLE IF NOT EXISTS run_log ( total_stored INTEGER DEFAULT 0, sources_ok TEXT, -- JSON list of sources that succeeded sources_failed TEXT, -- JSON list of sources that errored/skipped + failure_class TEXT, -- 4xx / 5xx / 429 / zero_fetch / ok notes TEXT );