From 4e64ded027610e12972692b531f57abfe72ce9bc Mon Sep 17 00:00:00 2001 From: Xuxchloris <7482714452@qq.com> Date: Fri, 14 Aug 2026 17:43:43 +0000 Subject: [PATCH] fix(ws): per-instance event loop instead of module-level global (fixes #119, fixes #133) --- .../channel/tests/test_ws_client_loop.py | 55 +++++++++++++++++++ lark_oapi/ws/client.py | 44 +++++++++++---- 2 files changed, 88 insertions(+), 11 deletions(-) create mode 100644 lark_oapi/channel/tests/test_ws_client_loop.py diff --git a/lark_oapi/channel/tests/test_ws_client_loop.py b/lark_oapi/channel/tests/test_ws_client_loop.py new file mode 100644 index 000000000..9d8f35a46 --- /dev/null +++ b/lark_oapi/channel/tests/test_ws_client_loop.py @@ -0,0 +1,55 @@ +"""ws.Client event-loop isolation (issues #119 / #133). + +Regression tests: each ws.Client must own a dedicated event loop (created +lazily, per instance) instead of sharing a module-level global loop, so +multi-bot setups work and a client constructed inside an already-running +loop is not bound to it. +""" + +import asyncio + +from lark_oapi.ws.client import Client + + +def make_client(app_id="cli_0000000000000001"): + return Client(app_id=app_id, app_secret="secret") + + +def test_client_owns_a_dedicated_loop(): + c1 = make_client() + c2 = make_client(app_id="cli_0000000000000002") + l1 = c1._get_loop() + l2 = c2._get_loop() + assert l1 is not None + # distinct instances never share a loop (the multi-bot race in #119) + assert l1 is not l2 + # stable per instance + assert c1._get_loop() is l1 + assert c2._get_loop() is l2 + l1.close() + l2.close() + + +def test_no_module_level_loop_capture(): + import lark_oapi.ws.client as ws_mod + + assert not hasattr(ws_mod, "loop") + + +def test_loop_created_inside_running_loop_is_dedicated(): + async def inner(): + c = make_client() + loop = c._get_loop() + # never the caller's running loop (#133) + assert loop is not asyncio.get_running_loop() + loop.close() + + asyncio.run(inner()) + + +def test_lock_created_lazily(): + c = make_client() + assert c._lock is None + lock = c._get_lock() + assert isinstance(lock, asyncio.Lock) + assert c._lock is lock diff --git a/lark_oapi/ws/client.py b/lark_oapi/ws/client.py index 8ee991838..41d4f0d61 100644 --- a/lark_oapi/ws/client.py +++ b/lark_oapi/ws/client.py @@ -28,12 +28,6 @@ from lark_oapi.ws.pb.google.protobuf.internal.containers import RepeatedCompositeFieldContainer from lark_oapi.ws.pb.pbbp2_pb2 import Frame -try: - loop = asyncio.get_event_loop() -except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - def _get_by_key(headers: RepeatedCompositeFieldContainer, key: str) -> str: for header in headers: @@ -151,7 +145,13 @@ def __init__(self, self._reconnect_interval: int = 120 self._ping_interval: int = 120 self._cache: ExpiringCache = ExpiringCache(clear_interval=30) - self._lock = asyncio.Lock() + # Event loop owned by this client instance (created lazily on first + # use). Previously a module-level global was shared by every client, + # which broke multi-bot setups (issues #119 / #133). + self._loop: Optional[asyncio.AbstractEventLoop] = None + # asyncio primitives bind to an event loop; created lazily on the + # client's own loop so it never binds to a foreign loop. + self._lock: Optional[asyncio.Lock] = None # Observer hooks for higher-level wrappers (e.g. FeishuChannel) to # react to reconnect lifecycle. ``on_reconnecting`` fires when the # client decides a connection was lost and starts retrying; @@ -161,7 +161,28 @@ def __init__(self, self.on_reconnected: Callable[[], None] = lambda: None logger.setLevel(log_level.value) + def _get_loop(self) -> asyncio.AbstractEventLoop: + """Return the event loop this client runs on. + + Every client owns a dedicated loop, created lazily on first use, so + multiple clients (multi-bot, one thread per bot) never share a loop, + and a client constructed inside an already-running loop (e.g. inside + ``asyncio.run``) is not tied to it. See issues #119 / #133. + """ + if self._loop is None or self._loop.is_closed(): + self._loop = asyncio.new_event_loop() + return self._loop + + def _get_lock(self) -> asyncio.Lock: + # asyncio primitives bind to the running loop when created; create the + # lock lazily (from a coroutine running on the client's own loop) so + # it never binds to a foreign loop. + if self._lock is None: + self._lock = asyncio.Lock() + return self._lock + def start(self) -> None: + loop = self._get_loop() try: loop.run_until_complete(self._connect()) except ClientException as e: @@ -191,7 +212,8 @@ async def _ping_loop(self): await asyncio.sleep(self._ping_interval) async def _connect(self) -> None: - await self._lock.acquire() + lock = self._get_lock() + await lock.acquire() if self._conn is not None: return try: @@ -208,11 +230,11 @@ async def _connect(self) -> None: self._service_id = service_id logger.info(self._fmt_log("connected to {}", conn_url)) - loop.create_task(self._receive_message_loop()) + asyncio.get_running_loop().create_task(self._receive_message_loop()) except InvalidHandshake as e: _parse_ws_conn_exception(e) finally: - self._lock.release() + lock.release() async def _receive_message_loop(self): try: @@ -220,7 +242,7 @@ async def _receive_message_loop(self): if self._conn is None: raise ConnectionClosedException("connection is closed") msg = await self._conn.recv() - loop.create_task(self._handle_message(msg)) + asyncio.get_running_loop().create_task(self._handle_message(msg)) except Exception as e: logger.error(self._fmt_log("receive message loop exit, err: {}", e)) await self._disconnect()