From 9731bcf28a5fe7fda012030784db4a2fb00b8a37 Mon Sep 17 00:00:00 2001 From: inchang-ing <291394088@qq.com> Date: Thu, 5 Nov 2026 16:04:45 +0800 Subject: [PATCH] fix(browserless): construct the fetch and screenshot clients off the event loop (#7913) * fix(browserless): construct the fetch and screenshot clients off the event loop * fix(browserless): close the client a cancelled construction still builds Keeping the constructor off the loop is right, but awaiting `asyncio.to_thread(httpx.AsyncClient, ...)` directly is not cancellation-safe: the await can be cancelled while the worker thread is still inside the constructor. The await is abandoned, the thread runs on and hands back a live client, and execution never enters the `async with` that would have closed it, so `is_closed` stays False and the connection pool leaks on that cancellation. Own the construction as a task instead. When the caller is cancelled, wait for the task with the shield loop this package already uses for cleanup, close whatever client it produced, and only then let the cancellation propagate. A failure while closing is logged rather than raised, so it cannot mask the cancellation the caller asked for. The new regression test gates the constructor on a thread event, cancels the operation while it is parked there, and releases the worker afterwards. Against the previous commit it fails for both fetch_html_with_status and capture_screenshot with "the client built after the cancellation was leaked". --------- Co-authored-by: inchang-ing <197932532+inchang-ing@users.noreply.github.com> --- backend/AGENTS.md | 8 +- .../browserless/browserless_client.py | 49 +++++++- .../test_browserless_client_ssl.py | 119 ++++++++++++++++++ 3 files changed, 171 insertions(+), 5 deletions(-) create mode 100644 backend/tests/blocking_io/test_browserless_client_ssl.py diff --git a/backend/AGENTS.md b/backend/AGENTS.md index d0d1bd7700..236f67134a 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -136,9 +136,11 @@ Read the nearest scoped `AGENTS.md` for subsystem ownership. Read-only knowledge: [contract](docs/KNOWLEDGE.md) covers independent retrieval tools, scope intersection and extension-owned providers. -HTTPX clients for RAGFlow, LightRAG, You.com, GroundRoute, Webz and Crawl4AI use -`asyncio.to_thread` for synchronous TLS setup; requests and cleanup stay on-loop. -The list is non-exhaustive: jsonify, SearXNG, Browserless, Jina and InfoQuest +HTTPX clients for RAGFlow, LightRAG, You.com, GroundRoute, Webz, Crawl4AI and +Browserless use `asyncio.to_thread` for synchronous TLS setup; requests and +cleanup stay on-loop, and the construction is owned as a task so cancelling +during setup still closes the client it builds. +The list is non-exhaustive: jsonify, SearXNG, Jina and InfoQuest still construct their clients on the loop and should follow the same rule when touched. diff --git a/backend/packages/harness/deerflow/community/browserless/browserless_client.py b/backend/packages/harness/deerflow/community/browserless/browserless_client.py index 75c339f250..7640a85a4f 100644 --- a/backend/packages/harness/deerflow/community/browserless/browserless_client.py +++ b/backend/packages/harness/deerflow/community/browserless/browserless_client.py @@ -1,3 +1,4 @@ +import asyncio import logging from dataclasses import dataclass from typing import Any @@ -30,6 +31,46 @@ def _get_header(headers: Any, name: str) -> str: return str(headers.get(name.lower(), "")) +async def _settle(future: "asyncio.Future[Any]") -> Any: + """Await a future that owns a resource, even if this task is cancelled again. + + Cleanup must not be the thing that gets abandoned: the caller remembers its + own cancellation and re-raises it once the resource is released. + """ + while True: + try: + return await asyncio.shield(future) + except asyncio.CancelledError: + if future.cancelled(): + raise + + +async def _open_client(**kwargs: Any) -> httpx.AsyncClient: + """Build an ``httpx.AsyncClient`` without loading TLS trust on the event loop. + + The constructor resolves the system trust store synchronously, so it has to + run in a thread. That alone is not enough: a cancellation can arrive while + the worker is still inside the constructor, and then the await is abandoned + while the thread goes on to hand back a live client that nobody closes. So + own the construction as a task and, once cancelled, wait for it and close + whatever it produced before the cancellation propagates. + """ + construction: asyncio.Task[httpx.AsyncClient] = asyncio.create_task(asyncio.to_thread(httpx.AsyncClient, **kwargs)) + try: + return await asyncio.shield(construction) + except asyncio.CancelledError: + try: + late_client = await _settle(construction) + except Exception as error: + logger.error(f"Browserless HTTPX client construction failed after cancellation: {error!s}") + else: + try: + await _settle(asyncio.create_task(late_client.aclose())) + except Exception as error: + logger.error(f"Closing a Browserless HTTPX client built after cancellation failed: {error!s}") + raise + + class BrowserlessClient: """Client for Browserless headless Chrome API.""" @@ -142,7 +183,9 @@ class BrowserlessClient: logger.debug(f"Fetching URL via Browserless: {url}") try: - async with httpx.AsyncClient(timeout=self.timeout_s) as client: + # Construction loads TLS trust synchronously, off the loop, and owns the + # client even when this coroutine is cancelled mid-construction. + async with await _open_client(timeout=self.timeout_s) as client: resp = await client.post( f"{self.base_url}/content", json=payload, @@ -232,7 +275,9 @@ class BrowserlessClient: logger.debug(f"Capturing URL screenshot via Browserless: {url}") try: - async with httpx.AsyncClient(timeout=self.timeout_s) as client: + # Construction loads TLS trust synchronously, off the loop, and owns the + # client even when this coroutine is cancelled mid-construction. + async with await _open_client(timeout=self.timeout_s) as client: resp = await client.post( f"{self.base_url}/screenshot", json=payload, diff --git a/backend/tests/blocking_io/test_browserless_client_ssl.py b/backend/tests/blocking_io/test_browserless_client_ssl.py new file mode 100644 index 0000000000..edb70c7059 --- /dev/null +++ b/backend/tests/blocking_io/test_browserless_client_ssl.py @@ -0,0 +1,119 @@ +"""Browserless must load HTTPX TLS trust off the agent event loop.""" + +import asyncio +import ssl +import threading + +import httpx +import pytest + +from deerflow.community.browserless.browserless_client import BrowserlessClient + + +@pytest.mark.asyncio +@pytest.mark.parametrize("operation", ["fetch_html_with_status", "capture_screenshot"]) +async def test_browserless_client_tls_setup_runs_off_loop(operation, monkeypatch): + loop_thread = threading.get_ident() + original_context = ssl.create_default_context + contexts = [] + clients = [] + + def create_context(*args, **kwargs): + assert threading.get_ident() != loop_thread, "TLS trust loading blocks the event loop" + context = original_context(*args, **kwargs) + contexts.append(context) + return context + + async def send(self, request, **kwargs): + assert threading.get_ident() == loop_thread + clients.append(self) + assert request.url.host == "browserless.example" + if operation == "capture_screenshot": + assert request.url.path == "/screenshot" + return httpx.Response( + 200, + request=request, + content=b"\x89PNG synthetic", + headers={"Content-Type": "image/png", "X-Response-Code": "200"}, + ) + assert request.url.path == "/content" + return httpx.Response( + 200, + request=request, + text="synthetic page", + headers={"X-Response-Code": "200", "X-Response-Status": "OK"}, + ) + + monkeypatch.setattr(ssl, "create_default_context", create_context) + monkeypatch.setattr(httpx.AsyncClient, "send", send) + + client = BrowserlessClient("https://browserless.example", token="token") + + if operation == "capture_screenshot": + result = await client.capture_screenshot("https://example.com") + assert result.content == b"\x89PNG synthetic" + else: + result = await client.fetch_html_with_status("https://example.com") + assert result.html == "synthetic page" + + assert contexts + assert len(clients) == 1 + assert clients[0].is_closed + + +@pytest.mark.asyncio +@pytest.mark.parametrize("operation", ["fetch_html_with_status", "capture_screenshot"]) +async def test_browserless_client_closes_client_built_after_cancellation(operation, monkeypatch): + """Cancelling during construction must not leak the client it eventually builds. + + ``httpx.AsyncClient`` cannot be interrupted inside its constructor, so the + worker keeps running after the await is cancelled and then hands back a live + client. Nothing can enter the ``async with`` for it, so whoever owns the + construction task has to close it, and the operation must still end cancelled. + """ + real_client = httpx.AsyncClient + started = threading.Event() + release = threading.Event() + finished = threading.Event() + built = [] + sent = [] + + def gated_client(*args, **kwargs): + started.set() + assert release.wait(10), "the constructor was never released" + client = real_client(*args, **kwargs) + built.append(client) + finished.set() + return client + + async def send(self, request, **kwargs): + sent.append(request.url) + return httpx.Response(200, request=request, text="unreachable") + + monkeypatch.setattr(httpx, "AsyncClient", gated_client) + monkeypatch.setattr(real_client, "send", send) + + client = BrowserlessClient("https://browserless.example", token="token") + if operation == "capture_screenshot": + pending = asyncio.ensure_future(client.capture_screenshot("https://example.com")) + else: + pending = asyncio.ensure_future(client.fetch_html_with_status("https://example.com")) + + assert await asyncio.to_thread(started.wait, 10), "the constructor never started" + # The operation is suspended on the constructor now, so the cancellation lands + # there and never reaches the request: release the worker only afterwards. + pending.cancel() + await asyncio.sleep(0) + await asyncio.sleep(0) + release.set() + + with pytest.raises(asyncio.CancelledError): + await pending + + # The worker thread cannot be stopped, so wait for it before judging what it + # built; a leaked client is only visible once the constructor has returned. + assert await asyncio.to_thread(finished.wait, 10), "the constructor never finished" + + assert sent == [], "a cancelled construction must not send a request" + assert len(built) == 1 + assert built[0].is_closed, "the client built after the cancellation was leaked"