RedisStreamBroker: receive latency grows with the number of streams because listen() runs the reclaim scan between reads #699

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):

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 XAUTOCLAIM per 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):

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,
)

repro.py