Repository navigation
RedisStreamBroker: receive latency grows with the number of streams because listen() runs the reclaim scan between reads #699
Description
What happened?
We run a worker that consumes about 160 streams through RedisStreamBroker, against a hosted Redis with a 28 ms round trip. Every message took 4 to 7 seconds to reach a handler, and one stream fell more than three hours behind during a burst.
The cause is the shape of listen(). It reads, yields what it got, and then goes through every stream to look for abandoned messages (take a lock, XAUTOCLAIM, release it, one stream at a time) before it reads again. A message published while that scan is running has to wait for it to finish, so the wait is roughly the number of streams times the round trips per stream times the latency to Redis. With one stream you never notice. With many streams, or any Redis that isn't on localhost, every message pays for it.
The same loop also limits throughput. Each pass claims at most xread_count messages per stream, so a stream can drain at xread_count divided by the scan time. A burst bigger than that builds a backlog, and the backlog only shrinks once traffic drops below that rate.
I reproduced it on the latest release (1.2.4) with a standalone script, attached. It needs Docker for redis:7-alpine, and it puts a small TCP proxy in front of Redis that delays each direction, so --one-way-ms 14 gives a 28 ms round trip:
uv run --no-project --with taskiq-redis==1.2.4 python repro.py --streams 1 10 50 161 --one-way-ms 14
This measures the time from publishing a message to listen() yielding it (15 messages per row):
| Streams | p50 | p95 | Redis commands/s |
|---|---|---|---|
| 1 | 31 ms | 32 ms | 3.1 |
| 10 | 31 ms | 37 ms | 22.4 |
| 50 | 1,697 ms | 2,898 ms | 72.8 |
| 161 | 4,454 ms | 8,452 ms | 93.4 |
On 1.2.2 in production we saw p50 of 4 to 7 s and p95 of 11 to 16 s.
What fixed it for us
We subclassed RedisStreamBroker and overrode only listen():
- The read loop only reads.
- The reclaim scan runs in its own task every N seconds (30 by default). It sends one
XAUTOCLAIMper stream in a single pipeline, and it keeps going on any stream that returned a full batch. - The scan takes a lease (
SET NX PX <interval>) in place of the per-stream locks, so several workers end up scanning about once per interval between them. - Both loops feed one bounded queue. If either loop dies, its error is raised from
listen().
The same script with --patched runs this version (the subclass is in repro.py):
| Streams | p50 | p95 | Redis commands/s |
|---|---|---|---|
| 1 | 32 ms | 35 ms | 1.6 |
| 10 | 32 ms | 35 ms | 1.9 |
| 50 | 32 ms | 33 ms | 3.2 |
| 161 | 32 ms | 33 ms | 6.7 |
In production, p95 receive latency went from about 19 s to about 30 ms, and the backlog cleared within minutes of deploying.
Two things change for users. A dead consumer's messages are recovered up to N seconds after idle_timeout, where before it was within one loop iteration. And the per-stream autoclaim:<group>:<stream> locks are replaced by a single lease key.
Would you want this as an opt-in, for example a reclaim_interval argument that moves the scan off the read path? Or would you rather keep listen() as it is? I'm happy to send a PR in whichever shape you prefer. It's also related to #131, which is about the autoclaim lock having no default timeout.
Taskiq version
0.13.0 (taskiq-redis 1.2.4, redis-py 8.1.0, Redis 7). We first saw it on taskiq 0.12.4 and taskiq-redis 1.2.2.
Python version
3.13
OS
macOS (arm64) with Docker 29.8.2 for the reproduction, Linux in production.
Broker initialization code
names = [f"stream-{i}" for i in range(n)] # n = 1, 10, 50, 161
broker = RedisStreamBroker(
"redis://<host>:<port>",
queue_name=names[0],
additional_streams=dict.fromkeys(names[1:], ">"),
consumer_group_name="g",
consumer_name="c",
consumer_id="0",
xread_block=5000,
xread_count=8,
idle_timeout=1_800_000,
unacknowledged_batch_size=5,
unacknowledged_lock_timeout=60.0,
)
Activity
atharvashukla098-tech commentedon Oct 20, 2026
Verified the diagnosis in
taskiq_redis/redis_broker.py—RedisStreamBroker.listen()runs the reclaim scan inline between everyxreadgroup(lines 296–312): a per-stream lock acquire/release plus oneXAUTOCLAIMper stream, one at a time, before the loop can read again. With N streams and R ms round-trip that's an N×R pause ahead of every batch, which is precisely the 161-stream / 28 ms row of the table (161 × ~28 ms ≈ 4.5 s, matching the measured p50). The throughput ceiling fromxread_count / scan_timefalls out of the same shape.The two-loop design you describe fixes both at once: read loop that only reads, reclaim scan as its own periodic task with a single lease and one pipelined
XAUTOCLAIMbatch per interval, both feeding a bounded queue. The two behavior changes you call out (recovery within N seconds ofidle_timeoutinstead of one loop iteration, and the single lease key replacing per-stream locks) are reasonable trade-offs for a 4.5 s → ~30 ms p50.On opt-in vs default: since the recovery-time change is observable, I'd suggest
reclaim_intervalas an opt-in at first (None/0keeps today's behavior exactly), so existing deployments choose when to take the new semantics — then it can graduate to a default later.If that shape works for you, could you assign this to me? I'd implement the opt-in
reclaim_intervalvariant in taskiq-redis with a regression test that fails on the current inline scan (e.g. asserting no XAUTOCLAIM is issued between two consecutive xreadgroup batches when the interval elapses), and attach before/after numbers from your repro script shape.atharvashukla098-tech commentedon Oct 20, 2026
Implementation update: I've built the opt-in
reclaim_intervalvariant in a local branch of taskiq-redis and verified it against a real Redis:RedisStreamBroker(..., reclaim_interval=30.0)runslisten()as two loops feeding one bounded queue: a read task that onlyXREADGROUPs, and a reclaim task that takes a singleSET NX PXlease and pipelines oneXAUTOCLAIMper stream per interval, looping while any stream returns a full batch. Several workers share the lease, so together they scan about once per interval.reclaim_interval=None(the default) keeps the historical inline scan exactly as it is — the old_reclaim_passcode is extracted unchanged, so no behavior change for existing deployments.tests/test_reclaim_interval.py), both against real Redis:reclaim_intervalset, noXAUTOCLAIMis issued on the read path (fails on the current code);idle_timeout(guards the default path).Still happy to adjust the shape (naming, lease key, defaults) to your preference — the branch is ready to push and open as a PR in taskiq-redis as soon as you'd like it, or I can wait for your call on the opt-in-vs-default question first.