From 69554494db846ca36f2c51627345c8a60479365f Mon Sep 17 00:00:00 2001 From: Wu Fei <63766429+wufei-png@users.noreply.github.com> Date: Fri, 14 Aug 2026 18:21:31 +0800 Subject: [PATCH 1/2] fix(channel): stop websocket clients on their own loop --- lark_channel/channel/channel.py | 4 +--- .../channel/tests/test_client_lifecycle.py | 18 ++++++++++++++++++ 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/lark_channel/channel/channel.py b/lark_channel/channel/channel.py index 2e7765c..b1990fe 100644 --- a/lark_channel/channel/channel.py +++ b/lark_channel/channel/channel.py @@ -1071,9 +1071,7 @@ def stop(self, *, join_timeout: float = 5.0) -> None: def _stop_private_ws_client(self, ws: Any) -> None: disconnect = getattr(ws, "_disconnect", None) try: - from lark_channel.ws import client as ws_client_module - - ws_loop = getattr(ws_client_module, "loop", None) + ws_loop = getattr(ws, "_loop", None) if callable(disconnect) and ws_loop is not None: if ws_loop.is_running(): try: diff --git a/lark_channel/channel/tests/test_client_lifecycle.py b/lark_channel/channel/tests/test_client_lifecycle.py index 2d7aba5..4435f23 100644 --- a/lark_channel/channel/tests/test_client_lifecycle.py +++ b/lark_channel/channel/tests/test_client_lifecycle.py @@ -57,6 +57,24 @@ def stop(self): assert calls["kwargs"]["handshake_timeout"] == 4.0 +def test_private_ws_shutdown_uses_the_client_instance_loop(): + c = _client() + ws_loop = asyncio.new_event_loop() + disconnected = [] + + class _PrivateWS: + _loop = ws_loop + + async def _disconnect(self): + disconnected.append(True) + + c._ws_client = _PrivateWS() + c.stop() + + assert disconnected == [True] + ws_loop.close() + + def test_connection_snapshot_initial_state(): c = _client() From e32f304e33a6f1855a7f116c28350a30a36a52f9 Mon Sep 17 00:00:00 2001 From: Wu Fei <63766429+wufei-png@users.noreply.github.com> Date: Fri, 14 Aug 2026 18:21:31 +0800 Subject: [PATCH 2/2] fix(ws): isolate event loops per client --- lark_channel/core/cache/expiring_cache.py | 49 ++++++-- lark_channel/ws/client.py | 66 +++++++---- lark_channel/ws/tests/test_loop_ownership.py | 107 ++++++++++++++++++ .../ws/tests/test_websockets_compat.py | 8 +- lark_channel/ws/tests/test_ws_security.py | 9 +- tests/bridge/test_ws_proxy_keepalive.py | 12 +- 6 files changed, 197 insertions(+), 54 deletions(-) create mode 100644 lark_channel/ws/tests/test_loop_ownership.py diff --git a/lark_channel/core/cache/expiring_cache.py b/lark_channel/core/cache/expiring_cache.py index 653a7cc..b3ccced 100644 --- a/lark_channel/core/cache/expiring_cache.py +++ b/lark_channel/core/cache/expiring_cache.py @@ -1,6 +1,6 @@ import asyncio import time -from typing import Dict, Tuple, Any +from typing import Any, Dict, Optional, Tuple class ExpiringCache(object): @@ -8,16 +8,35 @@ class ExpiringCache(object): def __init__(self, clear_interval=60): self._cache: Dict[str, Tuple[Any, float]] = {} self._clear_interval: int = clear_interval + self._loop: Optional[asyncio.AbstractEventLoop] = None + self._cron: Optional[asyncio.TimerHandle] = None - try: - loop = asyncio.get_event_loop() - except RuntimeError: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - self._cron = loop.create_task(self._start_clear_cron()) + def bind_loop(self, loop: asyncio.AbstractEventLoop) -> None: + """Run expiry cleanup on the loop that owns this cache's client. + + Construction deliberately has no event-loop side effects. Callers can + therefore create a cache from synchronous code or from an unrelated + running loop without capturing that ambient loop. + """ + if ( + self._loop is loop + and self._cron is not None + and not self._cron.cancelled() + ): + return + self.close() + self._loop = loop + self._schedule_clear() + + def close(self) -> None: + cron = self._cron + if cron is not None and not cron.cancelled(): + cron.cancel() + self._cron = None + self._loop = None def __del__(self): - self._cron.cancel() + self.close() def get(self, key: str) -> Any: elem = self._cache.get(key) @@ -41,7 +60,13 @@ def _clear(self): for key in expired_keys: del self._cache[key] - async def _start_clear_cron(self): - while True: - await asyncio.sleep(self._clear_interval) - self._clear() + def _schedule_clear(self) -> None: + loop = self._loop + if loop is None or loop.is_closed(): + return + self._cron = loop.call_later(self._clear_interval, self._run_clear_cron) + + def _run_clear_cron(self) -> None: + self._cron = None + self._clear() + self._schedule_clear() diff --git a/lark_channel/ws/client.py b/lark_channel/ws/client.py index c9148a4..77788e7 100644 --- a/lark_channel/ws/client.py +++ b/lark_channel/ws/client.py @@ -37,13 +37,6 @@ from lark_channel.channel.config import SecurityConfig -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: if header.key == key: @@ -195,7 +188,7 @@ def __init__(self, self._conn_url: str = "" self._service_id: str = "" self._conn_id: str = "" - self._loop = loop + self._loop: Optional[asyncio.AbstractEventLoop] = None self._reconnect_task = None # Local defaults; the Feishu WS endpoint authoritatively replaces these # via _configure() on every handshake (and may push updates mid-session @@ -223,21 +216,29 @@ def __init__(self, logger.setLevel(log_level.value) def start(self) -> None: + # ``start`` is a blocking API and is commonly run in an executor + # thread. A private loop keeps this client independent from whichever + # loop happened to be active when the module or instance was created. + self._set_loop(asyncio.new_event_loop()) + asyncio.set_event_loop(self._loop) try: - loop.run_until_complete(self._connect()) - except ClientException as e: - logger.error(self._fmt_log("connect failed, err: {}", e)) - raise e - except Exception as e: - logger.error(self._fmt_log("connect failed, err: {}", e)) - if self._auto_reconnect: - loop.run_until_complete(self._disconnect_and_reconnect()) - else: - loop.run_until_complete(self._disconnect()) + try: + self._loop.run_until_complete(self._connect()) + except ClientException as e: + logger.error(self._fmt_log("connect failed, err: {}", e)) raise e + except Exception as e: + logger.error(self._fmt_log("connect failed, err: {}", e)) + if self._auto_reconnect: + self._loop.run_until_complete(self._disconnect_and_reconnect()) + else: + self._loop.run_until_complete(self._disconnect()) + raise e - loop.create_task(self._ping_loop()) - loop.run_until_complete(_select()) + self._loop.create_task(self._ping_loop()) + self._loop.run_until_complete(_select()) + finally: + self._close_loop() async def _ping_loop(self): while True: @@ -276,7 +277,7 @@ 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(conn)) + self._create_task(self._receive_message_loop(conn)) except InvalidHandshake as e: _parse_ws_conn_exception(e) @@ -295,10 +296,10 @@ async def _receive_message_loop(self, conn): async def _schedule_handle_message(self, msg) -> None: if self._handler_semaphore is None: - loop.create_task(self._handle_message(msg)) + self._create_task(self._handle_message(msg)) return await self._handler_semaphore.acquire() - loop.create_task(self._handle_message_with_limit(msg)) + self._create_task(self._handle_message_with_limit(msg)) async def _handle_message_with_limit(self, msg) -> None: try: @@ -380,8 +381,27 @@ def _is_local_insecure_ws(self, parsed) -> bool: def _set_loop(self, loop_) -> None: self._loop = loop_ + self._cache.bind_loop(loop_) self._reconnect_task = None + def _create_task(self, coro): + return asyncio.get_running_loop().create_task(coro) + + def _close_loop(self) -> None: + loop_ = self._loop + if loop_ is None or loop_.is_closed() or loop_.is_running(): + return + self._cache.close() + pending = asyncio.all_tasks(loop_) + for task in pending: + task.cancel() + if pending: + loop_.run_until_complete( + asyncio.gather(*pending, return_exceptions=True) + ) + asyncio.set_event_loop(None) + loop_.close() + def probe_endpoint(self, *, timeout: float) -> bool: try: self._get_conn_url(timeout=timeout) diff --git a/lark_channel/ws/tests/test_loop_ownership.py b/lark_channel/ws/tests/test_loop_ownership.py new file mode 100644 index 0000000..94535c6 --- /dev/null +++ b/lark_channel/ws/tests/test_loop_ownership.py @@ -0,0 +1,107 @@ +import asyncio +import subprocess +import sys +import textwrap +import threading + +from lark_channel.core.cache import ExpiringCache +from lark_channel.ws import client as ws_client + + +def test_import_inside_running_loop_does_not_capture_caller_loop(): + code = textwrap.dedent( + """ + import asyncio + + async def main(): + from lark_channel.ws import client as ws_client + + caller_loop = asyncio.get_running_loop() + client = ws_client.Client("cli_test", "secret", auto_reconnect=False) + + async def no_op(): + return None + + client._connect = no_op + client._ping_loop = no_op + ws_client._select = no_op + + await caller_loop.run_in_executor(None, client.start) + assert client._loop is not caller_loop + assert client._loop.is_closed() + + asyncio.run(main()) + """ + ) + + result = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + timeout=10, + ) + + assert result.returncode == 0, result.stderr + + +def test_clients_started_in_parallel_own_distinct_loops(monkeypatch): + clients = [ + ws_client.Client("cli_one", "secret", auto_reconnect=False), + ws_client.Client("cli_two", "secret", auto_reconnect=False), + ] + barrier = threading.Barrier(2) + observed_loops = [] + errors = [] + + async def connect_and_meet(): + observed_loops.append(asyncio.get_running_loop()) + barrier.wait(timeout=2) + + async def no_op(): + return None + + monkeypatch.setattr(ws_client, "_select", no_op) + for client in clients: + client._connect = connect_and_meet + client._ping_loop = no_op + + def start(client): + try: + client.start() + except Exception as exc: # pragma: no cover - asserted below + errors.append(exc) + + threads = [threading.Thread(target=start, args=(client,)) for client in clients] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5) + + assert all(not thread.is_alive() for thread in threads) + assert errors == [] + assert len(observed_loops) == 2 + assert clients[0]._loop is not clients[1]._loop + assert set(observed_loops) == {client._loop for client in clients} + assert all(client._loop.is_closed() for client in clients) + + +def test_expiring_cache_binds_only_when_owner_loop_is_known(): + cache = ExpiringCache(clear_interval=0.01) + + assert cache._loop is None + assert cache._cron is None + + owner_loop = asyncio.new_event_loop() + cache.bind_loop(owner_loop) + + assert cache._loop is owner_loop + assert cache._cron is not None + assert not cache._cron.cancelled() + + cache.set("expired", "value", -1) + owner_loop.run_until_complete(asyncio.sleep(0.02)) + assert "expired" not in cache._cache + + cache.close() + owner_loop.run_until_complete(asyncio.sleep(0)) + owner_loop.close() diff --git a/lark_channel/ws/tests/test_websockets_compat.py b/lark_channel/ws/tests/test_websockets_compat.py index a179853..1409f68 100644 --- a/lark_channel/ws/tests/test_websockets_compat.py +++ b/lark_channel/ws/tests/test_websockets_compat.py @@ -99,8 +99,8 @@ async def fake_connect(uri, *, proxy=True): ) monkeypatch.setattr(ws_client.websockets, "connect", fake_connect) monkeypatch.setattr( - ws_client.loop, - "create_task", + client, + "_create_task", lambda coro: coro.close() if hasattr(coro, "close") else None, ) @@ -129,8 +129,8 @@ async def fake_connect(uri): ) monkeypatch.setattr(ws_client.websockets, "connect", fake_connect) monkeypatch.setattr( - ws_client.loop, - "create_task", + client, + "_create_task", lambda coro: coro.close() if hasattr(coro, "close") else None, ) diff --git a/lark_channel/ws/tests/test_ws_security.py b/lark_channel/ws/tests/test_ws_security.py index 51b878f..9510bb9 100644 --- a/lark_channel/ws/tests/test_ws_security.py +++ b/lark_channel/ws/tests/test_ws_security.py @@ -39,7 +39,7 @@ def close_task(coro): lambda: "ws://example.test/callback?device_id=device&service_id=42", ) monkeypatch.setattr(ws_client.websockets, "connect", fake_connect) - monkeypatch.setattr(ws_client, "loop", SimpleNamespace(create_task=close_task)) + monkeypatch.setattr(client, "_create_task", close_task) await client._connect() @@ -72,7 +72,7 @@ def close_task(coro): lambda: "ws://example.test/callback?device_id=device&service_id=42", ) monkeypatch.setattr(ws_client.websockets, "connect", fake_connect) - monkeypatch.setattr(ws_client, "loop", SimpleNamespace(create_task=close_task)) + monkeypatch.setattr(client, "_create_task", close_task) with caplog.at_level(logging.WARNING, logger="Lark"): await client._connect() @@ -133,7 +133,7 @@ def close_task(coro): lambda: "ws://example.test/callback?device_id=device&service_id=42", ) monkeypatch.setattr(ws_client.websockets, "connect", fake_connect) - monkeypatch.setattr(ws_client, "loop", SimpleNamespace(create_task=close_task)) + monkeypatch.setattr(client, "_create_task", close_task) await client._connect() @@ -164,7 +164,7 @@ def close_task(coro): lambda: "ws://127.0.0.1/callback?device_id=device&service_id=42", ) monkeypatch.setattr(ws_client.websockets, "connect", fake_connect) - monkeypatch.setattr(ws_client, "loop", SimpleNamespace(create_task=close_task)) + monkeypatch.setattr(client, "_create_task", close_task) await client._connect() @@ -271,7 +271,6 @@ async def fake_handle_message(message): async def noop_disconnect(*_args, **_kwargs): return None - monkeypatch.setattr(ws_client, "loop", asyncio.get_running_loop()) monkeypatch.setattr(client, "_handle_message", fake_handle_message) monkeypatch.setattr(client, "_disconnect", noop_disconnect) diff --git a/tests/bridge/test_ws_proxy_keepalive.py b/tests/bridge/test_ws_proxy_keepalive.py index 3052dd0..4b2927b 100644 --- a/tests/bridge/test_ws_proxy_keepalive.py +++ b/tests/bridge/test_ws_proxy_keepalive.py @@ -82,11 +82,7 @@ def conn_url(): client = ws_client.Client(app_id="cli_x", app_secret="s") monkeypatch.setattr(client, "_get_conn_url", conn_url) monkeypatch.setattr(ws_client.websockets, "connect", fake_connect) - monkeypatch.setattr( - ws_client, - "loop", - SimpleNamespace(create_task=close_task), - ) + monkeypatch.setattr(client, "_create_task", close_task) await client._connect() @@ -118,11 +114,7 @@ def conn_url(*, timeout=None): ) monkeypatch.setattr(client, "_get_conn_url", conn_url) monkeypatch.setattr(ws_client.websockets, "connect", fake_connect) - monkeypatch.setattr( - ws_client, - "loop", - SimpleNamespace(create_task=close_task), - ) + monkeypatch.setattr(client, "_create_task", close_task) await client._connect()