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()