diff --git a/gateway/platforms/weixin.py b/gateway/platforms/weixin.py index d78eb4aad6dc..5de8ff287bbf 100644 --- a/gateway/platforms/weixin.py +++ b/gateway/platforms/weixin.py @@ -43,6 +43,12 @@ aiohttp = None # type: ignore[assignment] AIOHTTP_AVAILABLE = False +# Fallback exception class when aiohttp is not available +if AIOHTTP_AVAILABLE: + _ClientError = aiohttp.ClientError +else: + _ClientError = Exception + try: from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes @@ -123,6 +129,11 @@ def _make_ssl_connector() -> Optional["aiohttp.TCPConnector"]: When ``certifi`` is installed, use its Mozilla CA bundle to guarantee verification. Otherwise fall back to aiohttp's default (which honors ``SSL_CERT_FILE`` env var via ``trust_env=True``). + + ``enable_cleanup_closed=True`` prevents connection pool poisoning + (aiohttp#12795) when the iLink server closes idle keep-alive connections + after ~2-3 minutes. Without this, the stale connection remains in the pool, + gets handed out to the next request, and causes an immediate read timeout. """ try: import ssl @@ -132,7 +143,7 @@ def _make_ssl_connector() -> Optional["aiohttp.TCPConnector"]: if not AIOHTTP_AVAILABLE: return None ssl_ctx = ssl.create_default_context(cafile=certifi.where()) - return aiohttp.TCPConnector(ssl=ssl_ctx) + return aiohttp.TCPConnector(ssl=ssl_ctx, enable_cleanup_closed=True) ITEM_TEXT = 1 ITEM_IMAGE = 2 @@ -387,7 +398,13 @@ async def _do() -> Dict[str, Any]: if not response.ok: raise RuntimeError(f"iLink POST {endpoint} HTTP {response.status}: {raw[:200]}") return json.loads(raw) - return await asyncio.wait_for(_do(), timeout=timeout_ms / 1000) + try: + return await asyncio.wait_for(_do(), timeout=timeout_ms / 1000) + except (asyncio.TimeoutError, _ClientError): + # Stale pooled connection (aiohttp#12795): the TCPConnector may hand + # out a server-closed keep-alive connection. Retry once — the second + # attempt creates a fresh TCP connection. + return await asyncio.wait_for(_do(), timeout=timeout_ms / 1000) async def _api_get( @@ -411,7 +428,10 @@ async def _do() -> Dict[str, Any]: if not response.ok: raise RuntimeError(f"iLink GET {endpoint} HTTP {response.status}: {raw[:200]}") return json.loads(raw) - return await asyncio.wait_for(_do(), timeout=timeout_ms / 1000) + try: + return await asyncio.wait_for(_do(), timeout=timeout_ms / 1000) + except (asyncio.TimeoutError, _ClientError): + return await asyncio.wait_for(_do(), timeout=timeout_ms / 1000) async def _get_updates(