refactor: enhance Google search scraping capabilities with warm sticky-IP pool and improved captcha handling

surfsense_backend/scripts/e2e_google_search.py

@@ -28,9 +28,12 @@ sys.path.insert(0, str(_ROOT))
 load_dotenv(_ROOT / ".env")
 
 logging.basicConfig(level=logging.WARNING)
-logging.getLogger("app.proprietary.platforms.google_search.scraper").setLevel(
-    logging.INFO
-)
+for _name in (
+    "app.proprietary.platforms.google_search.scraper",
+    "app.proprietary.platforms.google_search.fetch",
+    "app.proprietary.platforms.google_search.captcha",
+):
+    logging.getLogger(_name).setLevel(logging.INFO)
 
 from app.proprietary.platforms.google_search import (  # noqa: E402
     GoogleSearchScrapeInput,
@@ -146,6 +149,9 @@ async def run(
 
 _CASES = {
     "plain": lambda: run("plain query", queries="python asyncio tutorial"),
+    # Prod incident 2026-07-17: single-word brand/navigational queries were
+    # exhausting all 24 IPs (precheck 200 but the browser render 429-walled).
+    "brand": lambda: run("brand query (notebooklm, prod repro)", queries="notebooklm"),
     "site": lambda: run("site: filter", queries="machine learning", site="arxiv.org"),
     "ads": lambda: run("text ads", queries="car insurance quotes", expect_ads=True),
     "products": lambda: run(

surfsense_backend/scripts/scale_google_search.py

@@ -0,0 +1,229 @@
+"""Zero-cost concurrency simulation for the Google Search fetch seam.
+
+Drives many concurrent ``fetch_serp_html`` calls with the network I/O stubbed
+out (no live Google, no proxy spend, no captcha solves) so we can measure the
+*structure* of the scraper under load without paying for it:
+
+* **throughput** — completions/sec at steady state (the per-process ceiling is
+  the render gate divided by warm-render latency; nothing above it is real),
+* **latency** — p50/p95/p99 wait as the gate queues excess renders,
+* **solves** — how many paid solves the run would cost (the number we most want
+  the sticky-IP pool to shrink), and
+* **per-IP peak concurrency** — the funneling metric: how many renders pile onto
+  a single sticky IP at once (a hot IP is what Google re-walls).
+
+The stub keeps the *real* ``fetch_serp_html`` loop intact (sticky reuse,
+exemption skip-precheck, the render gate, inflight accounting) and only replaces
+``_precheck`` / ``_get_session`` / ``get_proxy_url``. Timings are compressed
+(warm≈0.1 s vs ~14 s live) and extrapolated to real seconds via the warm-render
+ratio, since the system is gate-bounded sleeps + a semaphore (delays scale
+linearly).
+
+    .venv/Scripts/python.exe scripts/scale_google_search.py --count 400 --rate 60
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import statistics
+import sys
+import threading
+import time
+from pathlib import Path
+
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+
+_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(_ROOT))
+
+from app.proprietary.platforms.google_search import fetch  # noqa: E402
+
+# Live-measured render latencies (README "Timings"); used only to translate the
+# compressed sim clock back to real seconds for the human-readable report.
+_REAL_WARM_S = 14.0
+_REAL_SOLVE_S = 45.0
+
+_RESULTS_HTML = '<div id="rso">ok</div>'  # trips fetch._has_results
+
+
+class _Metrics:
+    """Browser-loop-side counters (fetch runs on two loops → guard with a lock)."""
+
+    def __init__(self) -> None:
+        self.lock = threading.Lock()
+        self.solves = 0
+        self.renders = 0
+        self.ip_hits: dict[str, int] = {}
+        self.ip_now: dict[str, int] = {}
+        self.ip_peak: dict[str, int] = {}
+
+    def enter(self, proxy: str, solved: bool) -> None:
+        with self.lock:
+            self.renders += 1
+            if solved:
+                self.solves += 1
+            self.ip_hits[proxy] = self.ip_hits.get(proxy, 0) + 1
+            n = self.ip_now.get(proxy, 0) + 1
+            self.ip_now[proxy] = n
+            self.ip_peak[proxy] = max(self.ip_peak.get(proxy, 0), n)
+
+    def exit(self, proxy: str) -> None:
+        with self.lock:
+            self.ip_now[proxy] = self.ip_now.get(proxy, 1) - 1
+
+
+class _FakePage:
+    def __init__(self) -> None:
+        self.status = 200
+        self.html_content = _RESULTS_HTML
+
+
+class _FakeSession:
+    """Stands in for AsyncStealthySession: sleeps like a render, simulates the
+    solve on a cold IP by seeding fetch._exemption_jar (what page_action does)."""
+
+    def __init__(self, metrics: _Metrics, warm_s: float, solve_s: float) -> None:
+        self.m = metrics
+        self.warm_s = warm_s
+        self.solve_s = solve_s
+
+    async def fetch(self, url, proxy=None, **kwargs):
+        key = proxy or ""
+        # Cold IP (no cached exemption) pays the solve and warms the jar; a warm
+        # IP just pays the render. Mirrors _make_page_action's cache write.
+        cold = key not in fetch._exemption_jar
+        self.m.enter(key, solved=cold)
+        try:
+            if cold:
+                await asyncio.sleep(self.solve_s)
+                fetch._exemption_jar[key] = [{"name": "GOOGLE_ABUSE_EXEMPTION"}]
+            await asyncio.sleep(self.warm_s)
+            return _FakePage()
+        finally:
+            self.m.exit(key)
+
+
+def _install_stubs(metrics: _Metrics, warm_s: float, solve_s: float, precheck_s: float):
+    base = "http://u:p@gw.dataimpulse.com:823"  # hostname in fetch._STICKY_HOSTS
+    session = _FakeSession(metrics, warm_s, solve_s)
+
+    async def fake_precheck(url, proxy):
+        await asyncio.sleep(precheck_s)
+        return True
+
+    async def fake_get_session(mobile):
+        return session
+
+    fetch.get_proxy_url = lambda: base
+    fetch._precheck = fake_precheck
+    fetch._get_session = fake_get_session
+    fetch.captcha_enabled = lambda: True
+    fetch.get_captcha_config = lambda: object()
+    fetch._captcha.solver_latched = lambda: False
+    # The pool's backpressure poll is negligible vs a real 14 s render; keep it
+    # proportional under the compressed sim clock so waiters don't idle.
+    fetch._POOL_WAIT_S = max(warm_s * 0.05, 0.001)
+
+    # Isolate the sim to the LOCAL pool: the cross-process store (Redis) has its
+    # own unit test and would otherwise add ping latency / cross-run bleed here.
+    async def _no_adopt(exclude):
+        return None
+
+    async def _noop(*a, **k):
+        return None
+
+    fetch._store.adopt = _no_adopt
+    fetch._store.publish = _noop
+    fetch._store.evict = _noop
+
+
+def _reset_state() -> None:
+    fetch._exemption_jar.clear()
+    for name in ("_pool", "_pool_inflight"):
+        obj = getattr(fetch, name, None)
+        if isinstance(obj, dict):
+            obj.clear()
+    if hasattr(fetch, "_pool_pending"):
+        fetch._pool_pending = 0
+    if hasattr(fetch, "_last_good_proxy"):  # pre-pool revision
+        fetch._last_good_proxy = None
+
+
+async def _drive(count: int, rate: float) -> list[float]:
+    """Fire ``count`` fetches at ``rate``/sec (steady arrival), return latencies."""
+    lat: list[float] = []
+    interval = 1.0 / rate if rate > 0 else 0.0
+
+    async def one() -> None:
+        t0 = time.perf_counter()
+        await fetch.fetch_serp_html("https://www.google.com/search?q=notebooklm")
+        lat.append(time.perf_counter() - t0)
+
+    tasks: list[asyncio.Task] = []
+    for _ in range(count):
+        tasks.append(asyncio.create_task(one()))
+        if interval:
+            await asyncio.sleep(interval)
+    await asyncio.gather(*tasks)
+    return lat
+
+
+def _pct(xs: list[float], p: float) -> float:
+    if not xs:
+        return 0.0
+    xs = sorted(xs)
+    return xs[min(len(xs) - 1, int(p / 100 * len(xs)))]
+
+
+async def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--count", type=int, default=400)
+    ap.add_argument("--rate", type=float, default=60.0, help="arrivals/sec (sim clock)")
+    ap.add_argument("--warm", type=float, default=0.10, help="warm render seconds (sim)")
+    ap.add_argument("--solve", type=float, default=0.45, help="solve seconds (sim)")
+    ap.add_argument("--precheck", type=float, default=0.01)
+    args = ap.parse_args()
+
+    metrics = _Metrics()
+    _install_stubs(metrics, args.warm, args.solve, args.precheck)
+    _reset_state()
+
+    scale = _REAL_WARM_S / args.warm  # sim seconds → real seconds
+    gate = fetch._MAX_CONCURRENT_PAGES
+    pool = getattr(fetch, "_WARM_POOL_TARGET", None)
+
+    wall0 = time.perf_counter()
+    lat = await _drive(args.count, args.rate)
+    wall = time.perf_counter() - wall0
+    await fetch._in_browser_loop(asyncio.sleep(0))  # let last exit() settle
+
+    hot_peak = max(metrics.ip_peak.values()) if metrics.ip_peak else 0
+    top_hits = max(metrics.ip_hits.values()) if metrics.ip_hits else 0
+    top_share = 100.0 * top_hits / metrics.renders if metrics.renders else 0.0
+    thru_sim = args.count / wall if wall else 0.0
+    thru_real = thru_sim / scale
+
+    print("\n=== Google Search scale simulation ===")
+    print(f"  gate (_MAX_CONCURRENT_PAGES) = {gate}"
+          + (f", warm-pool target = {pool}" if pool else "  (single-slot sticky IP)"))
+    print(f"  requests={args.count} arrival_rate={args.rate}/s (sim)")
+    print(f"  --- structure (scale-free) ---")
+    print(f"  paid solves           = {metrics.solves}  (want ~pool size, not ~requests)")
+    print(f"  distinct sticky IPs   = {len(metrics.ip_hits)}")
+    print(f"  busiest IP carried    = {top_hits}/{metrics.renders} renders "
+          f"({top_share:.0f}%)  (funneling; want ~even spread)")
+    print(f"  peak renders on 1 IP  = {hot_peak}  (concurrency; capped by gate)")
+    print(f"  --- throughput / latency (extrapolated to live @ warm={_REAL_WARM_S}s) ---")
+    print(f"  throughput  = {thru_real*60:.0f} SERP/min  ({thru_real:.2f}/s)")
+    print(f"  latency p50 = {_pct(lat,50)*scale:6.1f}s   p95 = {_pct(lat,95)*scale:6.1f}s   "
+          f"p99 = {_pct(lat,99)*scale:6.1f}s")
+    print(f"  ceiling (gate/warm) = {gate/_REAL_WARM_S*60:.0f} SERP/min per process")
+    need = 500 / (gate / _REAL_WARM_S * 60)
+    print(f"  -> to sustain 500 SERP/min you need ~{need:.0f} such processes "
+          f"(or a larger gate)\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

surfsense_backend/scripts/stress_google_search.py

@@ -0,0 +1,207 @@
+"""Live stress test for the Google Search scraper (REAL solves + proxy spend).
+
+Simulates a burst of concurrent "users" hitting the real ``scrape_serps``
+pipeline with a mix of single-query and multi-query requests. Unlike the offline
+sim (``scale_google_search.py``), this exercises the actual DataImpulse IPs,
+CapSolver solves, and the shared browser — so it surfaces what the sim can't:
+real warm-vs-cold latency, whether the warm-IP pool amortizes solves across real
+IPs without Google re-walling them, and browser stability under concurrency.
+
+    .venv/Scripts/python.exe scripts/stress_google_search.py --users 30 --gate 8
+
+It COSTS money (each cold IP = one CapSolver solve, bounded by the pool target)
+and proxy bandwidth. Metrics are tallied from the live perf/captcha logs.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+import os
+import random
+import re
+import sys
+import time
+from pathlib import Path
+
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+
+_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(_ROOT))
+
+from dotenv import load_dotenv  # noqa: E402
+
+load_dotenv(_ROOT / ".env")
+
+# Brand/navigational terms reliably trip the reCAPTCHA wall (a solve/pool grow);
+# informational terms are the lighter, often-warm path. A realistic mix.
+_BRAND = ["notebooklm", "figma", "notion", "linear app", "vercel", "perplexity ai"]
+_INFO = [
+    "python asyncio tutorial",
+    "best mechanical keyboard 2026",
+    "kubernetes vs docker",
+    "typescript generics explained",
+    "how to make sourdough bread",
+    "rust ownership model",
+]
+
+
+class _LogTally(logging.Handler):
+    """Counts solves / renders / walls / pool-reuse from the live log stream."""
+
+    _RENDER = re.compile(
+        r"has_results=(\w+).*from_pool=(\w+) pool=(\d+)"
+    )
+    _SOLVE = re.compile(r"\[captcha\] solve (OK|did not)")
+
+    def __init__(self) -> None:
+        super().__init__(level=logging.INFO)
+        self.solves_ok = 0
+        self.solves_fail = 0
+        self.renders_ok = 0
+        self.renders_walled = 0
+        self.reuse = 0
+        self.grow = 0
+        self.max_pool = 0
+
+    def emit(self, record: logging.LogRecord) -> None:
+        msg = record.getMessage()
+        m = self._RENDER.search(msg)
+        if m:
+            ok, from_pool, pool = m.group(1), m.group(2), int(m.group(3))
+            if ok == "True":
+                self.renders_ok += 1
+            else:
+                self.renders_walled += 1
+            if from_pool == "True":
+                self.reuse += 1
+            else:
+                self.grow += 1
+            self.max_pool = max(self.max_pool, pool)
+            return
+        s = self._SOLVE.search(msg)
+        if s:
+            if s.group(1) == "OK":
+                self.solves_ok += 1
+            else:
+                self.solves_fail += 1
+
+
+def _make_request(rng: random.Random, multi_ratio: float) -> tuple[str, int]:
+    """Build one user's request: a single term or a 2-3 term multi-query.
+    Returns (newline-joined queries, expected SERP count)."""
+    pool = _BRAND + _INFO
+    if rng.random() < multi_ratio:
+        n = rng.randint(2, 3)
+        terms = rng.sample(pool, n)
+        return "\n".join(terms), n
+    return rng.choice(pool), 1
+
+
+async def _user(uid: int, queries: str, want: int, results: list[dict]) -> None:
+    from app.proprietary.platforms.google_search import (
+        GoogleSearchScrapeInput,
+        scrape_serps,
+    )
+
+    t0 = time.perf_counter()
+    inp = GoogleSearchScrapeInput(queries=queries, countryCode="us", languageCode="en")
+    try:
+        items = await scrape_serps(inp, limit=want)
+        got = len(items)
+        err = None
+    except Exception as e:  # a user request should never crash the whole run
+        got, err = 0, repr(e)
+    results.append(
+        {
+            "uid": uid,
+            "want": want,
+            "got": got,
+            "secs": time.perf_counter() - t0,
+            "err": err,
+        }
+    )
+
+
+async def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--users", type=int, default=30, help="concurrent requests")
+    ap.add_argument("--gate", type=int, default=8, help="MAX_CONCURRENT_PAGES for run")
+    ap.add_argument("--multi-ratio", type=float, default=0.5)
+    ap.add_argument("--ramp", type=float, default=0.3, help="secs between user starts")
+    ap.add_argument("--seed", type=int, default=7)
+    args = ap.parse_args()
+
+    # Must be set BEFORE fetch.py is imported (the gate is read at import).
+    os.environ["GOOGLE_SEARCH_MAX_CONCURRENT_PAGES"] = str(args.gate)
+
+    logging.basicConfig(level=logging.WARNING)
+    tally = _LogTally()
+    for name in (
+        "app.proprietary.platforms.google_search.fetch",
+        "app.proprietary.platforms.google_search.captcha",
+    ):
+        lg = logging.getLogger(name)
+        lg.setLevel(logging.INFO)
+        lg.addHandler(tally)
+
+    from app.proprietary.platforms.google_search.fetch import (  # noqa: E402
+        _WARM_POOL_TARGET,
+        close_sessions,
+    )
+
+    rng = random.Random(args.seed)
+    plans = [_make_request(rng, args.multi_ratio) for _ in range(args.users)]
+    total_serps = sum(w for _, w in plans)
+    n_multi = sum(1 for _, w in plans if w > 1)
+
+    print(f"\n=== LIVE stress: {args.users} users "
+          f"({args.users - n_multi} single / {n_multi} multi), "
+          f"~{total_serps} SERPs, gate={args.gate}, pool_target={_WARM_POOL_TARGET} ===")
+    print("  (real solves + proxy spend; warming up...)\n")
+
+    results: list[dict] = []
+    wall0 = time.perf_counter()
+    tasks = []
+    for uid, (queries, want) in enumerate(plans):
+        tasks.append(asyncio.create_task(_user(uid, queries, want, results)))
+        await asyncio.sleep(args.ramp)  # gentle ramp, not a thundering herd
+    await asyncio.gather(*tasks)
+    wall = time.perf_counter() - wall0
+
+    await close_sessions()
+
+    lat = sorted(r["secs"] for r in results)
+    got_serps = sum(r["got"] for r in results)
+    fails = [r for r in results if r["err"]]
+    short = [r for r in results if not r["err"] and r["got"] < r["want"]]
+
+    def pct(p: float) -> float:
+        return lat[min(len(lat) - 1, int(p / 100 * len(lat)))] if lat else 0.0
+
+    print("=== results ===")
+    print(f"  wall time            = {wall:.0f}s")
+    print(f"  requests ok/failed   = {len(results) - len(fails)}/{len(fails)}")
+    print(f"  SERPs got/expected   = {got_serps}/{total_serps}"
+          + (f"  ({len(short)} short)" if short else ""))
+    print(f"  throughput           = {got_serps / wall * 60:.0f} SERP/min")
+    print(f"  request latency p50={pct(50):.0f}s  p95={pct(95):.0f}s  "
+          f"max={lat[-1] if lat else 0:.0f}s")
+    print("  --- pipeline (from live logs) ---")
+    print(f"  paid solves ok/fail  = {tally.solves_ok}/{tally.solves_fail}  "
+          f"(bounded by pool target {_WARM_POOL_TARGET})")
+    print(f"  renders ok/walled    = {tally.renders_ok}/{tally.renders_walled}")
+    print(f"  pool reuse/grow      = {tally.reuse}/{tally.grow}  "
+          f"(reuse share {100 * tally.reuse / max(1, tally.reuse + tally.grow):.0f}%)")
+    print(f"  peak pool size       = {tally.max_pool}")
+    if fails:
+        print("  --- failures ---")
+        for r in fails[:5]:
+            print(f"    user{r['uid']}: {r['err']}")
+    print()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())