From 47a73f7dcf0c6dd960b69319d16feb2503c04238 Mon Sep 17 00:00:00 2001 From: debpalash <4178343+debpalash@users.noreply.github.com> Date: Tue, 13 Oct 2026 23:43:14 +0530 Subject: [PATCH] perf(leadgen): reuse pooled httpx clients in plain fetch StealthClient._fetch_plain built and tore down a new httpx.AsyncClient per fetch, paying a fresh TCP/TLS handshake every time. Pool clients keyed by (running event loop, proxy) with a bounded LRU (16 per loop), close evicted clients once idle, and add aclose_plain_clients() wired into app shutdown. Security properties are unchanged: follow_redirects=False, check_url on every hop, _MAX_REDIRECTS enforced, timeout passed per request. The shared client uses a reject-all cookie jar so no cookie can persist across fetches or tenants; cookies within a single redirect chain are replayed from a per-fetch jar, matching the old per-call behaviour. Part of #33 Signed-off-by: debpalash Co-Authored-By: Claude Opus 5.5 --- apps/api/main.py | 7 + apps/api/services/leadgen/http.py | 130 +++++++++++++- tests/test_leadgen_http_client_reuse.py | 224 ++++++++++++++++++++++++ 3 files changed, 353 insertions(+), 8 deletions(-) create mode 100644 tests/test_leadgen_http_client_reuse.py diff --git a/apps/api/main.py b/apps/api/main.py index dfbd14e1..00d1ca3e 100644 --- a/apps/api/main.py +++ b/apps/api/main.py @@ -147,6 +147,13 @@ async def lifespan(app: FastAPI): if run_inline_worker: await queue_service.stop_worker() print("✓ Queue Worker Stopped") + # Release pooled Tier-1 lead-gen HTTP clients (keep-alive connections). + try: + from apps.api.services.leadgen.http import aclose_plain_clients + + await aclose_plain_clients() + except Exception: + logger.warning("Closing pooled leadgen HTTP clients failed", exc_info=True) app = FastAPI(title="OpenGTM", version="3.0.0", lifespan=lifespan) diff --git a/apps/api/services/leadgen/http.py b/apps/api/services/leadgen/http.py index 41349df7..23f8bf14 100644 --- a/apps/api/services/leadgen/http.py +++ b/apps/api/services/leadgen/http.py @@ -26,8 +26,12 @@ """ import asyncio +import http.cookiejar import logging import re +import weakref +from collections import OrderedDict +from contextlib import asynccontextmanager from dataclasses import dataclass, field from typing import Optional from urllib.parse import urljoin, urlparse @@ -180,6 +184,116 @@ def extract_phones(self) -> list[str]: } +# ── Pooled Tier-1 httpx clients ────────────────────────────────────── +# +# ``_fetch_plain`` used to build (and tear down) a fresh ``httpx.AsyncClient`` +# per fetch, paying a new TCP/TLS handshake every time. Clients are now pooled +# and reused, with these invariants: +# +# * Keyed by event loop AND proxy. httpx clients are bound to the loop they +# were first used on, and the proxy is a client-level setting. Loops are +# held weakly so a finished loop (tests, ``asyncio.run`` in workers) drops +# its clients instead of handing them to a different loop. +# * Bounded: at most ``_MAX_PLAIN_CLIENTS`` per loop (LRU). An evicted client +# is closed once its in-flight requests finish. +# * Stateless across requests: the client's cookie jar rejects every cookie, +# so nothing a response sets can reach a later request (or another tenant / +# workspace). Cookies set during ONE fetch's redirect chain are carried in a +# per-fetch jar, matching the old per-call-client behaviour. +# * Security checks are unchanged: ``follow_redirects=False`` and +# ``check_url`` runs on every hop; timeouts are applied per request. + +_MAX_PLAIN_CLIENTS = 16 + + +class _RejectAllCookiesPolicy(http.cookiejar.DefaultCookiePolicy): + """Cookie policy that never stores or returns a cookie.""" + + def set_ok(self, cookie, request): + return False + + def return_ok(self, cookie, request): + return False + + +@dataclass +class _PooledClient: + client: object + in_use: int = 0 + evicted: bool = False + + +_plain_clients: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, OrderedDict[str, _PooledClient]]" = ( + weakref.WeakKeyDictionary() +) + + +def _new_plain_client(proxy: str | None): + import httpx + + return httpx.AsyncClient( + headers=_DEFAULT_HEADERS, + follow_redirects=False, + proxy=proxy or None, + verify=True, + # Pass the raw CookieJar: httpx copies an ``httpx.Cookies`` argument into + # a fresh default-policy jar, which would silently drop the policy. + cookies=http.cookiejar.CookieJar(policy=_RejectAllCookiesPolicy()), + ) + + +@asynccontextmanager +async def _plain_client(proxy: str | None): + """Borrow the pooled httpx client for (running loop, proxy).""" + loop = asyncio.get_running_loop() + pool = _plain_clients.get(loop) + if pool is None: + pool = OrderedDict() + _plain_clients[loop] = pool + + key = proxy or "" + entry = pool.get(key) + if entry is None or entry.client.is_closed: + entry = _PooledClient(client=_new_plain_client(proxy)) + pool[key] = entry + while len(pool) > _MAX_PLAIN_CLIENTS: + _, old = pool.popitem(last=False) + old.evicted = True + if old.in_use == 0: + await old.client.aclose() + else: + pool.move_to_end(key) + + entry.in_use += 1 + try: + yield entry.client + finally: + entry.in_use -= 1 + if entry.evicted and entry.in_use == 0: + await entry.client.aclose() + + +async def aclose_plain_clients() -> None: + """Close the pooled Tier-1 clients owned by the running event loop. + + Call on app/worker shutdown. Clients of other (possibly closed) loops can't + be awaited from here; they are released when their loop is collected. + """ + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return + pool = _plain_clients.pop(loop, None) + if not pool: + return + for entry in pool.values(): + entry.evicted = True + try: + await entry.client.aclose() + except Exception: # pragma: no cover - best-effort shutdown + logger.debug("error closing pooled httpx client", exc_info=True) + + class StealthClient: """ Unified stealth HTTP client with automatic browser-tier escalation. @@ -422,17 +536,17 @@ async def _fetch_plain(self, url: str, proxy: Optional[str], timeout: int) -> Fe try: import httpx - async with httpx.AsyncClient( - headers=_DEFAULT_HEADERS, - timeout=timeout, - follow_redirects=False, - proxy=proxy or None, - verify=True, - ) as client: + async with _plain_client(proxy) as client: + # Per-fetch cookie jar: cookies set during this redirect chain + # are replayed within it, but never stored on the shared client. + fetch_cookies = httpx.Cookies() current = url for _hop in range(_MAX_REDIRECTS + 1): check_url(current, allow_http=True, resolve=True) - resp = await client.get(current) + request = client.build_request("GET", current, timeout=timeout) + fetch_cookies.set_cookie_header(request) + resp = await client.send(request) + fetch_cookies.extract_cookies(resp) if resp.status_code not in (301, 302, 303, 307, 308): break location = resp.headers.get("location") diff --git a/tests/test_leadgen_http_client_reuse.py b/tests/test_leadgen_http_client_reuse.py new file mode 100644 index 00000000..7045b5ea --- /dev/null +++ b/tests/test_leadgen_http_client_reuse.py @@ -0,0 +1,224 @@ +"""Tier-1 (``_fetch_plain``) httpx client pooling — offline tests. + +Covers: client reuse per (loop, proxy), isolation across proxies and loops, +no cookie leakage between fetches, per-hop ``check_url`` on redirects, the +LRU bound, and ``aclose_plain_clients``. +""" + +import asyncio + +import httpx +import pytest + +from apps.api.services.leadgen import http as H +from apps.api.services.leadgen.http import StealthClient + + +@pytest.fixture +def mock_httpx(monkeypatch): + """Route every new AsyncClient through a MockTransport; record clients/requests.""" + state = {"clients": [], "proxies": [], "requests": [], "handler": None} + + def default_handler(request): + return httpx.Response(200, text="ok") + + real_init = httpx.AsyncClient.__init__ + + def init(self, *args, **kwargs): + state["proxies"].append(kwargs.pop("proxy", None)) + + def handler(request): + state["requests"].append(request) + return (state["handler"] or default_handler)(request) + + kwargs["transport"] = httpx.MockTransport(handler) + real_init(self, *args, **kwargs) + state["clients"].append(self) + + monkeypatch.setattr(httpx.AsyncClient, "__init__", init) + # Public-looking literal IPs don't need DNS; keep check_url real but offline. + return state + + +URL = "http://93.184.216.34/page" + + +def test_client_reused_across_fetches_same_proxy(mock_httpx): + async def run(): + c = StealthClient() + r1 = await c._fetch_plain(URL, None, 5) + r2 = await c._fetch_plain(URL, None, 5) + await H.aclose_plain_clients() + return r1, r2 + + r1, r2 = asyncio.run(run()) + assert r1.status_code == r2.status_code == 200 and not r1.error + assert len(mock_httpx["clients"]) == 1 + assert len(mock_httpx["requests"]) == 2 + + +def test_different_proxies_get_different_clients(mock_httpx): + async def run(): + c = StealthClient() + await c._fetch_plain(URL, "http://proxy-a:8080", 5) + await c._fetch_plain(URL, "http://proxy-b:8080", 5) + await c._fetch_plain(URL, "http://proxy-a:8080", 5) + await H.aclose_plain_clients() + + asyncio.run(run()) + assert mock_httpx["proxies"] == ["http://proxy-a:8080", "http://proxy-b:8080"] + + +def test_clients_not_shared_across_event_loops(mock_httpx): + async def run(): + await StealthClient()._fetch_plain(URL, None, 5) + + asyncio.run(run()) + asyncio.run(run()) + assert len(mock_httpx["clients"]) == 2 + + +def test_timeout_applied_per_request(mock_httpx): + async def run(): + c = StealthClient() + await c._fetch_plain(URL, None, 3) + await c._fetch_plain(URL, None, 11) + await H.aclose_plain_clients() + + asyncio.run(run()) + t1, t2 = (r.extensions["timeout"] for r in mock_httpx["requests"]) + assert t1["read"] == 3 and t2["read"] == 11 + + +def test_cookies_do_not_leak_between_fetches(mock_httpx): + def handler(request): + if request.url.path == "/set": + return httpx.Response( + 200, text="x" * 600, headers={"set-cookie": "session=tenantA; Path=/"} + ) + return httpx.Response(200, text="ok") + + mock_httpx["handler"] = handler + + async def run(): + c = StealthClient() + await c._fetch_plain("http://93.184.216.34/set", None, 5) + await c._fetch_plain("http://93.184.216.34/other", None, 5) + client = mock_httpx["clients"][0] + jar_size = len(client.cookies.jar) + await H.aclose_plain_clients() + return jar_size + + jar_size = asyncio.run(run()) + assert len(mock_httpx["clients"]) == 1 + assert jar_size == 0 + assert "cookie" not in mock_httpx["requests"][1].headers + + +def test_cookies_carried_within_one_redirect_chain_only(mock_httpx): + def handler(request): + if request.url.path == "/login": + return httpx.Response( + 302, headers={"location": "/home", "set-cookie": "sid=abc; Path=/"} + ) + return httpx.Response(200, text="home") + + mock_httpx["handler"] = handler + + async def run(): + c = StealthClient() + await c._fetch_plain("http://93.184.216.34/login", None, 5) + await c._fetch_plain("http://93.184.216.34/home", None, 5) + await H.aclose_plain_clients() + + asyncio.run(run()) + reqs = mock_httpx["requests"] + assert [r.url.path for r in reqs] == ["/login", "/home", "/home"] + assert reqs[1].headers.get("cookie") == "sid=abc" # same fetch: replayed + assert "cookie" not in reqs[2].headers # later fetch: not leaked + + +def test_check_url_runs_on_every_redirect_hop(mock_httpx, monkeypatch): + def handler(request): + n = int(request.url.path.strip("/") or 0) + if n < 3: + return httpx.Response(302, headers={"location": f"/{n + 1}"}) + return httpx.Response(200, text="done") + + mock_httpx["handler"] = handler + checked = [] + real_check = H.check_url + + def spy(url, **kw): + checked.append((url, kw)) + return real_check(url, **kw) + + monkeypatch.setattr(H, "check_url", spy) + + async def run(): + r = await StealthClient()._fetch_plain("http://93.184.216.34/0", None, 5) + await H.aclose_plain_clients() + return r + + r = asyncio.run(run()) + assert r.status_code == 200 + assert [u for u, _ in checked] == [f"http://93.184.216.34/{i}" for i in range(4)] + assert all(kw == {"allow_http": True, "resolve": True} for _, kw in checked) + + +def test_redirect_to_private_target_blocked_with_pooled_client(mock_httpx): + mock_httpx["handler"] = lambda req: httpx.Response( + 302, headers={"location": "http://127.0.0.1/admin"} + ) + + async def run(): + c = StealthClient() + r = await c._fetch_plain(URL, None, 5) + await H.aclose_plain_clients() + return r + + r = asyncio.run(run()) + assert r.status_code == 0 and "private/blocked" in r.error + assert [str(q.url) for q in mock_httpx["requests"]] == [URL] + + +def test_too_many_redirects_still_enforced(mock_httpx): + mock_httpx["handler"] = lambda req: httpx.Response(302, headers={"location": URL}) + + async def run(): + r = await StealthClient()._fetch_plain(URL, None, 5) + await H.aclose_plain_clients() + return r + + r = asyncio.run(run()) + assert r.status_code == 0 and "too many redirects" in r.error + assert len(mock_httpx["requests"]) == H._MAX_REDIRECTS + 1 + + +def test_pool_is_bounded_and_evicted_clients_closed(mock_httpx, monkeypatch): + monkeypatch.setattr(H, "_MAX_PLAIN_CLIENTS", 2) + + async def run(): + c = StealthClient() + for p in ("http://p1:1", "http://p2:1", "http://p3:1"): + await c._fetch_plain(URL, p, 5) + pool = H._plain_clients[asyncio.get_running_loop()] + keys = list(pool) + await H.aclose_plain_clients() + return keys + + keys = asyncio.run(run()) + assert keys == ["http://p2:1", "http://p3:1"] + assert mock_httpx["clients"][0].is_closed + + +def test_aclose_plain_clients_closes_and_clears(mock_httpx): + async def run(): + await StealthClient()._fetch_plain(URL, None, 5) + loop = asyncio.get_running_loop() + assert loop in H._plain_clients + await H.aclose_plain_clients() + return loop in H._plain_clients + + assert asyncio.run(run()) is False + assert mock_httpx["clients"][0].is_closed