fix(socket-mode): only release connect_operation_lock when this task acquired it

[ckarnell]

Summary

AsyncBaseSocketModeClient.connect_to_new_endpoint() releases connect_operation_lock whenever the lock is merely locked(), rather than when this coroutine actually acquired it:

try:
    await self.connect_operation_lock.acquire()
    ...
finally:
    if self.connect_operation_lock.locked() is True:
        self.connect_operation_lock.release()

asyncio.Lock has no notion of ownership, unlike threading.Lock. A release() call from a task that never held the lock succeeds and frees it for everyone.

Why this is reachable on an ordinary reconnect, not just at shutdown

connect() cancels both background tasks on every successful reconnection:

if self.current_session_monitor is not None:
    self.current_session_monitor.cancel()
...
if self.message_receiver is not None:
    self.message_receiver.cancel()

and both of those tasks call connect_to_new_endpoint() (aiohttp/__init__.py, in monitor_current_session() and receive_messages()). So:

1. monitor_current_session calls connect_to_new_endpoint(), acquires the lock, and is inside await self.connect().
2. receive_messages reaches its own reconnect path and suspends inside connect_operation_lock.acquire().
3. Still inside the first task's connect(), self.message_receiver.cancel() cancels the second task while it is waiting on the lock.
4. CancelledError propagates out of acquire(), so that task never acquired anything, but locked() is True because the first task still holds it.
5. The finally block releases the first task's lock, mid-reconnect.

Mutual exclusion around the reconnect is gone, so another reconnect can start while one is still in flight.

Reproduction

tests/slack_sdk_async/socket_mode/test_async_client_lock.py drives the real AsyncBaseSocketModeClient with a connect() that blocks until released, cancels a second caller while it waits on the lock, and asserts the lock is still held.

Without the change:

AssertionError: False is not true : the cancelled waiter released a lock it never acquired

With the change it passes.

The fix

Track acquisition in a local, which is exactly what the synchronous BaseSocketModeClient already does:

acquired = self.connect_operation_lock.acquire(blocking=True, timeout=5)
...
finally:
    if acquired:
        self.connect_operation_lock.release()

So this makes the async client consistent with its sync counterpart. Three lines, no behaviour change on the happy path: when the coroutine does acquire the lock, acquired is True and it releases exactly as before.

What I verified

• The new test fails without the change and passes with it, with the assertion message above.
• tests/slack_sdk_async/socket_mode/ passes: 14 passed.
• black --line-length 125 (the setting in pyproject.toml) leaves both files unchanged.
• The same pattern is present on main, not only in the released package.

What I did not check

Whether this is the cause of any specific reported disconnection. The area has prior history (#1110, #1112) whose symptoms are consistent with losing mutual exclusion around reconnects, but I have not tied this to a particular report, and I have not tried to reproduce it against live Slack infrastructure.

A note on severity

Not a security issue, and it needs a cancellation to land in a narrow window, so it is unlikely to be a frequent cause of trouble. The consequence when it does happen is two concurrent reconnects rather than data loss. The sync client already gets this right, which is the main argument for the change.

Category

• slack_sdk.socket_mode (Socket Mode client)

Requirements

• I've read and understood the Contributing Guidelines and have done my best effort to follow them.
• I've read and agree to the Code of Conduct.
• I've run ./scripts/run_validation.sh after making the changes. Not run in full. What I did run is listed under "What I verified" above: the tests/slack_sdk_async/socket_mode/ suite (14 passed) and black --line-length 125 on both changed files. Happy to run the full script if you would like it before review.

[salesforce-cla]

Thanks for the contribution! Before we can merge this, we need @ckarnell to sign the Salesforce Inc. Contributor License Agreement.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 84.15%. Comparing base (79c528e) to head (114b1e1).
✅ All tests successful. No failed tests found.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1926      +/-   ##
==========================================
- Coverage   84.17%   84.15%   -0.03%     
==========================================
  Files         118      118              
  Lines       13425    13427       +2     
==========================================
- Hits        11301    11299       -2     
- Misses       2124     2128       +4     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

[ckarnell]

Hi I've signed the CLA, can this please get reviewed?

[WilliamBergamin]

Hello 👋 thanks for your thoughtful contribution 🙏

I left a few comments mainly around aligning the logic with the sync implementation

[WilliamBergamin]

Could we mirror the same pattern found ion the sync implementation

Suggested change

	await self.connect_operation_lock.acquire()
	acquired = True
	acquired = await self.connect_operation_lock.acquire(blocking=True, timeout=5)

[WilliamBergamin]

Similar here

Suggested change

	if force or not await self.is_connected():
	if force or (acquired and not await self.is_connected()):

[WilliamBergamin]

Suggested change

	self.logger.debug(f"For reconnection, the connect_operation_lock was acquired (session: {session_id})")
	self.logger.debug(f"For reconnection, the connect_operation_lock was acquired: {acquired} (session: {session_id})")

[WilliamBergamin]

Could we mirror the same testing pattern found in the sync implementation 🙏

[ckarnell]

Thanks for the review. The acquire(blocking=True, timeout=5) form is the sync threading.Lock signature. asyncio.Lock.acquire takes no arguments, so that line raises TypeError: acquire() got an unexpected keyword argument 'blocking', which is the same sync/async gap this PR is about. I did what it was aiming at with asyncio.wait_for(self.connect_operation_lock.acquire(), timeout=5): on timeout acquired stays False and the current holder keeps the lock, on success it is True.

There was no coverage for the bounded acquire, so I added a test. When the lock cannot be acquired, the endpoint stays un-rotated and connect() is never called, and the task releases nothing since it holds nothing. That test fails on the old code and passes on the new.

On mirroring the sync test pattern: I could not find a sync lock test to mirror, and the sync socket-mode tests are integration-style against a mock server. Did you have a specific test in mind, or is the async unit test above sufficient here?