feat(redis): support cluster publishing pipelines

docs/docs/en/redis/cluster.md

@@ -20,7 +20,7 @@ search:
 |---|---|
 | Single Redis instance | Multi-node cluster |
 | Development / testing | Production with HA |
-| Need `pipeline` support | Can tolerate no pipeline |
+| Single-node transactions | Same-slot transactions |
 
 ## Connecting
 
@@ -51,7 +51,7 @@ broker = RedisClusterBroker(
 | List | ✅ | ✅ |
 | Stream + XAUTOCLAIM | ✅ | ✅ |
 | Pub/Sub | ✅ | ✅ (via sync cluster) |
-| Pipeline | ✅ | ❌ |
+| Pipeline | ✅ | ✅ (lists and streams) |
 
 ## Stream Location
 
@@ -76,7 +76,8 @@ broker = RedisClusterBroker(url="redis://localhost:7000")
 
 ## Limitations
 
-- **Pipeline** is not supported in Redis Cluster.
+- **Pipelines** support lists and streams, but redis-py blocks channel `PUBLISH` commands in cluster pipelines. See [Redis Pipeline](pipeline.md){.internal-link} for a tested example.
+- **Transactions** require `redis-py >= 6.2.0`, `transaction=True`, and keys in the same hash slot. A shared hash tag such as `{orders}` keeps related keys together. Pipelining without a transaction is not atomic.
 - **XAUTOCLAIM** with `min_idle_time` requires a consumer group with `group` and `consumer` parameters on `StreamSub`.
 - **Pub/Sub** uses a synchronous `RedisCluster` client (via `ThreadPoolExecutor`) because the async client does not expose `publish`/`pubsub` until `redis-py >= 8.0.0`.
 

docs/docs/en/redis/pipeline.md

@@ -10,7 +10,7 @@ search:
 
 # Redis Pipeline
 
-**FastStream** supports [**Redis** pipelining](https://redis.io/docs/latest/develop/use/pipelining/){.external-link target="_blank"} to optimize performance when publishing multiple messages in a batch. This allows you to queue several **Redis** operations and execute them in one network round-trip, reducing latency significantly.
+**FastStream** supports [**Redis** pipelining](https://redis.io/docs/latest/develop/use/pipelining/){.external-link target="_blank"} to optimize performance when publishing multiple messages in a batch. This allows you to queue several **Redis** operations and execute them together, reducing network round-trips.
 
 ## Usage Example
 
@@ -36,11 +36,26 @@ When using `#!python broker.publish_batch()` in combination with the `pipeline`
 
 ## Notes
 
-- Pipelining is supported for all **Redis** queue types, including channels, lists, and streams.
-- You can combine multiple queue types in a single pipeline.
+- With `RedisBroker`, pipelining is supported for all **Redis** queue types, including channels, lists, and streams.
+- You can combine supported queue types in a single pipeline.
 
-!!! warning "Redis Cluster"
-    Pipeline is **not supported** in Redis Cluster. If you are using `RedisClusterBroker`, the `pipeline` parameter is not available. Consider using `publish_batch()` with individual requests instead.
+## Redis Cluster
+
+`RedisClusterBroker` accepts a `redis.asyncio.cluster.ClusterPipeline` through the same `pipeline` parameter. Create it from the client returned by `#!python await broker.connect()`. You can queue list and stream publications with `broker.publish()` or a publisher's `publish()`, and list batches with `broker.publish_batch()` or a batch publisher.
+
+Pipelining alone does not make commands atomic. To combine a state update and a publication atomically, use `transaction=True` and keep every key in the same hash slot. The example uses the shared `{orders}` hash tag for the counter and stream:
+
+```python linenums="1"
+{!> docs_src/redis/pipeline/cluster_pipeline.py !}
+```
+
+The publication stays queued alongside `INCR` until `#!python await pipe.execute()`. Its result list contains the command results in order: the updated counter followed by the stream entry ID. `transaction=True` requires `redis-py >= 6.2.0`; ordinary cluster pipelines do not provide transaction semantics.
+
+!!! warning "Cluster pipeline restrictions"
+    redis-py blocks `PUBLISH` in cluster pipelines. Passing `pipeline=pipe` with a channel raises `RedisClusterException`; FastStream does not publish outside the pipeline as a fallback. Publish to channels without a pipeline instead.
+
+!!! note "WATCH and MULTI"
+    With a watched transaction, commands issued after `WATCH` but before `MULTI` execute immediately, following redis-py's behavior. Call `pipe.multi()` before queueing publications that must belong to the transaction.
 
 ## Benefits
 

docs/docs_src/redis/pipeline/cluster_pipeline.py

@@ -0,0 +1,29 @@
+import asyncio
+
+from faststream.redis import RedisClusterBroker
+
+broker = RedisClusterBroker("redis://127.0.0.1:7001")
+
+
+async def increment_and_publish(
+    broker: RedisClusterBroker,
+    key: str = "orders",
+) -> list[int | bytes]:
+    client = await broker.connect()
+    async with client.pipeline(transaction=True) as pipe:
+        pipe.incr(f"{{{key}}}:count")
+        await broker.publish(
+            "created",
+            stream=f"{{{key}}}:events",
+            pipeline=pipe,
+        )
+        return await pipe.execute()
+
+
+async def main() -> None:
+    async with broker:
+        await increment_and_publish(broker)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

faststream/redis/broker/broker.py

@@ -43,6 +43,7 @@
     from types import TracebackType
 
     from redis.asyncio.client import Pipeline, Redis
+    from redis.asyncio.cluster import ClusterPipeline
 
     from faststream._internal.basic_types import SendableMessage
     from faststream.redis.message import RedisChannelMessage
@@ -204,7 +205,7 @@ async def publish(
         list: str | None = None,
         stream: None = None,
         maxlen: int | None = None,
-        pipeline: Optional["Pipeline[bytes]"] = None,
+        pipeline: None = None,
     ) -> int: ...
 
     @overload
@@ -219,9 +220,24 @@ async def publish(
         list: str | None = None,
         stream: str = ...,
         maxlen: int | None = None,
-        pipeline: Optional["Pipeline[bytes]"] = None,
+        pipeline: None = None,
     ) -> bytes: ...
 
+    @overload
+    async def publish(
+        self,
+        message: "SendableMessage" = None,
+        channel: str | None = None,
+        *,
+        reply_to: str = "",
+        headers: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
+        list: str | None = None,
+        stream: str | None = None,
+        maxlen: int | None = None,
+        pipeline: "Pipeline[bytes]",
+    ) -> "Pipeline[bytes]": ...
+
     @override
     async def publish(
         self,
@@ -234,8 +250,8 @@ async def publish(
         list: str | None = None,
         stream: str | None = None,
         maxlen: int | None = None,
-        pipeline: Optional["Pipeline[bytes]"] = None,
-    ) -> int | bytes:
+        pipeline: Optional["Pipeline[bytes] | ClusterPipeline[bytes]"] = None,
+    ) -> "int | bytes | Pipeline[bytes] | ClusterPipeline[bytes]":
         """Publish message directly.
 
         This method allows you to publish a message in a non-AsyncAPI-documented way.
@@ -259,10 +275,11 @@ async def publish(
             maxlen:
                 Redis Stream maxlen publish option. Remove eldest message if maxlen exceeded.
             pipeline:
-                Redis pipeline to use for publishing messages.
+                Redis pipeline to use for publishing messages. Queued commands run on
+                pipeline.execute().
 
         Returns:
-            int: The result of the publish operation, typically the number of messages published.
+            The publish result, or the pipeline when the command is queued.
         """
         cmd = RedisPublishCommand(
             message,
@@ -278,7 +295,9 @@ async def publish(
             message_format=self.message_format,
         )
 
-        result: int | bytes = await super()._basic_publish(
+        result: (
+            int | bytes | Pipeline[bytes] | ClusterPipeline[bytes]
+        ) = await super()._basic_publish(
             cmd,
             producer=self.config.producer,
         )
@@ -315,16 +334,38 @@ async def request(  # type: ignore[override]
         )
         return msg
 
+    @overload  # type: ignore[override]
+    async def publish_batch(
+        self,
+        *messages: "SendableMessage",
+        list: str,
+        correlation_id: str | None = None,
+        reply_to: str = "",
+        headers: dict[str, Any] | None = None,
+        pipeline: None = None,
+    ) -> int: ...
+
+    @overload
+    async def publish_batch(
+        self,
+        *messages: "SendableMessage",
+        list: str,
+        correlation_id: str | None = None,
+        reply_to: str = "",
+        headers: dict[str, Any] | None = None,
+        pipeline: "Pipeline[bytes]",
+    ) -> "Pipeline[bytes]": ...
+
     @override
-    async def publish_batch(  # type: ignore[override]
+    async def publish_batch(
         self,
         *messages: "SendableMessage",
         list: str,
         correlation_id: str | None = None,
         reply_to: str = "",
         headers: dict[str, Any] | None = None,
-        pipeline: Optional["Pipeline[bytes]"] = None,
-    ) -> int:
+        pipeline: Optional["Pipeline[bytes] | ClusterPipeline[bytes]"] = None,
+    ) -> "int | Pipeline[bytes] | ClusterPipeline[bytes]":
         """Publish multiple messages to Redis List by one request.
 
         Args:
@@ -336,7 +377,7 @@ async def publish_batch(  # type: ignore[override]
             pipeline: Redis pipeline to use for publishing messages.
 
         Returns:
-            int: The result of the batch publish operation.
+            The batch publish result, or the pipeline when it is queued.
         """
         cmd = RedisPublishCommand(
             *messages,
@@ -349,7 +390,9 @@ async def publish_batch(  # type: ignore[override]
             message_format=self.message_format,
         )
 
-        result: int = await self._basic_publish_batch(
+        result: (
+            int | Pipeline[bytes] | ClusterPipeline[bytes]
+        ) = await self._basic_publish_batch(
             cmd,
             producer=self.config.producer,
         )