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

slack_sdk/socket_mode/aiohttp/__init__.py

@@ -369,6 +369,10 @@ 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 and not self.closed:
+                    self.logger.info("The aiohttp client session is closed; creating a new one")
+                    self.aiohttp_client_session = aiohttp.ClientSession()
+
                 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,62 @@ 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_connect_does_not_recreate_session_when_closed_during_reconnect(self):
+        client = SocketModeClient(
+            app_token="xapp-A111-222-xyz",
+            web_client=self.web_client,
+            auto_reconnect_enabled=False,
+        )
+        client.wss_uri = "ws://localhost:8888/link"
+        old_session = client.aiohttp_client_session
+
+        async def close_during_reconnect():
+            client.closed = True
+            await old_session.close()
+
+        client.current_session = MagicMock()
+        client.current_session.close = close_during_reconnect
+        try:
+            with patch.object(aiohttp, "ClientSession") as new_session:
+                await asyncio.wait_for(client.connect(), timeout=1.0)
+                new_session.assert_not_called()
+            self.assertIs(client.aiohttp_client_session, old_session)
+            self.assertTrue(old_session.closed)
+        finally:
+            await client.close()
+
     @async_test
     async def test_init_with_loop(self):
         client = SocketModeClient(