diff --git a/.github/wordlist.txt b/.github/wordlist.txt index c66c9310e6..b72c49fa1b 100644 --- a/.github/wordlist.txt +++ b/.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 diff --git a/docs/lua_scripting.rst b/docs/lua_scripting.rst index bd7b9bc01d..9f6d96352f 100644 --- a/docs/lua_scripting.rst +++ b/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() diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py index 917dc5a8d8..9c445b5e57 100644 --- a/redis/asyncio/cluster.py +++ b/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): diff --git a/redis/cluster.py b/redis/cluster.py index ffd1885be6..8e96cd35b9 100644 --- a/redis/cluster.py +++ b/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,6 +4597,8 @@ 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) @@ -4582,6 +4606,35 @@ def __init__(self, pipe: ClusterPipeline): 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( diff --git a/redis/commands/core.py b/redis/commands/core.py index 42e7b9cde9..af53cb2d12 100644 --- a/redis/commands/core.py +++ b/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: diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py index 7ab028ddc3..9fbe93b88e 100644 --- a/tests/test_asyncio/test_cluster.py +++ b/tests/test_asyncio/test_cluster.py @@ -43,6 +43,7 @@ AskError, ClusterDownError, ConnectionError, + CrossSlotTransactionError, DataError, MaxConnectionsError, MovedError, @@ -3918,6 +3919,196 @@ 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_evalsha_zero_keys_then_cross_slot_raises_at_execute(self) -> None: + """ + Zero-key EVALSHA retarget must not swallow a later true cross-slot + conflict: two keyed commands on different slots still raise at execute(). + """ + r = await get_mocked_redis_client(host=default_host, port=default_port) + try: + sha = "a" * 40 + slot_a = key_slot(b"{foo}a") + slot_b = key_slot(b"{bar}b") + assert slot_a != slot_b + async with r.pipeline(transaction=True) as pipe: + slots = iter([111, slot_a, slot_b]) + + async def _fake_determine_slot(*_args, **_kwargs): + return next(slots) + + with mock.patch.object( + pipe.cluster_client, + "_determine_slot", + side_effect=_fake_determine_slot, + ): + pipe.evalsha(sha, 0) + pipe.set("{foo}a", "1") + pipe.set("{bar}b", "2") + assert pipe._execution_strategy._pipeline_slots == {slot_a, slot_b} + with pytest.raises( + CrossSlotTransactionError, + match=( + "All keys involved in a cluster transaction " + "must map to the same slot" + ), + ): + await pipe.execute() + finally: + await r.aclose() + + async def test_async_script_queues_evalsha_on_cluster_pipeline(self) -> None: + """AsyncScript must queue EVALSHA on ClusterPipeline without dropping it.""" + r = await get_mocked_redis_client(host=default_host, port=default_port) + try: + script = r.register_script("return 1") + async with r.pipeline() as pipe: + await script(client=pipe) + queue = pipe._execution_strategy._command_queue + assert len(queue) == 1 + assert queue[0].args[0] == "EVALSHA" + assert queue[0].args[1] == script.sha + finally: + await r.aclose() + + async def test_evalsha_in_pipeline(self, r: RedisCluster) -> None: + """ + EVALSHA is allowed in ClusterPipeline when keys map to one slot + and the script is already loaded on cluster primaries (#2914). + """ + multiply = "return redis.call('GET', KEYS[1]) * ARGV[1]" + sha = await r.script_load(multiply) + await r.set("{user}a", 2) + async with r.pipeline() as pipe: + pipe.evalsha(sha, 1, "{user}a", 3) + assert await pipe.execute() == [6] + + async def test_evalsha_zero_keys_in_pipeline(self, r: RedisCluster) -> None: + """ + EVALSHA with numkeys=0 must execute in ClusterPipeline without + calling COMMAND GETKEYS (broken on Redis <7 for this case). + """ + sha = await r.script_load("return 42") + async with r.pipeline() as pipe: + pipe.evalsha(sha, 0) + assert await pipe.execute() == [42] + + async def test_evalsha_zero_keys_in_transaction_pipeline( + self, r: RedisCluster + ) -> None: + """ + Multiple zero-key EVALSHA calls in a transactional pipeline share + one slot and execute successfully. + """ + sha = await r.script_load("return 42") + async with r.pipeline(transaction=True) as pipe: + pipe.evalsha(sha, 0) + pipe.evalsha(sha, 0) + assert await pipe.execute() == [42, 42] + 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() diff --git a/tests/test_cluster.py b/tests/test_cluster.py index 4d97dae557..e0509adc18 100644 --- a/tests/test_cluster.py +++ b/tests/test_cluster.py @@ -21,6 +21,7 @@ ) from redis.cluster import ( PRIMARY, + PIPELINE_BLOCKED_COMMANDS, REDIS_CLUSTER_HASH_SLOTS, REPLICA, ClusterNode, @@ -41,6 +42,7 @@ AskError, ClusterDownError, ConnectionError, + CrossSlotTransactionError, DataError, MovedError, NoPermissionError, @@ -3886,6 +3888,155 @@ def test_disconnect_calls_disconnect_on_existing_connections(self, r): mock_node_conn.disconnect.assert_called_once() +@pytest.mark.onlycluster +def test_evalsha_not_in_pipeline_blocked_commands(): + """EVALSHA must be usable in ClusterPipeline (see #2914).""" + assert "EVALSHA" not in PIPELINE_BLOCKED_COMMANDS + + +@pytest.mark.onlycluster +def test_evalsha_can_be_queued_on_cluster_pipeline(): + """ + Queuing EVALSHA on a ClusterPipeline must not raise the historical + 'blocked when running redis in cluster mode' error. Slot selection for + EVALSHA is already handled by RedisCluster.determine_slot(). + """ + r = 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") + + assert r.determine_slot("EVALSHA", sha, 1, "foo", "bar") == key_slot(b"foo") + finally: + r.close() + + +@pytest.mark.onlycluster +def test_evalsha_zero_keys_pipeline_execute_skips_get_command_keys(): + """ + Sync ClusterPipeline must route EVALSHA via determine_slot, not + COMMAND GETKEYS. Redis <7 fails on GETKEYS for EVALSHA with numkeys=0. + """ + r = get_mocked_redis_client(host=default_host, port=default_port) + try: + mock_all_nodes_resp(r, 42) + sha = "a" * 40 + with r.pipeline() as pipe: + pipe.evalsha(sha, 0) + with patch.object( + pipe, + "_get_command_keys", + side_effect=AssertionError( + "COMMAND GETKEYS must not be used for EVALSHA" + ), + ): + assert pipe.execute() == [42] + finally: + r.close() + + +@pytest.mark.onlycluster +def test_evalsha_zero_keys_reuses_slot_in_transaction(): + """ + Zero-key EVALSHA in a transactional pipeline must reuse one slot so + execute() does not raise CrossSlotTransactionError. + """ + r = get_mocked_redis_client(host=default_host, port=default_port) + try: + sha = "a" * 40 + 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 follow-up must retarget without needing a live COMMAND map. + keyed_slot = key_slot(b"foo") + with patch.object(pipe, "determine_slot", return_value=keyed_slot): + pipe.set("foo", "bar") + assert pipe._execution_strategy._pipeline_slots == {keyed_slot} + assert pipe._execution_strategy._transaction_has_keyed_slot is True + finally: + r.close() + + +@pytest.mark.onlycluster +def test_evalsha_zero_keys_follows_keyed_slot_in_transaction(): + """Zero-key EVALSHA after a keyed slot is fixed reuses that slot.""" + r = get_mocked_redis_client(host=default_host, port=default_port) + try: + sha = "a" * 40 + 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: + r.close() + + +@pytest.mark.onlycluster +def test_slotless_command_does_not_lock_keyed_slot_flag(): + """Slotless commands must not block later keyed retargeting.""" + r = get_mocked_redis_client(host=default_host, port=default_port) + try: + sha = "a" * 40 + with r.pipeline(transaction=True) as pipe: + pipe.evalsha(sha, 0) + assert pipe._execution_strategy._transaction_has_keyed_slot is False + + with patch.object(pipe, "determine_slot", return_value=None): + pipe.execute_command("CLIENT TRACKING", "ON") + assert pipe._execution_strategy._transaction_has_keyed_slot is False + + keyed_slot = key_slot(b"foo") + with patch.object(pipe, "determine_slot", return_value=keyed_slot): + pipe.set("foo", "bar") + assert pipe._execution_strategy._pipeline_slots == {keyed_slot} + assert pipe._execution_strategy._transaction_has_keyed_slot is True + finally: + r.close() + + +@pytest.mark.onlycluster +def test_evalsha_zero_keys_then_cross_slot_raises_at_execute(): + """ + Zero-key EVALSHA retarget must not swallow a later true cross-slot + conflict: two keyed commands on different slots still raise at execute(). + """ + r = get_mocked_redis_client(host=default_host, port=default_port) + try: + sha = "a" * 40 + slot_a = key_slot(b"{foo}a") + slot_b = key_slot(b"{bar}b") + assert slot_a != slot_b + with r.pipeline(transaction=True) as pipe: + with patch.object( + pipe, "determine_slot", side_effect=[111, slot_a, slot_b] + ): + pipe.evalsha(sha, 0) + pipe.set("{foo}a", "1") + pipe.set("{bar}b", "2") + assert pipe._execution_strategy._pipeline_slots == {slot_a, slot_b} + with pytest.raises( + CrossSlotTransactionError, + match=( + "All keys involved in a cluster transaction " + "must map to the same slot" + ), + ): + pipe.execute() + finally: + r.close() + + @pytest.mark.onlycluster class TestClusterPipeline: """ @@ -3921,6 +4072,40 @@ def test_blocked_arguments(self, r): str(ex.value).startswith("shard_hint is deprecated in cluster mode") is True ) + def test_evalsha_in_pipeline(self, r): + """ + EVALSHA is allowed in ClusterPipeline when keys map to one slot + and the script is already loaded on cluster primaries (#2914). + """ + multiply = "return redis.call('GET', KEYS[1]) * ARGV[1]" + sha = r.script_load(multiply) + # hash tag keeps the key on a known slot for the pipeline + r.set("{user}a", 2) + with r.pipeline() as pipe: + pipe.evalsha(sha, 1, "{user}a", 3) + assert pipe.execute() == [6] + + def test_evalsha_zero_keys_in_pipeline(self, r): + """ + EVALSHA with numkeys=0 must execute in ClusterPipeline without + calling COMMAND GETKEYS (broken on Redis <7 for this case). + """ + sha = r.script_load("return 42") + with r.pipeline() as pipe: + pipe.evalsha(sha, 0) + assert pipe.execute() == [42] + + def test_evalsha_zero_keys_in_transaction_pipeline(self, r): + """ + Multiple zero-key EVALSHA calls in a transactional pipeline share + one slot and execute successfully. + """ + sha = r.script_load("return 42") + with r.pipeline(transaction=True) as pipe: + pipe.evalsha(sha, 0) + pipe.evalsha(sha, 0) + assert pipe.execute() == [42, 42] + def test_redis_cluster_pipeline(self, r): """ Test that we can use a pipeline with the RedisCluster class