fix: allow EVALSHA in ClusterPipeline (#2914)

docs/lua_scripting.rst

@@ -112,4 +112,45 @@ The following commands are not supported:
 - ``EVAL_RO``
 - ``EVALSHA_RO``
 
-Using scripting within pipelines in cluster mode is **not supported**.
+``EVALSHA`` can be used inside a ``ClusterPipeline``. Keys must map to the
+same hash slot (or ``numkeys`` may be ``0``, in which case the command is
+routed to a random primary, exactly as in the non-pipelined case above). In a
+transactional pipeline (``pipeline(transaction=True)``), zero-key ``EVALSHA``
+reuses the transaction's existing slot when one is already chosen, so multiple
+zero-key scripts (or a mix with keyed commands) stay single-slot.
+
+``EVAL``, ``load_scripts``, and ``script_load_for_pipeline`` remain
+**not supported** on cluster pipelines.
+
+Important caveats when using ``EVALSHA`` in a cluster pipeline:
+
+- The Lua script cache in Redis is **per-node and is not replicated**.
+  ``SCRIPT LOAD`` loads the script onto the *current* primaries only, at the
+  moment it is called.
+- Any topology change -- a failover that promotes a replica, a rolling
+  upgrade that replaces nodes, resharding, or adding a new shard -- can route
+  an ``EVALSHA`` to a node whose cache does not contain the script, which
+  fails with ``NOSCRIPT`` (``redis.exceptions.NoScriptError``).
+- Unlike the non-pipelined ``Script`` object, **a cluster pipeline performs no
+  automatic reload or retry** on ``NOSCRIPT``. Recovery is the caller's
+  responsibility: catch ``NoScriptError``, re-run ``SCRIPT LOAD``, and
+  re-execute the pipeline. Because zero-key ``EVALSHA`` is routed to a random
+  primary, the script must be present on **all** primaries.
+- In a **transactional** pipeline, a ``NOSCRIPT`` from ``EVALSHA`` is raised at
+  ``EXEC`` time and does **not** roll back the other commands (this follows
+  Redis ``MULTI``/``EXEC`` semantics), so partial application is possible.
+
+.. code:: python
+
+   >>> from redis.exceptions import NoScriptError
+   >>> sha = rc.script_load(lua)  # loads on all current primaries
+   >>> def run():
+   ...     with rc.pipeline() as pipe:
+   ...         pipe.evalsha(sha, 1, "{user}:1")
+   ...         return pipe.execute()
+   >>> try:
+   ...     result = run()
+   ... except NoScriptError:
+   ...     # a node was missing the script (e.g. after failover/upgrade)
+   ...     sha = rc.script_load(lua)  # reload on current primaries
+   ...     result = run()

redis/asyncio/cluster.py

@@ -75,6 +75,7 @@
     LoadBalancingStrategy,
     block_pipeline_command,
     get_node_name,
+    is_zero_key_eval_command,
     parse_cluster_shards,
     parse_cluster_shards_unified,
     parse_cluster_shards_with_str_keys,
@@ -2917,6 +2918,8 @@ def __init__(self, pipe: ClusterPipeline) -> None:
         self._explicit_transaction = False
         self._watching = False
         self._pipeline_slots: Set[int] = set()
+        # True once a keyed (non-slot-agnostic) command has fixed the slot
+        self._transaction_has_keyed_slot = False
         self._transaction_node: Optional[ClusterNode] = None
         self._transaction_connection: Optional[Connection] = None
         self._executing = False
@@ -2925,6 +2928,35 @@ def __init__(self, pipe: ClusterPipeline) -> None:
             RedisCluster.ERRORS_ALLOW_RETRY + self.SLOT_REDIRECT_ERRORS
         )
 
+    async def _resolve_transaction_slot(self, *args) -> Optional[int]:
+        """
+        Pick a slot for a transactional pipeline command.
+
+        Zero-key EVAL/EVALSHA can run on any primary. Reuse an existing
+        transaction slot when present so multiple zero-key scripts (or a
+        mix with keyed commands) stay single-slot.
+        """
+        if args[0] in self.NO_SLOTS_COMMANDS:
+            return None
+
+        if is_zero_key_eval_command(*args):
+            if self._pipeline_slots:
+                return next(iter(self._pipeline_slots))
+            return await self._pipe.cluster_client._determine_slot(*args)
+
+        slot_number = await self._pipe.cluster_client._determine_slot(*args)
+        if (
+            slot_number is not None
+            and self._pipeline_slots
+            and slot_number not in self._pipeline_slots
+            and not self._transaction_has_keyed_slot
+        ):
+            # Prior slots came only from zero-key EVAL/EVALSHA; retarget.
+            self._pipeline_slots.clear()
+        if slot_number is not None:
+            self._transaction_has_keyed_slot = True
+        return slot_number
+
     def _get_client_and_connection_for_transaction(
         self,
     ) -> Tuple[ClusterNode, Connection]:
@@ -2982,7 +3014,7 @@ async def _execute_command(
 
         slot_number: Optional[int] = None
         if args[0] not in self.NO_SLOTS_COMMANDS:
-            slot_number = await self._pipe.cluster_client._determine_slot(*args)
+            slot_number = await self._resolve_transaction_slot(*args)
 
         if (
             self._watching or args[0] in self.IMMEDIATE_EXECUTE_COMMANDS
@@ -3327,6 +3359,7 @@ async def reset(self):
             self._watching = False
             self._explicit_transaction = False
             self._pipeline_slots = set()
+            self._transaction_has_keyed_slot = False
             self._executing = False
 
     def multi(self):

redis/cluster.py

@@ -3623,6 +3623,20 @@ def inner(*args, **kwargs):
     return inner
 
 
+def is_zero_key_eval_command(*args) -> bool:
+    """
+    True for EVAL/EVALSHA with numkeys=0 (may run on any primary).
+    """
+    if len(args) < 3:
+        return False
+    if str(args[0]).upper() not in ("EVAL", "EVALSHA"):
+        return False
+    try:
+        return int(args[2]) == 0
+    except (TypeError, ValueError):
+        return False
+
+
 # Blocked pipeline commands
 PIPELINE_BLOCKED_COMMANDS = (
     "BGREWRITEAOF",
@@ -3641,7 +3655,6 @@ def inner(*args, **kwargs):
     "CONFIG",
     "DBSIZE",
     "ECHO",
-    "EVALSHA",
     "FLUSHALL",
     "FLUSHDB",
     "INFO",
@@ -4112,18 +4125,27 @@ def _send_cluster_commands(
                         # in a list of pre-defined request policies
                         command_flag = self.command_flags.get(command)
                         if not command_flag:
-                            # Fallback to default policy
-                            if not self._pipe.get_default_node():
-                                keys = None
-                            else:
-                                keys = self._pipe._get_command_keys(*c.args)
-                            if not keys or len(keys) == 0:
-                                command_policies = CommandPolicies()
-                            else:
+                            # Fallback to default policy.
+                            # EVAL/EVALSHA must not use _get_command_keys(): Redis
+                            # <7 breaks on COMMAND GETKEYS when numkeys is 0.
+                            # Other unflagged commands keep the keyless fallback.
+                            if command in ("EVAL", "EVALSHA"):
                                 command_policies = CommandPolicies(
                                     request_policy=RequestPolicy.DEFAULT_KEYED,
                                     response_policy=ResponsePolicy.DEFAULT_KEYED,
                                 )
+                            else:
+                                if not self._pipe.get_default_node():
+                                    keys = None
+                                else:
+                                    keys = self._pipe._get_command_keys(*c.args)
+                                if not keys or len(keys) == 0:
+                                    command_policies = CommandPolicies()
+                                else:
+                                    command_policies = CommandPolicies(
+                                        request_policy=RequestPolicy.DEFAULT_KEYED,
+                                        response_policy=ResponsePolicy.DEFAULT_KEYED,
+                                    )
                         else:
                             if command_flag in self._pipe._command_flags_mapping:
                                 command_policies = CommandPolicies(
@@ -4417,13 +4439,44 @@ def __init__(self, pipe: ClusterPipeline):
         self._explicit_transaction = False
         self._watching = False
         self._pipeline_slots: Set[int] = set()
+        # True once a keyed (non-slot-agnostic) command has fixed the slot
+        self._transaction_has_keyed_slot = False
         self._transaction_connection: Optional[Connection] = None
         self._executing = False
         self._retry = copy(self._pipe.retry)
         self._retry.update_supported_errors(
             RedisCluster.ERRORS_ALLOW_RETRY + self.SLOT_REDIRECT_ERRORS
         )
 
+    def _resolve_transaction_slot(self, *args) -> Optional[int]:
+        """
+        Pick a slot for a transactional pipeline command.
+
+        Zero-key EVAL/EVALSHA can run on any primary. Reuse an existing
+        transaction slot when present so multiple zero-key scripts (or a
+        mix with keyed commands) stay single-slot.
+        """
+        if args[0] in ClusterPipeline.NO_SLOTS_COMMANDS:
+            return None
+
+        if is_zero_key_eval_command(*args):
+            if self._pipeline_slots:
+                return next(iter(self._pipeline_slots))
+            return self._pipe.determine_slot(*args)
+
+        slot_number = self._pipe.determine_slot(*args)
+        if (
+            slot_number is not None
+            and self._pipeline_slots
+            and slot_number not in self._pipeline_slots
+            and not self._transaction_has_keyed_slot
+        ):
+            # Prior slots came only from zero-key EVAL/EVALSHA; retarget.
+            self._pipeline_slots.clear()
+        if slot_number is not None:
+            self._transaction_has_keyed_slot = True
+        return slot_number
+
     def _get_client_and_connection_for_transaction(self) -> Tuple[Redis, Connection]:
         """
         Find a connection for a pipeline transaction.
@@ -4460,7 +4513,7 @@ def _get_client_and_connection_for_transaction(self) -> Tuple[Redis, Connection]
     def execute_command(self, *args, **kwargs):
         slot_number: Optional[int] = None
         if args[0] not in ClusterPipeline.NO_SLOTS_COMMANDS:
-            slot_number = self._pipe.determine_slot(*args)
+            slot_number = self._resolve_transaction_slot(*args)
 
         if (
             self._watching or args[0] in self.IMMEDIATE_EXECUTE_COMMANDS
@@ -4785,6 +4838,7 @@ def reset(self):
         self._watching = False
         self._explicit_transaction = False
         self._pipeline_slots = set()
+        self._transaction_has_keyed_slot = False
         self._executing = False
 
     def send_cluster_commands(

tests/test_asyncio/test_cluster.py

@@ -3863,6 +3863,110 @@ async def test_blocked_methods(self, r: RedisCluster) -> None:
                 "when running redis in cluster mode..."
             )
 
+    async def test_evalsha_not_blocked_on_cluster_pipeline(self) -> None:
+        """EVALSHA must be usable on async ClusterPipeline (see #2914)."""
+        assert "EVALSHA" not in PIPELINE_BLOCKED_COMMANDS
+        r = await get_mocked_redis_client(host=default_host, port=default_port)
+        try:
+            pipe = r.pipeline()
+            sha = "a" * 40
+            returned = pipe.evalsha(sha, 1, "foo", "bar")
+            assert returned is pipe
+            queue = pipe._execution_strategy._command_queue
+            assert len(queue) == 1
+            assert queue[0].args == ("EVALSHA", sha, 1, "foo", "bar")
+        finally:
+            await r.aclose()
+
+    async def test_evalsha_zero_keys_pipeline_execute(self) -> None:
+        """Async ClusterPipeline must execute EVALSHA with numkeys=0."""
+        r = await get_mocked_redis_client(host=default_host, port=default_port)
+        try:
+            mock_all_nodes_resp(r, 42)
+            sha = "a" * 40
+            async with r.pipeline() as pipe:
+                pipe.evalsha(sha, 0)
+                assert await pipe.execute() == [42]
+        finally:
+            await r.aclose()
+
+    async def test_evalsha_zero_keys_reuses_slot_in_transaction(self) -> None:
+        """Zero-key EVALSHA in a transactional async pipeline reuses one slot."""
+        r = await get_mocked_redis_client(host=default_host, port=default_port)
+        try:
+            sha = "a" * 40
+            async with r.pipeline(transaction=True) as pipe:
+                pipe.evalsha(sha, 0)
+                pipe.evalsha(sha, 0)
+                slots = pipe._execution_strategy._pipeline_slots
+                assert len(slots) == 1
+
+                keyed_slot = key_slot(b"foo")
+
+                async def _fake_determine_slot(*_args, **_kwargs):
+                    return keyed_slot
+
+                with mock.patch.object(
+                    pipe.cluster_client,
+                    "_determine_slot",
+                    side_effect=_fake_determine_slot,
+                ):
+                    pipe.set("foo", "bar")
+                assert pipe._execution_strategy._pipeline_slots == {keyed_slot}
+                assert pipe._execution_strategy._transaction_has_keyed_slot is True
+        finally:
+            await r.aclose()
+
+    async def test_evalsha_zero_keys_follows_keyed_slot_in_transaction(self) -> None:
+        """Zero-key EVALSHA after a keyed slot is fixed reuses that slot."""
+        r = await get_mocked_redis_client(host=default_host, port=default_port)
+        try:
+            sha = "a" * 40
+            async with r.pipeline(transaction=True) as pipe:
+                strategy = pipe._execution_strategy
+                strategy._pipeline_slots = {key_slot(b"foo")}
+                strategy._transaction_has_keyed_slot = True
+                pipe.evalsha(sha, 0)
+                assert strategy._pipeline_slots == {key_slot(b"foo")}
+        finally:
+            await r.aclose()
+
+    async def test_slotless_command_does_not_lock_keyed_slot_flag(self) -> None:
+        """Slotless commands must not block later keyed retargeting."""
+        r = await get_mocked_redis_client(host=default_host, port=default_port)
+        try:
+            sha = "a" * 40
+            async with r.pipeline(transaction=True) as pipe:
+                pipe.evalsha(sha, 0)
+                assert pipe._execution_strategy._transaction_has_keyed_slot is False
+
+                async def _no_slot(*_args, **_kwargs):
+                    return None
+
+                with mock.patch.object(
+                    pipe.cluster_client,
+                    "_determine_slot",
+                    side_effect=_no_slot,
+                ):
+                    pipe.execute_command("CLIENT TRACKING", "ON")
+                assert pipe._execution_strategy._transaction_has_keyed_slot is False
+
+                keyed_slot = key_slot(b"foo")
+
+                async def _fake_determine_slot(*_args, **_kwargs):
+                    return keyed_slot
+
+                with mock.patch.object(
+                    pipe.cluster_client,
+                    "_determine_slot",
+                    side_effect=_fake_determine_slot,
+                ):
+                    pipe.set("foo", "bar")
+                assert pipe._execution_strategy._pipeline_slots == {keyed_slot}
+                assert pipe._execution_strategy._transaction_has_keyed_slot is True
+        finally:
+            await r.aclose()
+
     async def test_empty_stack(self, r: RedisCluster) -> None:
         """If a pipeline is executed with no commands it should return a empty list."""
         p = r.pipeline()