Add a circuit breaker to the multi-backend search fallback (item from review)

/serp/search (and the MCP `search` tool) tries DuckDuckGo, then Brave,
then Bing per query with no memory between calls - if a backend is
rate-limiting or has blocked this deployment's IP, every single query
still pays its full timeout hammering that backend before falling
through, which only makes the block worse.

- circuit_breaker.py: a small per-key CircuitBreaker (new module, no third-
party dependency) - opens after `failure_threshold` consecutive failures
for a key, short-circuits further calls during `cooldown_seconds`, then
allows one half-open trial; success closes it, another failure reopens
it for a fresh cooldown. Unit-tested with a fake clock for determinism.
- app.py: wired a module-level `_backend_circuit_breaker` (3 failures,
60s cooldown) into `_search_one`'s backend loop.

_search_one/`/serp/search` had zero test coverage before this - added
characterization tests for its current fallback behavior first (first-
success, falls-through-on-failure, all-fail error format), confirmed
green against the pre-existing code, then added the breaker under that
safety net. Also caught and fixed a latent test-isolation gap while doing
this: `_backend_circuit_breaker` is a real module-level singleton, and
several of the new characterization tests recorded real failures against
it - across enough tests this could have silently pushed a backend's
failure count toward the threshold and made an unrelated later test skip
it. Added an autouse fixture giving every test in test_app.py its own
fresh breaker instance.

Verified the new coverage isn't vacuous: temporarily removed the
before_call check and confirmed the wiring test fails, then restored it.
97 tests pass.

app.py

@@ -13,6 +13,7 @@ from pydantic import BaseModel, Field, StringConstraints
 from playwright.async_api import async_playwright, Browser, BrowserContext, Page
 import uvicorn
 
+from circuit_breaker import CircuitBreaker, CircuitOpenError
 from scrap import PatentScrapBulkResponse, PatentScrapResult, scrap_patent_async, scrap_patent_bulk_async
 from serp import PATENT_ID_CORE, SerpQuery, SerpResults, query_arxiv, query_bing_search, query_brave_search, query_ddg_search, query_google_patents, query_google_scholar
 from ops import OPSBulkResponse, OPSNotConfigured, ops_scrap_patent, ops_scrap_patent_bulk, ops_search, token_manager as ops_token_manager
@@ -206,6 +207,14 @@ async def search_duck(params: SerpQuery) -> SerpResults:
         lambda q, n: query_ddg_search(q, n), params, "duckduckgo search")
 
 
+# Shared across all queries and requests: once a backend has failed
+# `failure_threshold` times in a row, skip it for `cooldown_seconds` instead
+# of attempting (and paying the timeout cost of) another call that's very
+# likely to fail - and, more importantly, stop hammering a backend that may
+# already be rate-limiting or blocking this deployment's IP.
+_backend_circuit_breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60.0)
+
+
 async def _search_one(q: str, n_results: int) -> tuple[str, list[dict], Optional[str]]:
     """Try DDG, then Brave, then Bing for a single query; stop at the first success."""
     backends = [
@@ -215,11 +224,21 @@ async def _search_one(q: str, n_results: int) -> tuple[str, list[dict], Optional
     ]
     last_error: Optional[Exception] = None
     for name, call in backends:
+        try:
+            _backend_circuit_breaker.before_call(name)
+        except CircuitOpenError as e:
+            logging.info(f"Skipping {name} for query `{q}`: {e}")
+            last_error = e
+            continue
+
         try:
             logging.info(f"Querying {name} with query: `{q}`")
-            return q, await call(), None
+            result = await call()
+            _backend_circuit_breaker.record_success(name)
+            return q, result, None
         except Exception as e:
             logging.error(f"Failed to query {name} with query `{q}`: {e}")
+            _backend_circuit_breaker.record_failure(name)
             last_error = e
 
     return q, [], f"All backends failed for query '{q}': {last_error}"

circuit_breaker.py

@@ -0,0 +1,67 @@
+"""A minimal per-key circuit breaker for outbound scraping backends.
+
+When a backend (e.g. one search engine) is rate-limiting or blocking us,
+retrying it on every single query just makes that worse - it's the
+opposite of what you want when the goal is to stop looking suspicious to
+that backend. This tracks consecutive failures per key and, once a
+threshold is hit, short-circuits further calls to that key for a cooldown
+period instead of attempting (and paying the timeout cost of) a call that
+is very likely to fail anyway. After the cooldown it lets exactly one
+trial call through (half-open); success closes the circuit again, failure
+re-opens it for another full cooldown.
+
+Not thread-safe (no locking) - fine given every caller in this codebase
+only ever runs on a single asyncio event loop.
+"""
+
+import time
+from dataclasses import dataclass
+
+
+class CircuitOpenError(Exception):
+    """Raised by `before_call` instead of letting a call through while a
+    key's circuit is open."""
+
+    def __init__(self, key: str, retry_after: float):
+        self.key = key
+        self.retry_after = retry_after
+        super().__init__(f"Circuit open for '{key}'; retry after {retry_after:.0f}s")
+
+
+@dataclass
+class _BreakerState:
+    failure_count: int = 0
+    opened_at: float = 0.0
+
+
+class CircuitBreaker:
+    def __init__(self, failure_threshold: int = 3, cooldown_seconds: float = 60.0):
+        self.failure_threshold = failure_threshold
+        self.cooldown_seconds = cooldown_seconds
+        self._states: dict[str, _BreakerState] = {}
+
+    def _state(self, key: str) -> _BreakerState:
+        return self._states.setdefault(key, _BreakerState())
+
+    def before_call(self, key: str) -> None:
+        """Raise CircuitOpenError if `key` is currently open and its
+        cooldown hasn't elapsed. Call this before attempting the call.
+        """
+        state = self._state(key)
+        if state.failure_count < self.failure_threshold:
+            return
+        remaining = self.cooldown_seconds - (time.monotonic() - state.opened_at)
+        if remaining > 0:
+            raise CircuitOpenError(key, remaining)
+        # Cooldown elapsed: half-open - let this one call through. A
+        # renewed failure (record_failure) re-opens it for another cooldown;
+        # a success (record_success) clears it entirely.
+
+    def record_success(self, key: str) -> None:
+        self._states.pop(key, None)
+
+    def record_failure(self, key: str) -> None:
+        state = self._state(key)
+        state.failure_count += 1
+        if state.failure_count >= self.failure_threshold:
+            state.opened_at = time.monotonic()

tests/test_app.py

@@ -14,6 +14,7 @@ import pytest
 from httpx import ASGITransport
 
 import app as app_module
+from circuit_breaker import CircuitBreaker
 from scrap import PatentScrapResult
 from serp import SerpQuery
 
@@ -25,6 +26,16 @@ async def client():
         yield c
 
 
+@pytest.fixture(autouse=True)
+def _fresh_backend_circuit_breaker(monkeypatch):
+    """`_backend_circuit_breaker` is a module-level singleton shared across
+    every request in production, which is the point - but it means a
+    backend failure recorded by one test would otherwise leak into the
+    next. Give every test its own instance.
+    """
+    monkeypatch.setattr(app_module, "_backend_circuit_breaker", CircuitBreaker())
+
+
 # ------------------------- _shape_serp_results / _run_serp_queries -------------------------
 
 
@@ -238,3 +249,96 @@ async def test_api_lifespan_launches_chromium_without_a_sandbox(monkeypatch):
 
     assert launch_calls[0]["headless"] is True
     assert "--no-sandbox" in launch_calls[0].get("args", [])
+
+
+# --------------------------------- _search_one fallback ---------------------------------
+
+
+async def test_search_one_returns_first_backend_that_succeeds(monkeypatch):
+    calls = []
+
+    async def fake_ddg(q, n):
+        calls.append("ddg")
+        return [{"title": "from ddg"}]
+
+    async def fake_brave(browser, q, n):
+        calls.append("brave")
+        return [{"title": "from brave"}]
+
+    monkeypatch.setattr(app_module, "query_ddg_search", fake_ddg)
+    monkeypatch.setattr(app_module, "query_brave_search", fake_brave)
+
+    q, results, err = await app_module._search_one("widget", 5)
+
+    assert calls == ["ddg"]
+    assert results == [{"title": "from ddg"}]
+    assert err is None
+
+
+async def test_search_one_falls_through_to_the_next_backend_on_failure(monkeypatch):
+    calls = []
+
+    async def failing_ddg(q, n):
+        calls.append("ddg")
+        raise RuntimeError("ddg blocked")
+
+    async def fake_brave(browser, q, n):
+        calls.append("brave")
+        return [{"title": "from brave"}]
+
+    monkeypatch.setattr(app_module, "query_ddg_search", failing_ddg)
+    monkeypatch.setattr(app_module, "query_brave_search", fake_brave)
+
+    q, results, err = await app_module._search_one("widget", 5)
+
+    assert calls == ["ddg", "brave"]
+    assert results == [{"title": "from brave"}]
+    assert err is None
+
+
+async def test_search_one_reports_an_error_when_every_backend_fails(monkeypatch):
+    async def failing(*args):
+        raise RuntimeError("blocked")
+
+    monkeypatch.setattr(app_module, "query_ddg_search", failing)
+    monkeypatch.setattr(app_module, "query_brave_search", failing)
+    monkeypatch.setattr(app_module, "query_bing_search", failing)
+
+    q, results, err = await app_module._search_one("widget", 5)
+
+    assert results == []
+    assert "All backends failed for query 'widget'" in err
+    assert "blocked" in err
+
+
+async def test_search_one_skips_a_backend_whose_circuit_is_open(monkeypatch):
+    """A backend that's failed `failure_threshold` times in a row shouldn't
+    be attempted again until its cooldown elapses - repeatedly hammering a
+    backend that's already rate-limiting us only makes that worse.
+    """
+    from circuit_breaker import CircuitBreaker
+
+    fresh_breaker = CircuitBreaker(failure_threshold=1, cooldown_seconds=60)
+    monkeypatch.setattr(app_module, "_backend_circuit_breaker", fresh_breaker)
+
+    ddg_calls = []
+
+    async def failing_ddg(q, n):
+        ddg_calls.append(q)
+        raise RuntimeError("ddg blocked")
+
+    async def fake_brave(browser, q, n):
+        return [{"title": "from brave"}]
+
+    monkeypatch.setattr(app_module, "query_ddg_search", failing_ddg)
+    monkeypatch.setattr(app_module, "query_brave_search", fake_brave)
+
+    # First call: DDG fails, opening its circuit (threshold=1); Brave picks it up.
+    await app_module._search_one("first query", 5)
+    assert ddg_calls == ["first query"]
+
+    # Second call: DDG's circuit is now open, so it should be skipped
+    # entirely (not called again) and go straight to Brave.
+    q, results, err = await app_module._search_one("second query", 5)
+    assert ddg_calls == ["first query"]  # unchanged - DDG was skipped
+    assert results == [{"title": "from brave"}]

tests/test_circuit_breaker.py

@@ -0,0 +1,100 @@
+import circuit_breaker as cb_module
+from circuit_breaker import CircuitBreaker, CircuitOpenError
+
+import pytest
+
+
+class _FakeClock:
+    """Deterministic stand-in for time.monotonic()."""
+
+    def __init__(self, start: float = 0.0):
+        self.now = start
+
+    def __call__(self) -> float:
+        return self.now
+
+    def advance(self, seconds: float) -> None:
+        self.now += seconds
+
+
+@pytest.fixture
+def clock(monkeypatch):
+    fake = _FakeClock()
+    monkeypatch.setattr(cb_module.time, "monotonic", fake)
+    return fake
+
+
+def test_before_call_is_a_noop_below_the_failure_threshold(clock):
+    breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60)
+    breaker.record_failure("bing")
+    breaker.record_failure("bing")
+
+    breaker.before_call("bing")  # should not raise - only 2 failures so far
+
+
+def test_before_call_raises_once_threshold_is_reached(clock):
+    breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60)
+    for _ in range(3):
+        breaker.record_failure("bing")
+
+    with pytest.raises(CircuitOpenError):
+        breaker.before_call("bing")
+
+
+def test_record_success_closes_the_circuit(clock):
+    breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60)
+    for _ in range(3):
+        breaker.record_failure("bing")
+
+    breaker.record_success("bing")
+
+    breaker.before_call("bing")  # back to closed - should not raise
+
+
+def test_before_call_allows_a_trial_after_the_cooldown_elapses(clock):
+    breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60)
+    for _ in range(3):
+        breaker.record_failure("bing")
+
+    clock.advance(61)
+
+    breaker.before_call("bing")  # cooldown elapsed - half-open, should not raise
+
+
+def test_a_failed_trial_after_cooldown_reopens_for_another_full_cooldown(clock):
+    breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=60)
+    for _ in range(3):
+        breaker.record_failure("bing")
+
+    clock.advance(61)
+    breaker.before_call("bing")  # half-open trial begins
+    breaker.record_failure("bing")  # ...and it failed
+
+    clock.advance(30)  # only 30s into the new cooldown
+    with pytest.raises(CircuitOpenError):
+        breaker.before_call("bing")
+
+    clock.advance(31)  # now past the new 60s cooldown
+    breaker.before_call("bing")  # should not raise
+
+
+def test_keys_are_independent(clock):
+    breaker = CircuitBreaker(failure_threshold=2, cooldown_seconds=60)
+    for _ in range(2):
+        breaker.record_failure("bing")
+
+    breaker.before_call("brave")  # untouched key - should not raise
+    with pytest.raises(CircuitOpenError):
+        breaker.before_call("bing")
+
+
+def test_circuit_open_error_carries_the_key_and_remaining_seconds(clock):
+    breaker = CircuitBreaker(failure_threshold=1, cooldown_seconds=60)
+    breaker.record_failure("bing")
+
+    clock.advance(10)
+    with pytest.raises(CircuitOpenError) as exc_info:
+        breaker.before_call("bing")
+
+    assert exc_info.value.key == "bing"
+    assert exc_info.value.retry_after == pytest.approx(50, abs=0.01)