Truncate pipeline exception message to a sane size

Fixes #20234.

Author: rohansingh

redis/asyncio/client.py

@@ -1501,9 +1501,17 @@ def annotate_exception(
         self, exception: Exception, number: int, command: Iterable[object]
     ) -> None:
         cmd = " ".join(map(safe_str, command))
-        msg = f"Command # {number} ({cmd}) of pipeline caused error: {exception.args}"
+        msg = (
+            f"Command # {number} ({self._truncate_command(cmd)}) "
+            "of pipeline caused error: {exception.args}"
+        )
         exception.args = (msg,) + exception.args[1:]
 
+    def _truncate_command(self, command, max_length=100):
+        if len(command) > max_length:
+            return command[: max_length - 3] + "..."
+        return command
+
     async def parse_response(
         self, connection: Connection, command_name: Union[str, bytes], **options
     ):

redis/asyncio/cluster.py

@@ -1154,9 +1154,7 @@ def get_node(
             return self.nodes_cache.get(node_name)
         else:
             raise DataError(
-                "get_node requires one of the following: "
-                "1. node name "
-                "2. host and port"
+                "get_node requires one of the following: 1. node name 2. host and port"
             )
 
     def set_nodes(
@@ -1338,7 +1336,7 @@ async def initialize(self) -> None:
                             if len(disagreements) > 5:
                                 raise RedisClusterException(
                                     f"startup_nodes could not agree on a valid "
-                                    f'slots cache: {", ".join(disagreements)}'
+                                    f"slots cache: {', '.join(disagreements)}"
                                 )
 
             # Validate if all slots are covered or if we should try next startup node
@@ -1593,8 +1591,9 @@ async def _execute(
                     if isinstance(result, Exception):
                         command = " ".join(map(safe_str, cmd.args))
                         msg = (
-                            f"Command # {cmd.position + 1} ({command}) of pipeline "
-                            f"caused error: {result.args}"
+                            f"Command # {cmd.position + 1} "
+                            f"({self._truncate_command(command)}) "
+                            f"of pipeline caused error: {result.args}"
                         )
                         result.args = (msg,) + result.args[1:]
                         raise result
@@ -1637,6 +1636,11 @@ def mset_nonatomic(
 
         return self
 
+    def _truncate_command(self, command, max_length=100):
+        if len(command) > max_length:
+            return command[: max_length - 3] + "..."
+        return command
+
 
 for command in PIPELINE_BLOCKED_COMMANDS:
     command = command.replace(" ", "_").lower()

redis/client.py

@@ -1514,11 +1514,16 @@ def raise_first_error(self, commands, response):
     def annotate_exception(self, exception, number, command):
         cmd = " ".join(map(safe_str, command))
         msg = (
-            f"Command # {number} ({cmd}) of pipeline "
+            f"Command # {number} ({self._truncate_command(cmd)}) of pipeline "
             f"caused error: {exception.args[0]}"
         )
         exception.args = (msg,) + exception.args[1:]
 
+    def _truncate_command(self, command, max_length=100):
+        if len(command) > max_length:
+            return command[: max_length - 3] + "..."
+        return command
+
     def parse_response(self, connection, command_name, **options):
         result = Redis.parse_response(self, connection, command_name, **options)
         if command_name in self.UNWATCH_COMMANDS:

redis/cluster.py

@@ -1628,7 +1628,7 @@ def initialize(self):
                             if len(disagreements) > 5:
                                 raise RedisClusterException(
                                     f"startup_nodes could not agree on a valid "
-                                    f'slots cache: {", ".join(disagreements)}'
+                                    f"slots cache: {', '.join(disagreements)}"
                                 )
 
             fully_covered = self.check_slots_coverage(tmp_slots)
@@ -2047,11 +2047,16 @@ def annotate_exception(self, exception, number, command):
         """
         cmd = " ".join(map(safe_str, command))
         msg = (
-            f"Command # {number} ({cmd}) of pipeline "
+            f"Command # {number} ({self._truncate_command(cmd)}) of pipeline "
             f"caused error: {exception.args[0]}"
         )
         exception.args = (msg,) + exception.args[1:]
 
+    def _truncate_command(self, command, max_length=100):
+        if len(command) > max_length:
+            return command[: max_length - 3] + "x.."
+        return command
+
     def execute(self, raise_on_error: bool = True) -> List[Any]:
         """
         Execute all the commands in the current pipeline

tests/test_asyncio/test_cluster.py

@@ -146,7 +146,6 @@ async def get_mocked_redis_client(
     with mock.patch.object(ClusterNode, "execute_command") as execute_command_mock:
 
         async def execute_command(*_args, **_kwargs):
-
             if _args[0] == "CLUSTER SLOTS":
                 if cluster_slots_raise_error:
                     raise ResponseError()
@@ -1577,23 +1576,23 @@ async def test_cluster_bitop_not_empty_string(self, r: RedisCluster) -> None:
 
     @skip_if_server_version_lt("2.6.0")
     async def test_cluster_bitop_not(self, r: RedisCluster) -> None:
-        test_str = b"\xAA\x00\xFF\x55"
+        test_str = b"\xaa\x00\xff\x55"
         correct = ~0xAA00FF55 & 0xFFFFFFFF
         await r.set("{foo}a", test_str)
         await r.bitop("not", "{foo}r", "{foo}a")
         assert int(binascii.hexlify(await r.get("{foo}r")), 16) == correct
 
     @skip_if_server_version_lt("2.6.0")
     async def test_cluster_bitop_not_in_place(self, r: RedisCluster) -> None:
-        test_str = b"\xAA\x00\xFF\x55"
+        test_str = b"\xaa\x00\xff\x55"
         correct = ~0xAA00FF55 & 0xFFFFFFFF
         await r.set("{foo}a", test_str)
         await r.bitop("not", "{foo}a", "{foo}a")
         assert int(binascii.hexlify(await r.get("{foo}a")), 16) == correct
 
     @skip_if_server_version_lt("2.6.0")
     async def test_cluster_bitop_single_string(self, r: RedisCluster) -> None:
-        test_str = b"\x01\x02\xFF"
+        test_str = b"\x01\x02\xff"
         await r.set("{foo}a", test_str)
         await r.bitop("and", "{foo}res1", "{foo}a")
         await r.bitop("or", "{foo}res2", "{foo}a")
@@ -1604,8 +1603,8 @@ async def test_cluster_bitop_single_string(self, r: RedisCluster) -> None:
 
     @skip_if_server_version_lt("2.6.0")
     async def test_cluster_bitop_string_operands(self, r: RedisCluster) -> None:
-        await r.set("{foo}a", b"\x01\x02\xFF\xFF")
-        await r.set("{foo}b", b"\x01\x02\xFF")
+        await r.set("{foo}a", b"\x01\x02\xff\xff")
+        await r.set("{foo}b", b"\x01\x02\xff")
         await r.bitop("and", "{foo}res1", "{foo}a", "{foo}b")
         await r.bitop("or", "{foo}res2", "{foo}a", "{foo}b")
         await r.bitop("xor", "{foo}res3", "{foo}a", "{foo}b")
@@ -2803,6 +2802,25 @@ async def test_asking_error(self, r: RedisCluster) -> None:
             assert ask_node._free.pop().read_response.await_count
             assert res == ["MOCK_OK"]
 
+    async def test_error_is_truncated(self, r) -> None:
+        """
+        Test that an error from the pipeline is truncated correctly.
+        """
+        key = "a" * 5000
+
+        async with r.pipeline() as pipe:
+            pipe.set(key, 1)
+            pipe.llen(key)
+            pipe.expire(key, 100)
+
+            with pytest.raises(Exception) as ex:
+                await pipe.execute()
+
+            expected = (
+                "Command # 2 (LLEN " + ("a" * 92) + "...) of pipeline caused error: "
+            )
+            assert str(ex.value).startswith(expected)
+
     async def test_moved_redirection_on_slave_with_default(
         self, r: RedisCluster
     ) -> None:

tests/test_asyncio/test_pipeline.py

@@ -368,6 +368,21 @@ async def test_exec_error_in_no_transaction_pipeline_unicode_command(self, r):
 
         assert await r.get(key) == b"1"
 
+    async def test_exec_error_in_pipeline_truncated(self, r):
+        key = "a" * 5000
+        await r.set(key, 1)
+        async with r.pipeline(transaction=False) as pipe:
+            pipe.llen(key)
+            pipe.expire(key, 100)
+
+            with pytest.raises(redis.ResponseError) as ex:
+                await pipe.execute()
+
+            expected = (
+                "Command # 1 (LLEN " + ("a" * 92) + "...) of pipeline caused error: "
+            )
+            assert str(ex.value).startswith(expected)
+
     async def test_pipeline_with_bitfield(self, r):
         async with r.pipeline() as pipe:
             pipe.set("a", "1")

tests/test_cluster.py

@@ -1692,23 +1692,23 @@ def test_cluster_bitop_not_empty_string(self, r):
 
     @skip_if_server_version_lt("2.6.0")
     def test_cluster_bitop_not(self, r):
-        test_str = b"\xAA\x00\xFF\x55"
+        test_str = b"\xaa\x00\xff\x55"
         correct = ~0xAA00FF55 & 0xFFFFFFFF
         r["{foo}a"] = test_str
         r.bitop("not", "{foo}r", "{foo}a")
         assert int(binascii.hexlify(r["{foo}r"]), 16) == correct
 
     @skip_if_server_version_lt("2.6.0")
     def test_cluster_bitop_not_in_place(self, r):
-        test_str = b"\xAA\x00\xFF\x55"
+        test_str = b"\xaa\x00\xff\x55"
         correct = ~0xAA00FF55 & 0xFFFFFFFF
         r["{foo}a"] = test_str
         r.bitop("not", "{foo}a", "{foo}a")
         assert int(binascii.hexlify(r["{foo}a"]), 16) == correct
 
     @skip_if_server_version_lt("2.6.0")
     def test_cluster_bitop_single_string(self, r):
-        test_str = b"\x01\x02\xFF"
+        test_str = b"\x01\x02\xff"
         r["{foo}a"] = test_str
         r.bitop("and", "{foo}res1", "{foo}a")
         r.bitop("or", "{foo}res2", "{foo}a")
@@ -1719,8 +1719,8 @@ def test_cluster_bitop_single_string(self, r):
 
     @skip_if_server_version_lt("2.6.0")
     def test_cluster_bitop_string_operands(self, r):
-        r["{foo}a"] = b"\x01\x02\xFF\xFF"
-        r["{foo}b"] = b"\x01\x02\xFF"
+        r["{foo}a"] = b"\x01\x02\xff\xff"
+        r["{foo}b"] = b"\x01\x02\xff"
         r.bitop("and", "{foo}res1", "{foo}a", "{foo}b")
         r.bitop("or", "{foo}res2", "{foo}a", "{foo}b")
         r.bitop("xor", "{foo}res3", "{foo}a", "{foo}b")
@@ -3260,6 +3260,25 @@ def raise_ask_error():
             assert ask_node.redis_connection.connection.read_response.called
             assert res == ["MOCK_OK"]
 
+    def test_error_is_truncated(self, r):
+        """
+        Test that an error from the pipeline is truncated correctly.
+        """
+        key = "a" * 5000
+
+        with r.pipeline() as pipe:
+            pipe.set(key, 1)
+            pipe.llen(key)
+            pipe.expire(key, 100)
+
+            with pytest.raises(Exception) as ex:
+                pipe.execute()
+
+            expected = (
+                "Command # 2 (LLEN " + ("a" * 92) + "...) of pipeline caused error: "
+            )
+            assert str(ex.value).startswith(expected)
+
     def test_return_previously_acquired_connections(self, r):
         # in order to ensure that a pipeline will make use of connections
         #   from different nodes

tests/test_pipeline.py

@@ -369,6 +369,21 @@ def test_exec_error_in_no_transaction_pipeline_unicode_command(self, r):
 
         assert r[key] == b"1"
 
+    def test_exec_error_in_pipeline_truncated(self, r):
+        key = "a" * 5000
+        r[key] = 1
+        with r.pipeline(transaction=False) as pipe:
+            pipe.llen(key)
+            pipe.expire(key, 100)
+
+            with pytest.raises(redis.ResponseError) as ex:
+                pipe.execute()
+
+            expected = (
+                "Command # 1 (LLEN " + ("a" * 92) + "...) of pipeline caused error: "
+            )
+            assert str(ex.value).startswith(expected)
+
     def test_pipeline_with_bitfield(self, r):
         with r.pipeline() as pipe:
             pipe.set("a", "1")