RedisStreamBroker: autoclaim lock has no timeout, so a SIGKILLed worker permanently blocks message reclaim

[I-am-Saiff]

Summary

RedisStreamBroker.listen() guards its XAUTOCLAIM with a Redis lock created without a timeout. redis-py's Lock defaults to timeout=None, so the key never expires. A worker killed (SIGKILL, OOM, container eviction) while holding that lock leaves it set permanently, and every subsequent worker takes the continue branch and never reclaims pending messages again.

The practical effect: crash recovery works after a graceful shutdown and fails silently after a hard kill — which is the case the pending-entries list exists for.

Location

taskiq_redis/redis_broker.py, RedisStreamBroker.listen():

lock = redis_conn.lock(
    f"autoclaim:{self.consumer_group_name}:{stream}",
)
if await lock.locked():
    continue
async with lock:
    pending = await redis_conn.xautoclaim(
        name=stream,
        groupname=self.consumer_group_name,
        consumername=self.consumer_name,
        min_idle_time=self.idle_timeout,
        count=self.unacknowledged_batch_size,
    )

redis_conn.lock(name) is called with no timeout; redis.asyncio.Redis.lock has timeout: Optional[float] = None.

Reproduction

import asyncio, redis.asyncio as r

async def main():
    c = r.from_url("redis://localhost:6379", decode_responses=True)
    await c.xadd("t:stream", {"data": "x"})
    await c.xgroup_create("t:stream", "g", id="0", mkstream=True)
    await c.xreadgroup("g", "dead-consumer", {"t:stream": ">"}, count=10)
    print("pending:", (await c.xpending("t:stream", "g"))["pending"])   # 1

    # a worker SIGKILLed inside `async with lock:` leaves this behind
    lock = c.lock("autoclaim:g:t:stream")          # timeout=None, as in listen()
    await lock.acquire()
    print("lock TTL:", await c.ttl("autoclaim:g:t:stream"))            # -1, never expires

    # a fresh worker follows the same guard
    fresh = c.lock("autoclaim:g:t:stream")
    if await fresh.locked():
        print("fresh worker skips xautoclaim -> reclaimed 0")
    await c.delete("autoclaim:g:t:stream")
    print("after clearing the lock -> reclaimed",
          len((await c.xautoclaim("t:stream", "g", "fresh", min_idle_time=0))[1]))  # 1

asyncio.run(main())

Output:

pending: 1
lock TTL: -1
fresh worker skips xautoclaim -> reclaimed 0
after clearing the lock -> reclaimed 1

Observed end to end as well: a worker running with --ack-type when_executed, SIGKILLed mid-task, leaves its message unacked; a replacement worker never picks it up.

Versions

taskiq-redis 1.1.2, taskiq 0.11.20, redis 6.4.0, Redis server 8.10.1, Python 3.9, macOS.

Suggested fix

Give the lock a bounded lifetime, e.g. redis_conn.lock(key, timeout=self.idle_timeout / 1000) or a small fixed value. Deleting a stale key from outside isn't safe for a consumer to do on its own — a live holder is byte-identical to an orphaned one — so the expiry needs to come from the construction site.

Happy to open a PR if the approach looks right.