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

#1926 fixed this in AsyncBaseSocketModeClient.connect_to_new_endpoint, but
the same `if self.connect_operation_lock.locked() is True: release()` shape
is still in send_message() in both async backends, and neither had a test.

websockets is the worse of the two: its retry block never acquires the lock
at all, it only releases it. So whenever a reconnect holds the lock and a
send_message() falls into the retry path, send_message() releases the
reconnect's lock and two reconnects can run at once. The aiohttp sibling
does acquire here, which is what makes the omission visible.

aiohttp acquires but releases on locked(), so a task cancelled while waiting
on acquire() releases whichever task does hold the lock. That is exactly the
case #1926 describes.

Both now track whether this task acquired, matching #1926 and the sync
backends, which take the lock with `with` and cannot get this wrong.

Author: dylanpulver

slack_sdk/socket_mode/aiohttp/__init__.py

@@ -438,15 +438,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

@@ -192,3 +192,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()