diff --git a/bot_bottle/gateway/egress/addon.py b/bot_bottle/gateway/egress/addon.py index a5292f2c..89fcb07b 100644 --- a/bot_bottle/gateway/egress/addon.py +++ b/bot_bottle/gateway/egress/addon.py @@ -612,7 +612,11 @@ class EgressAddon: source_ip = conn.peername[0] if conn is not None and conn.peername else "" token = self._conn_tokens.get(getattr(conn, "id", ""), "") if conn is not None else "" try: - proposal_id = self._resolver.propose_supervise( + # The resolver RPC is synchronous (blocking urllib); run it off the + # event loop so a slow/unreachable orchestrator can't freeze the + # proxy for every other bottle's traffic (#471 review). + proposal_id = await asyncio.to_thread( + self._resolver.propose_supervise, source_ip, token, tool=TOOL_EGRESS_TOKEN_ALLOW, proposed_file=payload, @@ -675,7 +679,13 @@ class EgressAddon: deadline = loop.time() + self._token_allow_timeout while True: try: - result = self._resolver.poll_supervise(source_ip, token, proposal_id) + # Poll off the event loop — the resolver RPC blocks on urllib, + # and this loop runs for the whole operator-approval window, so + # calling it inline would repeatedly freeze all proxy traffic + # until a decision or timeout (#471 review). + result = await asyncio.to_thread( + self._resolver.poll_supervise, source_ip, token, proposal_id, + ) except PolicyResolveError: result = None if result is not None and result.get("status") in STATUSES: diff --git a/tests/unit/test_egress_addon_request_flow.py b/tests/unit/test_egress_addon_request_flow.py index 415bb8d1..15843f68 100644 --- a/tests/unit/test_egress_addon_request_flow.py +++ b/tests/unit/test_egress_addon_request_flow.py @@ -21,6 +21,7 @@ import asyncio import json import os import sys +import threading import types import unittest from io import StringIO @@ -599,6 +600,35 @@ class TestSuperviseBranch(unittest.TestCase): assert flow.response is not None self.assertEqual(403, flow.response.status_code) + def test_supervise_rpcs_run_off_the_event_loop_thread(self) -> None: + # The propose/poll RPCs block on urllib, and the poll loop runs for the + # whole operator-approval window; calling them inline would freeze the + # proxy event loop for every other bottle's traffic. They must run in a + # worker thread instead (#471 review). Prove it by recording the thread + # the RPCs execute on and asserting it isn't the loop's (main) thread. + loop_ident = threading.get_ident() + rpc_idents: list[int] = [] + + class _ThreadRecordingResolver(_StaticResolver): + def propose_supervise(self, *a: Any, **k: Any) -> str: + rpc_idents.append(threading.get_ident()) + return super().propose_supervise(*a, **k) + + def poll_supervise(self, *a: Any, **k: Any) -> "dict[str, object]": + rpc_idents.append(threading.get_ident()) + return super().poll_supervise(*a, **k) + + config = Config(routes=(Route(host="api.example.com"),)) + addon = _addon(config, slug="test-bottle") + resolver = _ThreadRecordingResolver(config, bottle_id="test-bottle") + resolver.supervise_status = "approved" + addon._resolver = cast(Any, resolver) + addon._token_allow_timeout = 0.05 + flow = _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")) + _run_request(addon, flow) + self.assertTrue(rpc_idents, "propose/poll were never invoked") + self.assertNotIn(loop_ident, rpc_idents) # every RPC ran off the loop thread + # --------------------------------------------------------------------------- # Inbound DLP on responses