fix: release async cluster transaction connections to their real owner

[ahmed5145]

When the slot map changes while an async TransactionStrategy holds a connection, release it to its actual ClusterNode owner before acquiring from the newly resolved node. Apply the same ownership resolution in reset() and _reinitialize_on_error() for sync/async parity (#4253 item 1).

Description of change

Addresses #4253 (item 1). Follow-up from the review of #4206.

Sync TransactionStrategy already checks pool ownership before reusing a held transaction connection and releases via find_connection_owner. Async did not: _get_client_and_connection_for_transaction() recomputed _transaction_node from the current slot map but reused _transaction_connection unconditionally, and reset() / _reinitialize_on_error() released through _transaction_node. If another operation refreshed the shared nodes_manager topology while this pipeline held a connection during WATCH (without this pipeline itself raising MOVED/ASK), the next command could send on a stale connection and reset() could push that connection onto the wrong node's free queue.

This PR brings async in line with sync:

• ClusterNode.owns_connection() / NodesManager.find_connection_owner()
• ownership check + release-to-real-owner before reuse in _get_client_and_connection_for_transaction()
• reset() and _reinitialize_on_error() resolve the owner explicitly before release

No public API changes. Sync is unchanged (already correct). Item 2 of #4253 (sync Script → cluster pipeline EVALSHA coverage) stays in #4206 and is not included here.

Pull Request check-list

• Do tests and lints pass with this change?
• Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
• Is the new or changed code fully tested?
• Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
• Is there an example added to the examples folder (if applicable)?

---

Note

Medium Risk
Touches async cluster connection pooling during WATCH/MULTI transactions after topology refresh. Incorrect ownership handling can leak or cross-pool connections, but the change is localized and covered by a new test.

Overview
Fixes async cluster TransactionStrategy so a connection held across a concurrent slot-map refresh is released to its real ClusterNode instead of being reused or returned via a stale _transaction_node.

Adds ClusterNode.owns_connection() and NodesManager.find_connection_owner(). _get_client_and_connection_for_transaction() now drops the old connection when the resolved node does not own it; reset() and _reinitialize_on_error() look up the owner by host/port before release().

Brings async in line with sync for #4253 (item 1). No public API change. A unit test covers mid-WATCH retargeting and reset() with a stale node pointer.

^{Reviewed by Cursor Bugbot for commit 16a7976. Bugbot is set up for automated code reviews on this repo. Configure here.}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 0ef10fe0e9

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 3 potential issues.

^{Reviewed by Cursor Bugbot for commit 0ef10fe. Configure here.}

[petyaslavova]

Hey @ahmed5145, thank you for your fast reaction on this! I'll review it later this week. Meanwhile, can you please fix the conflicts and rebase with master?

[ahmed5145]

Thanks! Rebased onto latest master and resolved the conflict in tests/test_asyncio/test_cluster.py (kept the merged #4206 EVALSHA coverage and our ownership regression test). Ready for review whenever you have time.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 8629be18d4

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 16a7976064

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[ahmed5145]

Hey @petyaslavova, friendly ping. Branch is updated with master. Ready for review whenever you have time.

CI is nearly green (365/368). The two failures are the known flaky standalone pubsub test (test_resubscribe_to_patterns_on_reconnection) on Redis 8.10 / Python 3.10 unified-responses cells, unrelated to this PR. One hiredis fixed-clients job was cancelled after ~1h. I don't have permission to re-run failed jobs on the redis/redis-py repo. Thanks!

[Mukller]

Verified locally on the PR branch (Python 3.13) as far as possible without a live cluster:

What I checked statically and found correct:

1. Owner-resolution key consistency: find_connection_owner() builds get_node_name(connection.host, connection.port) — exactly the format nodes_manager.nodes_cache is keyed by (node.name). The lookup is sound.
2. owns_connection contract: ClusterNode._connections is appended to in the acquire path (self._connections.append(connection)), so membership testing is a valid ownership predicate; __slots__-declared, list-backed, cheap for pool-sized n.
3. The three release sites are now owner-correct: _get_client_and_connection_for_transaction (stale connection after MOVED refresh → released to real owner, fresh one acquired), _reinitialize_on_error, and reset()'s finally-block. The detach-before-release discipline is preserved, and both new fallbacks (elif self._transaction_node: in reset; silent drop when no owner exists in the topology-change path) are reasonable — a connection whose node vanished from topology belongs to a pool that's being torn down anyway.

Tests: all mock-based transaction tests pass locally (-k "transaction or Transaction" → 3 passed). One new test fails locally (test_evalsha_zero_keys_in_transaction_pipeline[pool]) plus two setup errors — but those are server-dependent tests failing with ConnectionRefusedError: localhost:6379, identical to how server-dependent tests fail on upstream main in this environment; not related to the diff.

CI shows 365 green with 2-3 scattered matrix failures — same flaky-infra pattern visible on neighboring PRs today.

One residual question for maintainers (no blocker): in _get_client_and_connection_for_transaction, if the held connection's node was removed from the topology (find_connection_owner → None), the reference is dropped without release. That looks intentional (the old pool is being disposed), but it might be worth a one-line comment there so future readers don't "fix" it into a double-release.

Good catch on the stale-owner leak — this matches the sync TransactionStrategy behavior now.