fix(socket-mode): only release connect_operation_lock in send_message when acquired

slack_sdk/socket_mode/aiohttp/__init__.py

@@ -434,15 +434,16 @@ async def send_message(self, message: str):
                 )
             # Although acquiring self.connect_operation_lock also for the first method call is the safest way,
             # we avoid synchronizing a lot for better performance. That's why we are doing a retry here.
+            acquired = False
             try:
-                await self.connect_operation_lock.acquire()
+                acquired = await self.connect_operation_lock.acquire()
                 if await self.is_connected():
                     await self.current_session.send_str(message)  # type: ignore[union-attr]
                 else:
                     self.logger.warning(f"The current session ({session_id}) is no longer active. Failed to send a message")
                     raise e
             finally:
-                if self.connect_operation_lock.locked() is True:
+                if acquired:
                     self.connect_operation_lock.release()
 
     async def close(self):

slack_sdk/socket_mode/websockets/__init__.py

@@ -245,14 +245,16 @@ async def send_message(self, message: str):
                 )
             # Although acquiring self.connect_operation_lock also for the first method call is the safest way,
             # we avoid synchronizing a lot for better performance. That's why we are doing a retry here.
+            acquired = False
             try:
+                acquired = await self.connect_operation_lock.acquire()
                 if await self.is_connected():
                     await self.current_session.send(message)  # type: ignore[union-attr]
                 else:
                     self.logger.warning(f"The current session ({session_id}) is no longer active. Failed to send a message")
                     raise e
             finally:
-                if self.connect_operation_lock.locked() is True:
+                if acquired:
                     self.connect_operation_lock.release()
 
     async def close(self):

tests/slack_sdk_async/socket_mode/test_aiohttp.py

@@ -134,3 +134,52 @@ async def test_enqueue_message(self):
 
 async def listener(self, message, raw_message):
     pass
+
+
+class _LockProbeClient(SocketModeClient):
+    """Just enough state for send_message(), with no real connection."""
+
+    def __init__(self):
+        self.connect_operation_lock = asyncio.Lock()
+        self.logger = logging.getLogger(__name__)
+        self.closed = False
+        self.current_session = MagicMock()
+
+    async def is_connected(self) -> bool:
+        return True
+
+    @classmethod
+    def build_session_id(cls, session) -> str:
+        return "test-session"
+
+    async def session_id(self) -> str:
+        return "test-session"
+
+
+class TestAiohttpSendMessageLock(unittest.TestCase):
+    @async_test
+    async def test_send_message_leaves_another_tasks_lock_alone(self):
+        client = _LockProbeClient()
+
+        async def send_str(message):
+            raise ConnectionError("the underlying connection was replaced")
+
+        client.current_session.send_str = send_str
+
+        # Stand in for a reconnect holding the lock for the whole call.
+        await client.connect_operation_lock.acquire()
+
+        task = asyncio.ensure_future(client.send_message("hello"))
+        await asyncio.sleep(0.1)
+
+        task.cancel()
+        try:
+            await task
+        except (asyncio.CancelledError, ConnectionError):
+            pass
+
+        self.assertTrue(
+            client.connect_operation_lock.locked(),
+            "send_message() released the connect lock held by another task",
+        )
+        client.connect_operation_lock.release()

tests/slack_sdk_async/socket_mode/test_websockets.py

@@ -1,4 +1,9 @@
+import asyncio
+import logging
 import unittest
+from unittest.mock import MagicMock
+
+from websockets.exceptions import WebSocketException
 
 from slack_sdk.socket_mode.websockets import SocketModeClient
 from slack_sdk.web.async_client import AsyncWebClient
@@ -73,3 +78,49 @@ async def test_enqueue_message(self):
 
 async def listener(message, raw_message):
     pass
+
+
+class _LockProbeClient(SocketModeClient):
+    """Just enough state for send_message(), with no real connection."""
+
+    def __init__(self):
+        self.connect_operation_lock = asyncio.Lock()
+        self.logger = logging.getLogger(__name__)
+        self.closed = False
+        self.current_session = MagicMock()
+
+    async def is_connected(self) -> bool:
+        return True
+
+    @classmethod
+    def build_session_id(cls, session) -> str:
+        return "test-session"
+
+
+class TestWebsocketsSendMessageLock(unittest.TestCase):
+    @async_test
+    async def test_send_message_leaves_another_tasks_lock_alone(self):
+        client = _LockProbeClient()
+
+        async def send(message):
+            raise WebSocketException("the underlying connection was replaced")
+
+        client.current_session.send = send
+
+        # Stand in for a reconnect holding the lock for the whole call.
+        await client.connect_operation_lock.acquire()
+
+        task = asyncio.ensure_future(client.send_message("hello"))
+        await asyncio.sleep(0.1)
+
+        task.cancel()
+        try:
+            await task
+        except (asyncio.CancelledError, WebSocketException):
+            pass
+
+        self.assertTrue(
+            client.connect_operation_lock.locked(),
+            "send_message() released the connect lock held by another task",
+        )
+        client.connect_operation_lock.release()