Improve Redis stream reclaim and prefetch handling

[vvanglro]

Summary

This PR improves RedisStreamBroker reliability and fairness by adopting stronger Redis Streams handling patterns: per-task-deadline reclaim, lock-free XCLAIM, listener-local prefetch backpressure, and faster handoff of buffered messages on listener close.

What changed

• Replace lock-based pending recovery with timeout-aware XCLAIM recovery.
• Reclaim messages using the task's own timeout label plus reclaim_timeout_grace, falling back to idle_timeout for messages without a timeout label.
• Deprecate and ignore unacknowledged_lock_timeout; Redis XCLAIM min_idle_time now provides the atomic claim guard.
• Add reclaim_interval so pending-message scans are throttled instead of running on every listen loop.
• Add local prefetch/backpressure tracking so xread_count caps delivered-but-unacknowledged messages for a single-stream listener.
• Deprecate additional_streams. It remains supported for compatibility, but will be removed in a future major release; use one broker and worker process per stream instead.
• Track only messages delivered by the current listener instance as protected from reclaim, so a restarted worker using the same consumer_name can still recover pending messages from its predecessor.
• On listener close, claim entries that were fetched into the broker-local buffer but not yet yielded to taskiq to an internal abandoned consumer and stamp them as very idle, so the next reclaim sweep can recover them immediately.
• Update README documentation for stream broker reclaim/backpressure/close behavior.

Why

The previous reclaim path used a single global idle_timeout plus a broker-side Redis lock. That can reclaim long tasks too early (for example, a task with a 30m timeout reclaimed at the default 10m idle_timeout) and recover short crashed tasks slowly (for example, a 5s task waiting for the default 10m idle_timeout). The lock also adds a failure point (unacknowledged_lock_timeout) and extra round-trips, while Redis XCLAIM min_idle_time already deduplicates claims atomically on the server.

additional_streams makes strict listener-level prefetch limits ambiguous: Redis applies XREADGROUP COUNT per stream, not across all streams in a multi-stream read. One broker per stream keeps consumer ownership, reclaim behavior, and prefetch capacity independent.

The close-time handoff is deliberately limited to the broker's internal buffer: messages already yielded to taskiq may be executing, so they are not abandoned immediately to avoid duplicate execution during shutdown.

Behavior notes

• Reclaim sweeps run before XREADGROUP within a listen iteration, so overdue pending messages take priority over new messages when a sweep is due.
• reclaim_interval defaults to 30000 ms; set it to 0 to scan on every iteration.
• Only messages held by the current listener instance are protected from reclaim; pending entries from a dead predecessor (even with the same consumer_name) are recoverable.
• xread_count limits delivered-but-unacknowledged messages for a single-stream broker. xread_count=None disables the limit.
• additional_streams emits DeprecationWarning when configured.
• Broker-local buffered entries are handed to an internal abandoned consumer on listener close; already-yielded entries still follow their normal ack/reclaim lifecycle.

Tests

Added coverage for:

• task-timeout based reclaim
• idle-timeout fallback reclaim
• reclaim with shared consumer_name
• xread_count backpressure
• additional_streams deprecation warning
• buffered-message handoff on listener close

Validation run locally:

uv run ruff check taskiq_redis/redis_broker.py tests/test_broker.py
uv run mypy taskiq_redis/redis_broker.py tests/test_broker.py
uv run pytest tests/test_broker.py -k "not cluster and not sentinel" -q

Result: 16 passed, 7 deselected for the single-node Redis broker tests.

Reference

The reclaim/backpressure design was informed by the Redis Streams broker implementation in sylvinus/dramatiq-redis-streams, especially its use of deadline-aware pending recovery, bounded local prefetching, and abandoned-buffer handoff on close.

[vvanglro]

@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.

[codecov]

Codecov Report

❌ Patch coverage is 95.23810% with 8 lines in your changes missing coverage. Please review.
✅ Project coverage is 94.43%. Comparing base (090a00a) to head (4460660).
⚠️ Report is 36 commits behind head on main.

Files with missing lines	Patch %	Lines
taskiq_redis/redis_broker.py	95.23%	8 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #129      +/-   ##
==========================================
+ Coverage   91.70%   94.43%   +2.73%     
==========================================
  Files           7        8       +1     
  Lines         434      845     +411     
==========================================
+ Hits          398      798     +400     
- Misses         36       47      +11     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.