From 119e4a916e12ae1fd39474b924a786c46bafbdea Mon Sep 17 00:00:00 2001 From: ciregenz Date: Thu, 30 Jul 2026 16:24:58 -0700 Subject: [PATCH] [eric] web: stop paying a dead search engine its full tier budget on every query, and let a 403 still try the lite frontend --- .../apps/agents/tools/search/search_ddg.py | 9 ++ backend/apps/web/cascade.py | 19 +++ backend/apps/web/tier_breaker.py | 77 ++++++++++ backend/apps/web/web.py | 9 +- backend/tests/test_tier_breaker.py | 141 ++++++++++++++++++ backend/tests/test_web_search_ddg.py | 43 ++++++ backend/tests/web_cascade_fixtures.py | 9 ++ 7 files changed, 304 insertions(+), 3 deletions(-) create mode 100644 backend/apps/web/tier_breaker.py create mode 100644 backend/tests/test_tier_breaker.py diff --git a/backend/apps/agents/tools/search/search_ddg.py b/backend/apps/agents/tools/search/search_ddg.py index 00d8c472..630d32d1 100644 --- a/backend/apps/agents/tools/search/search_ddg.py +++ b/backend/apps/agents/tools/search/search_ddg.py @@ -51,7 +51,16 @@ async def search_ddg(query: str, num_results: int) -> str: if lite is None: raise DDGRateLimited(query) return lite + # A hard block (403 is what html escalates to after the 202s) used to skip lite entirely, so a whole second frontend went untried; measured 7 times in one 44-query round. if reply.status >= 400: + try: + lite = await search_ddg_lite(query, num_results) + except Exception as exc: + raise RuntimeError(f"DuckDuckGo html returned HTTP {reply.status}; lite: {exc}") from None + if lite is None: + raise DDGRateLimited(query) + if lite: + return lite raise RuntimeError(f"DuckDuckGo html returned HTTP {reply.status}") body = reply.text diff --git a/backend/apps/web/cascade.py b/backend/apps/web/cascade.py index 16229271..af5cf6eb 100644 --- a/backend/apps/web/cascade.py +++ b/backend/apps/web/cascade.py @@ -14,6 +14,12 @@ from typing import Awaitable, Callable, Dict, List, Optional from pydantic import BaseModel, ConfigDict, Field, InstanceOf from typeguard import typechecked +from backend.apps.web.tier_breaker import ( + record_tier_failure, + record_tier_success, + tier_cooldown_left, +) + # A tier handed less than this has no realistic chance, and reporting it as a timeout would be a lie; we say the budget ran out instead. MIN_TIER_SECONDS = 3.0 @@ -24,6 +30,8 @@ class CascadeTier(BaseModel): name: str run: InstanceOf[Callable[[], Awaitable[Optional[Dict]]]] budget: float + # Only for tiers whose failure is a property of the HOST, not of this request; see tier_breaker. + breaker: bool = False class CascadeOutcome(BaseModel): @@ -48,14 +56,25 @@ async def run_cascade(tiers: List[CascadeTier], total_budget: float) -> CascadeO f"{total_budget:.0f}s cascade budget spent; not attempted: {', '.join(skipped)}" ) break + cooling = tier_cooldown_left(tier.name) if tier.breaker else 0.0 + if cooling: + errors.append(f"{tier.name}: skipped, still failing (retry in {cooling:.0f}s)") + continue slice_seconds = min(tier.budget, remaining) try: result = await asyncio.wait_for(tier.run(), timeout=slice_seconds) except asyncio.TimeoutError: errors.append(f"{tier.name}: timed out after {slice_seconds:.0f}s") + if tier.breaker: + record_tier_failure(tier.name) except Exception as exc: errors.append(f"{tier.name}: {str(exc)[:150]}") + if tier.breaker: + record_tier_failure(tier.name) else: + # Answering "no hits" still proves the host is up, so it clears the failure streak. + if tier.breaker: + record_tier_success(tier.name) if result is not None: return CascadeOutcome(result=result, errors=errors) diff --git a/backend/apps/web/tier_breaker.py b/backend/apps/web/tier_breaker.py new file mode 100644 index 00000000..e574c0f0 --- /dev/null +++ b/backend/apps/web/tier_breaker.py @@ -0,0 +1,77 @@ +"""Short-lived circuit breaker for the cascade's fixed-host tiers. + +Measured on this machine over three 44-query rounds: DuckDuckGo answered the +first 7 searches, then served its bot challenge, then 403'd, then stopped +answering TCP altogether. Once that happened EVERY search paid the full 8s +DuckDuckGo tier budget before Startpage answered, so the keyless p50 went from +1.0s to 8.5s while the success rate stayed at 100%. The engine wasn't broken, +our retrying of a known-dead engine was. + +A tier only opts in when "closed" is a property of the HOST rather than of the +request, which is true for a search frontend and false for a page fetch (one +404 says nothing about the next URL). State is per-process and time-boxed, so +the worst a wrong guess costs is one cooldown window of a tier we skip. +""" + +import time +from typing import Dict, Optional + +from pydantic import BaseModel, ConfigDict +from typeguard import typechecked + +# Two failures can be one bad minute; three in a row is a closed door. +FAILURES_TO_OPEN = 3 +FIRST_COOLDOWN_SECONDS = 120.0 +MAX_COOLDOWN_SECONDS = 900.0 + + +class TierHealth(BaseModel): + model_config = ConfigDict(validate_assignment=True) + + consecutive_failures: int = 0 + open_until: float = 0.0 + cooldown: float = 0.0 + + +p_health: Dict[str, TierHealth] = {} + + +@typechecked +def p_entry(name: str) -> TierHealth: + if name not in p_health: + p_health[name] = TierHealth() + return p_health[name] + + +@typechecked +def tier_cooldown_left(name: str, now: Optional[float] = None) -> float: + """Seconds until this tier is worth trying again; 0 when it is open for business.""" + entry = p_health.get(name) + if entry is None: + return 0.0 + return max(0.0, entry.open_until - (time.monotonic() if now is None else now)) + + +@typechecked +def record_tier_failure(name: str, now: Optional[float] = None) -> None: + """A timeout or exception. Three in a row shuts the tier for a doubling cooldown.""" + stamp = time.monotonic() if now is None else now + entry = p_entry(name) + entry.consecutive_failures += 1 + if entry.consecutive_failures < FAILURES_TO_OPEN: + return + # The half-open probe that fails again doubles the wait, so a permanently dead engine stops costing anything. + entry.cooldown = min(max(entry.cooldown * 2, FIRST_COOLDOWN_SECONDS), MAX_COOLDOWN_SECONDS) + entry.open_until = stamp + entry.cooldown + + +@typechecked +def record_tier_success(name: str) -> None: + """The tier answered, even if the answer was 'no hits'. It is alive; forget the history.""" + if name in p_health: + p_health[name] = TierHealth() + + +@typechecked +def reset_tier_health() -> None: + p_health.clear() diff --git a/backend/apps/web/web.py b/backend/apps/web/web.py index fd0ac11b..62326793 100644 --- a/backend/apps/web/web.py +++ b/backend/apps/web/web.py @@ -187,8 +187,8 @@ async def search(body: SearchBody) -> Dict: "backend": "openai_subscription"} tiers = [ - CascadeTier(name="ddg", run=try_keyless, budget=KEYLESS_TIER_SECONDS), - CascadeTier(name="startpage", run=try_startpage, budget=KEYLESS_TIER_SECONDS), + CascadeTier(name="ddg", run=try_keyless, budget=KEYLESS_TIER_SECONDS, breaker=True), + CascadeTier(name="startpage", run=try_startpage, budget=KEYLESS_TIER_SECONDS, breaker=True), CascadeTier(name="browser_search", run=try_browser_search, budget=BROWSER_TIER_SECONDS), ] + p_grounded_tiers("search", body.primary, { "gemini_native": try_gemini, @@ -232,9 +232,12 @@ async def search(body: SearchBody) -> Dict: async def fetch(body: FetchBody) -> Dict: """Fetch a URL, primary-aware. Mirrors the /search cascade.""" # Belt-and-suspenders: even though we delegate to remote Gemini/OpenAI fetchers (which can't reach private IPs), validating the URL here means a private/metadata URL gets a 4xx instead of being silently forwarded. - from backend.apps.agents.tools.ssrf_guard import SSRFBlocked, assert_safe_url + from backend.apps.agents.tools.ssrf_guard import DomainUnreachable, SSRFBlocked, assert_safe_url try: await assert_safe_url(body.url) + except DomainUnreachable: + # A domain that no longer resolves is the archive's whole reason for existing, so let the cascade run instead of 400ing here. + pass except SSRFBlocked as exc: raise HTTPException(status_code=400, detail=f"Refused: {exc}") gemini_key = resolve_gemini_api_key() diff --git a/backend/tests/test_tier_breaker.py b/backend/tests/test_tier_breaker.py new file mode 100644 index 00000000..b2183518 --- /dev/null +++ b/backend/tests/test_tier_breaker.py @@ -0,0 +1,141 @@ +"""A dead search frontend must stop costing us its whole tier budget. + +Measured before this existed: once DuckDuckGo stopped answering TCP, three +consecutive 44-query rounds each paid the full 8s DuckDuckGo budget on EVERY +query, so keyless p50 went 1.0s -> 8.5s while Startpage still served 40/40. +These pin the breaker that makes that unrepresentable, and pin the two ways it +could go wrong instead: skipping a healthy tier, or leaking a per-URL failure +into a fixed-host one. +""" + +import asyncio + +import pytest + +import backend.apps.web.cascade as C +from backend.apps.web.cascade import CascadeTier, run_cascade +from backend.apps.web.tier_breaker import ( + FAILURES_TO_OPEN, + FIRST_COOLDOWN_SECONDS, + MAX_COOLDOWN_SECONDS, + record_tier_failure, + record_tier_success, + reset_tier_health, + tier_cooldown_left, +) + + +@pytest.fixture(autouse=True) +def p_clean(): + reset_tier_health() + yield + reset_tier_health() + + +@pytest.fixture +def p_tiny_floor(monkeypatch): + monkeypatch.setattr(C, "MIN_TIER_SECONDS", 0.01) + + +async def p_boom(): + raise RuntimeError("engine closed") + + +async def p_hit(): + return {"backend": "second"} + + +def test_streak_opens_then_success_clears(): + for _ in range(FAILURES_TO_OPEN - 1): + record_tier_failure("ddg") + assert tier_cooldown_left("ddg") == 0.0 + record_tier_failure("ddg") + assert 0 < tier_cooldown_left("ddg") <= FIRST_COOLDOWN_SECONDS + record_tier_success("ddg") + assert tier_cooldown_left("ddg") == 0.0 + + +def test_cooldown_doubles_and_is_capped(): + seen = [] + now = 0.0 + for round_no in range(12): + for _ in range(FAILURES_TO_OPEN): + record_tier_failure("ddg", now=now) + seen.append(tier_cooldown_left("ddg", now=now)) + now += seen[-1] + 1 + record_tier_failure("ddg", now=now) # the half-open probe fails again + assert seen[0] == pytest.approx(FIRST_COOLDOWN_SECONDS) + assert seen[1] > seen[0] + assert max(seen) <= MAX_COOLDOWN_SECONDS + + +@pytest.mark.asyncio +async def test_dead_tier_is_skipped_instantly(p_tiny_floor): + calls = [] + + async def p_slow_boom(): + calls.append(1) + await asyncio.sleep(0.05) + raise RuntimeError("engine closed") + + tiers = [ + CascadeTier(name="ddg", run=p_slow_boom, budget=5.0, breaker=True), + CascadeTier(name="startpage", run=p_hit, budget=5.0, breaker=True), + ] + for _ in range(FAILURES_TO_OPEN): + out = await run_cascade(tiers, 5.0) + assert out.result == {"backend": "second"} + assert len(calls) == FAILURES_TO_OPEN + + out = await run_cascade(tiers, 5.0) + assert out.result == {"backend": "second"} + assert len(calls) == FAILURES_TO_OPEN, "a cooling tier must not be called at all" + assert any("skipped, still failing" in e for e in out.errors) + + +@pytest.mark.asyncio +async def test_timeout_counts_as_failure(p_tiny_floor): + async def p_hangs(): + await asyncio.sleep(30) + + tiers = [ + CascadeTier(name="ddg", run=p_hangs, budget=0.05, breaker=True), + CascadeTier(name="startpage", run=p_hit, budget=5.0, breaker=True), + ] + for _ in range(FAILURES_TO_OPEN): + await run_cascade(tiers, 5.0) + assert tier_cooldown_left("ddg") > 0 + + +@pytest.mark.asyncio +async def test_no_hits_is_not_a_failure(p_tiny_floor): + """An engine that answers 'nothing matched' is alive; only errors count against it.""" + async def p_empty(): + return None + + tiers = [ + CascadeTier(name="ddg", run=p_empty, budget=5.0, breaker=True), + CascadeTier(name="startpage", run=p_hit, budget=5.0, breaker=True), + ] + for _ in range(FAILURES_TO_OPEN + 2): + await run_cascade(tiers, 5.0) + assert tier_cooldown_left("ddg") == 0.0 + + +@pytest.mark.asyncio +async def test_tiers_without_breaker_always_run(p_tiny_floor): + """A fetch tier fails per-URL, so one dead page must never cool the tier for every other URL.""" + calls = [] + + async def p_fail(): + calls.append(1) + raise RuntimeError("404") + + tiers = [ + CascadeTier(name="local", run=p_fail, budget=5.0), + CascadeTier(name="wayback", run=p_hit, budget=5.0), + ] + for _ in range(FAILURES_TO_OPEN + 3): + await run_cascade(tiers, 5.0) + assert len(calls) == FAILURES_TO_OPEN + 3 + assert tier_cooldown_left("local") == 0.0 diff --git a/backend/tests/test_web_search_ddg.py b/backend/tests/test_web_search_ddg.py index b6897046..53ff9ac7 100644 --- a/backend/tests/test_web_search_ddg.py +++ b/backend/tests/test_web_search_ddg.py @@ -80,3 +80,46 @@ async def test_genuinely_empty_is_not_a_rate_limit(monkeypatch): p_patch_client(monkeypatch, p_reply(200, "nothing here")) out = await WebSearchTool.search_ddg("zxcvqwer no hits", 5) assert out == "" + + +P_LITE_RESULTS = """ + +
Lite Result
A snippet from the lite frontend.
+""" + + +def p_patch_split(monkeypatch, html_reply: HttpReply, lite_reply: HttpReply): + """Let the two DDG frontends answer differently, which is the whole point of having both.""" + async def p_html(url, **kw): + return html_reply + + async def p_lite(url, **kw): + return lite_reply + monkeypatch.setattr(SD, "browser_request", p_html) + monkeypatch.setattr(SDL, "browser_request", p_lite) + + +@pytest.mark.asyncio +async def test_html_403_still_tries_lite(monkeypatch): + """Measured 7 times in one 44-query round: html escalates from 202 to 403, and the old + code raised on the status without ever asking the second frontend.""" + p_patch_split(monkeypatch, p_reply(403, "blocked"), p_reply(200, P_LITE_RESULTS)) + out = await WebSearchTool.search_ddg("topic", 5) + assert "example.org/lite" in out + assert "Lite Result" in out + + +@pytest.mark.asyncio +async def test_html_403_and_lite_challenged_is_the_bot_challenge(monkeypatch): + p_patch_split(monkeypatch, p_reply(403, "blocked"), p_reply(202, "challenge")) + with pytest.raises(DDGRateLimited): + await WebSearchTool.search_ddg("topic", 5) + + +@pytest.mark.asyncio +async def test_both_frontends_erroring_names_both(monkeypatch): + p_patch_split(monkeypatch, p_reply(403, "blocked"), p_reply(500, "boom")) + with pytest.raises(RuntimeError) as exc: + await WebSearchTool.search_ddg("topic", 5) + assert "403" in str(exc.value) + assert "lite" in str(exc.value).lower() diff --git a/backend/tests/web_cascade_fixtures.py b/backend/tests/web_cascade_fixtures.py index eb7e90df..cfc0fb43 100644 --- a/backend/tests/web_cascade_fixtures.py +++ b/backend/tests/web_cascade_fixtures.py @@ -7,6 +7,15 @@ import backend.apps.agents.tools.search.search_startpage as SP import backend.apps.web.web as W from backend.apps.agents.tools.web import DDGRateLimited, WebSearchTool import backend.apps.agents.tools.ssrf_guard as p_ssrf +from backend.apps.web.tier_breaker import reset_tier_health + + +@pytest.fixture(autouse=True) +def fresh_breaker(): + # The breaker is per-process by design, so without this one test's forced outage cools the next test's tier. + reset_tier_health() + yield + reset_tier_health() @pytest.fixture(autouse=True)