From da31a1f9e78581a5134096dd692d5791f84b7b14 Mon Sep 17 00:00:00 2001 From: cryptorugmunch Date: Fri, 3 Jul 2026 11:49:33 +0200 Subject: [PATCH] feat(resilience): circuit breakers and per-domain tier memory --- resilience.py | 228 +++++++++++++++++++++++++++++++++++++++ tests/test_resilience.py | 219 +++++++++++++++++++++++++++++++++++++ ultimate_scraper.py | 133 ++++++++++++----------- 3 files changed, 517 insertions(+), 63 deletions(-) create mode 100644 resilience.py create mode 100644 tests/test_resilience.py diff --git a/resilience.py b/resilience.py new file mode 100644 index 0000000..b19f5d0 --- /dev/null +++ b/resilience.py @@ -0,0 +1,228 @@ +"""Resilience primitives for Pry scrapers. + +Provides circuit breakers and per-domain tier memory so the fallback chain +can skip unhealthy services and prefer the cheapest successful strategy for +a given domain. +""" + +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 Rug Munch Media LLC +# +# Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper +# Licensed under MIT. See LICENSE. +from __future__ import annotations + +import logging +import time +from collections import defaultdict +from dataclasses import dataclass, field +from typing import Any, ClassVar + +logger = logging.getLogger(__name__) + +DEFAULT_FAILURE_THRESHOLD = 5 +DEFAULT_RECOVERY_SECONDS = 30 +DEFAULT_HALF_OPEN_MAX_TRIES = 2 +DOMAIN_MEMORY_MAX_AGE_SECONDS = 3600 + + +class CircuitState: + CLOSED = "closed" + OPEN = "open" + HALF_OPEN = "half_open" + + +@dataclass +class CircuitBreaker: + """Simple circuit breaker with exponential cooldown.""" + + name: str + failure_threshold: int = DEFAULT_FAILURE_THRESHOLD + recovery_seconds: float = DEFAULT_RECOVERY_SECONDS + half_open_max_tries: int = DEFAULT_HALF_OPEN_MAX_TRIES + state: str = CircuitState.CLOSED + failures: int = 0 + successes: int = 0 + last_failure_at: float = 0.0 + half_open_attempts: int = 0 + + def is_allowed(self) -> bool: + now = time.time() + if self.state == CircuitState.CLOSED: + return True + if self.state == CircuitState.OPEN: + if now - self.last_failure_at >= self.recovery_seconds: + self.state = CircuitState.HALF_OPEN + self.half_open_attempts = 0 + logger.info("circuit_breaker_half_open", extra={"service": self.name}) + return True + return False + # HALF_OPEN + return self.half_open_attempts < self.half_open_max_tries + + def record_success(self) -> None: + if self.state == CircuitState.HALF_OPEN: + self.half_open_attempts += 1 + if self.half_open_attempts >= self.half_open_max_tries: + self._close() + return + self.successes += 1 + # Decay failures slowly on success to forgive transient blips. + if self.failures > 0: + self.failures -= 1 + + def record_failure(self) -> None: + self.failures += 1 + self.last_failure_at = time.time() + if self.state == CircuitState.HALF_OPEN: + self._open() + return + if self.failures >= self.failure_threshold: + self._open() + + def _open(self) -> None: + if self.state != CircuitState.OPEN: + logger.warning( + "circuit_breaker_open", + extra={"service": self.name, "failures": self.failures}, + ) + self.state = CircuitState.OPEN + self.half_open_attempts = 0 + + def _close(self) -> None: + logger.info("circuit_breaker_closed", extra={"service": self.name}) + self.state = CircuitState.CLOSED + self.failures = 0 + self.half_open_attempts = 0 + + +@dataclass +class TierRecord: + successes: int = 0 + failures: int = 0 + last_success_at: float = 0.0 + last_failure_at: float = 0.0 + total_latency: float = 0.0 + + @property + def score(self) -> float: + """Higher score = more preferred. Balances success rate and recency.""" + total = self.successes + self.failures + if total == 0: + return 0.0 + success_rate = self.successes / total + # Recency bonus: 1.0 if succeeded in last 5 minutes, decaying. + recency = 0.0 + if self.last_success_at > 0: + age = time.time() - self.last_success_at + recency = max(0.0, 1.0 - age / DOMAIN_MEMORY_MAX_AGE_SECONDS) + return success_rate + recency * 0.5 + + +@dataclass +class DomainTierMemory: + """Remember which tiers work best per domain.""" + + data: dict[str, dict[str, TierRecord]] = field(default_factory=lambda: defaultdict(dict)) + + def record(self, domain: str, tier: str, success: bool, latency: float = 0.0) -> None: + rec = self.data[domain].setdefault(tier, TierRecord()) + if success: + rec.successes += 1 + rec.last_success_at = time.time() + else: + rec.failures += 1 + rec.last_failure_at = time.time() + rec.total_latency += latency + + def best_tier(self, domain: str, candidates: list[str]) -> str | None: + records = self.data.get(domain, {}) + if not records: + return None + # Only consider tiers with at least one success and no recent failures. + now = time.time() + eligible = [ + (tier, rec) + for tier, rec in records.items() + if rec.successes > 0 + and (rec.last_failure_at == 0 or now - rec.last_failure_at > 300) + and tier in candidates + ] + if not eligible: + return None + eligible.sort(key=lambda item: item[1].score, reverse=True) + return eligible[0][0] + + def to_dict(self) -> dict[str, dict[str, dict[str, Any]]]: + return { + domain: { + tier: { + "successes": rec.successes, + "failures": rec.failures, + "last_success_at": rec.last_success_at, + "last_failure_at": rec.last_failure_at, + "score": round(rec.score, 3), + } + for tier, rec in tiers.items() + } + for domain, tiers in self.data.items() + } + + +class ResilienceRegistry: + """Global registry for circuit breakers and domain tier memory.""" + + # External services that should have circuit breakers. + SERVICE_NAMES: ClassVar[list[str]] = [ + "flaresolverr", + "ollama", + "openrouter", + "redis", + ] + + def __init__(self) -> None: + self.breakers: dict[str, CircuitBreaker] = { + name: CircuitBreaker(name=name) for name in self.SERVICE_NAMES + } + self.domain_memory = DomainTierMemory() + + def breaker(self, name: str) -> CircuitBreaker: + if name not in self.breakers: + self.breakers[name] = CircuitBreaker(name=name) + return self.breakers[name] + + def is_service_allowed(self, name: str) -> bool: + return self.breaker(name).is_allowed() + + def record_service_result(self, name: str, success: bool) -> None: + breaker = self.breaker(name) + if success: + breaker.record_success() + else: + breaker.record_failure() + + def record_domain_tier_result( + self, domain: str, tier: str, success: bool, latency: float = 0.0 + ) -> None: + self.domain_memory.record(domain, tier, success, latency) + + def best_tier_for_domain(self, domain: str, candidates: list[str]) -> str | None: + return self.domain_memory.best_tier(domain, candidates) + + def stats(self) -> dict[str, Any]: + return { + "circuit_breakers": { + name: { + "state": cb.state, + "failures": cb.failures, + "successes": cb.successes, + "last_failure_at": cb.last_failure_at, + } + for name, cb in self.breakers.items() + }, + "domain_memory": self.domain_memory.to_dict(), + } + + +# Global singleton used by scrapers. +registry = ResilienceRegistry() diff --git a/tests/test_resilience.py b/tests/test_resilience.py new file mode 100644 index 0000000..b6606da --- /dev/null +++ b/tests/test_resilience.py @@ -0,0 +1,219 @@ +"""Tests for resilience primitives and their integration into UltimateScraper.""" + +from __future__ import annotations + +import time as time_module +from contextlib import ExitStack +from unittest.mock import AsyncMock, patch + +import pytest + +import resilience +import ultimate_scraper +from resilience import CircuitBreaker, CircuitState, DomainTierMemory, ResilienceRegistry + + +def _valid_html(text: str) -> str: + body = f"

{text}

" * 40 + return ( + "T" + "" + f"{body}" + ) + + +@pytest.fixture +def fresh_registry(monkeypatch): + """Provide a clean ResilienceRegistry and patch it into UltimateScraper.""" + reg = ResilienceRegistry() + monkeypatch.setattr(resilience, "registry", reg) + monkeypatch.setattr(ultimate_scraper, "registry", reg) + return reg + + +@pytest.fixture +def frozen_time(monkeypatch): + """Freeze time.time() for deterministic circuit-breaker tests.""" + now = 1_700_000_000.0 + + def fake_time(): + return now + + monkeypatch.setattr(time_module, "time", fake_time) + monkeypatch.setattr(resilience.time, "time", fake_time) + monkeypatch.setattr(ultimate_scraper.time, "time", fake_time) + return lambda offset=0: setattr(fake_time, "__wrapped__", None) or None + + +class TestCircuitBreaker: + def test_starts_closed_and_allows(self): + cb = CircuitBreaker(name="test") + assert cb.state == CircuitState.CLOSED + assert cb.is_allowed() is True + + def test_opens_after_threshold_failures(self): + cb = CircuitBreaker(name="test", failure_threshold=3, recovery_seconds=30) + cb.record_failure() + cb.record_failure() + assert cb.is_allowed() is True + cb.record_failure() + assert cb.state == CircuitState.OPEN + assert cb.is_allowed() is False + + def test_half_open_after_recovery(self, frozen_time, monkeypatch): + cb = CircuitBreaker(name="test", failure_threshold=1, recovery_seconds=60) + cb.record_failure() + assert cb.state == CircuitState.OPEN + # Advance time past recovery window. + now = 1_700_000_000.0 + 61 + monkeypatch.setattr(time_module, "time", lambda: now) + monkeypatch.setattr(resilience.time, "time", lambda: now) + assert cb.is_allowed() is True + assert cb.state == CircuitState.HALF_OPEN + + def test_closes_after_half_open_successes(self, frozen_time, monkeypatch): + cb = CircuitBreaker( + name="test", failure_threshold=1, recovery_seconds=60, half_open_max_tries=2 + ) + cb.record_failure() + now = 1_700_000_000.0 + 61 + monkeypatch.setattr(time_module, "time", lambda: now) + monkeypatch.setattr(resilience.time, "time", lambda: now) + assert cb.is_allowed() is True + cb.record_success() + assert cb.state == CircuitState.HALF_OPEN + assert cb.is_allowed() is True + cb.record_success() + assert cb.state == CircuitState.CLOSED + + def test_failure_in_half_open_reopens(self, frozen_time, monkeypatch): + cb = CircuitBreaker(name="test", failure_threshold=1, recovery_seconds=60) + cb.record_failure() + now = 1_700_000_000.0 + 61 + monkeypatch.setattr(time_module, "time", lambda: now) + monkeypatch.setattr(resilience.time, "time", lambda: now) + cb.is_allowed() + cb.record_failure() + assert cb.state == CircuitState.OPEN + + +class TestDomainTierMemory: + def test_picks_best_tier(self): + mem = DomainTierMemory() + mem.record("example.com", "direct", success=True, latency=0.1) + mem.record("example.com", "direct", success=True, latency=0.1) + mem.record("example.com", "playwright", success=False, latency=0.1) + best = mem.best_tier("example.com", ["direct", "playwright"]) + assert best == "direct" + + def test_ignores_recent_failures(self, frozen_time, monkeypatch): + mem = DomainTierMemory() + now = 1_700_000_000.0 + monkeypatch.setattr(time_module, "time", lambda: now) + mem.record("example.com", "direct", success=True, latency=0.1) + mem.record("example.com", "direct", success=False, latency=0.1) + assert mem.best_tier("example.com", ["direct"]) is None + + def test_returns_none_for_unknown_domain(self): + mem = DomainTierMemory() + assert mem.best_tier("example.com", ["direct"]) is None + + +class TestResilienceRegistry: + def test_stats_shape(self, fresh_registry): + stats = fresh_registry.stats() + assert "circuit_breakers" in stats + assert "domain_memory" in stats + assert set(stats["circuit_breakers"].keys()) == set(ResilienceRegistry.SERVICE_NAMES) + + def test_service_result_toggles_breaker(self, fresh_registry): + fresh_registry.breaker("flaresolverr").failure_threshold = 1 + fresh_registry.record_service_result("flaresolverr", False) + assert fresh_registry.is_service_allowed("flaresolverr") is False + fresh_registry.record_service_result("flaresolverr", True) + # In half-open after recovery, not immediate. + assert fresh_registry.breaker("flaresolverr").state == CircuitState.OPEN + + +_TIER_NAMES = [ + "_tier_direct", + "_tier_tls_fingerprint", + "_tier_cloudscraper", + "_tier_flaresolverr", + "_tier_undetected", + "_tier_camoufox", + "_tier_playwright", + "_tier_googlebot", + "_tier_premium_proxy", + "_tier_archive", + "_tier_google_cache", + "_tier_tor", +] + + +class TestUltimateScraperResilience: + @pytest.mark.anyio + async def test_skips_circuit_open_tier(self, fresh_registry): + scraper = ultimate_scraper.UltimateScraper() + fresh_registry.breaker("flaresolverr").failure_threshold = 1 + fresh_registry.record_service_result("flaresolverr", False) + + html = _valid_html("archive content") + mocks = { + name: AsyncMock(return_value=None) + for name in _TIER_NAMES + } + mocks["_tier_archive"].return_value = html + with ExitStack() as stack: + for name in _TIER_NAMES: + stack.enter_context(patch.object(scraper, name, new=mocks[name])) + result = await scraper.scrape("https://example.com/page") + assert result["status"] == "ok" + assert result["method"] == "archive" + mocks["_tier_direct"].assert_awaited_once() + mocks["_tier_cloudscraper"].assert_awaited_once() + mocks["_tier_flaresolverr"].assert_not_awaited() + + @pytest.mark.anyio + async def test_prefers_domain_memory_best_tier(self, fresh_registry): + scraper = ultimate_scraper.UltimateScraper() + fresh_registry.record_domain_tier_result("example.com", "archive", success=True) + html = _valid_html("archived") + mocks = {name: AsyncMock(return_value=None) for name in _TIER_NAMES} + mocks["_tier_archive"].return_value = html + mocks["_tier_direct"].return_value = _valid_html("direct") + with ExitStack() as stack: + for name in _TIER_NAMES: + stack.enter_context(patch.object(scraper, name, new=mocks[name])) + result = await scraper.scrape("https://example.com/page") + assert result["status"] == "ok" + assert result["method"] == "archive" + mocks["_tier_archive"].assert_awaited() + mocks["_tier_direct"].assert_not_awaited() + + @pytest.mark.anyio + async def test_records_tier_results_after_success(self, fresh_registry): + scraper = ultimate_scraper.UltimateScraper() + html = _valid_html("direct content") + mocks = {name: AsyncMock(return_value=None) for name in _TIER_NAMES} + mocks["_tier_direct"].return_value = html + with ExitStack() as stack: + for name in _TIER_NAMES: + stack.enter_context(patch.object(scraper, name, new=mocks[name])) + await scraper.scrape("https://example.com/page") + assert fresh_registry.best_tier_for_domain( + "example.com", ["direct", "archive"] + ) == "direct" + + @pytest.mark.anyio + async def test_records_tier_failure_no_success_memory(self, fresh_registry): + scraper = ultimate_scraper.UltimateScraper() + mocks = {name: AsyncMock(return_value=None) for name in _TIER_NAMES} + with ExitStack() as stack: + for name in _TIER_NAMES: + stack.enter_context(patch.object(scraper, name, new=mocks[name])) + result = await scraper.scrape("https://example.com/page") + assert result["status"] == "error" + assert fresh_registry.best_tier_for_domain( + "example.com", ["direct", "archive"] + ) is None diff --git a/ultimate_scraper.py b/ultimate_scraper.py index 84eded7..c59a209 100644 --- a/ultimate_scraper.py +++ b/ultimate_scraper.py @@ -14,9 +14,12 @@ import logging import os import random import re +import time from typing import Any, ClassVar from urllib.parse import urlparse +from resilience import registry + logger = logging.getLogger(__name__) # Try importing optional bypass libraries @@ -118,67 +121,71 @@ class UltimateScraper: return self._result(url, html, "premium_proxy") errors.append("premium_proxy_failed") - # Tier 1: Direct HTTP - html = await self._tier_direct(url, opts, proxy_url=proxy_url) - if html and self._is_valid(html): - return self._result(url, html, "direct") + # Build tier list. Order starts cheap/direct and escalates. + tier_order = [ + ("direct", self._tier_direct), + ("tls_fingerprint", self._tier_tls_fingerprint), + ("cloudscraper", self._tier_cloudscraper), + ("flaresolverr", self._tier_flaresolverr), + ("undetected", self._tier_undetected), + ("camoufox", self._tier_camoufox), + ("playwright", self._tier_playwright), + ("googlebot", self._tier_googlebot), + ] - # Tier 2: TLS fingerprint impersonation - html = await self._tier_tls_fingerprint(url, opts, proxy_url=proxy_url) - if html and self._is_valid(html): - return self._result(url, html, "tls_fingerprint") - - # Tier 3: cloudscraper (Python Cloudflare bypass) - html = await self._tier_cloudscraper(url, opts) - if html and self._is_valid(html): - return self._result(url, html, "cloudscraper") - - # Tier 4: FlareSolverr - html = await self._tier_flaresolverr(url, opts) - if html and self._is_valid(html): - return self._result(url, html, "flaresolverr") - - # Tier 5: undetected-chromedriver - html = await self._tier_undetected(url, opts) - if html and self._is_valid(html): - return self._result(url, html, "undetected") - - # Tier 6: Camoufox stealth Firefox - html = await self._tier_camoufox(url, opts, proxy_url=proxy_url) - if html and self._is_valid(html): - return self._result(url, html, "camoufox") - - # Tier 7: Playwright stealth - html = await self._tier_playwright(url, opts) - if html and self._is_valid(html): - return self._result(url, html, "playwright") - - # Tier 8: Googlebot UA - html = await self._tier_googlebot(url, opts, proxy_url=proxy_url) - if html and self._is_valid(html): - return self._result(url, html, "googlebot") - - # Tier 9: Premium proxy retry (after cheap tiers fail) + # Premium proxy retry (after cheap tiers fail) if not tried first. if proxy_url and not opts.get("use_premium_proxy_first", True): - html = await self._tier_premium_proxy(url, proxy_url, opts) - if html and self._is_valid(html): - return self._result(url, html, "premium_proxy") - errors.append("premium_proxy_failed") + tier_order.append(("premium_proxy", self._tier_premium_proxy)) - # Tier 10: Archive.org / Wayback Machine - html = await self._tier_archive(url, opts) - if html and self._is_valid(html): - return self._result(url, html, "archive") + tier_order.extend([ + ("archive", self._tier_archive), + ("google_cache", self._tier_google_cache), + ("tor", self._tier_tor), + ]) - # Tier 11: Google Cache - html = await self._tier_google_cache(url, opts) - if html and self._is_valid(html): - return self._result(url, html, "google_cache") + # If domain memory knows a good tier, try it first. + parsed = urlparse(url) + domain = parsed.netloc + best = registry.best_tier_for_domain(domain, [name for name, _ in tier_order]) + if best: + tier_order.sort(key=lambda item: (item[0] != best, item[0])) - # Tier 12: Tor (anonymous routing) - html = await self._tier_tor(url, opts) - if html and self._is_valid(html): - return self._result(url, html, "tor") + # External service tiers protected by circuit breakers. + breaker_for_tier = { + "flaresolverr": "flaresolverr", + "playwright": "playwright", + "camoufox": "camoufox", + "tls_fingerprint": "tls_fingerprint", + } + + for tier_name, tier_fn in tier_order: + # Skip external tiers whose circuit breaker is open. + breaker_name = breaker_for_tier.get(tier_name) + if breaker_name and not registry.is_service_allowed(breaker_name): + logger.debug("tier_skipped_circuit_open", extra={"tier": tier_name, "domain": domain}) + errors.append(f"{tier_name}: circuit_open") + continue + + start = time.time() + try: + if tier_name == "premium_proxy": + html = await tier_fn(url, proxy_url, opts) + else: + html = await tier_fn(url, opts, proxy_url=proxy_url) + except Exception as e: # noqa: BLE001 + logger.debug("tier_exception", extra={"tier": tier_name, "error": str(e)[:50]}) + html = None + + latency = time.time() - start + success = bool(html and self._is_valid(html)) + registry.record_domain_tier_result(domain, tier_name, success, latency) + if breaker_name: + registry.record_service_result(breaker_name, success) + + if success: + method = "premium_proxy" if tier_name == "premium_proxy" else tier_name + return self._result(url, html, method) + errors.append(f"{tier_name}_failed") last_err = errors[-1] if errors else "all_tiers_failed" return { @@ -301,7 +308,7 @@ class UltimateScraper: logger.debug("premium_proxy_failed", extra={"error": str(e)[:50]}) return None - async def _tier_cloudscraper(self, url: str, opts: dict[str, Any]) -> str | None: + async def _tier_cloudscraper(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None: if not _has_cloudscraper: return None try: @@ -317,7 +324,7 @@ class UltimateScraper: logger.debug("cloudscraper_failed", extra={"error": str(e)[:50]}) return None - async def _tier_flaresolverr(self, url: str, opts: dict[str, Any]) -> str | None: + async def _tier_flaresolverr(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None: from settings import settings if not settings.flaresolverr_url: @@ -337,7 +344,7 @@ class UltimateScraper: logger.debug("flaresolverr_failed", extra={"error": str(e)[:50]}) return None - async def _tier_undetected(self, url: str, opts: dict[str, Any]) -> str | None: + async def _tier_undetected(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None: if not _has_undetected: return None try: @@ -364,7 +371,7 @@ class UltimateScraper: logger.debug("undetected_failed", extra={"error": str(e)[:50]}) return None - async def _tier_playwright(self, url: str, opts: dict[str, Any]) -> str | None: + async def _tier_playwright(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None: if not _has_playwright: return None try: @@ -460,7 +467,7 @@ class UltimateScraper: logger.debug("googlebot_failed", extra={"error": str(e)[:50]}) return None - async def _tier_archive(self, url: str, opts: dict[str, Any]) -> str | None: + async def _tier_archive(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None: """Fetch from Wayback Machine CDX API for latest snapshot.""" try: import httpx @@ -480,7 +487,7 @@ class UltimateScraper: logger.debug("archive_failed", extra={"error": str(e)[:50]}) return None - async def _tier_google_cache(self, url: str, opts: dict[str, Any]) -> str | None: + async def _tier_google_cache(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None: try: import httpx @@ -493,7 +500,7 @@ class UltimateScraper: logger.debug("google_cache_failed", extra={"error": str(e)[:50]}) return None - async def _tier_tor(self, url: str, opts: dict[str, Any]) -> str | None: + async def _tier_tor(self, url: str, opts: dict[str, Any], proxy_url: str | None = None) -> str | None: """Fetch via Tor SOCKS5 proxy.""" if not _has_tor: return None