From d0fa3976a716d09174471766d612aaab0938fb4c Mon Sep 17 00:00:00 2001
From: Adrian Dankiv <adrian.dankiv@webflow.com>
Date: Wed, 28 Oct 2026 12:40:39 +0100
Subject: [PATCH] fix: reclaim pending messages after empty stream reads

---
 taskiq_redis/redis_broker.py |  2 --
 tests/test_broker.py         | 50 ++++++++++++++++++++++++++++++++++++
 2 files changed, 50 insertions(+), 2 deletions(-)

diff --git a/taskiq_redis/redis_broker.py b/taskiq_redis/redis_broker.py
index b1ad461..b1d2d4f 100644
--- a/taskiq_redis/redis_broker.py
+++ b/taskiq_redis/redis_broker.py
@@ -284,8 +284,6 @@ async def listen(self) -> AsyncGenerator[AckableMessage, None]:
                     noack=False,
                     count=self.count,
                 )
-                if not fetched:
-                    continue
                 for stream, msg_list in fetched:  # type: ignore[str-unpack]
                     for msg_id, msg in msg_list:  # type: ignore[str-unpack,union-attr]
                         logger.debug("Received message: %s", msg)
diff --git a/tests/test_broker.py b/tests/test_broker.py
index 3ee4665..fcd86c4 100644
--- a/tests/test_broker.py
+++ b/tests/test_broker.py
@@ -23,6 +23,56 @@ def test_no_url_should_raise_typeerror() -> None:
         ListQueueBroker()  # type: ignore
 
 
+@pytest.mark.anyio
+async def test_stream_reclaims_without_new_messages(
+    valid_broker_message: BrokerMessage,
+    redis_url: str,
+) -> None:
+    stream = uuid.uuid4().hex
+    broker = RedisStreamBroker(
+        url=redis_url,
+        queue_name=stream,
+        consumer_group_name=stream,
+        xread_block=10,
+        idle_timeout=600_000,
+        unacknowledged_batch_size=1,
+        unacknowledged_lock_timeout=600,
+    )
+    await broker.startup()
+    listener = broker.listen()
+    try:
+        async with Redis(connection_pool=broker.connection_pool) as redis:
+            message_id = await redis.xadd(
+                stream,
+                {"data": valid_broker_message.message},
+            )
+            await redis.xreadgroup(
+                stream,
+                "dead-worker",
+                {stream: ">"},
+                count=1,
+            )
+            await redis.xclaim(
+                stream,
+                stream,
+                "dead-worker",
+                0,
+                [message_id],
+                idle=broker.idle_timeout + 1000,
+                justid=True,
+            )
+        recovered = await asyncio.wait_for(anext(listener), timeout=1)
+        assert recovered.data == valid_broker_message.message
+        pending_ack = recovered.ack()
+        assert pending_ack is not None
+        await pending_ack
+    finally:
+        await listener.aclose()
+        async with Redis(connection_pool=broker.connection_pool) as redis:
+            await redis.delete(stream, f"autoclaim:{stream}:{stream}")
+        await broker.shutdown()
+
+
 async def get_message(
     broker: AsyncBroker,
 ) -> bytes | AckableMessage: