fix(cluster): populate ClusterPipeline.command_stack via execution strategy (#3703)

redis/asyncio/cluster.py

@@ -2400,6 +2400,18 @@ def __bool__(self) -> bool:
     def __len__(self) -> int:
         return len(self._execution_strategy)
 
+    @property
+    def command_stack(self) -> List["PipelineCommand"]:
+        """
+        Deprecated. Returns the commands currently queued on the pipeline.
+
+        Retained for backwards compatibility with code that introspects the
+        queued commands directly (for example APM/tracing integrations such as
+        Datadog's ``ddtrace``). The commands are owned by the underlying
+        execution strategy; prefer ``len(pipeline)`` to check the queue size.
+        """
+        return self._execution_strategy.command_queue
+
     def execute_command(
         self, *args: Union[KeyT, EncodableT], **kwargs: Any
     ) -> "ClusterPipeline":
@@ -2504,6 +2516,12 @@ def __repr__(self) -> str:
 
 
 class ExecutionStrategy(ABC):
+    @property
+    @abstractmethod
+    def command_queue(self) -> List["PipelineCommand"]:
+        """The commands currently queued on the pipeline."""
+        pass
+
     @abstractmethod
     async def initialize(self) -> "ClusterPipeline":
         """
@@ -2608,6 +2626,10 @@ def __init__(self, pipe: ClusterPipeline) -> None:
         self._pipe: ClusterPipeline = pipe
         self._command_queue: List["PipelineCommand"] = []
 
+    @property
+    def command_queue(self) -> List["PipelineCommand"]:
+        return self._command_queue
+
     async def initialize(self) -> "ClusterPipeline":
         if self._pipe.cluster_client._initialize:
             await self._pipe.cluster_client.initialize()

redis/cluster.py

@@ -3419,7 +3419,6 @@ def __init__(
         **kwargs,
     ):
         """ """
-        self.command_stack = []
         self.nodes_manager = nodes_manager
         self.commands_parser = commands_parser
         self.refresh_table_asap = False
@@ -3518,6 +3517,18 @@ def __bool__(self):
         "Pipeline instances should  always evaluate to True on Python 3+"
         return True
 
+    @property
+    def command_stack(self):
+        """
+        Deprecated. Returns the commands currently queued on the pipeline.
+
+        Retained for backwards compatibility with code that introspects the
+        queued commands directly (for example APM/tracing integrations such as
+        Datadog's ``ddtrace``). The commands are owned by the underlying
+        execution strategy; prefer ``len(pipeline)`` to check the queue size.
+        """
+        return self._execution_strategy.command_queue
+
     def execute_command(self, *args, **kwargs):
         """
         Wrapper function for pipeline_execute_command

tests/test_asyncio/test_cluster.py

@@ -3809,6 +3809,29 @@ def custom_callback(response):
         finally:
             await r.aclose()
 
+    async def test_command_stack_reflects_queued_commands(self) -> None:
+        """
+        ClusterPipeline.command_stack should expose the commands queued on the
+        pipeline. Regression test for #3703: after the execution-strategy
+        refactor (#3611) the async pipeline had no command_stack at all, which
+        silently broke APM/tracing integrations (e.g. Datadog ddtrace) that
+        introspect it.
+        """
+        r = await get_mocked_redis_client(host=default_host, port=default_port)
+        try:
+            pipeline = r.pipeline()
+            assert pipeline.command_stack == []
+
+            pipeline.set("{foo}1", "1").set("{foo}2", "2")
+
+            assert len(pipeline.command_stack) == 2
+            # command_stack delegates to the execution strategy's command queue
+            assert pipeline.command_stack is pipeline._execution_strategy.command_queue
+            assert pipeline.command_stack[0].args == ("SET", "{foo}1", "1")
+            assert pipeline.command_stack[1].args == ("SET", "{foo}2", "2")
+        finally:
+            await r.aclose()
+
     async def test_blocked_arguments(self, r: RedisCluster) -> None:
         """Test handling for blocked pipeline arguments."""
 

tests/test_cluster.py

@@ -3845,6 +3845,34 @@ def test_redis_cluster_pipeline(self, r):
             pipe.get("foo")
             assert pipe.execute() == [True, b"bar"]
 
+    def test_command_stack_reflects_queued_commands(self):
+        """
+        ClusterPipeline.command_stack should expose the commands queued on the
+        pipeline. Regression test for #3703: after the execution-strategy
+        refactor (#3611) command_stack was always empty, which silently broke
+        APM/tracing integrations (e.g. Datadog ddtrace) that introspect it.
+        """
+        r = get_mocked_redis_client(host=default_host, port=default_port)
+        with r.pipeline() as pipe:
+            assert pipe.command_stack == []
+
+            pipe.set("{foo}1", "1").set("{foo}2", "2")
+
+            assert len(pipe.command_stack) == 2
+            # command_stack delegates to the execution strategy's command queue
+            assert pipe.command_stack is pipe._execution_strategy.command_queue
+            assert pipe.command_stack[0].args == ("SET", "{foo}1", "1")
+            assert pipe.command_stack[1].args == ("SET", "{foo}2", "2")
+
+    @pytest.mark.parametrize("transaction", [False, True])
+    def test_command_stack_delegates_to_execution_strategy(self, transaction):
+        """command_stack should delegate to the active execution strategy's
+        command queue for both the pipeline and transaction strategies."""
+        r = get_mocked_redis_client(host=default_host, port=default_port)
+        pipe = r.pipeline(transaction=transaction)
+        assert pipe.command_stack is pipe._execution_strategy.command_queue
+        assert pipe.command_stack == []
+
     def test_mget_disabled(self, r):
         """
         Test that mget is disabled for ClusterPipeline