From 9d636371c83a796f407347affec0f6b4b8abeeb6 Mon Sep 17 00:00:00 2001 From: Dylan Pulver Date: Wed, 9 Sep 2026 13:27:08 +0200 Subject: [PATCH] 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. --- slack_sdk/socket_mode/aiohttp/__init__.py | 5 +- slack_sdk/socket_mode/websockets/__init__.py | 4 +- .../socket_mode/test_aiohttp.py | 49 ++++++++++++++++++ .../socket_mode/test_websockets.py | 51 +++++++++++++++++++ 4 files changed, 106 insertions(+), 3 deletions(-) diff --git a/slack_sdk/socket_mode/aiohttp/__init__.py b/slack_sdk/socket_mode/aiohttp/__init__.py index 5fb1d1171..c0cbbe52b 100644 --- a/slack_sdk/socket_mode/aiohttp/__init__.py +++ b/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): diff --git a/slack_sdk/socket_mode/websockets/__init__.py b/slack_sdk/socket_mode/websockets/__init__.py index 3b217e4eb..62c04613b 100644 --- a/slack_sdk/socket_mode/websockets/__init__.py +++ b/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): diff --git a/tests/slack_sdk_async/socket_mode/test_aiohttp.py b/tests/slack_sdk_async/socket_mode/test_aiohttp.py index 4834c0c84..05d833071 100644 --- a/tests/slack_sdk_async/socket_mode/test_aiohttp.py +++ b/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() diff --git a/tests/slack_sdk_async/socket_mode/test_websockets.py b/tests/slack_sdk_async/socket_mode/test_websockets.py index 322bd8c73..aa9df6abc 100644 --- a/tests/slack_sdk_async/socket_mode/test_websockets.py +++ b/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()