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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
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")

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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

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)

Unified Consuming and Claiming with XREADGROUP CLAIM#

Redis 8.4 added the CLAIM option to XREADGROUP, merging claiming and consuming into a single command: each read first claims messages that have been pending for at least the given duration, then reads new ones. FastStream exposes it via the claim_min_idle_time parameter, so a single subscriber can process new messages and recover abandoned ones:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

from faststream import FastStream, Logger
from faststream.redis import RedisBroker, StreamSub
from faststream.redis.annotations import RedisStreamMessage

broker = RedisBroker()
app = FastStream(broker)


@broker.subscriber(
    stream=StreamSub(
        "orders",
        group="processors",
        consumer="worker-1",
        claim_min_idle_time=5000,  # Also claim messages idle for 5+ seconds
    )
)
async def handle(order_id: str, message: RedisStreamMessage, logger: Logger):
    # Previous deliveries: 0 for new messages, 1+ for reclaimed ones
    if message.raw_message["delivery_counts"][0]:
        idle_ms = message.raw_message["idle_times"][0]
        logger.info(f"Recovering order {order_id} after {idle_ms} ms idle")
    else:
        logger.info(f"Processing order: {order_id}")


@app.after_startup
async def publish_test():
    await broker.publish("order-123", stream="orders")

Claim Metadata#

When claim_min_idle_time is set, Redis extends every returned entry with two extra fields, which FastStream exposes via the raw message (as in the example above). Both are lists aligned with message_ids, so batch subscribers get one value per entry:

• message.raw_message["idle_times"] — milliseconds since the last delivery
• message.raw_message["delivery_counts"] — previous deliveries (0 for a new message)

This makes retry caps and dead-letter routing possible without an extra XPENDING call.

claim_min_idle_time vs min_idle_time#

The new option does not replace XAUTOCLAIM-based claiming — they complement each other:

If your stream is capped (maxlen / XTRIM), deleted entries can leave ghost records in the PEL that only XAUTOCLAIM cleans up.

Additional guidelines:

• The min_idle_time sizing warning below applies here as well: set claim_min_idle_time greater than your worst-case processing time, with a safety margin. This is especially important with max_workers, where the subscriber keeps reading while previous messages are still being processed — a too-low threshold lets it reclaim its own in-flight messages.
• Claimed entries share the COUNT budget with new ones and are returned first, so a large backlog of idle messages can fill whole reads. Consider setting max_records to bound a single delivery.
• Works with Redis Cluster as well: each stream maps to a single node, so the CLAIM option needs nothing cluster-specific.

Combining with Manual Acknowledgment#

You can combine min_idle_time with manual acknowledgment policies for fine-grained control:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

from faststream import AckPolicy, FastStream, Logger
from faststream.redis import RedisBroker, RedisStreamMessage, StreamSub, Redis

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, redis: Redis):
    try:
        # Process the claimed message
        logger.info(f"Processing: {msg.body}")
        # Explicitly acknowledge after successful processing
        await msg.ack(redis=redis, group="critical-tasks")
    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")

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

Deployment Patterns#

Pattern 1: Dedicated Claiming Worker#

Deploy a separate worker specifically for claiming abandoned messages:

# 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 process new messages and claim abandoned ones. This pattern requires claim_min_idle_time (Redis 8.4+, see Unified Consuming and Claiming) — with min_idle_time, subscribers only claim pending messages and never consume new ones:

# Each worker both processes new messages and claims abandoned ones
@broker.subscriber(
    stream=StreamSub(
        "tasks",
        group="workers",
        consumer=f"worker-{instance_id}",
        claim_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

Testing#

Claiming behaves the same way in tests, without waiting on real min_idle_time timeouts. See Testing Consumer Groups for details and runnable examples.

References#

For more information about Redis Streams message claiming:

• Redis XAUTOCLAIM Documentation
• Redis XREADGROUP CLAIM Documentation
• Redis Streams Claiming Guide