Redis Stream with the Consumer Group. Сonsumer does not receive the messages that are in the PEL after the restart.

[formatc13]

I use Redis Stream with the consumer group.
In the case of a consumer restart, I expected that the messages that got into the PEL would be read again. But as far as I can see, the message is not requested with last_id, but always with ">". As a result, the consumer does not receive the messages that it was processing at the time of the shutdown.

faststream/faststream/redis/subscriber/usecase.py

Line 678 in 7c5dffc

streams={stream.name: ">"},

Is this correct? how to handle this situation?

[powersemmi]

Hi there,

Thanks for raising this question! What you’re seeing in faststream/faststream/redis/subscriber/usecase.py is indeed expected behavior:

# Line 678 in 7c5dffc
streams = { stream.name: ">" }

Using ">" tells Redis to deliver only new messages that arrive after the call to XREADGROUP. Any messages that were fetched into the Pending Entries List (PEL) but not yet acknowledged will not be re-delivered by that call.

---

How the PEL (Pending Entries List) works

1. Message arrives in the stream → it’s added to the stream itself.
2. Consumer group reads it → Redis creates a PEL entry associating that message ID with your consumer.
3. Until you ACK it, that message remains pending, tied to that consumer indefinitely.

Because you’re passing ">", you’re telling Redis “give me only messages I haven’t seen yet.” It won’t automatically re-deliver anything sitting in the PEL.

---

Handling “stuck” or “dead” messages

To recover messages that have been pending for too long (e.g. because your consumer crashed before ACKing), you have a couple of options:

1. XAUTOCLAIM

Before your XREADGROUP call, use XAUTOCLAIM (or its newer alias XCLAIM in Redis 7.0+) to claim back messages that have been idle longer than a threshold. For example:

# Pseudocode
pending = redis.xautoclaim(
    stream,
    group_name,
    consumer_name,
    min_idle_time_ms=60000,   # e.g. 1 minute
    count=100,
    start_id="0-0"
)
# pending now contains messages reassigned to this consumer,
# so you can process them before reading new ones.

2. Custom scheduler

Run a background task (or sidecar) that periodically:

• Queries XPENDING for each stream/group,
• Identifies stale messages (using the “idle” time),
• Reassigns them via XCLAIM or by moving them back into the main stream for reprocessing.

---

Why you’re not seeing re-delivery

Most likely, your service isn’t performing a graceful shutdown. On a “hard kill” (e.g. process crash or power-off), any messages you fetched but didn’t ACK stay in the PEL and won’t be re-read until you explicitly claim them.

---

Recommendations

• Implement graceful shutdown
  Like there. And don't forget to add the 'graceful_timeout' parameter for your broker.

• Use XAUTOCLAIM (or XCLAIM)
  Right before your XREADGROUP, claim any messages that have been pending beyond a reasonable threshold.

For more details, see the Redis documentation on XAUTOCLAIM and XREADGROUP.

[formatc13]

Yes, but if the customer groups are not used, you can pass the last_id and get all the records, not just those that were added to the stream at the time of launch. Why is the last_id ignored when using customer group?

faststream/faststream/redis/subscriber/usecase.py

Line 723 in 7c5dffc

{self.stream_sub.name: self.last_id},

[Lancetnik]

Closed by #2309