From 114b1e155171e0ec3bfe1665b475de65287849ce Mon Sep 17 00:00:00 2001 From: Cohen Karnell Date: Thu, 30 Jul 2026 02:21:04 -0500 Subject: [PATCH 1/2] Only release connect_operation_lock when this task acquired it AsyncBaseSocketModeClient.connect_to_new_endpoint() releases the lock whenever connect_operation_lock.locked() is true, rather than when this coroutine actually acquired it. asyncio.Lock has no notion of ownership, so release() from a task that never held the lock succeeds and frees it for everyone. That is reachable on an ordinary reconnect. connect() cancels message_receiver and current_session_monitor on every successful reconnection, and both of those tasks call connect_to_new_endpoint(). So one of them can be cancelled while suspended inside acquire(), and its finally block then releases the lock belonging to the reconnect still in progress, dropping mutual exclusion around it. Track acquisition in a local instead, which is what the synchronous BaseSocketModeClient already does with its acquired flag. --- slack_sdk/socket_mode/async_client.py | 4 +- .../socket_mode/test_async_client_lock.py | 70 +++++++++++++++++++ 2 files changed, 73 insertions(+), 1 deletion(-) create mode 100644 tests/slack_sdk_async/socket_mode/test_async_client_lock.py diff --git a/slack_sdk/socket_mode/async_client.py b/slack_sdk/socket_mode/async_client.py index 5225dc285..317b74236 100644 --- a/slack_sdk/socket_mode/async_client.py +++ b/slack_sdk/socket_mode/async_client.py @@ -71,15 +71,17 @@ async def disconnect(self): async def connect_to_new_endpoint(self, force: bool = False): session_id = await self.session_id() + acquired = False try: await self.connect_operation_lock.acquire() + acquired = True if self.trace_enabled: self.logger.debug(f"For reconnection, the connect_operation_lock was acquired (session: {session_id})") if force or not await self.is_connected(): self.wss_uri = await self.issue_new_wss_url() await self.connect() finally: - if self.connect_operation_lock.locked() is True: + if acquired: self.connect_operation_lock.release() if self.trace_enabled: self.logger.debug(f"The connect_operation_lock for reconnection was released (session: {session_id})") diff --git a/tests/slack_sdk_async/socket_mode/test_async_client_lock.py b/tests/slack_sdk_async/socket_mode/test_async_client_lock.py new file mode 100644 index 000000000..824b99fb8 --- /dev/null +++ b/tests/slack_sdk_async/socket_mode/test_async_client_lock.py @@ -0,0 +1,70 @@ +import asyncio +import unittest + +from slack_sdk.socket_mode.async_client import AsyncBaseSocketModeClient +from tests.slack_sdk_async.helpers import async_test + + +class _FakeClient(AsyncBaseSocketModeClient): + """Minimal concrete client: connect() blocks until released, like a real reconnect.""" + + def __init__(self): + self.connect_operation_lock = asyncio.Lock() + self.trace_enabled = False + self.wss_uri = "wss://example.com/original" + self.connected = False + self.connect_started = asyncio.Event() + self.allow_connect_to_finish = asyncio.Event() + + async def is_connected(self) -> bool: + return self.connected + + async def issue_new_wss_url(self) -> str: + return "wss://example.com/new" + + async def connect(self): + self.connect_started.set() + await self.allow_connect_to_finish.wait() + + async def disconnect(self): + pass + + async def session_id(self) -> str: + return "test-session" + + +class TestAsyncClientConnectLock(unittest.TestCase): + @async_test + async def test_cancelled_waiter_does_not_release_another_task_lock(self): + """A cancelled waiter must not release the reconnect that is still in progress. + + `connect()` cancels `message_receiver` and `current_session_monitor` on every + successful reconnection, and both of those tasks call `connect_to_new_endpoint()`. + So a task can be cancelled while it is suspended inside + `connect_operation_lock.acquire()`. `asyncio.Lock` has no concept of ownership, so + releasing on `locked()` alone lets that cancelled task free a lock it never held, + which drops mutual exclusion around the reconnect. + """ + client = _FakeClient() + + holder = asyncio.ensure_future(client.connect_to_new_endpoint(force=True)) + await asyncio.wait_for(client.connect_started.wait(), timeout=5) + self.assertTrue(client.connect_operation_lock.locked()) + + waiter = asyncio.ensure_future(client.connect_to_new_endpoint(force=True)) + await asyncio.sleep(0.1) + + waiter.cancel() + with self.assertRaises(asyncio.CancelledError): + await waiter + + # The holder is still inside connect(), so the lock must still be held. + self.assertFalse(holder.done()) + self.assertTrue( + client.connect_operation_lock.locked(), + "the cancelled waiter released a lock it never acquired", + ) + + client.allow_connect_to_finish.set() + await asyncio.wait_for(holder, timeout=5) + self.assertFalse(client.connect_operation_lock.locked()) From 5335d6e5f61d3b267cd54c2f2f8e27ba2c2c0725 Mon Sep 17 00:00:00 2001 From: Cohen Karnell <7363269+ckarnell@users.noreply.github.com> Date: Wed, 12 Aug 2026 16:44:54 -0400 Subject: [PATCH 2/2] Align the async lock acquisition with the sync client Review feedback on #1926: mirror connect_to_new_endpoint's sync pattern, so the acquire is bounded and the reconnect is gated on having got the lock. asyncio.Lock.acquire() takes no arguments, unlike threading.Lock, so the sync client's acquire(blocking=True, timeout=5) is spelled with asyncio.wait_for here. The suggested acquire(blocking=True, timeout=5) raises TypeError on an asyncio lock. Adds a test for the branch this introduces: when the lock cannot be acquired, the endpoint is not rotated, connect() is not called, and no lock is released. The existing test passed either way, so the new behaviour was uncovered. --- slack_sdk/socket_mode/async_client.py | 15 +++++++++---- .../socket_mode/test_async_client_lock.py | 21 +++++++++++++++++++ 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/slack_sdk/socket_mode/async_client.py b/slack_sdk/socket_mode/async_client.py index 317b74236..304914858 100644 --- a/slack_sdk/socket_mode/async_client.py +++ b/slack_sdk/socket_mode/async_client.py @@ -73,11 +73,18 @@ async def connect_to_new_endpoint(self, force: bool = False): session_id = await self.session_id() acquired = False try: - await self.connect_operation_lock.acquire() - acquired = True + # asyncio.Lock.acquire() takes no arguments, so the sync client's + # acquire(blocking=True, timeout=5) is spelled with wait_for here. + try: + await asyncio.wait_for(self.connect_operation_lock.acquire(), timeout=5) + acquired = True + except asyncio.TimeoutError: + acquired = False if self.trace_enabled: - self.logger.debug(f"For reconnection, the connect_operation_lock was acquired (session: {session_id})") - if force or not await self.is_connected(): + self.logger.debug( + f"For reconnection, the connect_operation_lock was acquired: {acquired} (session: {session_id})" + ) + if force or (acquired and not await self.is_connected()): self.wss_uri = await self.issue_new_wss_url() await self.connect() finally: diff --git a/tests/slack_sdk_async/socket_mode/test_async_client_lock.py b/tests/slack_sdk_async/socket_mode/test_async_client_lock.py index 824b99fb8..17ad05ce1 100644 --- a/tests/slack_sdk_async/socket_mode/test_async_client_lock.py +++ b/tests/slack_sdk_async/socket_mode/test_async_client_lock.py @@ -1,5 +1,6 @@ import asyncio import unittest +from unittest.mock import patch from slack_sdk.socket_mode.async_client import AsyncBaseSocketModeClient from tests.slack_sdk_async.helpers import async_test @@ -68,3 +69,23 @@ async def test_cancelled_waiter_does_not_release_another_task_lock(self): client.allow_connect_to_finish.set() await asyncio.wait_for(holder, timeout=5) self.assertFalse(client.connect_operation_lock.locked()) + + @async_test + async def test_reconnect_is_skipped_when_the_lock_cannot_be_acquired(self): + """Mirrors the sync client: acquire is bounded, and a failed acquire skips the work. + + The sync client calls acquire(blocking=True, timeout=5) and gates the reconnect on + `acquired`, so a caller that never gets the lock does not reconnect and does not + release a lock it does not hold. + """ + client = _FakeClient() + # connect() must not block here: on the unpatched client the acquire succeeds and + # this test has to FAIL rather than hang, since a hanging control cannot be read. + client.allow_connect_to_finish.set() + + with patch("asyncio.wait_for", side_effect=asyncio.TimeoutError): + await client.connect_to_new_endpoint() + + self.assertFalse(client.connect_started.is_set(), "reconnected without holding the lock") + self.assertEqual(client.wss_uri, "wss://example.com/original", "the endpoint was rotated anyway") + self.assertFalse(client.connect_operation_lock.locked(), "released a lock it never acquired")