Repository navigation
RedisStreamClusterBroker never recovers unacknowledged messages #130
Description
Problem
RedisStreamClusterBroker.listen() only ever calls XREADGROUP ... >. Unlike RedisStreamBroker.listen(), it never reads the consumer group's pending-entries list, so there is no XAUTOCLAIM pass.
If a worker dies after a message is delivered but before it is acked, that message stays in the PEL indefinitely: it is never redelivered and never retried, so the task silently never completes. SmartRetryMiddleware doesn't help here, since nothing runs in a process that has already died.
The cluster variant is also missing the corresponding knobs: idle_timeout, unacknowledged_batch_size, unacknowledged_lock_timeout, and xread_count (so XREADGROUP runs uncapped).
Version: taskiq-redis 1.2.2, redis 7.4.1.
Impact
AWS ElastiCache Serverless runs cluster-mode-enabled only, so users on it are forced onto RedisStreamClusterBroker and lose crash recovery as a side effect of their infrastructure choice, with no signal that it happened. The same applies to other managed cluster-mode-only Redis/Valkey offerings.
There's no cluster-specific blocker
XAUTOCLAIM is a single-key command, and the reclaim lock key (autoclaim:{group}:{stream}) is a single key, so both hash to one slot and
work unchanged under cluster mode. RedisCluster exposes equivalent .lock() and .xautoclaim() methods.
Suggestion
Bring RedisStreamClusterBroker.listen() to parity with RedisStreamBroker.listen(), including the accompanying constructor
parameters. Ideally by factoring the reclaim logic into a shared mixin or helper used by both brokers, so the two don't drift again as the non-cluster path evolves.
Activity
Claire-ovo commentedon Oct 27, 2026
I reproduced the Cluster gap on current main 8f70bc8 with Python 3.14.7, redis-py 8.0.0, taskiq 0.13.0, and Redis 8.4.7 (three primary nodes, all 16,384 slots covered).
An old broker received a synthetic message without acknowledging it, then its listener and connections were closed. I set that pending entry's idle time to 900,000 ms using XCLAIM while retaining its old owner. A new broker, with XREADGROUP block=50 ms, delivered nothing during a 500 ms observation window; XPENDING still showed the old owner. A fresh message was delivered and acknowledged normally. Explicit XAUTOCLAIM with min_idle_time=600,000 retrieved the exact pending ID/payload, and XACK reduced the pending count to zero.
This is a bounded protocol-level reproduction, not an OS-killed Taskiq worker or failover test. The nodes ran in one isolated container; no TLS, managed service, or multi-stream validation was performed.
Before preparing a Cluster patch, would you prefer parity with the current idle-time reclaim API, or should this follow the recovery contract being discussed in #129? I would include recovery with no new traffic: copying the current single-node loop's empty-read continue would retain the idle-stream issue already reported there. I can help with Cluster regression coverage if that fits the existing work.
This investigation is AI-assisted. No product patch is prepared yet.