mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-02 19:58:46 +00:00
b6e82b095f
Review round 3 advisory (gpt-6-astra): ws._dispatch_extension_message still called a synchronous in-process handler directly on the event loop, so one skill's blocking callback (its barrier wait included) stalled unrelated HTTP and WebSocket work. The WS dispatcher now mirrors the HTTP one: a coroutine handler is awaited, a synchronous one runs through asyncio.to_thread. Regression tests/test_ws_dispatch_off_loop.py: a heartbeat coroutine keeps ticking while the handler blocks, and the handler observes the release the loop sets afterwards (fails on the previous tree).
82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
"""A synchronous extension WS handler never stalls the ASGI event loop.
|
|
|
|
#1195 F3 review follow-up: `ws._dispatch_extension_message` mirrors the HTTP
|
|
dispatcher — a coroutine handler is awaited, a synchronous one (including its
|
|
synchronous execution-barrier wait) runs in a worker thread. The proof is a
|
|
heartbeat coroutine that keeps ticking while the handler is blocked.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import builtins
|
|
import json
|
|
import threading
|
|
|
|
from ouroboros import extension_loader
|
|
from tests._extension_loader_shared import _prepare_extension
|
|
from tests._extension_loader_shared import ( # noqa: F401 (autouse fixture applies on import)
|
|
_clear_loader_state,
|
|
)
|
|
|
|
BLOCKING_WS_PLUGIN = (
|
|
"import builtins\n"
|
|
"def _slow(msg):\n"
|
|
" builtins._ouro_ws_entered.set()\n"
|
|
" released = builtins._ouro_ws_release.wait(timeout=2)\n"
|
|
" return {'released': released}\n"
|
|
"def register(api):\n"
|
|
" api.register_ws_handler('slow', _slow)\n"
|
|
)
|
|
|
|
|
|
def test_sync_ws_handler_runs_off_the_event_loop(tmp_path, monkeypatch):
|
|
from ouroboros.gateway import ws as ws_module
|
|
|
|
loaded, repo_root, drive_root = _prepare_extension(
|
|
tmp_path, "ws_off_loop", BLOCKING_WS_PLUGIN, permissions=["ws_handler"],
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(repo_root))
|
|
builtins._ouro_ws_entered = threading.Event()
|
|
builtins._ouro_ws_release = threading.Event()
|
|
sent: list = []
|
|
|
|
class _FakeApp:
|
|
class state: # noqa: N801
|
|
drive_root = None
|
|
|
|
class _FakeWebSocket:
|
|
def __init__(self, app):
|
|
self.app = app
|
|
|
|
async def send_text(self, text):
|
|
sent.append(json.loads(text))
|
|
|
|
app = _FakeApp()
|
|
app.state.drive_root = drive_root
|
|
msg_type = extension_loader.extension_surface_name(loaded.name, "slow")
|
|
|
|
async def main():
|
|
ticks = 0
|
|
dispatch = asyncio.create_task(
|
|
ws_module._dispatch_extension_message(_FakeWebSocket(app), {"type": msg_type}, msg_type)
|
|
)
|
|
# The handler is standing inside its blocking wait; the loop must still turn.
|
|
await asyncio.to_thread(builtins._ouro_ws_entered.wait, 5)
|
|
for _ in range(20):
|
|
await asyncio.sleep(0.01)
|
|
ticks += 1
|
|
builtins._ouro_ws_release.set()
|
|
assert await asyncio.wait_for(dispatch, 5) is True
|
|
return ticks
|
|
|
|
try:
|
|
ticks = asyncio.run(main())
|
|
finally:
|
|
del builtins._ouro_ws_entered
|
|
del builtins._ouro_ws_release
|
|
extension_loader.unload_extension(loaded.name)
|
|
# The release is set by the loop AFTER ticking; a handler that had frozen the
|
|
# loop could never see it and would report the timeout instead.
|
|
assert ticks == 20
|
|
assert sent and sent[-1]["type"] == msg_type + ".reply", sent
|
|
assert sent[-1]["data"] == {"released": True}, "the event loop was blocked while the sync handler ran"
|