Truncate pipeline exception message to a sane size (#3530)

Fixes #20234.

Author: rohansingh

redis/asyncio/client.py

@@ -77,6 +77,7 @@
     get_lib_version,
     safe_str,
     str_if_bytes,
+    truncate_text,
 )
 
 if TYPE_CHECKING and SSL_AVAILABLE:
@@ -1514,7 +1515,10 @@ 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} ({truncate_text(cmd)}) "
+            "of pipeline caused error: {exception.args}"
+        )
         exception.args = (msg,) + exception.args[1:]
 
     async def parse_response(

redis/asyncio/cluster.py

@@ -71,6 +71,7 @@
     get_lib_version,
     safe_str,
     str_if_bytes,
+    truncate_text,
 )
 
 if SSL_AVAILABLE:
@@ -1634,8 +1635,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"({truncate_text(command)}) "
+                            f"of pipeline caused error: {result.args}"
                         )
                         result.args = (msg,) + result.args[1:]
                         raise result

redis/client.py

@@ -61,6 +61,7 @@
     get_lib_version,
     safe_str,
     str_if_bytes,
+    truncate_text,
 )
 
 if TYPE_CHECKING:
@@ -1524,7 +1525,8 @@ 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} ({truncate_text(cmd)}) of pipeline "
+            f"caused error: {exception.args[0]}"
         )
         exception.args = (msg,) + exception.args[1:]
 

redis/cluster.py

@@ -47,6 +47,7 @@
     merge_result,
     safe_str,
     str_if_bytes,
+    truncate_text,
 )
 
 
@@ -2125,7 +2126,8 @@ 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} ({truncate_text(cmd)}) of pipeline "
+            f"caused error: {exception.args[0]}"
         )
         exception.args = (msg,) + exception.args[1:]
 

redis/utils.py

@@ -1,5 +1,6 @@
 import datetime
 import logging
+import textwrap
 from contextlib import contextmanager
 from functools import wraps
 from typing import Any, Dict, List, Mapping, Optional, Union
@@ -298,3 +299,9 @@ def extract_expire_flags(
         exp_options.extend(["PXAT", pxat])
 
     return exp_options
+
+
+def truncate_text(txt, max_length=100):
+    return textwrap.shorten(
+        text=txt, width=max_length, placeholder="...", break_long_words=True
+    )

tests/test_asyncio/test_cluster.py

@@ -2905,6 +2905,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" * 50
+        a_value = "a" * 20
+        b_value = "b" * 20
+
+        async with r.pipeline() as pipe:
+            pipe.set(key, 1)
+            pipe.hset(key, mapping={"field_a": a_value, "field_b": b_value})
+            pipe.expire(key, 100)
+
+            with pytest.raises(Exception) as ex:
+                await pipe.execute()
+
+            expected = f"Command # 2 (HSET {key} field_a {a_value} field_b...) 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,22 @@ 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" * 50
+        a_value = "a" * 20
+        b_value = "b" * 20
+
+        await r.set(key, 1)
+        async with r.pipeline(transaction=False) as pipe:
+            pipe.hset(key, mapping={"field_a": a_value, "field_b": b_value})
+            pipe.expire(key, 100)
+
+            with pytest.raises(redis.ResponseError) as ex:
+                await pipe.execute()
+
+            expected = f"Command # 1 (HSET {key} field_a {a_value} field_b...) 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

@@ -3315,6 +3315,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" * 50
+        a_value = "a" * 20
+        b_value = "b" * 20
+
+        with r.pipeline() as pipe:
+            pipe.set(key, 1)
+            pipe.hset(key, mapping={"field_a": a_value, "field_b": b_value})
+            pipe.expire(key, 100)
+
+            with pytest.raises(Exception) as ex:
+                pipe.execute()
+
+            expected = f"Command # 2 (HSET {key} field_a {a_value} field_b...) 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,22 @@ 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" * 50
+        a_value = "a" * 20
+        b_value = "b" * 20
+
+        r[key] = 1
+        with r.pipeline(transaction=False) as pipe:
+            pipe.hset(key, mapping={"field_a": a_value, "field_b": b_value})
+            pipe.expire(key, 100)
+
+            with pytest.raises(redis.ResponseError) as ex:
+                pipe.execute()
+
+            expected = f"Command # 1 (HSET {key} field_a {a_value} field_b...) 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")