From d50cbf8dc5b18fc716686710dd5a1222d7d00c3b Mon Sep 17 00:00:00 2001 From: mika Date: Fri, 30 Oct 2026 07:18:54 -0700 Subject: [PATCH] fix(browser): isolate egress DNS workers (#6301) * fix(browser): isolate egress DNS workers * fix(browser): release resolver slots after loop close * test(browser): cover resolver isolation on production loop * docs(browser): scope egress resolver guidance to browser automation * test: approve browser automation guidance path --------- Co-authored-by: mikamikasuki <211269698+mikamikasuki@users.noreply.github.com> Co-authored-by: Willem Jiang --- .../community/browser_automation/AGENTS.md | 10 + .../community/browser_automation/egress.py | 52 ++++- backend/tests/test_agent_guidance_check.py | 1 + backend/tests/test_browser_egress.py | 190 ++++++++++++++++++ 4 files changed, 252 insertions(+), 1 deletion(-) diff --git a/backend/packages/harness/deerflow/community/browser_automation/AGENTS.md b/backend/packages/harness/deerflow/community/browser_automation/AGENTS.md index 424eafc499..6160d84a51 100644 --- a/backend/packages/harness/deerflow/community/browser_automation/AGENTS.md +++ b/backend/packages/harness/deerflow/community/browser_automation/AGENTS.md @@ -1,3 +1,13 @@ +# Browser Automation + +### Browser Egress Resolution + +SOCKS DNS runs on the private Playwright loop. Use a dedicated bounded executor +so timed-out lookups cannot starve that loop's request guard. Retain permits until +threads exit, including after loop shutdown; reject at capacity and propagate +ContextVars. Keep the Gateway and browser loops separate in starvation tests. +See `backend/tests/test_browser_egress.py`. + ### Browser Progress Screenshots Hidden per-action browser progress frames use JPEG at quality 80 to keep their diff --git a/backend/packages/harness/deerflow/community/browser_automation/egress.py b/backend/packages/harness/deerflow/community/browser_automation/egress.py index 7c86f65852..56e7bb49a6 100644 --- a/backend/packages/harness/deerflow/community/browser_automation/egress.py +++ b/backend/packages/harness/deerflow/community/browser_automation/egress.py @@ -18,9 +18,12 @@ from __future__ import annotations import asyncio import contextlib +import contextvars import ipaddress import logging +import threading from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor logger = logging.getLogger(__name__) @@ -47,6 +50,49 @@ _HANDSHAKE_TIMEOUT_S = 15 _RESOLVE_TIMEOUT_S = 15 _CONNECT_TIMEOUT_S = 15 _CLOSE_TIMEOUT_S = 2 +_EGRESS_RESOLVER_WORKERS = 8 + +# A timed-out synchronous resolver keeps running in its worker thread. Keep +# those calls out of the browser loop's default executor, and acquire a +# permit before submission so the dedicated executor never accumulates an +# unbounded work queue. Saturated egress fails closed instead of queuing DNS +# behind work that may itself be stuck. +_RESOLVER_EXECUTOR = ThreadPoolExecutor(max_workers=_EGRESS_RESOLVER_WORKERS, thread_name_prefix="browser-egress-dns") +_RESOLVER_SLOTS = threading.BoundedSemaphore(_EGRESS_RESOLVER_WORKERS) + + +class _ResolverCapacityExceeded(RuntimeError): + pass + + +def _consume_resolver_result(future: asyncio.Future[object]) -> None: + """Consume abandoned results after a timeout or cancelled SOCKS handler.""" + try: + future.exception() + except (asyncio.CancelledError, Exception): + pass + + +async def _resolve_with_timeout(resolve: EgressResolver, host: str) -> list[str]: + slots = _RESOLVER_SLOTS + if not slots.acquire(blocking=False): + raise _ResolverCapacityExceeded("browser egress resolver capacity is full") + + loop = asyncio.get_running_loop() + context = contextvars.copy_context() + try: + worker = _RESOLVER_EXECUTOR.submit(context.run, resolve, host) + except BaseException: + slots.release() + raise + # Release from the concurrent future's callback, which runs with the worker + # completion and does not depend on the event loop still being alive. + worker.add_done_callback(lambda _completed: slots.release()) + future = asyncio.wrap_future(worker, loop=loop) + future.add_done_callback(_consume_resolver_result) + # Shield the executor future: cancellation of the SOCKS handler must not + # release its permit before the already-started resolver thread exits. + return await asyncio.wait_for(asyncio.shield(future), timeout=_RESOLVE_TIMEOUT_S) class _RefusedRequest(Exception): @@ -97,11 +143,15 @@ class BrowserEgressProxy: await _reply(writer, exc.reply) return try: - addresses = await asyncio.wait_for(asyncio.to_thread(self._resolve, host), timeout=_RESOLVE_TIMEOUT_S) + addresses = await _resolve_with_timeout(self._resolve, host) except ValueError as exc: logger.warning("browser egress refused for %s:%d: %s", host, port, exc) await _reply(writer, _REPLY_NOT_ALLOWED) return + except _ResolverCapacityExceeded: + logger.warning("browser egress resolution capacity full for %s:%d", host, port) + await _reply(writer, _REPLY_HOST_UNREACHABLE) + return except TimeoutError: logger.warning("browser egress resolution timed out for %s:%d", host, port) await _reply(writer, _REPLY_HOST_UNREACHABLE) diff --git a/backend/tests/test_agent_guidance_check.py b/backend/tests/test_agent_guidance_check.py index 500498c196..c3af1c716d 100644 --- a/backend/tests/test_agent_guidance_check.py +++ b/backend/tests/test_agent_guidance_check.py @@ -34,6 +34,7 @@ EXPECTED_GUIDANCE_PATHS = { "backend/packages/harness/deerflow/community/ddg_search/AGENTS.md", "backend/packages/harness/deerflow/community/e2b_sandbox/AGENTS.md", "backend/packages/harness/deerflow/community/aio_sandbox/AGENTS.md", + "backend/packages/harness/deerflow/community/browser_automation/AGENTS.md", "backend/packages/harness/deerflow/community/opensandbox/AGENTS.md", "backend/packages/harness/deerflow/config/AGENTS.md", "backend/packages/harness/deerflow/extensions/AGENTS.md", diff --git a/backend/tests/test_browser_egress.py b/backend/tests/test_browser_egress.py index c9cff3cba5..3f49d0f856 100644 --- a/backend/tests/test_browser_egress.py +++ b/backend/tests/test_browser_egress.py @@ -14,12 +14,15 @@ import ipaddress import socket import sys import threading +from concurrent.futures import ThreadPoolExecutor +from contextvars import ContextVar from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from types import ModuleType, SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest +from app.gateway import context_usage from deerflow.community import url_safety from deerflow.community.browser_automation import egress from deerflow.community.browser_automation import session as session_mod @@ -223,6 +226,193 @@ async def test_proxy_bounds_a_stuck_resolution(monkeypatch): assert reply == 0x04 +def test_timed_out_egress_resolution_does_not_starve_browser_request_guard(monkeypatch): + browser_loop = session_mod._PlaywrightLoopThread() + browser_executor = ThreadPoolExecutor(max_workers=1) + gateway_executor = ThreadPoolExecutor(max_workers=1) + resolver_executor = ThreadPoolExecutor(max_workers=1) + monkeypatch.setattr(egress, "_RESOLVER_EXECUTOR", resolver_executor, raising=False) + monkeypatch.setattr(egress, "_RESOLVER_SLOTS", threading.BoundedSemaphore(1), raising=False) + monkeypatch.setattr(egress, "_RESOLVE_TIMEOUT_S", 0.1) + resolver_started = threading.Event() + release_resolver = threading.Event() + resolver_calls: list[str] = [] + + def resolve(host: str) -> list[str]: + resolver_calls.append(host) + resolver_started.set() + release_resolver.wait(timeout=10) + return ["127.0.0.1"] + + guard = MagicMock(return_value=None) + session = BrowserSession(browser_loop, headless=True, timeout_ms=1000, viewport={}, url_guard=guard, egress_resolver=resolve) + session._context = SimpleNamespace(route=AsyncMock(), close=AsyncMock()) + + async def start_session_proxy(): + asyncio.get_running_loop().set_default_executor(browser_executor) + await session._install_request_guard() + return (await session._egress_proxy_settings())["server"] + + async def scenario(): + gateway_loop = asyncio.get_running_loop() + gateway_loop.set_default_executor(gateway_executor) + assert gateway_loop is not browser_loop._loop + assert gateway_executor is not browser_executor + + messages = [SimpleNamespace(content="hello")] + snapshot = SimpleNamespace(values={"messages": messages}) + + class Accessor: + async def aget(self, _config): + return snapshot + + model = SimpleNamespace(name="test-model", context_window=1000) + app_config = SimpleNamespace(models=[model], get_model_config=lambda _name: model) + + class RunStore: + async def list_by_thread(self, *_args, **_kwargs): + return [] + + async def build_accessor(_request, *, thread_id): + return Accessor(), {"configurable": {"thread_id": thread_id}} + + monkeypatch.setattr(context_usage, "_count_messages_approximately", lambda _messages: 250) + monkeypatch.setattr(context_usage, "build_thread_checkpoint_state_accessor", build_accessor) + monkeypatch.setattr(context_usage, "get_config", lambda: app_config) + + connect_task = None + guard_task = None + try: + # Use the production session entry point: SOCKS DNS runs on the + # private Playwright loop, not the caller's Gateway loop. + proxy_url = await browser_loop.run(start_session_proxy()) + connect_task = asyncio.create_task(_socks_connect(proxy_url, 0x03, _domain("wedged.example"), 80)) + assert await asyncio.wait_for(asyncio.to_thread(resolver_started.wait, 2), timeout=3) + _method, reply, _reader, writer = await asyncio.wait_for(connect_task, timeout=2) + writer.close() + await writer.wait_closed() + assert reply == 0x04 + assert not release_resolver.is_set() + + _method, overloaded_reply, _reader, overloaded_writer = await asyncio.wait_for(_socks_connect(proxy_url, 0x03, _domain("second.example"), 80), timeout=2) + overloaded_writer.close() + await overloaded_writer.wait_closed() + assert overloaded_reply == 0x04 + assert resolver_calls == ["wedged.example"] + + # Gateway token counting has its own default pool and succeeds on + # both the base and PR head while the browser resolver is held. + assert await asyncio.wait_for( + context_usage.build_context_usage(request=SimpleNamespace(app=SimpleNamespace()), thread_id="thread-1", run_store=RunStore()), + timeout=2, + ) == {"token_count": 250, "max_context_tokens": 1000, "percentage": 25.0} + + # The actual contention is on the browser loop: its installed route + # callback must still run the synchronous request guard. + route = SimpleNamespace(request=SimpleNamespace(url="https://public.example/"), continue_=AsyncMock(), abort=AsyncMock()) + route_handler = session._context.route.await_args.args[1] + guard_task = asyncio.create_task(browser_loop.run(route_handler(route))) + await asyncio.wait_for(guard_task, timeout=2) + guard.assert_called_once_with("https://public.example/") + route.continue_.assert_awaited_once() + route.abort.assert_not_awaited() + assert not release_resolver.is_set() + finally: + release_resolver.set() + for task in (connect_task, guard_task): + if task is not None: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await browser_loop.run(session._close()) + await browser_loop.run(browser_loop._loop.shutdown_default_executor()) + + try: + asyncio.run(scenario()) + finally: + release_resolver.set() + resolver_executor.shutdown(wait=True, cancel_futures=True) + browser_executor.shutdown(wait=True, cancel_futures=True) + gateway_executor.shutdown(wait=True, cancel_futures=True) + browser_loop._loop.call_soon_threadsafe(browser_loop._loop.stop) + browser_loop._thread.join(timeout=3) + assert not browser_loop._thread.is_alive() + browser_loop._loop.close() + + +@pytest.mark.asyncio +async def test_egress_resolver_propagates_caller_context(monkeypatch): + executor = ThreadPoolExecutor(max_workers=1) + monkeypatch.setattr(egress, "_RESOLVER_EXECUTOR", executor) + monkeypatch.setattr(egress, "_RESOLVER_SLOTS", threading.BoundedSemaphore(1)) + resolver_context = ContextVar("test_browser_egress_context", default="worker-default") + token = resolver_context.set("request-context") + try: + assert await egress._resolve_with_timeout(lambda _host: [resolver_context.get()], "public.example") == ["request-context"] + finally: + resolver_context.reset(token) + executor.shutdown(wait=True, cancel_futures=True) + assert resolver_context.get() == "worker-default" + + +@pytest.mark.asyncio +async def test_egress_resolver_submission_failure_releases_slot(monkeypatch): + executor = ThreadPoolExecutor(max_workers=1) + slots = threading.BoundedSemaphore(1) + monkeypatch.setattr(egress, "_RESOLVER_EXECUTOR", executor) + monkeypatch.setattr(egress, "_RESOLVER_SLOTS", slots) + try: + executor.shutdown(wait=True) + with pytest.raises(RuntimeError, match="cannot schedule new futures after shutdown"): + await egress._resolve_with_timeout(MagicMock(), "public.example") + assert slots.acquire(blocking=False), "failed submission must return its permit" + slots.release() + finally: + executor.shutdown(wait=True, cancel_futures=True) + + +def test_timed_out_resolver_releases_slot_after_event_loop_closes(monkeypatch): + executor = ThreadPoolExecutor(max_workers=1) + slots = threading.BoundedSemaphore(1) + monkeypatch.setattr(egress, "_RESOLVER_EXECUTOR", executor) + monkeypatch.setattr(egress, "_RESOLVER_SLOTS", slots) + monkeypatch.setattr(egress, "_RESOLVE_TIMEOUT_S", 0.05) + + resolver_started = threading.Event() + release_resolver = threading.Event() + resolver_finished = threading.Event() + + def resolve(_host: str) -> list[str]: + resolver_started.set() + release_resolver.wait(timeout=10) + resolver_finished.set() + return ["127.0.0.1"] + + async def time_out_resolver(): + task = asyncio.create_task(egress._resolve_with_timeout(resolve, "wedged.example")) + try: + # Cancelling to_thread's awaiter cannot interrupt Event.wait itself. + assert await asyncio.wait_for(asyncio.to_thread(resolver_started.wait, 2), timeout=3) + with pytest.raises(TimeoutError): + await task + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + loop = asyncio.new_event_loop() + try: + try: + loop.run_until_complete(time_out_resolver()) + finally: + loop.close() + release_resolver.set() + assert resolver_finished.wait(timeout=2) + assert slots.acquire(timeout=2), "resolver worker completion must release capacity without its event loop" + slots.release() + finally: + release_resolver.set() + executor.shutdown(wait=True, cancel_futures=True) + + @pytest.mark.asyncio async def test_proxy_close_is_bounded_by_a_tunnel_it_could_not_sweep(monkeypatch): monkeypatch.setattr(egress, "_CLOSE_TIMEOUT_S", 0.2)