Cluster transaction/script follow-ups for handling EVALSHA in async cluster transaction; async connection ownership and missing sync Script test

Description

Two sync/async parity gaps were found while reviewing #4206. Grouping them here since both live in the cluster transaction/script area.

1. Async cluster transaction reuses and releases the previous node's connection when the transaction node changes

redis.asyncio.cluster's TransactionStrategy does not track which node its transaction connection was acquired from, while the sync redis.cluster.TransactionStrategy does. When the node that owns the transaction slot changes while a connection is held, the async strategy both sends commands on the stale connection and releases it into the wrong node's pool.

Two places diverge from sync:

a. TransactionStrategy._get_client_and_connection_for_transaction() (redis/asyncio/cluster.py) recomputes self._transaction_node from the current slot map, but reuses self._transaction_connection unconditionally when it is already set:

node = self._pipe.cluster_client.nodes_manager.get_node_from_slot(...)
self._transaction_node = node
if not self._transaction_connection:
    self._transaction_connection = self._transaction_node.acquire_connection()
return self._transaction_node, self._transaction_connection

The sync counterpart checks ownership first, releases the connection back to its real owner, and acquires a new one from the new node:

if self._transaction_connection:
    if not redis_node.connection_pool.owns_connection(self._transaction_connection):
        previous_node = self._nodes_manager.find_connection_owner(self._transaction_connection)
        previous_node.connection_pool.release(self._transaction_connection)
        self._transaction_connection = None

b. TransactionStrategy.reset() releases with self._transaction_node.release(connection), i.e. to the current node. If the node changed, a connection acquired from node A is pushed onto node B's free queue. The sync reset() resolves the owner explicitly via self._nodes_manager.find_connection_owner(...) before releasing.

Reachability

_reinitialize_on_error() already disconnects and releases the connection on SLOT_REDIRECT_ERRORS / CONNECTION_ERRORS, so the common redirection path is safe. The gap is the case where the slot map changes without this pipeline raising: nodes_manager is shared, so a concurrent operation that observes a MOVED (failover promoting a replica, resharding, a rolling upgrade) can refresh the topology while an async cluster transaction is holding a connection during WATCH. The next immediate command then resolves node B, sends on node A's connection, and reset() releases node A's connection into node B's pool — the connection is lost from node A's accounting and node B can hand out a connection to the wrong server.

Expected

The async strategy should carry the same ownership check as sync: verify that the held connection belongs to the newly resolved node before reusing it, release it to its actual owner otherwise, and resolve the owner (not self._transaction_node) when releasing in reset().

Suggested test

An async test that holds a transaction connection through WATCH, mutates the slot map so the slot resolves to a different ClusterNode, then asserts that the next command uses a connection owned by the new node and that the original connection is returned to the original node's pool.

2. Missing sync test for a registered Script queuing EVALSHA on a cluster pipeline

#4206 added test_async_script_queues_evalsha_on_cluster_pipeline in tests/test_asyncio/test_cluster.py, which asserts that await script(client=cluster_pipeline) queues one EVALSHA command instead of dropping it. There is no sync counterpart.

The sync path works for a different reason and is worth pinning down: Script.__call__ (redis/commands/core.py) only calls client.scripts.add(self) for redis.client.Pipeline, and redis.cluster.ClusterPipeline is not a subclass, so it falls through to client.evalsha(...), which queues and returns the pipeline. Before #4206 that raised RedisClusterException; now it queues, and nothing guards the behavior.

Suggested test

In tests/test_cluster.py, mirroring the async test: register a script on a mocked RedisCluster, call script(client=pipe) on a ClusterPipeline, and assert the command queue holds exactly one entry whose args start with ("EVALSHA", script.sha).

Notes

Both items come out of the review of #4206. Item 1 became visible because that PR makes a mid-transaction slot retarget possible for zero-key EVAL/EVALSHA; that specific path is not reachable today (a keyed WATCH fixes the slot before any connection is acquired), so it is tracked here as a parity gap rather than a blocker. Item 2 is test-only, no behavior change expected.