diff --git a/backend/core/event_bus.py b/backend/core/event_bus.py index 74fd662c..a9242ed3 100644 --- a/backend/core/event_bus.py +++ b/backend/core/event_bus.py @@ -64,31 +64,27 @@ def emit(kind: str, payload: dict[str, Any] | None = None) -> None: **(payload or {}), } event_str = json.dumps(event) - loop = asyncio.get_running_loop() if _on_event_loop() else _serving_loop - if loop is None: + try: + caller_loop = asyncio.get_running_loop() + except RuntimeError: + caller_loop = None + target_loop = _serving_loop or caller_loop + if target_loop is None: # No serving loop yet — nobody to notify; dropping is correct. logger.debug("No event loop — event dropped: %s", kind) return try: - if _on_event_loop(): - loop.create_task(_broadcast(event_str)) + if caller_loop is target_loop: + target_loop.create_task(_broadcast(event_str)) else: - # Threadpool worker (sync endpoint body): the only thread-safe way in. - loop.call_soon_threadsafe(_schedule_broadcast, event_str) + # Sync endpoints and async producers on a foreign loop must both + # hand off: the lock and listener queues belong to serving_loop. + target_loop.call_soon_threadsafe(_schedule_broadcast, event_str) except RuntimeError: # The serving loop closed between capture and use (app shutdown). logger.debug("Event loop closed — event dropped: %s", kind) -def _on_event_loop() -> bool: - """True when called from the running event loop (async-context emit).""" - try: - asyncio.get_running_loop() - except RuntimeError: - return False - return True - - def _schedule_broadcast(event_str: str) -> None: """Run `_broadcast` on the serving loop; called via call_soon_threadsafe.""" asyncio.get_running_loop().create_task(_broadcast(event_str)) diff --git a/tests/test_event_bus_thread_emit.py b/tests/test_event_bus_thread_emit.py index 49b9960c..16002fc3 100644 --- a/tests/test_event_bus_thread_emit.py +++ b/tests/test_event_bus_thread_emit.py @@ -66,6 +66,58 @@ def test_emit_from_thread_reaches_serving_loop(bus, tmp_path): loop.close() +def test_emit_from_foreign_running_loop_reaches_serving_loop(bus): + """An async producer may run on a worker loop, but listener state belongs + to the WebSocket serving loop and must only be touched there.""" + serving_loop = asyncio.new_event_loop() + serving_loop.set_debug(True) + received: list[str] = [] + started = threading.Event() + done = threading.Event() + + async def serve_with_waiter_ready(): + q = await bus.subscribe() + waiter = asyncio.create_task(q.get()) + await asyncio.sleep(0) # q.get() has installed its serving-loop Future + started.set() + try: + received.append(await asyncio.wait_for(waiter, 0.5)) + except asyncio.TimeoutError: + pass + finally: + done.set() + await bus.unsubscribe(q) + + def run_serving_loop(): + asyncio.set_event_loop(serving_loop) + serving_loop.run_until_complete(serve_with_waiter_ready()) + + serving_thread = threading.Thread( + target=run_serving_loop, name="test-serving-loop" + ) + serving_thread.start() + try: + assert started.wait(2.0), "serving loop never subscribed" + + async def foreign_async_caller(): + assert asyncio.get_running_loop() is not serving_loop + bus.emit("profiles", {"action": "updated", "id": "foreign-loop"}) + await asyncio.sleep(0.05) + + asyncio.run(foreign_async_caller()) + + assert done.wait(2.0), ( + "emit() ran listener delivery on the caller's foreign loop" + ) + assert received, "foreign-loop event never reached the serving loop" + payload = json.loads(received[0]) + assert payload["id"] == "foreign-loop" + finally: + serving_loop.call_soon_threadsafe(done.set) + serving_thread.join(2.0) + serving_loop.close() + + async def _serve(bus, received: list[str], started: threading.Event, done: threading.Event): q = await bus.subscribe() started.set()