Add documentation for Redis Stream message claiming and min_idle_time usage

Signed-off-by: Victor Maleca <powersemmi@gmail.com>

Author: powersemmi

docs/docs/en/redis/streams/claiming.md

@@ -0,0 +1,149 @@
+---
+# 0.5 - API
+# 2 - Release
+# 3 - Contributing
+# 5 - Template Page
+# 10 - Default
+search:
+  boost: 10
+---
+
+# Redis Stream Message Claiming
+
+When working with Redis Stream Consumer Groups, there may be situations where messages remain in a pending state because a consumer failed to process them. FastStream provides a mechanism to automatically claim these pending messages using Redis's `XAUTOCLAIM` command through the `min_idle_time` parameter.
+
+## What is Message Claiming?
+
+In Redis Streams, when a consumer reads a message from a consumer group but fails to acknowledge it (due to a crash, network issue, or processing error), the message remains in the **Pending Entries List (PEL)** of that consumer group. These unacknowledged messages are associated with the original consumer and have an "idle time" - the duration since they were last delivered.
+
+Message claiming allows another consumer to take ownership of these pending messages that have been idle for too long, ensuring that messages don't get stuck and workload can be redistributed among healthy consumers.
+
+## Using `min_idle_time` for Automatic Claiming
+
+FastStream's `StreamSub` provides a `min_idle_time` parameter that enables automatic claiming of pending messages via Redis's `XAUTOCLAIM` command. When set, the consumer will automatically scan for and claim messages that have been pending for at least the specified duration (in milliseconds).
+
+### Basic Example
+
+Here's a simple example that demonstrates automatic message claiming:
+
+```python linenums="1"
+{! docs_src/redis/stream/claiming_basic.py !}
+```
+
+## How It Works
+
+When `min_idle_time` is set:
+
+1. **Circular Scanning**: Instead of using `XREADGROUP` to read new messages, the consumer uses `XAUTOCLAIM` to scan the Pending Entries List
+2. **Idle Time Check**: Only messages that have been pending for at least `min_idle_time` milliseconds are claimed
+3. **Ownership Transfer**: Claimed messages are automatically transferred from the failing consumer to the claiming consumer
+4. **Continuous Processing**: The scanning process is circular - after reaching the end of the PEL, it starts over from the beginning
+
+### Practical Use Case
+
+Consider a scenario where you have multiple workers processing orders:
+
+```python linenums="1"
+from faststream import FastStream
+from faststream.redis import RedisBroker, StreamSub
+
+broker = RedisBroker()
+app = FastStream(broker)
+
+# Worker that might fail
+@broker.subscriber(
+    stream=StreamSub(
+        "orders",
+        group="order-processors",
+        consumer="worker-1",
+    )
+)
+async def worker_that_might_fail(order_id: str):
+    # Process order - might crash before acknowledging
+    await process_complex_order(order_id)
+    # If crash happens here, message stays pending
+
+# Backup worker with message claiming
+@broker.subscriber(
+    stream=StreamSub(
+        "orders",
+        group="order-processors",
+        consumer="worker-2",
+        min_idle_time=10000,  # 10 seconds
+    )
+)
+async def backup_worker(order_id: str):
+    # This worker will automatically pick up messages
+    # that worker-1 failed to process within 10 seconds
+    print(f"Recovering and processing order: {order_id}")
+    await process_complex_order(order_id)
+```
+
+## Combining with Manual Acknowledgment
+
+You can combine `min_idle_time` with manual acknowledgment policies for fine-grained control:
+
+```python linenums="1"
+{! docs_src/redis/stream/claiming_manual_ack.py !}
+```
+
+## Configuration Guidelines
+
+### Choosing `min_idle_time`
+
+The appropriate `min_idle_time` value depends on your use case:
+
+- **Short duration (1-5 seconds)**: For fast-processing tasks where quick failure recovery is needed
+- **Medium duration (10-60 seconds)**: For most general-purpose applications with moderate processing times
+- **Long duration (5-30 minutes)**: For long-running tasks where you want to ensure a consumer has truly failed
+
+!!! warning
+    Setting `min_idle_time` too low may cause messages to be unnecessarily transferred between healthy consumers. Set it based on your typical message processing time plus a safety buffer.
+
+### Deployment Patterns
+
+#### Pattern 1: Dedicated Claiming Worker
+Deploy a separate worker specifically for claiming abandoned messages:
+
+```python
+# Main workers (fast path)
+@broker.subscriber(
+    stream=StreamSub("tasks", group="workers", consumer="main-1")
+)
+async def main_worker(task): ...
+
+# Claiming worker (recovery path)
+@broker.subscriber(
+    stream=StreamSub("tasks", group="workers", consumer="claimer", min_idle_time=15000)
+)
+async def claiming_worker(task): ...
+```
+
+#### Pattern 2: All Workers Can Claim
+All workers can claim messages from each other:
+
+```python
+# Each worker can both process new messages and claim abandoned ones
+@broker.subscriber(
+    stream=StreamSub(
+        "tasks",
+        group="workers",
+        consumer=f"worker-{instance_id}",
+        min_idle_time=10000,
+    )
+)
+async def worker(task): ...
+```
+
+## Technical Details
+
+- **Start ID**: FastStream automatically manages the `start_id` parameter for `XAUTOCLAIM`, enabling circular scanning through the Pending Entries List
+- **Empty Results**: When no pending messages meet the idle time criteria, the consumer will continue polling
+- **ACK Handling**: Claimed messages must still be acknowledged using `msg.ack()` to be removed from the PEL
+
+## References
+
+For more information about Redis Streams message claiming:
+
+- [Redis XAUTOCLAIM Documentation](https://redis.io/docs/latest/commands/xautoclaim/){.external-link target="_blank"}
+- [Redis Streams Claiming Guide](https://redis.io/docs/latest/develop/data-types/streams/#claiming-and-the-delivery-counter){.external-link target="_blank"}

docs/docs_src/redis/stream/claiming_basic.py

@@ -0,0 +1,22 @@
+from faststream import FastStream, Logger
+from faststream.redis import RedisBroker, StreamSub
+
+broker = RedisBroker()
+app = FastStream(broker)
+
+
+@broker.subscriber(
+    stream=StreamSub(
+        "orders",
+        group="processors",
+        consumer="worker-1",
+        min_idle_time=5000,  # Claim messages idle for 5+ seconds
+    )
+)
+async def handle(order_id: str, logger: Logger):
+    logger.info(f"Processing order: {order_id}")
+
+
+@app.after_startup
+async def publish_test():
+    await broker.publish("order-123", stream="orders")

docs/docs_src/redis/stream/claiming_manual_ack.py

@@ -0,0 +1,30 @@
+from faststream import AckPolicy, FastStream, Logger
+from faststream.redis import RedisBroker, RedisStreamMessage, StreamSub
+
+broker = RedisBroker()
+app = FastStream(broker)
+
+
+@broker.subscriber(
+    stream=StreamSub(
+        "critical-tasks",
+        group="task-workers",
+        consumer="worker-failover",
+        min_idle_time=30000,  # 30 seconds
+    ),
+    ack_policy=AckPolicy.MANUAL,
+)
+async def handle(msg: RedisStreamMessage, logger: Logger):
+    try:
+        # Process the claimed message
+        logger.info(f"Processing: {msg.body}")
+        # Explicitly acknowledge after successful processing
+        await msg.ack()
+    except Exception as e:
+        # Don't acknowledge - let it be claimed by another consumer
+        logger.error(f"Failed to process: {e}")
+
+
+@app.after_startup
+async def publish_test():
+    await broker.publish("critical-task-1", stream="critical-tasks")

faststream/redis/subscriber/usecases/stream_subscriber.py

@@ -48,7 +48,7 @@ def __init__(
         self._stream_sub = config.stream_sub
         self.last_id = config.stream_sub.last_id
         self.min_idle_time = config.stream_sub.min_idle_time
-        self.autoclaim_start_id = "0-0"
+        self.autoclaim_start_id = b"0-0"
 
     @property
     def stream_sub(self) -> "StreamSub":
@@ -201,9 +201,7 @@ async def xautoclaim() -> tuple[
                     stream_name = self.stream_sub.name.encode()
                     (next_id, messages, _) = stream_message
                     # Update start_id for next call
-                    self.autoclaim_start_id = (
-                        next_id.decode() if isinstance(next_id, bytes) else next_id
-                    )
+                    self.autoclaim_start_id = next_id
                     if not messages:
                         return ()
                     return ((stream_name, messages),)
@@ -251,9 +249,7 @@ async def get_one(
             )
             (next_id, messages, _) = stream_message
             # Update start_id for next call
-            self.autoclaim_start_id = (
-                next_id.decode() if isinstance(next_id, bytes) else next_id
-            )
+            self.autoclaim_start_id = next_id
             if not messages:
                 return None
             stream_name = self.stream_sub.name.encode()
@@ -323,9 +319,7 @@ async def __aiter__(self) -> AsyncIterator["RedisStreamMessage"]:  # type: ignor
                 )
                 (next_id, messages, _) = stream_message
                 # Update start_id for next call
-                self.autoclaim_start_id = (
-                    next_id.decode() if isinstance(next_id, bytes) else next_id
-                )
+                self.autoclaim_start_id = next_id
                 if not messages:
                     continue
                 stream_name = self.stream_sub.name.encode()

tests/brokers/redis/test_config.py

@@ -47,6 +47,25 @@ def test_stream_with_group() -> None:
     assert config.ack_policy is AckPolicy.REJECT_ON_ERROR
 
 
+@pytest.mark.redis()
+def test_stream_sub_with_no_ack_group() -> None:
+    with pytest.warns(
+        RuntimeWarning,
+        match="`no_ack` is not supported by consumer group with last_id other than `>`",
+    ):
+        config = RedisSubscriberConfig(
+            _outer_config=MagicMock(),
+            stream_sub=StreamSub(
+                "test_stream",
+                group="test_group",
+                consumer="test_consumer",
+                no_ack=True,
+                last_id="$",
+            ),
+        )
+    assert config.ack_policy is AckPolicy.MANUAL
+
+
 @pytest.mark.redis()
 def test_stream_with_group_and_min_idle_time() -> None:
     config = RedisSubscriberConfig(