mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-07-26 14:26:06 +00:00
fix(lark): keep connection lookup off event loop
Move the Lark SDK synchronous connection URL lookup off the main asyncio loop, serialize reconnects, and cover the incident with a non-blocking regression test.
This commit is contained in:
@@ -242,6 +242,33 @@ def _lark_extract_action_form_inputs(action: typing.Any, action_value_obj: dict)
|
|||||||
return form_inputs
|
return form_inputs
|
||||||
|
|
||||||
|
|
||||||
|
class NonBlockingLarkWSClient(lark_oapi.ws.Client):
|
||||||
|
"""Keep the SDK's synchronous connection lookup off LangBot's event loop.
|
||||||
|
|
||||||
|
lark-oapi performs ``requests.post`` inside its async ``_connect`` method.
|
||||||
|
A stalled TLS handshake therefore freezes Quart and every other adapter in
|
||||||
|
the process. Pre-fetch the URL in a worker thread, then let the SDK finish
|
||||||
|
the WebSocket setup on the main loop with the already-resolved URL.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, *args, **kwargs):
|
||||||
|
super().__init__(*args, **kwargs)
|
||||||
|
self._langbot_connect_lock = asyncio.Lock()
|
||||||
|
|
||||||
|
async def _connect(self) -> None:
|
||||||
|
async with self._langbot_connect_lock:
|
||||||
|
if self._conn is not None:
|
||||||
|
return
|
||||||
|
|
||||||
|
conn_url = await asyncio.to_thread(self._get_conn_url)
|
||||||
|
original_get_conn_url = self._get_conn_url
|
||||||
|
self._get_conn_url = lambda: conn_url
|
||||||
|
try:
|
||||||
|
await super()._connect()
|
||||||
|
finally:
|
||||||
|
self._get_conn_url = original_get_conn_url
|
||||||
|
|
||||||
|
|
||||||
class AESCipher(object):
|
class AESCipher(object):
|
||||||
def __init__(self, key):
|
def __init__(self, key):
|
||||||
self.bs = AES.block_size
|
self.bs = AES.block_size
|
||||||
@@ -1240,7 +1267,9 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
|
|||||||
bot_account_id = config['bot_name']
|
bot_account_id = config['bot_name']
|
||||||
|
|
||||||
domain = self._resolve_domain(config)
|
domain = self._resolve_domain(config)
|
||||||
bot = lark_oapi.ws.Client(config['app_id'], config['app_secret'], event_handler=event_handler, domain=domain)
|
bot = NonBlockingLarkWSClient(
|
||||||
|
config['app_id'], config['app_secret'], event_handler=event_handler, domain=domain
|
||||||
|
)
|
||||||
api_client = self.build_api_client(config)
|
api_client = self.build_api_client(config)
|
||||||
cipher = AESCipher(config.get('encrypt-key', ''))
|
cipher = AESCipher(config.get('encrypt-key', ''))
|
||||||
self.request_app_ticket(api_client, config)
|
self.request_app_ticket(api_client, config)
|
||||||
|
|||||||
@@ -0,0 +1,39 @@
|
|||||||
|
import asyncio
|
||||||
|
import threading
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from langbot.pkg.platform.sources.lark import NonBlockingLarkWSClient
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_lark_connection_url_lookup_does_not_block_main_event_loop(monkeypatch):
|
||||||
|
client = NonBlockingLarkWSClient('app-id', 'app-secret')
|
||||||
|
lookup_started = threading.Event()
|
||||||
|
release_lookup = threading.Event()
|
||||||
|
base_connect_urls: list[str] = []
|
||||||
|
|
||||||
|
def blocking_get_conn_url() -> str:
|
||||||
|
lookup_started.set()
|
||||||
|
if not release_lookup.wait(timeout=2):
|
||||||
|
raise TimeoutError('test did not release the connection lookup')
|
||||||
|
return 'wss://example.invalid/connect?device_id=device&service_id=service'
|
||||||
|
|
||||||
|
async def fake_base_connect(self):
|
||||||
|
base_connect_urls.append(self._get_conn_url())
|
||||||
|
|
||||||
|
monkeypatch.setattr(client, '_get_conn_url', blocking_get_conn_url)
|
||||||
|
monkeypatch.setattr('lark_oapi.ws.Client._connect', fake_base_connect)
|
||||||
|
|
||||||
|
connect_task = asyncio.create_task(client._connect())
|
||||||
|
await asyncio.wait_for(asyncio.to_thread(lookup_started.wait, 1), timeout=1)
|
||||||
|
|
||||||
|
# If the SDK's synchronous requests.post still runs on the event-loop
|
||||||
|
# thread, this sleep cannot complete until release_lookup is set.
|
||||||
|
await asyncio.wait_for(asyncio.sleep(0.01), timeout=0.1)
|
||||||
|
assert not connect_task.done()
|
||||||
|
|
||||||
|
release_lookup.set()
|
||||||
|
await asyncio.wait_for(connect_task, timeout=1)
|
||||||
|
|
||||||
|
assert base_connect_urls == ['wss://example.invalid/connect?device_id=device&service_id=service']
|
||||||
Reference in New Issue
Block a user