diff --git a/adapters/arxiv.py b/adapters/arxiv.py index 419e329..e216444 100644 --- a/adapters/arxiv.py +++ b/adapters/arxiv.py @@ -143,8 +143,8 @@ class ArxivAdapter(SourceAdapter): return papers - def _request(self, query: str, max_results: int = 20, sort_by="submittedDate") -> list[dict]: - """Make an arXiv API request.""" + def _request(self, query: str, max_results: int = 20, sort_by="submittedDate", retries: int = 3) -> list[dict]: + """Make an arXiv API request with retry on 429.""" url = ( f"{ARXIV_API}" f"?search_query={urllib.parse.quote(query)}" @@ -153,18 +153,29 @@ class ArxivAdapter(SourceAdapter): f"&max_results={max_results}" ) - req = urllib.request.Request(url, headers=browser_headers()) + for attempt in range(retries): + req = urllib.request.Request(url, headers=browser_headers()) + try: + with urllib.request.urlopen(req, timeout=15) as resp: + xml_data = resp.read().decode("utf-8") + return self._parse_atom(xml_data) + except urllib.error.HTTPError as e: + if e.code == 429 and attempt < retries - 1: + wait = (attempt + 1) * self.rate_limit + print(f" HTTP 429 for arXiv query, retrying in {wait}s...") + jitter_sleep(wait) + continue + print(f" HTTP {e.code} for arXiv query") + return [] + except Exception as e: + if attempt < retries - 1: + print(f" arXiv request error: {e}, retrying...") + jitter_sleep(2) + continue + print(f" arXiv request error: {e}") + return [] - 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") - return [] - except Exception as e: - print(f" arXiv request error: {e}") - return [] + return [] def _score(self, paper: dict, age_days: float) -> float: """Score based on AI-methodology relevance, not structural metadata. @@ -385,12 +396,10 @@ class ArxivAdapter(SourceAdapter): papers = self._request(query, max_results=limit, sort_by="submittedDate") all_papers.extend(papers) else: - # Fetch from each configured category - for cat in self.categories: - q = f"cat:{cat}" - cat_papers = self._request(q, max_results=limit, sort_by="submittedDate") - all_papers.extend(cat_papers) - jitter_sleep(self.rate_limit) + # Batch all categories into ONE request to stay within timeout budget + batch_query = " OR ".join(f"cat:{cat}" for cat in self.categories) + cat_papers = self._request(batch_query, max_results=limit * len(self.categories), sort_by="submittedDate") + all_papers.extend(cat_papers) # Deduplicate by arxiv_id seen = set() diff --git a/adapters/reddit.py b/adapters/reddit.py index af26c06..26c1645 100644 --- a/adapters/reddit.py +++ b/adapters/reddit.py @@ -114,10 +114,10 @@ class RedditAdapter(SourceAdapter): url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50" req = urllib.request.Request(url, headers={"User-Agent": self.user_agent}) - backoff = [3, 8] # staggered backoff: 3s then 8s + backoff = [2, 5] # staggered backoff: 2s then 5s for attempt in range(3): # max 3 attempts try: - with urllib.request.urlopen(req, timeout=10) as resp: + with urllib.request.urlopen(req, timeout=8) as resp: xml_data = resp.read().decode("utf-8") break except urllib.error.HTTPError as e: @@ -128,7 +128,7 @@ class RedditAdapter(SourceAdapter): if attempt < len(backoff): delay = backoff[attempt] print(f" RSS 429 for r/{subreddit}, retrying in {delay}s") - time.sleep(delay) + jitter_sleep(delay) continue print(f" RSS rate-limited for r/{subreddit}, skip") return [] @@ -330,10 +330,10 @@ class RedditAdapter(SourceAdapter): url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50" req = urllib.request.Request(url, headers={"User-Agent": self.user_agent}) - backoff = [3, 8] # staggered backoff: 3s then 8s + backoff = [2, 5] # staggered backoff: 2s then 5s for attempt in range(3): # max 3 attempts try: - with urllib.request.urlopen(req, timeout=10) as resp: + with urllib.request.urlopen(req, timeout=8) as resp: xml_data = resp.read().decode("utf-8") break except urllib.error.HTTPError as e: @@ -344,7 +344,7 @@ class RedditAdapter(SourceAdapter): if attempt < len(backoff): delay = backoff[attempt] print(f" RSS 429 for r/{subreddit}, retrying in {delay}s") - time.sleep(delay) + jitter_sleep(delay) continue print(f" RSS rate-limited for r/{subreddit}, skip") return [] diff --git a/oracle/cli.py b/oracle/cli.py index 41cdeb3..118e731 100644 --- a/oracle/cli.py +++ b/oracle/cli.py @@ -91,7 +91,7 @@ def cmd_ingest(args): entries = [] error = None try: - entries = adapter.fetch(limit=args.limit, timeout=10) + entries = adapter.fetch(limit=args.limit, timeout=30) except TypeError: # Old adapter signature without timeout param — use thread-based fallback import threading @@ -103,9 +103,9 @@ def cmd_ingest(args): result["error"] = str(e) t = threading.Thread(target=_fetch, daemon=True) t.start() - t.join(timeout=10) + t.join(timeout=30) if t.is_alive(): - error = f"timeout after 10s" + error = f"timeout after 30s" else: entries = result["entries"] error = result["error"] @@ -527,7 +527,7 @@ def main(): p_weekly.add_argument("--json", action="store_true", help="Output as JSON") # metrics - p_metrics = subparsers.add_parser("metrics", help="Pipeline metrics and adapter health") + p_metrics = sub.add_parser("metrics", help="Pipeline metrics and adapter health") p_metrics.add_argument("--db", default="oracle.db") p_metrics.add_argument("--days", type=int, default=7) p_metrics.add_argument("--adapters", action="store_true", help="Show adapter health table")