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