Truncate pipeline exception message to a sane size

Fixes #20234.

Author: rohansingh

redis/asyncio/client.py

@@ -1508,9 +1508,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

@@ -1598,8 +1598,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
@@ -1648,6 +1649,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

@@ -1521,10 +1521,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 caused error: {exception.args[0]}"
+            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

@@ -2054,10 +2054,16 @@ def annotate_exception(self, exception, number, command):
         """
         cmd = " ".join(map(safe_str, command))
         msg = (
-            f"Command # {number} ({cmd}) of pipeline caused error: {exception.args[0]}"
+            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

@@ -2802,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

@@ -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")