| 1 | """HTTP utilities for last30days skill (stdlib only).""" |
| 2 | |
| 3 | import json |
| 4 | from collections import OrderedDict |
| 5 | import math |
| 6 | import os |
| 7 | import random |
| 8 | import re |
| 9 | import socket |
| 10 | import sys |
| 11 | import threading |
| 12 | import time |
| 13 | import urllib.error |
| 14 | import urllib.request |
| 15 | from concurrent.futures import Future |
| 16 | from contextlib import contextmanager |
| 17 | from contextvars import ContextVar, copy_context |
| 18 | from pathlib import Path |
| 19 | from typing import Any, Dict, Optional, Union |
| 20 | from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit, quote |
| 21 | |
| 22 | from . import health |
| 23 | from . import log as _log |
| 24 | |
| 25 | DEFAULT_TIMEOUT = 30 |
| 26 | |
| 27 | |
| 28 | def log(msg: str): |
| 29 | """Log debug message to stderr.""" |
| 30 | _log.debug(msg) |
| 31 | |
| 32 | |
| 33 | MAX_RETRIES = 5 |
| 34 | MAX_429_RETRIES = 2 |
| 35 | RETRY_DELAY = 2.0 |
| 36 | |
| 37 | |
| 38 | # Longest a 429 retry may sleep on any host. Reddit's x-ratelimit-reset can say |
| 39 | # 540s and GitHub's is an epoch timestamp; neither is worth parking a worker |
| 40 | # (or the main thread) for. Past this bound the retry is not worth taking, so |
| 41 | # the caller's own fallback backoff applies and the request fails fast. |
| 42 | MAX_RETRY_DELAY_SECONDS = 60.0 |
| 43 | # A reset value this large is an absolute epoch timestamp, not delta-seconds. |
| 44 | _EPOCH_RESET_THRESHOLD = 100_000_000.0 |
| 45 | |
| 46 | |
| 47 | def retry_delay_from_headers(headers, fallback): |
| 48 | """Seconds to wait after a 429, read from whichever header the host sent. |
| 49 | |
| 50 | ``Retry-After`` is the standard, but Reddit's search/RSS endpoints answer an |
| 51 | anonymous 429 with ``x-ratelimit-reset`` (seconds until the window rolls) and |
| 52 | no ``Retry-After`` at all:: |
| 53 | |
| 54 | HTTP/2 429 |
| 55 | x-ratelimit-used: 1 |
| 56 | x-ratelimit-remaining: 0.0 |
| 57 | x-ratelimit-reset: 42 |
| 58 | |
| 59 | Reading only ``Retry-After`` means the caller falls back to exponential |
| 60 | backoff -- 3s, 5s, 9s -- every one of which is shorter than the ~42s Reddit |
| 61 | actually requires. Each retry re-429s, the budget drains, and the source is |
| 62 | reported dead when it was merely early. Honouring the reset header turns a |
| 63 | guaranteed zero into a result at the cost of one wait. |
| 64 | |
| 65 | Returns ``fallback`` when neither header is present or parseable. |
| 66 | """ |
| 67 | if not headers: |
| 68 | return fallback |
| 69 | for name in ("Retry-After", "x-ratelimit-reset"): |
| 70 | raw = headers.get(name) |
| 71 | if raw is None: |
| 72 | continue |
| 73 | try: |
| 74 | value = float(raw) |
| 75 | except (TypeError, ValueError): |
| 76 | continue |
| 77 | if value >= _EPOCH_RESET_THRESHOLD: |
| 78 | # GitHub-style absolute reset time. |
| 79 | value = value - time.time() |
| 80 | if value > 0: |
| 81 | return min(value, MAX_RETRY_DELAY_SECONDS) |
| 82 | return fallback |
| 83 | |
| 84 | # DNS resolution failures (gaierror) are transient — typically resolved by a |
| 85 | # brief backoff and retry. Use a dedicated minimum attempt count + exponential |
| 86 | # delays (1s, 2s, 4s) so callers that pass a small `retries` value still get a |
| 87 | # meaningful chance to recover from a transient resolution failure. |
| 88 | MIN_DNS_RETRIES = 3 |
| 89 | USER_AGENT = "last30days-skill/3.0 (Assistant Skill)" |
| 90 | |
| 91 | # urllib copies almost all headers across 3xx; strip credentials when origin changes (#1062). |
| 92 | _CROSS_ORIGIN_AUTH_HEADERS = frozenset( |
| 93 | {"authorization", "x-api-key", "x-csrf-token", "x-subscription-token"} |
| 94 | ) |
| 95 | |
| 96 | |
| 97 | def _request_origin(url: str) -> tuple[str, str, int]: |
| 98 | parts = urlsplit(url) |
| 99 | scheme = parts.scheme.lower() |
| 100 | host = (parts.hostname or "").lower() |
| 101 | if parts.port is not None: |
| 102 | port = parts.port |
| 103 | elif scheme == "https": |
| 104 | port = 443 |
| 105 | elif scheme == "http": |
| 106 | port = 80 |
| 107 | else: |
| 108 | port = 0 |
| 109 | return scheme, host, port |
| 110 | |
| 111 | |
| 112 | class _StripAuthOnCrossOriginRedirect(urllib.request.HTTPRedirectHandler): |
| 113 | def redirect_request(self, req, fp, code, msg, headers, newurl): |
| 114 | new = super().redirect_request(req, fp, code, msg, headers, newurl) |
| 115 | if new is None: |
| 116 | return None |
| 117 | if _request_origin(req.full_url) != _request_origin(new.full_url): |
| 118 | for store in (new.headers, getattr(new, "unredirected_hdrs", None)): |
| 119 | if not store: |
| 120 | continue |
| 121 | for name in list(store): |
| 122 | if name.lower() in _CROSS_ORIGIN_AUTH_HEADERS: |
| 123 | del store[name] |
| 124 | return new |
| 125 | |
| 126 | |
| 127 | _opener = urllib.request.build_opener(_StripAuthOnCrossOriginRedirect) |
| 128 | _DEFAULT_URLOPEN = urllib.request.urlopen |
| 129 | |
| 130 | |
| 131 | def _open_request(req, timeout): |
| 132 | """Honor test patches of urllib.request.urlopen; otherwise use the strip opener.""" |
| 133 | current = urllib.request.urlopen |
| 134 | if current is not _DEFAULT_URLOPEN: |
| 135 | return current(req, timeout=timeout) |
| 136 | return _opener.open(req, timeout=timeout) |
| 137 | |
| 138 | _failure_sink: ContextVar[Optional[list["HTTPError"]]] = ContextVar( |
| 139 | "last30days_http_failure_sink", |
| 140 | default=None, |
| 141 | ) |
| 142 | _expected_miss_statuses: ContextVar[frozenset[int]] = ContextVar( |
| 143 | "last30days_http_expected_miss_statuses", |
| 144 | default=frozenset(), |
| 145 | ) |
| 146 | |
| 147 | _FIXTURE_FORMAT = "last30days-http-fixture/v1" |
| 148 | _FIXTURE_SECRET_KEYS = frozenset( |
| 149 | { |
| 150 | "api_key", "apikey", "authorization", "cookie", "key", "secret", "token", |
| 151 | "password", "passwd", "passphrase", "credential", "bearer", "jwt", |
| 152 | } |
| 153 | ) |
| 154 | # Suffixes are matched on the normalized key, where camelCase collapses without |
| 155 | # a separator ("accessJwt" -> "accessjwt"), so these are bare rather than |
| 156 | # underscore-prefixed. "key" is deliberately absent: it would redact "monkey". |
| 157 | _FIXTURE_SECRET_KEY_SUFFIXES = ( |
| 158 | "_api_key", "apikey", "_authorization", "_cookie", "_secret", "_token", |
| 159 | "password", "passwd", "passphrase", "credential", "jwt", |
| 160 | ) |
| 161 | _fixture_lock = threading.Lock() |
| 162 | _fixture_state: Optional[dict[str, Any]] = None |
| 163 | _NO_FIXTURE = object() |
| 164 | _fixture_module_capture: ContextVar[bool] = ContextVar( |
| 165 | "last30days_fixture_module_capture", |
| 166 | default=False, |
| 167 | ) |
| 168 | |
| 169 | |
| 170 | def _is_secret_key(value: object) -> bool: |
| 171 | key = re.sub(r"[^a-z0-9]+", "_", str(value).lower()).strip("_") |
| 172 | return ( |
| 173 | key in _FIXTURE_SECRET_KEYS |
| 174 | or key.endswith(_FIXTURE_SECRET_KEY_SUFFIXES) |
| 175 | ) |
| 176 | |
| 177 | |
| 178 | def _scrub_fixture_value( |
| 179 | value: Any, |
| 180 | *, |
| 181 | key: str = "", |
| 182 | redactions: frozenset[str] = frozenset(), |
| 183 | ) -> Any: |
| 184 | """Remove credentials before a recorded exchange reaches disk.""" |
| 185 | if key and _is_secret_key(key): |
| 186 | return "<redacted>" |
| 187 | if isinstance(value, dict): |
| 188 | return { |
| 189 | str(child_key): _scrub_fixture_value( |
| 190 | child_value, |
| 191 | key=str(child_key), |
| 192 | redactions=redactions, |
| 193 | ) |
| 194 | for child_key, child_value in value.items() |
| 195 | } |
| 196 | if isinstance(value, list): |
| 197 | return [_scrub_fixture_value(item, redactions=redactions) for item in value] |
| 198 | if isinstance(value, str): |
| 199 | scrubbed = value |
| 200 | for secret in sorted(redactions, key=len, reverse=True): |
| 201 | if len(secret) >= 4: |
| 202 | scrubbed = scrubbed.replace(secret, "<redacted>") |
| 203 | return scrubbed |
| 204 | return value |
| 205 | |
| 206 | |
| 207 | _AUTH_SCHEME_RE = re.compile(r"^(?:bearer|basic|token)\s+(\S+)$", re.IGNORECASE) |
| 208 | |
| 209 | |
| 210 | def _collect_secret_values(value: Any, *, key: str = "") -> set[str]: |
| 211 | values: set[str] = set() |
| 212 | if key and _is_secret_key(key) and value not in (None, ""): |
| 213 | text = str(value) |
| 214 | values.add(text) |
| 215 | # "Bearer <token>": the bare token is what a response body or an |
| 216 | # adapter error message echoes, so redact it on its own too. |
| 217 | scheme = _AUTH_SCHEME_RE.match(text.strip()) |
| 218 | if scheme: |
| 219 | values.add(scheme.group(1)) |
| 220 | return values |
| 221 | if isinstance(value, dict): |
| 222 | for child_key, child_value in value.items(): |
| 223 | values.update(_collect_secret_values(child_value, key=str(child_key))) |
| 224 | elif isinstance(value, list): |
| 225 | for child in value: |
| 226 | values.update(_collect_secret_values(child)) |
| 227 | return values |
| 228 | |
| 229 | |
| 230 | def _fixture_redactions( |
| 231 | url: str, |
| 232 | headers: dict[str, str], |
| 233 | json_data: Optional[Dict[str, Any]], |
| 234 | ) -> frozenset[str]: |
| 235 | values: set[str] = set() |
| 236 | try: |
| 237 | for key, value in parse_qsl(urlsplit(url).query, keep_blank_values=True): |
| 238 | if _is_secret_key(key) and value: |
| 239 | values.add(value) |
| 240 | except ValueError: |
| 241 | pass |
| 242 | values.update(_collect_secret_values(headers)) |
| 243 | values.update(_collect_secret_values(json_data)) |
| 244 | # Session-wide secret values (process env plus resolved config) so a |
| 245 | # bearer loaded from .env, Keychain, or pass is scrubbed at the HTTP |
| 246 | # seam even when this request carried it only in a header. |
| 247 | with _fixture_lock: |
| 248 | state = _fixture_state |
| 249 | if state is not None and state.get("redactions"): |
| 250 | values.update(state["redactions"]) |
| 251 | return frozenset(values) |
| 252 | |
| 253 | |
| 254 | def config_secret_values(config: Dict[str, Any]) -> frozenset[str]: |
| 255 | """Secret VALUES from a resolved config, for fixture redaction. |
| 256 | |
| 257 | Every key the Keychain/pass loader knows (``env.KEYCHAIN_KEYS``) plus any |
| 258 | secret-named key is a credential regardless of which layer supplied it. |
| 259 | """ |
| 260 | values: set[str] = set() |
| 261 | try: |
| 262 | from . import env as _env |
| 263 | secret_keys = set(_env.KEYCHAIN_KEYS) |
| 264 | except Exception: # pragma: no cover - env is always importable |
| 265 | secret_keys = set() |
| 266 | for key, value in (config or {}).items(): |
| 267 | if not isinstance(value, str) or len(value) < 4: |
| 268 | continue |
| 269 | if "://" in value: |
| 270 | # An endpoint (e.g. an API base URL) is an address, not a credential. |
| 271 | continue |
| 272 | if key in secret_keys or _is_secret_key(key): |
| 273 | values.add(value) |
| 274 | return frozenset(values) |
| 275 | |
| 276 | |
| 277 | def add_fixture_redactions(values) -> None: |
| 278 | """Register secret values with the active recording session, if any. |
| 279 | |
| 280 | ``env.get_config`` calls this with the resolved config's secrets so a |
| 281 | bearer loaded from a file or a credential store is redacted at both the |
| 282 | HTTP seam and the module seam. A no-op outside a recording session. |
| 283 | """ |
| 284 | extra = {v for v in values if isinstance(v, str) and len(v) >= 4} |
| 285 | if not extra: |
| 286 | return |
| 287 | with _fixture_lock: |
| 288 | state = _fixture_state |
| 289 | if state is None or state["mode"] != "record": |
| 290 | return |
| 291 | state["redactions"] = frozenset(state.get("redactions") or frozenset()) | extra |
| 292 | |
| 293 | |
| 294 | def _scrub_fixture_url(url: str) -> str: |
| 295 | try: |
| 296 | parts = urlsplit(url) |
| 297 | query = urlencode( |
| 298 | [ |
| 299 | (key, "<redacted>" if _is_secret_key(key) else value) |
| 300 | for key, value in parse_qsl(parts.query, keep_blank_values=True) |
| 301 | ] |
| 302 | ) |
| 303 | return urlunsplit((parts.scheme, parts.netloc, parts.path, query, parts.fragment)) |
| 304 | except ValueError: |
| 305 | return url |
| 306 | |
| 307 | |
| 308 | def _fixture_request( |
| 309 | method: str, |
| 310 | url: str, |
| 311 | json_data: Optional[Dict[str, Any]], |
| 312 | raw: bool, |
| 313 | ) -> dict[str, Any]: |
| 314 | request_data: dict[str, Any] = { |
| 315 | "method": method.upper(), |
| 316 | "url": _scrub_fixture_url(url), |
| 317 | "raw": bool(raw), |
| 318 | } |
| 319 | if json_data is not None: |
| 320 | request_data["json"] = _scrub_fixture_value(json_data) |
| 321 | return request_data |
| 322 | |
| 323 | |
| 324 | def _fixture_key(request_data: dict[str, Any]) -> str: |
| 325 | return json.dumps(request_data, sort_keys=True, separators=(",", ":"), ensure_ascii=False) |
| 326 | |
| 327 | |
| 328 | @contextmanager |
| 329 | def recording_requests(path: str | Path): |
| 330 | """Record scrubbed HTTP exchanges to ``path`` for offline eval replay. |
| 331 | |
| 332 | This process-global session is deliberate: source requests run in worker |
| 333 | threads, so a ContextVar would not observe the complete pipeline fan-out. |
| 334 | Nested or concurrent recording/replay sessions are rejected. |
| 335 | """ |
| 336 | global _fixture_state |
| 337 | target = Path(path).expanduser() |
| 338 | if target.suffix.lower() != ".json": |
| 339 | target = target / "http.json" |
| 340 | with _fixture_lock: |
| 341 | if _fixture_state is not None: |
| 342 | raise RuntimeError("An HTTP fixture session is already active") |
| 343 | _fixture_state = { |
| 344 | "mode": "record", |
| 345 | "path": target, |
| 346 | "exchanges": [], |
| 347 | "source_exchanges": [], |
| 348 | # Secret VALUES from the environment, so module-seam recordings |
| 349 | # scrub tokens echoed inside normal string fields (adapter error |
| 350 | # messages, parsed item text), not just secret-named keys. The |
| 351 | # resolved config's secrets join via add_fixture_redactions once |
| 352 | # env.get_config runs inside the session. |
| 353 | "redactions": frozenset( |
| 354 | value |
| 355 | for key, value in os.environ.items() |
| 356 | if _is_secret_key(key) and isinstance(value, str) and len(value) >= 4 |
| 357 | ), |
| 358 | } |
| 359 | completed = False |
| 360 | try: |
| 361 | yield target |
| 362 | completed = True |
| 363 | finally: |
| 364 | with _fixture_lock: |
| 365 | state = _fixture_state |
| 366 | _fixture_state = None |
| 367 | if state is not None and completed: |
| 368 | target.parent.mkdir(parents=True, exist_ok=True) |
| 369 | payload = { |
| 370 | "format": _FIXTURE_FORMAT, |
| 371 | "exchanges": state["exchanges"], |
| 372 | "source_exchanges": state["source_exchanges"], |
| 373 | } |
| 374 | # A recorded exchange is credential-adjacent by construction: |
| 375 | # redaction is key-name driven, so an unrecognized key name leaves |
| 376 | # a real value on disk. Create the temp file 0600 at open time |
| 377 | # rather than chmod-ing after the write, or the credentials sit in |
| 378 | # a world-readable file for the length of the write (the parent |
| 379 | # directory is caller-supplied and not guaranteed private). |
| 380 | # Mirrors last30days.save_output. Unlink first so a stale or |
| 381 | # pre-planted temp file cannot be reused with its own wider mode -- |
| 382 | # O_CREAT does not alter the mode of an existing file. |
| 383 | temporary = target.with_name(f".{target.name}.tmp") |
| 384 | temporary.unlink(missing_ok=True) |
| 385 | fd = os.open( |
| 386 | temporary, |
| 387 | os.O_CREAT | os.O_EXCL | os.O_WRONLY, |
| 388 | 0o600, |
| 389 | ) |
| 390 | with os.fdopen(fd, "w", encoding="utf-8") as handle: |
| 391 | handle.write( |
| 392 | json.dumps(payload, indent=2, ensure_ascii=False) + "\n" |
| 393 | ) |
| 394 | temporary.replace(target) |
| 395 | |
| 396 | |
| 397 | @contextmanager |
| 398 | def fixture_module_capture(enabled: bool): |
| 399 | """Suppress nested HTTP recording when a whole adapter result is captured.""" |
| 400 | token = _fixture_module_capture.set(enabled) |
| 401 | try: |
| 402 | yield |
| 403 | finally: |
| 404 | _fixture_module_capture.reset(token) |
| 405 | |
| 406 | |
| 407 | @contextmanager |
| 408 | def replaying_requests(path: str | Path): |
| 409 | """Replay recorded exchanges and fail closed on any unrecorded request.""" |
| 410 | global _fixture_state |
| 411 | target = Path(path).expanduser() |
| 412 | if target.is_dir(): |
| 413 | target = target / "http.json" |
| 414 | payload = json.loads(target.read_text(encoding="utf-8")) |
| 415 | if payload.get("format") != _FIXTURE_FORMAT: |
| 416 | raise ValueError(f"Unsupported HTTP fixture format in {target}") |
| 417 | queues: dict[str, list[dict[str, Any]]] = {} |
| 418 | for exchange in payload.get("exchanges") or []: |
| 419 | queues.setdefault(_fixture_key(exchange["request"]), []).append(exchange["response"]) |
| 420 | source_queues: dict[str, list[Any]] = {} |
| 421 | for exchange in payload.get("source_exchanges") or []: |
| 422 | source_queues.setdefault(_fixture_key(exchange["request"]), []).append(exchange) |
| 423 | with _fixture_lock: |
| 424 | if _fixture_state is not None: |
| 425 | raise RuntimeError("An HTTP fixture session is already active") |
| 426 | _fixture_state = { |
| 427 | "mode": "replay", |
| 428 | "path": target, |
| 429 | "queues": queues, |
| 430 | "source_queues": source_queues, |
| 431 | } |
| 432 | try: |
| 433 | yield target |
| 434 | with _fixture_lock: |
| 435 | unused = sum(len(values) for values in queues.values()) + sum( |
| 436 | len(values) for values in source_queues.values() |
| 437 | ) |
| 438 | if unused: |
| 439 | raise AssertionError(f"HTTP fixture replay left {unused} unused exchange(s): {target}") |
| 440 | finally: |
| 441 | with _fixture_lock: |
| 442 | _fixture_state = None |
| 443 | |
| 444 | |
| 445 | def _fixture_replay(request_data: dict[str, Any]) -> Any: |
| 446 | with _fixture_lock: |
| 447 | state = _fixture_state |
| 448 | if state is None or state["mode"] != "replay": |
| 449 | return _NO_FIXTURE |
| 450 | queue = state["queues"].get(_fixture_key(request_data)) |
| 451 | if not queue: |
| 452 | raise AssertionError( |
| 453 | "Unrecorded HTTP request during fixture replay: " |
| 454 | f"{request_data['method']} {request_data['url']}" |
| 455 | ) |
| 456 | response = queue.pop(0) |
| 457 | if response.get("error"): |
| 458 | error = response["error"] |
| 459 | recorded_error = HTTPError( |
| 460 | str(error.get("message") or "Recorded HTTP error"), |
| 461 | status_code=error.get("status_code"), |
| 462 | body=error.get("body"), |
| 463 | outcome_state=error.get("outcome_state"), |
| 464 | ) |
| 465 | _raise(recorded_error) |
| 466 | return response.get("value") |
| 467 | |
| 468 | |
| 469 | def _fixture_record( |
| 470 | request_data: dict[str, Any], |
| 471 | *, |
| 472 | value: Any = None, |
| 473 | error: Optional["HTTPError"] = None, |
| 474 | redactions: frozenset[str] = frozenset(), |
| 475 | ) -> None: |
| 476 | if _fixture_module_capture.get(): |
| 477 | return |
| 478 | with _fixture_lock: |
| 479 | state = _fixture_state |
| 480 | if state is None or state["mode"] != "record": |
| 481 | return |
| 482 | # Union the session's env-derived secret VALUES in, so a credential |
| 483 | # echoed back inside an ordinary response field is scrubbed on this |
| 484 | # path too, not only on the source-record path. Response scrubbing does |
| 485 | # not feed _fixture_key, so this cannot make a replay key |
| 486 | # machine-dependent. |
| 487 | redactions = redactions | (state.get("redactions") or frozenset()) |
| 488 | response: dict[str, Any] |
| 489 | if error is None: |
| 490 | response = {"value": _scrub_fixture_value(value, redactions=redactions)} |
| 491 | else: |
| 492 | response = { |
| 493 | "error": _scrub_fixture_value( |
| 494 | { |
| 495 | "message": str(error), |
| 496 | "status_code": error.status_code, |
| 497 | "body": error.body, |
| 498 | "outcome_state": error.outcome_state, |
| 499 | }, |
| 500 | redactions=redactions, |
| 501 | ) |
| 502 | } |
| 503 | state["exchanges"].append({"request": request_data, "response": response}) |
| 504 | |
| 505 | |
| 506 | def fixture_source_replay(request_data: dict[str, Any]) -> tuple[bool, Any]: |
| 507 | """Return a recorded CLI-backed source result when replay is active.""" |
| 508 | scrubbed = _scrub_fixture_value(request_data) |
| 509 | with _fixture_lock: |
| 510 | state = _fixture_state |
| 511 | if state is None or state["mode"] != "replay": |
| 512 | return False, None |
| 513 | queue = state["source_queues"].get(_fixture_key(scrubbed)) |
| 514 | if not queue: |
| 515 | raise AssertionError( |
| 516 | "Unrecorded CLI-backed source request during fixture replay: " |
| 517 | f"{request_data.get('source', 'unknown')}" |
| 518 | ) |
| 519 | exchange = queue.pop(0) |
| 520 | if exchange.get("type") == "error": |
| 521 | error = exchange.get("error") or {} |
| 522 | raise RecordedSourceError( |
| 523 | str(error.get("message") or "Recorded source error"), |
| 524 | exception_type=str(error.get("exception_type") or "Exception"), |
| 525 | outcome_state=error.get("outcome_state"), |
| 526 | ) |
| 527 | return True, exchange.get("value") |
| 528 | |
| 529 | |
| 530 | def fixture_source_record(request_data: dict[str, Any], value: Any) -> None: |
| 531 | """Record the parsed output of a source adapter that bypasses http.py.""" |
| 532 | with _fixture_lock: |
| 533 | state = _fixture_state |
| 534 | if state is None or state["mode"] != "record": |
| 535 | return |
| 536 | session_redactions = state.get("redactions") or frozenset() |
| 537 | state["source_exchanges"].append( |
| 538 | { |
| 539 | "request": _scrub_fixture_value(request_data, redactions=session_redactions), |
| 540 | "value": _scrub_fixture_value(value, redactions=session_redactions), |
| 541 | } |
| 542 | ) |
| 543 | |
| 544 | |
| 545 | def fixture_source_record_error(request_data: dict[str, Any], error: Exception) -> None: |
| 546 | """Record a replayable failure from a source adapter that bypasses http.py.""" |
| 547 | with _fixture_lock: |
| 548 | state = _fixture_state |
| 549 | if state is None or state["mode"] != "record": |
| 550 | return |
| 551 | session_redactions = state.get("redactions") or frozenset() |
| 552 | state["source_exchanges"].append( |
| 553 | { |
| 554 | "request": _scrub_fixture_value(request_data, redactions=session_redactions), |
| 555 | "type": "error", |
| 556 | "error": _scrub_fixture_value( |
| 557 | { |
| 558 | "exception_type": type(error).__name__, |
| 559 | "message": str(error), |
| 560 | "outcome_state": getattr(error, "outcome_state", None), |
| 561 | } |
| 562 | , redactions=session_redactions), |
| 563 | } |
| 564 | ) |
| 565 | |
| 566 | |
| 567 | class RecordedSourceError(RuntimeError): |
| 568 | """Failure restored from a recorded module-backed source exchange.""" |
| 569 | |
| 570 | def __init__( |
| 571 | self, |
| 572 | message: str, |
| 573 | *, |
| 574 | exception_type: str, |
| 575 | outcome_state: Optional[str] = None, |
| 576 | ): |
| 577 | super().__init__(message) |
| 578 | self.exception_type = exception_type |
| 579 | self.outcome_state = outcome_state |
| 580 | |
| 581 | |
| 582 | def _is_dns_failure(err: urllib.error.URLError) -> bool: |
| 583 | """Return True if a URLError was caused by DNS resolution (gaierror).""" |
| 584 | return isinstance(getattr(err, "reason", None), socket.gaierror) |
| 585 | |
| 586 | |
| 587 | class HTTPError(Exception): |
| 588 | """HTTP request error with status code.""" |
| 589 | def __init__( |
| 590 | self, |
| 591 | message: str, |
| 592 | status_code: Optional[int] = None, |
| 593 | body: Optional[str] = None, |
| 594 | outcome_state: Optional[str] = None, |
| 595 | ): |
| 596 | super().__init__(message) |
| 597 | self.status_code = status_code |
| 598 | self.body = body |
| 599 | self.outcome_state = outcome_state or classify_failure( |
| 600 | status_code=status_code, |
| 601 | message=message, |
| 602 | ) |
| 603 | |
| 604 | |
| 605 | class DeadlineExceeded(HTTPError): |
| 606 | """The caller's shared wall deadline expired across request retries.""" |
| 607 | |
| 608 | def __init__(self): |
| 609 | super().__init__( |
| 610 | "Request deadline exceeded", |
| 611 | outcome_state=health.TIMEOUT, |
| 612 | ) |
| 613 | |
| 614 | |
| 615 | @contextmanager |
| 616 | def capture_failures(): |
| 617 | """Capture terminal request failures in the current retrieval context. |
| 618 | |
| 619 | Source modules historically catch ``HTTPError`` and return an empty result. |
| 620 | The context-local sink lets the pipeline retain that failure without shared |
| 621 | mutable state across its worker threads. |
| 622 | """ |
| 623 | failures: list[HTTPError] = [] |
| 624 | token = _failure_sink.set(failures) |
| 625 | try: |
| 626 | yield failures |
| 627 | finally: |
| 628 | _failure_sink.reset(token) |
| 629 | |
| 630 | |
| 631 | @contextmanager |
| 632 | def tee_failures(): |
| 633 | """Observe failures locally WITHOUT hiding them from the enclosing sink. |
| 634 | |
| 635 | ``capture_failures()`` *replaces* the context-local sink, so nesting it |
| 636 | inside a retrieval context swallows the very failure the pipeline needs. |
| 637 | This yields a local list and forwards its contents to the parent sink on |
| 638 | exit, so a swallow site (``get_text`` returns None and drops the status) |
| 639 | can recover what it lost while the pipeline still sees the failure. |
| 640 | """ |
| 641 | parent = _failure_sink.get() |
| 642 | local: list[HTTPError] = [] |
| 643 | token = _failure_sink.set(local) |
| 644 | try: |
| 645 | yield local |
| 646 | finally: |
| 647 | _failure_sink.reset(token) |
| 648 | if parent is not None: |
| 649 | parent.extend(local) |
| 650 | |
| 651 | |
| 652 | @contextmanager |
| 653 | def expected_misses(*status_codes: int): |
| 654 | """Exclude adapter-declared probe misses from captured run failures.""" |
| 655 | token = _expected_miss_statuses.set( |
| 656 | _expected_miss_statuses.get().union(status_codes) |
| 657 | ) |
| 658 | try: |
| 659 | yield |
| 660 | finally: |
| 661 | _expected_miss_statuses.reset(token) |
| 662 | |
| 663 | |
| 664 | def submit_with_context(executor, func, /, *args, **kwargs) -> Future: |
| 665 | """Submit a worker with the caller's failure-capture context.""" |
| 666 | context = copy_context() |
| 667 | return executor.submit(context.run, func, *args, **kwargs) |
| 668 | |
| 669 | |
| 670 | def _record_failure(error: HTTPError) -> None: |
| 671 | if error.status_code in _expected_miss_statuses.get(): |
| 672 | return |
| 673 | sink = _failure_sink.get() |
| 674 | if sink is not None: |
| 675 | sink.append(error) |
| 676 | |
| 677 | |
| 678 | def _raise(error: HTTPError) -> None: |
| 679 | _record_failure(error) |
| 680 | raise error |
| 681 | |
| 682 | |
| 683 | def classify_failure(*, status_code: Optional[int] = None, message: str = "") -> str: |
| 684 | """Map a request failure to the doctor-aligned per-run vocabulary.""" |
| 685 | text = message.lower() |
| 686 | if status_code == 429 or any( |
| 687 | marker in text for marker in ("http 429", "status 429", "rate limit", "too many requests") |
| 688 | ): |
| 689 | return health.RATE_LIMITED |
| 690 | # Credit exhaustion is checked before the auth branch: a 402 (or a body |
| 691 | # saying the account has no credits) asks the user to top up, not to |
| 692 | # re-authenticate. Markers stay narrow on purpose: the bare word "credits" |
| 693 | # is not one ("10,000 free credits" is onboarding copy, not a failure). |
| 694 | if status_code == 402 or any( |
| 695 | marker in text |
| 696 | for marker in ( |
| 697 | "http 402", |
| 698 | "status 402", |
| 699 | "payment required", |
| 700 | "insufficient credits", |
| 701 | "does not have any credits", |
| 702 | "out of credits", |
| 703 | ) |
| 704 | ): |
| 705 | return health.PAYMENT_REQUIRED |
| 706 | if status_code in (401, 403) or any( |
| 707 | marker in text |
| 708 | for marker in ( |
| 709 | "http 401", |
| 710 | "http 403", |
| 711 | "status 401", |
| 712 | "status 403", |
| 713 | "unauthorized", |
| 714 | "forbidden", |
| 715 | "authentication failed", |
| 716 | "expired token", |
| 717 | "not signed in", |
| 718 | "not logged in", |
| 719 | "invalid_grant", |
| 720 | "refresh token", |
| 721 | "session expired", |
| 722 | "grok session expired", |
| 723 | ) |
| 724 | ): |
| 725 | return health.AUTH_FAILED |
| 726 | if status_code == 408 or "timed out" in text or "timeout" in text: |
| 727 | return health.TIMEOUT |
| 728 | if any( |
| 729 | marker in text |
| 730 | for marker in ( |
| 731 | "invalid json", |
| 732 | "json decode", |
| 733 | "schema", |
| 734 | "interstitial", |
| 735 | "non-json", |
| 736 | ) |
| 737 | ): |
| 738 | return health.SCHEMA_DRIFT |
| 739 | if any( |
| 740 | marker in text |
| 741 | for marker in ( |
| 742 | "url error", |
| 743 | "connection error", |
| 744 | "connection refused", |
| 745 | "connection reset", |
| 746 | "name or service not known", |
| 747 | "temporary failure in name resolution", |
| 748 | "nodename nor servname", |
| 749 | "dns", |
| 750 | "network is unreachable", |
| 751 | ) |
| 752 | ): |
| 753 | return health.UNREACHABLE |
| 754 | return health.ERROR |
| 755 | |
| 756 | |
| 757 | def request( |
| 758 | method: str, |
| 759 | url: str, |
| 760 | headers: Optional[Dict[str, str]] = None, |
| 761 | json_data: Optional[Dict[str, Any]] = None, |
| 762 | params: Optional[Dict[str, Any]] = None, |
| 763 | timeout: float = DEFAULT_TIMEOUT, |
| 764 | retries: int = MAX_RETRIES, |
| 765 | max_429_retries: int = MAX_429_RETRIES, |
| 766 | raw: bool = False, |
| 767 | deadline_monotonic: float | None = None, |
| 768 | ) -> Union[Dict[str, Any], str]: |
| 769 | """Make an HTTP request and return JSON response. |
| 770 | |
| 771 | Args: |
| 772 | method: HTTP method (GET, POST, etc.) |
| 773 | url: Request URL |
| 774 | headers: Optional headers dict |
| 775 | json_data: Optional JSON body (for POST) |
| 776 | params: Optional query-string params. Values are stringified. None values |
| 777 | are dropped. If ``url`` already has a query string, ``params`` is appended. |
| 778 | timeout: Request timeout in seconds |
| 779 | retries: Number of retries on failure |
| 780 | max_429_retries: Maximum 429 retries before giving up (separate cap) |
| 781 | raw: If True, return raw response text instead of parsed JSON |
| 782 | deadline_monotonic: Optional absolute monotonic deadline shared by all |
| 783 | attempts and retry delays. |
| 784 | |
| 785 | Returns: |
| 786 | Parsed JSON response as dict, or raw text string if raw=True. |
| 787 | |
| 788 | Raises: |
| 789 | HTTPError: On request failure |
| 790 | """ |
| 791 | headers = headers or {} |
| 792 | headers.setdefault("User-Agent", USER_AGENT) |
| 793 | |
| 794 | if params: |
| 795 | filtered = {k: str(v) for k, v in params.items() if v is not None} |
| 796 | if filtered: |
| 797 | separator = "&" if ("?" in url) else "?" |
| 798 | url = f"{url}{separator}{urlencode(filtered)}" |
| 799 | # Encode any non-ASCII characters to prevent UnicodeEncodeError from |
| 800 | # http.client.HTTPConnection.putrequest (which uses latin-1 internally). |
| 801 | # Only encode path, query, and fragment — not the hostname (netloc), which |
| 802 | # needs IDNA encoding instead of percent-encoding for non-ASCII domains. |
| 803 | parts = urlsplit(url) |
| 804 | safe = '/:@!$&\'()*+,;=-._~%?#[]=+' |
| 805 | url = urlunsplit(( |
| 806 | parts.scheme, |
| 807 | parts.netloc, |
| 808 | quote(parts.path, safe=safe), |
| 809 | quote(parts.query, safe=safe), |
| 810 | quote(parts.fragment, safe=safe), |
| 811 | )) |
| 812 | |
| 813 | fixture_request = _fixture_request(method, url, json_data, raw) |
| 814 | fixture_redactions = _fixture_redactions(url, headers, json_data) |
| 815 | replayed = _fixture_replay(fixture_request) |
| 816 | if replayed is not _NO_FIXTURE: |
| 817 | return replayed |
| 818 | |
| 819 | data = None |
| 820 | if json_data is not None: |
| 821 | data = json.dumps(json_data).encode('utf-8') |
| 822 | headers.setdefault("Content-Type", "application/json") |
| 823 | |
| 824 | req = urllib.request.Request(url, data=data, headers=headers, method=method) |
| 825 | |
| 826 | safe_url = re.sub(r'([?&])(key|api_key|token|secret)=[^&]*', r'\1\2=***', url) |
| 827 | log(f"{method} {safe_url}") |
| 828 | |
| 829 | last_error = None |
| 830 | rate_limit_count = 0 |
| 831 | # DNS failures get a dedicated minimum attempt count + exponential backoff. |
| 832 | # `effective_retries` is the actual loop bound; we expand it on the first |
| 833 | # gaierror if the caller passed a smaller `retries` value than MIN_DNS_RETRIES. |
| 834 | effective_retries = retries |
| 835 | dns_attempts = 0 |
| 836 | attempt = 0 |
| 837 | |
| 838 | def raise_recorded(error: HTTPError) -> None: |
| 839 | _fixture_record(fixture_request, error=error, redactions=fixture_redactions) |
| 840 | _raise(error) |
| 841 | |
| 842 | def deadline_error() -> HTTPError: |
| 843 | return DeadlineExceeded() |
| 844 | |
| 845 | def sleep_before_retry(delay: float) -> bool: |
| 846 | """Sleep only when the full delay fits inside the caller's deadline.""" |
| 847 | nonlocal last_error |
| 848 | if deadline_monotonic is not None: |
| 849 | remaining = deadline_monotonic - time.monotonic() |
| 850 | if remaining <= 0 or delay >= remaining: |
| 851 | last_error = deadline_error() |
| 852 | return False |
| 853 | time.sleep(delay) |
| 854 | return True |
| 855 | |
| 856 | def open_and_read(request_timeout: float) -> tuple[int, str]: |
| 857 | with _open_request(req, request_timeout) as response: |
| 858 | return response.status, response.read().decode('utf-8') |
| 859 | |
| 860 | def open_and_read_before_deadline( |
| 861 | request_timeout: float, |
| 862 | ) -> tuple[int, str]: |
| 863 | """Stop waiting at the wall deadline, even during DNS or body reads.""" |
| 864 | if deadline_monotonic is None: |
| 865 | return open_and_read(request_timeout) |
| 866 | remaining = deadline_monotonic - time.monotonic() |
| 867 | if remaining <= 0: |
| 868 | raise deadline_error() |
| 869 | future: Future = Future() |
| 870 | |
| 871 | def worker() -> None: |
| 872 | try: |
| 873 | future.set_result(open_and_read(request_timeout)) |
| 874 | except BaseException as exc: |
| 875 | future.set_exception(exc) |
| 876 | |
| 877 | threading.Thread(target=worker, daemon=True).start() |
| 878 | try: |
| 879 | return future.result(timeout=remaining) |
| 880 | except TimeoutError as exc: |
| 881 | # A worker-side socket TimeoutError is a transport failure, not |
| 882 | # proof that the command-wide wall deadline expired. Re-read a |
| 883 | # completed future so its original exception reaches the normal |
| 884 | # transport classifier below. |
| 885 | if future.done(): |
| 886 | return future.result() |
| 887 | raise deadline_error() from exc |
| 888 | |
| 889 | while attempt < effective_retries: |
| 890 | request_timeout = timeout |
| 891 | if deadline_monotonic is not None: |
| 892 | remaining = deadline_monotonic - time.monotonic() |
| 893 | if remaining <= 0: |
| 894 | last_error = deadline_error() |
| 895 | break |
| 896 | request_timeout = min(timeout, remaining) |
| 897 | try: |
| 898 | response_status, body = open_and_read_before_deadline(request_timeout) |
| 899 | if ( |
| 900 | deadline_monotonic is not None |
| 901 | and time.monotonic() >= deadline_monotonic |
| 902 | ): |
| 903 | raise_recorded(deadline_error()) |
| 904 | log(f"Response: {response_status} ({len(body)} bytes)") |
| 905 | if raw: |
| 906 | _fixture_record(fixture_request, value=body, redactions=fixture_redactions) |
| 907 | return body |
| 908 | parsed = json.loads(body) if body else {} |
| 909 | _fixture_record(fixture_request, value=parsed, redactions=fixture_redactions) |
| 910 | return parsed |
| 911 | except DeadlineExceeded as exc: |
| 912 | raise_recorded(exc) |
| 913 | except urllib.error.HTTPError as e: |
| 914 | body = None |
| 915 | try: |
| 916 | body = e.read().decode('utf-8') |
| 917 | except (OSError, UnicodeDecodeError): |
| 918 | pass |
| 919 | log(f"HTTP Error {e.code}: {e.reason}") |
| 920 | if body: |
| 921 | snippet = " ".join(body.split()) |
| 922 | log(f"Error body: {snippet[:200]}") |
| 923 | last_error = HTTPError(f"HTTP {e.code}: {e.reason}", e.code, body) |
| 924 | |
| 925 | # Don't retry client errors (4xx) except rate limits |
| 926 | if 400 <= e.code < 500 and e.code != 429: |
| 927 | raise_recorded(last_error) |
| 928 | |
| 929 | # Cap 429 retries separately to avoid wasting latency |
| 930 | if e.code == 429: |
| 931 | rate_limit_count += 1 |
| 932 | if rate_limit_count >= max_429_retries: |
| 933 | raise_recorded(last_error) |
| 934 | |
| 935 | # HTTP errors respect the caller's original `retries`; only DNS |
| 936 | # failures get the widened `effective_retries` budget. |
| 937 | if attempt < retries - 1: |
| 938 | if e.code == 429: |
| 939 | # Respect Retry-After or x-ratelimit-reset (Reddit sends the |
| 940 | # latter), falling back to exponential backoff: 3s, 5s, 9s... |
| 941 | delay = retry_delay_from_headers( |
| 942 | getattr(e, "headers", None), |
| 943 | RETRY_DELAY * (2 ** attempt) + 1, |
| 944 | ) |
| 945 | log(f"Rate limited (429). Waiting {delay:.1f}s before retry {attempt + 2}/{retries}") |
| 946 | else: |
| 947 | delay = RETRY_DELAY * (2 ** attempt) |
| 948 | if not sleep_before_retry(delay): |
| 949 | break |
| 950 | else: |
| 951 | # Caller's original retry budget exhausted; an earlier DNS |
| 952 | # failure may have widened `effective_retries`, but that |
| 953 | # widening is DNS-only — don't grant extra HTTP attempts. |
| 954 | break |
| 955 | except urllib.error.URLError as e: |
| 956 | log(f"URL Error: {e.reason}") |
| 957 | reason = getattr(e, "reason", None) |
| 958 | # urllib commonly wraps socket.timeout (an alias of TimeoutError |
| 959 | # since 3.10) in URLError; classify those as timeouts, not |
| 960 | # unreachable hosts, so the recovery guidance is right. |
| 961 | wrapped_timeout = isinstance(reason, TimeoutError) or "timed out" in str(reason).lower() |
| 962 | last_error = HTTPError( |
| 963 | f"URL Error: {e.reason}", |
| 964 | outcome_state=health.TIMEOUT if wrapped_timeout else health.UNREACHABLE, |
| 965 | ) |
| 966 | if _is_dns_failure(e): |
| 967 | # DNS resolution failures are transient; expand the retry budget |
| 968 | # to MIN_DNS_RETRIES if the caller passed fewer, and use |
| 969 | # exponential backoff (1s, 2s, 4s, ...) instead of the linear |
| 970 | # default. Counts DNS attempts separately so other URLError |
| 971 | # causes don't bypass the regular retry budget. |
| 972 | dns_attempts += 1 |
| 973 | if effective_retries < MIN_DNS_RETRIES: |
| 974 | log( |
| 975 | f"DNS resolution failed; expanding retry budget from " |
| 976 | f"{effective_retries} to {MIN_DNS_RETRIES}" |
| 977 | ) |
| 978 | effective_retries = MIN_DNS_RETRIES |
| 979 | if attempt < effective_retries - 1: |
| 980 | delay = 2 ** (dns_attempts - 1) # 1s, 2s, 4s, 8s, ... |
| 981 | log( |
| 982 | f"DNS resolution failure (attempt {dns_attempts}); " |
| 983 | f"retrying in {delay:.1f}s" |
| 984 | ) |
| 985 | if not sleep_before_retry(delay): |
| 986 | break |
| 987 | elif attempt < retries - 1: |
| 988 | # Non-DNS URLError (e.g. ConnectionRefused) respects the |
| 989 | # caller's original retry budget, not the DNS-widened bound. |
| 990 | if not sleep_before_retry(RETRY_DELAY * (attempt + 1)): |
| 991 | break |
| 992 | else: |
| 993 | # Caller's original retry budget exhausted; an earlier DNS |
| 994 | # failure widening `effective_retries` does not carry over |
| 995 | # to non-DNS error paths. |
| 996 | break |
| 997 | except json.JSONDecodeError as e: |
| 998 | log(f"JSON decode error: {e}") |
| 999 | last_error = HTTPError( |
| 1000 | f"Invalid JSON response: {e}", |
| 1001 | outcome_state=health.SCHEMA_DRIFT, |
| 1002 | ) |
| 1003 | raise_recorded(last_error) |
| 1004 | except (OSError, TimeoutError, ConnectionResetError) as e: |
| 1005 | # Handle socket-level errors (connection reset, timeout, etc.) |
| 1006 | log(f"Connection error: {type(e).__name__}: {e}") |
| 1007 | state = health.TIMEOUT if isinstance(e, TimeoutError) else health.UNREACHABLE |
| 1008 | last_error = HTTPError( |
| 1009 | f"Connection error: {type(e).__name__}: {e}", |
| 1010 | outcome_state=state, |
| 1011 | ) |
| 1012 | if attempt < retries - 1: |
| 1013 | # Socket errors respect the caller's original retry budget. |
| 1014 | if not sleep_before_retry(RETRY_DELAY * (attempt + 1)): |
| 1015 | break |
| 1016 | else: |
| 1017 | # Original budget exhausted; DNS widening doesn't apply here. |
| 1018 | break |
| 1019 | |
| 1020 | attempt += 1 |
| 1021 | |
| 1022 | if last_error: |
| 1023 | raise_recorded(last_error) |
| 1024 | error = HTTPError("Request failed with no error details") |
| 1025 | raise_recorded(error) |
| 1026 | |
| 1027 | |
| 1028 | def get(url: str, headers: Optional[Dict[str, str]] = None, **kwargs) -> Dict[str, Any]: |
| 1029 | """Make a GET request.""" |
| 1030 | return request("GET", url, headers=headers, **kwargs) |
| 1031 | |
| 1032 | |
| 1033 | def post(url: str, json_data: Dict[str, Any], headers: Optional[Dict[str, str]] = None, **kwargs) -> Dict[str, Any]: |
| 1034 | """Make a POST request with JSON body.""" |
| 1035 | return request("POST", url, headers=headers, json_data=json_data, **kwargs) |
| 1036 | |
| 1037 | |
| 1038 | def post_raw(url: str, json_data: Dict[str, Any], headers: Optional[Dict[str, str]] = None, **kwargs) -> str: |
| 1039 | """Make a POST request with JSON body and return raw text.""" |
| 1040 | return request("POST", url, headers=headers, json_data=json_data, raw=True, **kwargs) |
| 1041 | |
| 1042 | |
| 1043 | BROWSER_USER_AGENT = ( |
| 1044 | "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " |
| 1045 | "AppleWebKit/537.36 (KHTML, like Gecko) " |
| 1046 | "Chrome/124.0.0.0 Safari/537.36" |
| 1047 | ) |
| 1048 | |
| 1049 | |
| 1050 | def get_text( |
| 1051 | url: str, |
| 1052 | timeout: int = DEFAULT_TIMEOUT, |
| 1053 | retries: int = 2, |
| 1054 | accept: str = "*/*", |
| 1055 | headers: Optional[Dict[str, str]] = None, |
| 1056 | ) -> Optional[str]: |
| 1057 | """Fetch a URL and return decoded text, or None on any failure. |
| 1058 | |
| 1059 | Keyless helper for Reddit RSS and shreddit HTML endpoints — the free path |
| 1060 | that replaced the now-403 ``.json`` endpoints. Sends a browser User-Agent |
| 1061 | and never raises: returns None on HTTP error, network failure, or timeout |
| 1062 | so tiered callers can fall through to the next source. |
| 1063 | |
| 1064 | Args: |
| 1065 | url: Request URL |
| 1066 | timeout: HTTP timeout per attempt in seconds |
| 1067 | retries: Number of retries on failure (kept low — these tiers fail fast) |
| 1068 | accept: Accept header value (e.g. "application/atom+xml", "text/html") |
| 1069 | headers: Optional extra headers merged over the defaults |
| 1070 | |
| 1071 | Returns: |
| 1072 | Decoded response body as text, or None on failure. |
| 1073 | """ |
| 1074 | merged = { |
| 1075 | "User-Agent": BROWSER_USER_AGENT, |
| 1076 | "Accept": accept, |
| 1077 | "Accept-Language": "en-US,en;q=0.9", |
| 1078 | } |
| 1079 | if headers: |
| 1080 | merged.update(headers) |
| 1081 | try: |
| 1082 | return request( |
| 1083 | "GET", url, headers=merged, timeout=timeout, retries=retries, raw=True |
| 1084 | ) |
| 1085 | except HTTPError as e: |
| 1086 | log(f"get_text failed ({e}): {url}") |
| 1087 | return None |
| 1088 | |
| 1089 | |
| 1090 | class RateLimiter: |
| 1091 | """Thread-safe token-bucket throttle for an endpoint family. |
| 1092 | |
| 1093 | The keyless source tiers run under the pipeline's ThreadPoolExecutor, so a |
| 1094 | multi-subquery run can fire many requests at the same host at once. A bare |
| 1095 | per-request retry budget does not prevent that stampede — it only reacts |
| 1096 | after a 429. A token bucket bounds the *sustained* rate while still allowing |
| 1097 | a short burst, so legitimate parallelism is preserved (unlike a strict |
| 1098 | min-interval gate that would serialize every concurrent caller and could |
| 1099 | push later futures past their result timeouts). |
| 1100 | |
| 1101 | ``rate_per_sec`` tokens refill per second; ``burst`` is the bucket capacity |
| 1102 | (max simultaneous calls before throttling kicks in). The lock is released |
| 1103 | while sleeping so waiting threads don't serialize on each other. |
| 1104 | """ |
| 1105 | |
| 1106 | def __init__(self, rate_per_sec: float, burst: int | None = None): |
| 1107 | self.rate = rate_per_sec |
| 1108 | self.capacity = burst if burst is not None else max(1, int(rate_per_sec)) |
| 1109 | self._tokens = float(self.capacity) |
| 1110 | self._last = time.monotonic() |
| 1111 | self._lock = threading.Lock() |
| 1112 | # Threads currently blocked in acquire(). Callers that wait on a batch |
| 1113 | # of throttled futures size their timeouts from this queue depth. |
| 1114 | self._waiting = 0 |
| 1115 | |
| 1116 | @property |
| 1117 | def waiting(self) -> int: |
| 1118 | """Threads currently blocked in :meth:`acquire`.""" |
| 1119 | with self._lock: |
| 1120 | return self._waiting |
| 1121 | |
| 1122 | def acquire(self) -> None: |
| 1123 | """Consume one token, blocking only when the bucket is empty.""" |
| 1124 | queued = False |
| 1125 | try: |
| 1126 | while True: |
| 1127 | with self._lock: |
| 1128 | now = time.monotonic() |
| 1129 | # Clamp elapsed to >= 0: a backward clock reading must never |
| 1130 | # drive tokens negative (which would spin this loop forever). |
| 1131 | elapsed = max(0.0, now - self._last) |
| 1132 | self._tokens = min(self.capacity, self._tokens + elapsed * self.rate) |
| 1133 | self._last = now |
| 1134 | if self._tokens >= 1.0: |
| 1135 | self._tokens -= 1.0 |
| 1136 | return |
| 1137 | if not queued: |
| 1138 | self._waiting += 1 |
| 1139 | queued = True |
| 1140 | wait = (1.0 - self._tokens) / self.rate |
| 1141 | time.sleep(wait) |
| 1142 | finally: |
| 1143 | if queued: |
| 1144 | with self._lock: |
| 1145 | self._waiting -= 1 |
| 1146 | |
| 1147 | |
| 1148 | # Shared across all keyless Reddit tiers (RSS, listing, shreddit) so their |
| 1149 | # combined fan-out is throttled as one family. Burst lets the parallel |
| 1150 | # enrichment workers proceed; sustained rate caps the stampede. |
| 1151 | # 1 req/sec is slow enough that home IPs survive RSS + listing + shreddit |
| 1152 | # fan-out; raise LAST30DAYS_REDDIT_KEYLESS_RATE to trade 429s for wall-clock. |
| 1153 | REDDIT_KEYLESS_RATE_ENV = "LAST30DAYS_REDDIT_KEYLESS_RATE" |
| 1154 | DEFAULT_REDDIT_KEYLESS_RATE = 1.0 |
| 1155 | DEFAULT_REDDIT_KEYLESS_BURST = 2 |
| 1156 | _REDDIT_429_RETRY_SLEEP_SEC = 1.0 |
| 1157 | _REDDIT_429_RETRY_JITTER_SEC = 0.5 |
| 1158 | |
| 1159 | |
| 1160 | def parse_reddit_keyless_rate(raw: Optional[str]) -> float: |
| 1161 | """Parse LAST30DAYS_REDDIT_KEYLESS_RATE; invalid/non-positive -> default.""" |
| 1162 | text = (raw or "").strip() |
| 1163 | if not text: |
| 1164 | return DEFAULT_REDDIT_KEYLESS_RATE |
| 1165 | try: |
| 1166 | rate = float(text) |
| 1167 | except (TypeError, ValueError): |
| 1168 | return DEFAULT_REDDIT_KEYLESS_RATE |
| 1169 | if not math.isfinite(rate) or rate <= 0: |
| 1170 | return DEFAULT_REDDIT_KEYLESS_RATE |
| 1171 | return rate |
| 1172 | |
| 1173 | |
| 1174 | def make_reddit_keyless_limiter( |
| 1175 | environ: Optional[Dict[str, str]] = None, |
| 1176 | ) -> RateLimiter: |
| 1177 | envmap = os.environ if environ is None else environ |
| 1178 | return RateLimiter( |
| 1179 | rate_per_sec=parse_reddit_keyless_rate(envmap.get(REDDIT_KEYLESS_RATE_ENV)), |
| 1180 | burst=DEFAULT_REDDIT_KEYLESS_BURST, |
| 1181 | ) |
| 1182 | |
| 1183 | |
| 1184 | REDDIT_KEYLESS_LIMITER = make_reddit_keyless_limiter() |
| 1185 | |
| 1186 | |
| 1187 | def _sync_reddit_keyless_rate() -> None: |
| 1188 | """Apply a process-env override without resetting in-flight tokens.""" |
| 1189 | rate = parse_reddit_keyless_rate(os.environ.get(REDDIT_KEYLESS_RATE_ENV)) |
| 1190 | if REDDIT_KEYLESS_LIMITER.rate != rate: |
| 1191 | REDDIT_KEYLESS_LIMITER.rate = rate |
| 1192 | |
| 1193 | |
| 1194 | def _failures_are_429(failures: list[HTTPError]) -> bool: |
| 1195 | if not failures: |
| 1196 | return False |
| 1197 | last = failures[-1] |
| 1198 | return last.status_code == 429 or last.outcome_state == health.RATE_LIMITED |
| 1199 | |
| 1200 | |
| 1201 | def _sleep_reddit_429_retry() -> None: |
| 1202 | """Short jittered pause before the single in-lane 429 retry.""" |
| 1203 | time.sleep( |
| 1204 | _REDDIT_429_RETRY_SLEEP_SEC |
| 1205 | + random.uniform(0.0, _REDDIT_429_RETRY_JITTER_SEC) |
| 1206 | ) |
| 1207 | |
| 1208 | |
| 1209 | # Run-scoped memo for keyless Reddit GETs. Subreddit listing partials, listing |
| 1210 | # RSS feeds, arctic supplements, and shreddit comment pages depend only on the |
| 1211 | # subreddit and sort, and the Reddit lane is dispatched with the raw topic for |
| 1212 | # every subquery, so a four-subquery run requested each of them four times. |
| 1213 | # Memoizing successful bodies for the life of one command turns ~184 requests |
| 1214 | # into ~50 on the measured 2026-08-31 run shape. Concurrent requesters for the |
| 1215 | # same URL wait on the first fetch instead of issuing their own (all four |
| 1216 | # subquery streams start at once, so a result-only cache would miss). |
| 1217 | REDDIT_KEYLESS_MEMO_MAX = 512 |
| 1218 | _REDDIT_KEYLESS_MEMO: "OrderedDict[str, str]" = OrderedDict() |
| 1219 | _REDDIT_KEYLESS_INFLIGHT: Dict[str, threading.Event] = {} |
| 1220 | _REDDIT_KEYLESS_MEMO_LOCK = threading.Lock() |
| 1221 | |
| 1222 | |
| 1223 | # Queue depth only counts threads already blocked in acquire(). The other |
| 1224 | # lanes' workers submit their requests as they go, so a batch's last fetch can |
| 1225 | # start well after the depth seen at wait time. This flat allowance covers |
| 1226 | # that (the 2026-08-31 smoke lost three feeds at ~35s with the depth term |
| 1227 | # alone; a full run's ~50 distinct keyless requests take ~50s at 1 req/s). |
| 1228 | REDDIT_KEYLESS_CONTENTION_SECONDS = 45.0 |
| 1229 | |
| 1230 | |
| 1231 | def reddit_keyless_wait_allowance(batch_size: int) -> float: |
| 1232 | """Seconds a batch of *batch_size* throttled fetches may spend waiting for tokens. |
| 1233 | |
| 1234 | Every keyless Reddit lane in a run shares one bucket, so a lane's futures |
| 1235 | can sit behind other lanes' requests before their own fetch starts. Size |
| 1236 | per-future result timeouts as ``base + this`` instead of a fixed number; |
| 1237 | at 1 req/s a fixed 20-second timeout expired on real runs while the fetch |
| 1238 | was still queued (issue #985 follow-up). |
| 1239 | """ |
| 1240 | _sync_reddit_keyless_rate() |
| 1241 | limiter = REDDIT_KEYLESS_LIMITER |
| 1242 | rate = limiter.rate if limiter.rate > 0 else 1.0 |
| 1243 | return (limiter.waiting + max(0, batch_size)) / rate + REDDIT_KEYLESS_CONTENTION_SECONDS |
| 1244 | |
| 1245 | |
| 1246 | def reset_reddit_keyless_memo() -> None: |
| 1247 | """Forget memoized keyless Reddit bodies. Called once per command, and by tests.""" |
| 1248 | with _REDDIT_KEYLESS_MEMO_LOCK: |
| 1249 | _REDDIT_KEYLESS_MEMO.clear() |
| 1250 | _REDDIT_KEYLESS_INFLIGHT.clear() |
| 1251 | |
| 1252 | |
| 1253 | def _reddit_memo_get(url: str) -> Optional[str]: |
| 1254 | with _REDDIT_KEYLESS_MEMO_LOCK: |
| 1255 | text = _REDDIT_KEYLESS_MEMO.get(url) |
| 1256 | if text is not None: |
| 1257 | _REDDIT_KEYLESS_MEMO.move_to_end(url) |
| 1258 | return text |
| 1259 | |
| 1260 | |
| 1261 | def _reddit_memo_put(url: str, text: str) -> None: |
| 1262 | with _REDDIT_KEYLESS_MEMO_LOCK: |
| 1263 | _REDDIT_KEYLESS_MEMO[url] = text |
| 1264 | _REDDIT_KEYLESS_MEMO.move_to_end(url) |
| 1265 | while len(_REDDIT_KEYLESS_MEMO) > REDDIT_KEYLESS_MEMO_MAX: |
| 1266 | _REDDIT_KEYLESS_MEMO.popitem(last=False) |
| 1267 | |
| 1268 | |
| 1269 | def reddit_keyless_get_text( |
| 1270 | url: str, |
| 1271 | timeout: int = DEFAULT_TIMEOUT, |
| 1272 | retries: int = 2, |
| 1273 | accept: str = "*/*", |
| 1274 | headers: Optional[Dict[str, str]] = None, |
| 1275 | ) -> Optional[str]: |
| 1276 | """get_text for the keyless Reddit tiers, memoized per run and throttled. |
| 1277 | |
| 1278 | Same contract as :func:`get_text` (returns None on any failure) but a URL |
| 1279 | already fetched this command is served from the run memo without spending |
| 1280 | a limiter token, concurrent requesters for one URL share the in-flight |
| 1281 | fetch, and cold fetches are spaced via :data:`REDDIT_KEYLESS_LIMITER` so a |
| 1282 | broad multi-query run does not stampede Reddit's keyless endpoints. |
| 1283 | """ |
| 1284 | cached = _reddit_memo_get(url) |
| 1285 | if cached is not None: |
| 1286 | return cached |
| 1287 | # Elect one owner per URL. A waiter whose owner failed re-enters the |
| 1288 | # election rather than fetching un-gated, so a failed fetch costs one |
| 1289 | # retry for the whole group, not one per waiter. |
| 1290 | for _round in range(3): |
| 1291 | with _REDDIT_KEYLESS_MEMO_LOCK: |
| 1292 | cached = _REDDIT_KEYLESS_MEMO.get(url) |
| 1293 | if cached is not None: |
| 1294 | return cached |
| 1295 | gate = _REDDIT_KEYLESS_INFLIGHT.get(url) |
| 1296 | owner = gate is None |
| 1297 | if owner: |
| 1298 | gate = threading.Event() |
| 1299 | _REDDIT_KEYLESS_INFLIGHT[url] = gate |
| 1300 | if owner: |
| 1301 | break |
| 1302 | # The owner may itself be queued in the shared bucket; wait for that |
| 1303 | # queue, not just for one socket timeout. |
| 1304 | gate.wait( |
| 1305 | timeout=timeout * max(1, retries) + reddit_keyless_wait_allowance(1) |
| 1306 | ) |
| 1307 | cached = _reddit_memo_get(url) |
| 1308 | if cached is not None: |
| 1309 | return cached |
| 1310 | else: |
| 1311 | # Three failed owners in a row: give up quietly rather than pile on. |
| 1312 | return None |
| 1313 | try: |
| 1314 | _sync_reddit_keyless_rate() |
| 1315 | REDDIT_KEYLESS_LIMITER.acquire() |
| 1316 | text = get_text(url, timeout=timeout, retries=retries, accept=accept, headers=headers) |
| 1317 | if text is not None: |
| 1318 | _reddit_memo_put(url, text) |
| 1319 | return text |
| 1320 | finally: |
| 1321 | with _REDDIT_KEYLESS_MEMO_LOCK: |
| 1322 | _REDDIT_KEYLESS_INFLIGHT.pop(url, None) |
| 1323 | gate.set() |
| 1324 | |
| 1325 | |
| 1326 | def reddit_keyless_get_text_retry_429( |
| 1327 | url: str, |
| 1328 | timeout: int = DEFAULT_TIMEOUT, |
| 1329 | accept: str = "*/*", |
| 1330 | headers: Optional[Dict[str, str]] = None, |
| 1331 | ) -> tuple[Optional[str], Optional[str]]: |
| 1332 | """Limiter-throttled GET with one extra limiter-respecting retry on 429. |
| 1333 | |
| 1334 | Returns ``(body, error)``. The first attempt is captured locally so a |
| 1335 | recovered 429 is not left in the pipeline sink. A second 429, or any |
| 1336 | non-429 miss, is recorded as before. Internal ``get_text`` retries are |
| 1337 | skipped (``retries=1``) so the in-lane retry is the one that re-acquires |
| 1338 | the bucket. |
| 1339 | """ |
| 1340 | # retries=1 on purpose: letting request() sleep out a 42-60s |
| 1341 | # x-ratelimit-reset inside a lane worker starves the whole batch (the |
| 1342 | # 2026-08-31 smoke lost 14 feeds to future timeouts with retries=2 versus |
| 1343 | # 6 with 1). A keyless 429 fails fast, the lane retries once after a short |
| 1344 | # jittered pause through the bucket, and the memo keeps the other streams |
| 1345 | # from re-requesting the same URL. |
| 1346 | kwargs: Dict[str, Any] = { |
| 1347 | "timeout": timeout, |
| 1348 | "retries": 1, |
| 1349 | "accept": accept, |
| 1350 | "headers": headers, |
| 1351 | } |
| 1352 | with capture_failures() as first: |
| 1353 | text = reddit_keyless_get_text(url, **kwargs) |
| 1354 | if text is not None: |
| 1355 | return text, None |
| 1356 | if _failures_are_429(first): |
| 1357 | _sleep_reddit_429_retry() |
| 1358 | with tee_failures() as second: |
| 1359 | text = reddit_keyless_get_text(url, **kwargs) |
| 1360 | if text is not None: |
| 1361 | return text, None |
| 1362 | err = second[-1] if second else (first[-1] if first else None) |
| 1363 | return None, str(err) if err is not None else "no response" |
| 1364 | for err in first: |
| 1365 | _record_failure(err) |
| 1366 | err = first[-1] if first else None |
| 1367 | return None, str(err) if err is not None else "no response" |
| 1368 | |
| 1369 | |
| 1370 | def scrapecreators_headers(token: str) -> Dict[str, str]: |
| 1371 | """Build ScrapeCreators request headers (x-api-key + JSON content type).""" |
| 1372 | return { |
| 1373 | "x-api-key": token, |
| 1374 | "Content-Type": "application/json", |
| 1375 | } |
| 1376 | |
| 1377 | |
| 1378 | def get_reddit_json(path: str, timeout: int = DEFAULT_TIMEOUT, retries: int = MAX_RETRIES) -> Dict[str, Any]: |
| 1379 | """Fetch Reddit thread JSON. |
| 1380 | |
| 1381 | Args: |
| 1382 | path: Reddit path (e.g., /r/subreddit/comments/id/title) |
| 1383 | timeout: HTTP timeout per attempt in seconds |
| 1384 | retries: Number of retries on failure |
| 1385 | |
| 1386 | Returns: |
| 1387 | Parsed JSON response |
| 1388 | """ |
| 1389 | # Ensure path starts with / |
| 1390 | if not path.startswith('/'): |
| 1391 | path = '/' + path |
| 1392 | |
| 1393 | # Remove trailing slash and add .json |
| 1394 | path = path.rstrip('/') |
| 1395 | if not path.endswith('.json'): |
| 1396 | path = path + '.json' |
| 1397 | |
| 1398 | url = f"https://www.reddit.com{path}?raw_json=1" |
| 1399 | |
| 1400 | headers = { |
| 1401 | "User-Agent": USER_AGENT, |
| 1402 | "Accept": "application/json", |
| 1403 | } |
| 1404 | |
| 1405 | return get(url, headers=headers, timeout=timeout, retries=retries) |
| 1406 |