feat(redis): support cluster publishing pipelines by lxingy3 · Pull Request #3096 · ag2ai/faststream

lxingy3 · 2026-09-04T17:48:34Z

Description

RedisClusterBroker currently ignores pipeline=, so a state update and a publication cannot share a transaction. Simply forwarding it is not enough: awaiting the ClusterPipeline returned by a queued command reinitializes the pipeline and clears its queued commands.

This change forwards the supplied pipeline through broker and publisher calls for lists, streams, and list batches. Queued calls return the pipeline without awaiting it; commands issued after WATCH but before MULTI still execute immediately. Publishing without a pipeline keeps its existing behavior, including automatic connection for cluster batch publishing.

Channel PUBLISH remains blocked by redis-py in cluster pipelines. There is no immediate-publish fallback. Atomic transactions require redis-py 6.2.0 or later and keys in the same hash slot; ordinary pipelines are not atomic.

Type of change

New feature
Documentation update

Validation

110 related tests passed, covering standalone Redis publishing, cluster publishing, producers, pipelines, and the documentation example. Redis 7.0.15 ran as a standalone server and a three-node cluster. Five existing slow tests were excluded.
The standalone test broker used 127.0.0.1 locally. Windows/WSL's localhost IPv6 fallback exceeded existing timeouts; the same RPC timeout reproduced on unchanged main. No production code or test timeout was changed for this environment issue.
redis-py 6.2.0: all nine focused cluster tests passed, including same-slot transactions, WATCH/MULTI, cross-slot rejection, and the runnable example.
redis-py 5.0.0: all four non-connected queue-preservation and command-restriction cases passed. Transactions are not claimed for this version.
All four deterministic regression cases fail on unchanged main.
Ruff, codespell, Mypy, Pyright, Bandit, and Semgrep passed. The remaining pre-commit hooks passed; local just wrappers were checked through their underlying commands.

Checklist

Reviewed the implementation and preserved existing no-pipeline behavior
Added regression tests and public API type checks
Updated the cluster and pipeline documentation with a tested example
Ran the affected tests and static analysis

lxingy3 · 2026-09-04T17:57:40Z

The Redis and Redis Cluster integration jobs passed, including the new transaction tests. The Kafka job finished with 359 passed and one failure in tests/brokers/kafka/future/test_fastapi.py::TestRouter::test_base_real.

That case did not receive its message within the existing three-second wait; aiokafka logged that the generated topic was not found in cluster metadata. The same inherited test passed later in the job as TestKafkaRouter::test_base_real. This PR does not change Kafka, shared execution code, dependencies, or the workflow.

Could a maintainer rerun the failed jobs in this run? I only have read access to the upstream repository, so I cannot trigger the rerun.

IvanKirpichnikov

In FastStream, we support three types of Redis brokers:

Default Redis
Redis Cluster
Redis Sentinel

For regular Redis and Redis Sentinel, the pipeline type is redis.asyncio.client.Pipeline, while for Redis Cluster it is redis.asyncio.cluster.ClusterPipeline.

We need to add the correct type hints for the pipeline argument in the following methods:

publish
request
publish_batch
e.g...

We also need to add overloads for these methods so that the return type depends on whether a pipeline is passed.

The expected behavior is:

Without pipeline → keep the existing return type.
With pipeline → return the corresponding broker pipeline type, as in redis-py.

In other words, the overloads should reflect the following behavior:

For regular Redis and Sentinel: redis.asyncio.client.Pipeline
For Redis Cluster: redis.asyncio.cluster.ClusterPipeline

Return the matching Redis pipeline from publishing overloads and keep RedisClusterBroker restricted to ClusterPipeline. Request remains pipeline-free because request-reply must publish immediately before awaiting a response.

lxingy3 · 2026-09-04T22:49:23Z

Addressed in 679e5152:

RedisBroker and RedisSentinelBroker now expose Pipeline overloads, while RedisClusterBroker exposes ClusterPipeline overloads.
Calls without a pipeline keep their existing int/bytes return types; calls with a pipeline return that exact pipeline type. Publisher overloads follow the same rule.
Runtime identity checks now cover regular and cluster pipelines, and the public types are covered by mypy and Pyright assertions.

I intentionally kept request pipeline-free. A request subscribes, publishes, and waits for the reply in the same call; queueing that publish would prevent the caller from executing the pipeline until after request returns, so it would time out.

@IvanKirpichnikov, could you take another look?

IvanKirpichnikov · 2026-09-04T22:57:49Z

+        pipeline: Optional["Pipeline[bytes] | ClusterPipeline[bytes]"] = None,
+    ) -> "int | bytes | Pipeline[bytes] | ClusterPipeline[bytes]":


only Pipeline. It can only be used with ClusterRedis

IvanKirpichnikov · 2026-09-04T22:57:59Z


-        result: int | bytes = await super()._basic_publish(
+        result: (
+            int | bytes | Pipeline[bytes] | ClusterPipeline[bytes]


only Pipeline. It can only be used with ClusterRedis

IvanKirpichnikov · 2026-09-04T22:58:10Z

+        pipeline: Optional["Pipeline[bytes] | ClusterPipeline[bytes]"] = None,
+    ) -> "int | Pipeline[bytes] | ClusterPipeline[bytes]":


only Pipeline. It can only be used with ClusterRedis

IvanKirpichnikov · 2026-09-04T22:58:17Z


-        result: int = await self._basic_publish_batch(
+        result: (
+            int | Pipeline[bytes] | ClusterPipeline[bytes]


only Pipeline. It can only be used with ClusterRedis

IvanKirpichnikov · 2026-09-06T21:30:52Z

+
+        A cross-slot transaction must fail without publishing outside it.
+        """
+        from redis.exceptions import CrossSlotTransactionError


Make the import at the module level.

IvanKirpichnikov · 2026-09-06T21:40:57Z

+        broker = self.get_broker()
+        counter = f"{{first}}:{queue}:count"
+        stream = f"{{second}}:{queue}:stream"
+
+        async with broker:
+            client = await broker.connect()
+            try:
+                async with client.pipeline(transaction=True) as pipe:
+                    pipe.incr(counter)
+                    await broker.publish("one", stream=stream, pipeline=pipe)
+                    with pytest.raises(CrossSlotTransactionError):
+                        await pipe.execute()
+
+                assert await client.get(counter) is None
+                assert await client.xlen(stream) == 0
+            finally:
+                await client.delete(counter)
+                await client.delete(stream)


broker = self.get_broker() counter = f"{{first}}:{queue}:count" stream = f"{{second}}:{queue}:stream" async with self.patch_broker(broker): client = broker.config.broker_config.connection.client async with client.pipeline(transaction=True) as pipe: pipe.incr(counter) await broker.publish("one", stream=stream, pipeline=pipe) with pytest.raises(CrossSlotTransactionError): await pipe.execute() assert await client.get(counter) is None assert await client.xlen(stream) == 0

IvanKirpichnikov · 2026-09-06T21:47:41Z

+                maxlen=cmd.maxlen,
+            )
+        else:
+            raise UnreachablePathError


maybe use assert_never(cmd.destination_type)

IvanKirpichnikov · 2026-09-06T21:50:45Z

+        broker = self.get_broker()
+        counter = f"{{{queue}}}:count"
+        stream = f"{{{queue}}}:stream"
+        publisher = broker.publisher(stream=stream)
+
+        async with broker:
+            client = await broker.connect()
+            try:
+                async with client.pipeline(transaction=True) as pipe:
+                    await pipe.watch(counter)
+                    result = await broker.publish(
+                        "immediate", stream=stream, pipeline=pipe
+                    )
+                    assert isinstance(result, bytes)
+                    assert await client.xlen(stream) == 1
+
+                    pipe.multi()
+                    pipe.incr(counter)
+                    await publisher.publish("queued", pipeline=pipe)
+                    assert await client.get(counter) is None
+                    assert await client.xlen(stream) == 1
+
+                    results = await pipe.execute()
+
+                assert results[0] == 1
+                assert len(results) == 2
+                assert await client.get(counter) == b"1"
+                entries = await client.xrange(stream)
+                assert [
+                    broker.message_format.parse(fields[b"__data__"])[0]
+                    for _, fields in entries
+                ] == [b"immediate", b"queued"]
+            finally:
+                await client.delete(counter, stream)


broker = self.get_broker() counter = f"{{{queue}}}:count" stream = f"{{{queue}}}:stream" publisher = broker.publisher(stream=stream) async with self.patch_broker(broker): client = broker.config.broker_config.connection.client async with client.pipeline(transaction=True) as pipe: await pipe.watch(counter) result = await broker.publish( "immediate", stream=stream, pipeline=pipe ) assert isinstance(result, bytes) assert await client.xlen(stream) == 1 pipe.multi() pipe.incr(counter) await publisher.publish("queued", pipeline=pipe) assert await client.get(counter) is None assert await client.xlen(stream) == 1 results = await pipe.execute() assert results[0] == 1 assert len(results) == 2 assert await client.get(counter) == b"1" entries = await client.xrange(stream) assert [ broker.message_format.parse(fields[b"__data__"])[0] for _, fields in entries ] == [b"immediate", b"queued"]

IvanKirpichnikov · 2026-09-06T21:52:11Z

+        broker = self.get_broker()
+        counter = f"{{{queue}}}:count"
+        destination = f"{{{queue}}}:list"
+        publisher = broker.publisher(list=ListSub(destination, batch=True))
+
+        async with broker:
+            client = await broker.connect()
+            try:
+                async with client.pipeline(transaction=True) as pipe:
+                    pipe.incr(counter)
+                    await broker.publish_batch(
+                        "one", "two", list=destination, pipeline=pipe
+                    )
+                    await publisher.publish("three", "four", pipeline=pipe)
+
+                    assert await client.exists(counter, destination) == 0
+                    assert await pipe.execute() == [1, 2, 4]
+
+                assert await client.get(counter) == b"1"
+                messages = await client.lrange(destination, 0, -1)
+                assert [broker.message_format.parse(msg)[0] for msg in messages] == [
+                    b"one",
+                    b"two",
+                    b"three",
+                    b"four",
+                ]
+            finally:
+                await client.delete(counter, destination)


broker = self.get_broker() counter = f"{{{queue}}}:count" destination = f"{{{queue}}}:list" publisher = broker.publisher(list=ListSub(destination, batch=True)) async with self.patch_broker(broker): client = broker.config.broker_config.connection.client async with client.pipeline(transaction=True) as pipe: pipe.incr(counter) await broker.publish_batch( "one", "two", list=destination, pipeline=pipe ) await publisher.publish("three", "four", pipeline=pipe) assert await client.exists(counter, destination) == 0 assert await pipe.execute() == [1, 2, 4] assert await client.get(counter) == b"1" messages = await client.lrange(destination, 0, -1) assert [broker.message_format.parse(msg)[0] for msg in messages] == [ b"one", b"two", b"three", b"four", ]

IvanKirpichnikov · 2026-09-06T21:53:07Z

+        broker = self.get_broker()
+        counter = f"{{{queue}}}:count"
+        stream = f"{{{queue}}}:stream"
+        publisher = broker.publisher(stream=stream)
+
+        async with broker:
+            client = await broker.connect()
+            try:
+                async with client.pipeline(transaction=True) as pipe:
+                    pipe.incr(counter)
+                    await broker.publish(
+                        "one",
+                        stream=stream,
+                        correlation_id=queue,
+                        headers={"source": "broker"},
+                        pipeline=pipe,
+                    )
+                    await publisher.publish("two", pipeline=pipe)
+
+                    assert await client.exists(counter, stream) == 0
+                    results = await pipe.execute()
+
+                assert results[0] == 1
+                assert len(results) == 3
+                assert await client.get(counter) == b"1"
+                entries = await client.xrange(stream)
+                messages = [
+                    broker.message_format.parse(fields[b"__data__"])
+                    for _, fields in entries
+                ]
+                assert [body for body, _ in messages] == [b"one", b"two"]
+                assert messages[0][1]["correlation_id"] == queue
+                assert messages[0][1]["source"] == "broker"
+            finally:
+                await client.delete(counter, stream)


broker = self.get_broker() counter = f"{{{queue}}}:count" stream = f"{{{queue}}}:stream" publisher = broker.publisher(stream=stream) async with self.patch_broker(broker): client = broker.config.broker_config.connection.client async with client.pipeline(transaction=True) as pipe: pipe.incr(counter) await broker.publish( "one", stream=stream, correlation_id=queue, headers={"source": "broker"}, pipeline=pipe, ) await publisher.publish("two", pipeline=pipe) assert await client.exists(counter, stream) == 0 results = await pipe.execute() assert results[0] == 1 assert len(results) == 3 assert await client.get(counter) == b"1" entries = await client.xrange(stream) messages = [ broker.message_format.parse(fields[b"__data__"]) for _, fields in entries ] assert [body for body, _ in messages] == [b"one", b"two"] assert messages[0][1]["correlation_id"] == queue assert messages[0][1]["source"] == "broker"

lxingy3 requested review from IvanKirpichnikov and powersemmi as code owners September 4, 2026 17:48

github-actions Bot added documentation Improvements or additions to documentation Redis Issues related to `faststream.redis` module and Redis features labels Sep 4, 2026

IvanKirpichnikov requested changes Sep 4, 2026

View reviewed changes

IvanKirpichnikov assigned lxingy3 Sep 4, 2026

lxingy3 and others added 2 commits September 4, 2026 18:31

feat(redis): support cluster publishing pipelines

f2fded3

fix(redis): refine pipeline result types

679e515

Return the matching Redis pipeline from publishing overloads and keep RedisClusterBroker restricted to ClusterPipeline. Request remains pipeline-free because request-reply must publish immediately before awaiting a response.

lxingy3 force-pushed the feat/3044-redis-cluster-pipeline branch from c09a056 to 679e515 Compare September 4, 2026 22:41

IvanKirpichnikov requested changes Sep 6, 2026

View reviewed changes

Navigation Menu

Uh oh!

feat(redis): support cluster publishing pipelines - #3096

feat(redis): support cluster publishing pipelines#3096
lxingy3 wants to merge 2 commits into
ag2ai:mainfrom
lxingy3:feat/3044-redis-cluster-pipeline

lxingy3 commented Sep 4, 2026

Uh oh!

lxingy3 commented Sep 4, 2026

Uh oh!

IvanKirpichnikov left a comment

Uh oh!

lxingy3 commented Sep 4, 2026

Uh oh!

IvanKirpichnikov Sep 4, 2026

Uh oh!

IvanKirpichnikov Sep 4, 2026

Uh oh!

IvanKirpichnikov Sep 4, 2026

Uh oh!

IvanKirpichnikov Sep 4, 2026

Uh oh!

IvanKirpichnikov Sep 6, 2026

Uh oh!

IvanKirpichnikov Sep 6, 2026

Uh oh!

IvanKirpichnikov Sep 6, 2026

Uh oh!

IvanKirpichnikov Sep 6, 2026

Uh oh!

IvanKirpichnikov Sep 6, 2026

Uh oh!

IvanKirpichnikov Sep 6, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		pipeline: Optional["Pipeline[bytes] \| ClusterPipeline[bytes]"] = None,
		) -> "int \| bytes \| Pipeline[bytes] \| ClusterPipeline[bytes]":

		pipeline: Optional["Pipeline[bytes] \| ClusterPipeline[bytes]"] = None,
		) -> "int \| Pipeline[bytes] \| ClusterPipeline[bytes]":

Uh oh!

Conversation

lxingy3 commented Sep 4, 2026

Description

Type of change

Validation

Checklist

Uh oh!

lxingy3 commented Sep 4, 2026

Uh oh!

IvanKirpichnikov left a comment

Choose a reason for hiding this comment

Uh oh!

lxingy3 commented Sep 4, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants