fix: allow EVALSHA in ClusterPipeline (#2914)

.github/wordlist.txt

@@ -132,6 +132,7 @@ gmail
 himport
 hiredis
 http
+idempotent
 idx
 iff
 ini
@@ -182,6 +183,7 @@ redismodules
 reinitialization
 replicaof
 repo
+resharding
 runtime
 sedrik
 serializers

docs/lua_scripting.rst

@@ -112,4 +112,50 @@ 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.
+
+``load_scripts`` and ``script_load_for_pipeline`` remain **not supported**
+on cluster pipelines. On the sync client, ``ClusterPipeline.eval()`` is also
+blocked; the async cluster pipeline has no ``eval`` override.
+
+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``, then retry
+  only when replay is safe (for example an idempotent or single-command
+  pipeline). Blindly re-executing a multi-command pipeline can duplicate
+  side effects: Redis still runs the rest of a non-transactional batch when
+  one ``EVALSHA`` returns ``NOSCRIPT``, and the client raises only after
+  reading every response. 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:
+   ...     # single-command pipeline: safe to reload and retry
+   ...     sha = rc.script_load(lua)  # reload on current primaries
+   ...     result = run()

redis/asyncio/cluster.py

@@ -77,6 +77,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,
@@ -3074,6 +3075,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
@@ -3082,6 +3085,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 scripts; 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]:
@@ -3139,7 +3171,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
@@ -3509,6 +3541,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

@@ -3764,6 +3764,20 @@ def inner(*args, **kwargs):
     return inner
 
 
+def is_zero_key_eval_command(*args) -> bool:
+    """
+    True for EVAL/EVALSHA with numkeys=0 (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",
@@ -3782,7 +3796,6 @@ def inner(*args, **kwargs):
     "CONFIG",
     "DBSIZE",
     "ECHO",
-    "EVALSHA",
     "FLUSHALL",
     "FLUSHDB",
     "INFO",
@@ -4260,18 +4273,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(
@@ -4575,13 +4597,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 scripts; 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.
@@ -4618,7 +4671,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
@@ -4966,6 +5019,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(

redis/commands/core.py

@@ -11003,10 +11003,15 @@ async def __call__(
         args = tuple(keys) + tuple(args)
         # make sure the Redis server knows about the script
         from redis.asyncio.client import Pipeline
+        from redis.asyncio.cluster import ClusterPipeline
 
         if isinstance(client, Pipeline):
             # Make sure the pipeline can register the script before executing.
             client.scripts.add(self)
+        if isinstance(client, ClusterPipeline):
+            # ClusterPipeline.evalsha queues synchronously. Awaiting the
+            # pipeline would call initialize() and drop the queued command.
+            return client.evalsha(self.sha, len(keys), *args)
         try:
             return await client.evalsha(self.sha, len(keys), *args)
         except NoScriptError: