mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-30 15:10:41 +00:00
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 <willem.jiang@gmail.com>
This commit is contained in:
mika
GitHub
parent
882433ae50
commit
d50cbf8dc5
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user