fix(socket_mode): recreate a closed aiohttp session when reconnecting

slack_sdk/socket_mode/aiohttp/__init__.py

@@ -127,6 +127,7 @@ def __init__(
         # over the lifetime of your application,
         # it is suggested you use a single session for the lifetime of your application
         # to benefit from connection pooling.
+        self._loop = loop
         self.aiohttp_client_session = aiohttp.ClientSession(loop=loop)
 
         self.on_message_listeners = on_message_listeners or []
@@ -322,6 +323,7 @@ async def is_connected(self) -> bool:
         connected: bool = (
             not self.closed
             and not self.stale
+            and not self.aiohttp_client_session.closed
             and self.current_session is not None
             and not self.current_session.closed
             and not await self.is_ping_pong_failing()
@@ -369,6 +371,12 @@ async def connect(self):
                 except Exception as e:
                     self.logger.exception(f"Failed to close the old session : {e}")
 
+                if self.aiohttp_client_session.closed:
+                    # All connections share this session. Once it has been closed,
+                    # every connection attempt fails, so it has to be recreated.
+                    self.logger.info("The aiohttp client session is closed; creating a new one")
+                    self.aiohttp_client_session = aiohttp.ClientSession(loop=self._loop)
+
                 if self.wss_uri is None:
                     # If the underlying WSS URL does not exist,
                     # acquiring a new active WSS URL from the server-side first

tests/slack_sdk_async/socket_mode/test_aiohttp.py

@@ -1,7 +1,9 @@
 import asyncio
 import logging
 import unittest
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
+
+import aiohttp
 
 from slack_sdk.socket_mode.aiohttp import SocketModeClient
 from slack_sdk.web.async_client import AsyncWebClient
@@ -68,6 +70,37 @@ async def close_then_raise(*args, **kwargs):
         self.assertTrue(client.closed)
         client.logger.exception.assert_not_called()
 
+    @async_test
+    async def test_connect_recreates_closed_aiohttp_session(self):
+        # Regression test for #1922: a closed aiohttp session made every reconnect attempt fail forever.
+        client = SocketModeClient(
+            app_token="xapp-A111-222-xyz",
+            web_client=self.web_client,
+            auto_reconnect_enabled=False,
+            ping_interval=0.01,
+        )
+        client.wss_uri = "ws://localhost:8888/link"
+        old_session = client.aiohttp_client_session
+        await old_session.close()
+        self.assertFalse(await client.is_connected())
+
+        used_sessions = []
+
+        async def ws_connect(session, *args, **kwargs):
+            used_sessions.append((session, session.closed))
+            await client.close()
+            raise RuntimeError("stop connecting")
+
+        with patch.object(aiohttp.ClientSession, "ws_connect", ws_connect):
+            await asyncio.wait_for(client.connect(), timeout=1.0)
+
+        self.assertEqual(len(used_sessions), 1)
+        session, closed = used_sessions[0]
+        self.assertIsNot(session, old_session)
+        self.assertFalse(closed)
+        self.assertIs(client.aiohttp_client_session, session)
+        self.assertTrue(session.closed)  # closed again by client.close()
+
     @async_test
     async def test_init_with_loop(self):
         client = SocketModeClient(