Improve Redis stream reclaim and prefetch handling - #129

@s3rius I would appreciate your feedback on the direction of this PR.

It improves RedisStreamBroker recovery and worker fairness by:

• reclaiming pending entries according to each task's timeout label plus a grace period, with idle_timeout as the fallback;
• removing the broker-side reclaim lock and relying on Redis XCLAIM min_idle_time for the atomic claim check;
• limiting a single-stream listener's delivered-but-unacknowledged entries through xread_count;
• allowing a restarted listener with the same consumer_name to recover pending work from its predecessor;
• handing off fetched-but-not-yet-yielded entries to an internal abandoned consumer on shutdown, making them immediately reclaimable;
• deprecating additional_streams, since Redis applies XREADGROUP COUNT per stream and it conflicts with a strict listener-level outstanding limit.

The design was partly informed by dramatiq-redis-streams, adapted for taskiq's async model.

Since opening this PR, I continued the Redis Streams-specific work in a standalone package: https://github.com/vvanglro/taskiq-redis-streams

That package is now the reference implementation for the direction I would like to explore. Its main design difference is liveness-based recovery: each generated consumer keeps a TTL heartbeat, and pending entries are reclaimed only after the previous owner's heartbeat expires. This means a live worker can run a task for an arbitrary duration without using Taskiq's timeout label as a Redis reclaim deadline.

It also uses a dedicated heartbeat connection, separates read batch size from outstanding-message capacity (xread_count and max_pending), keeps the buffered-message handoff to an abandoned consumer, and uses an atomic Lua check of both the current PEL owner and the missing heartbeat before XCLAIM.

This intentionally differs from #129's compatibility-oriented design: per-task timeout reclaim with idle_timeout fallback, configurable consumer/group settings, and the deprecated-but-still-supported additional_streams API.

I do not plan to keep expanding #129 into a full replacement without agreement on that direction. Would you prefer a smaller, compatibility-preserving upstream PR for selected improvements, or is the standalone broker a better place to continue this stream-specific design?

I'd love to read your implementation first to understand the magnitude of incompatibility. And I would also want to fully understand how it works.

Sorry for the delay. I'll try to make it this week.

One effect of this PR that isn't called out in the description, and that I think justifies it on its own: it makes the reclaim pass reachable on an idle stream.

Today RedisStreamBroker.listen() only reaches XAUTOCLAIM after a successful fetch:

fetched = await redis_conn.xreadgroup(...)
if not fetched:
 continue # <- everything below, including the XAUTOCLAIM pass, is unreachable
for stream, msg_list in fetched:
 ...
# XAUTOCLAIM pass lives down here

Reclaim is therefore gated on delivery, not on time. If a stream goes quiet, a crashed worker's pending entries aren't recovered slowly — they aren't recovered at all until some unrelated message happens to arrive.

Repro against taskiq-redis==1.2.3, redis==8.1.0, Redis 7.4:

import asyncio
from redis.asyncio import Redis
from taskiq_redis import RedisStreamBroker
URL, STREAM, GROUP = "redis://localhost:6379", "repro:stream", "taskiq"
async def main():
 r = Redis.from_url(URL)
 await r.delete(STREAM)
 broker = RedisStreamBroker(
 url=URL, queue_name=STREAM, consumer_group_name=GROUP,
 idle_timeout=1000, # 1s: anything pending is instantly reclaimable
 xread_block=500, xread_count=10,
 )
 await broker.startup()
 # a worker that took a message and died without acking
 await r.xadd(STREAM, {b"data": b"orphan"})
 await r.xreadgroup(GROUP, "dead-worker", {STREAM: ">"}, count=10)
 print("pending after simulated crash:", (await r.xpending(STREAM, GROUP))["pending"])
 received = []
 async def consume():
 async for m in broker.listen():
 received.append(m.data)
 task = asyncio.create_task(consume())
 await asyncio.sleep(10) # 10x idle_timeout, stream idle
 print("after 10s idle -> delivered to listener:", received)
 await r.xadd(STREAM, {b"data": b"unrelated"}) # one unrelated message
 await asyncio.sleep(3)
 print("after 1 new msg -> delivered to listener:", received)
 task.cancel()
 await r.delete(STREAM); await r.aclose(); await broker.shutdown()
asyncio.run(main())

pending after simulated crash: 1
after 10s idle -> delivered to listener: []
after 1 new msg -> delivered to listener: [b'unrelated', b'orphan']

idle_timeout is 1s and the entry sits unreclaimed through 10x that. It only comes back as a side effect of unrelated traffic.

Nothing else covers this case:

• retry middleware fires from on_error, which never runs in a process that was killed;
• restarting the worker doesn't help — listen() reads with >, which returns only entries no consumer in the group has seen, so pending entries stay invisible to it whether or not consumer_name is pinned.

The exposure is worst for bursty or scale-to-zero deployments: a crash during traffic is recovered within one poll, while a crash as traffic stops can strand work indefinitely. Anything that makes worker death periodic rather than random makes it much more likely to be hit — we ran into it with Entra ID token expiry on Azure Managed Redis, where an unreauthenticated connection is dropped on a fixed ~77 minute cadence and takes the worker process with it.

This PR fixes it by construction: moving the reclaim ahead of the fetch and throttling it with reclaim_interval means an idle listener still sweeps.

while True:
 ...
 last_reclaim, buffered = await self._build_due_reclaimed_messages(...) # time-gated
 ...
 buffered = await self._build_new_ackable_messages(...) # then fetch

That's the same ordering RQ uses — run_maintenance_tasks() runs before the blocking dequeue_any(), on a timer, so an idle worker still reaps its registries.

I couldn't find an existing issue for this specific gate; #131 describes a different mechanism (the un-expiring autoclaim lock) reaching a similar outcome. Happy to open it as a standalone issue if that's more useful for tracking than a comment here.