fix(ws): isolate transport event loops per client

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:

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()
 

lark_channel/core/cache/expiring_cache.py

@@ -1,23 +1,42 @@
 import asyncio
 import time
-from typing import Dict, Tuple, Any
+from typing import Any, Dict, Optional, Tuple
 
 
 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()

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)

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()

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,
     )
 

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)