The consumer group · dive 02

Recover self, then peers

Two mechanisms, one loop — drain your own backlog the instant you restart, reclaim a dead peer's after it goes idle.

A group's reliability comes down to one question: when a reader dies, what happens to the entries it was holding? The group answers with the PEL — the per-consumer list of entries delivered but not yet acked. EchoMQ.StreamConsumer recovers held work with two complementary mechanisms, and the difference between them is who died and whether they came back: a PEL-drain on (re)start recovers self, and the XAUTOCLAIM beat recovers dead peers. Pick a recovery scenario on the right to see which mechanism owns it.

Drain self on (re)start Which mechanism? Reclaim dead peers Park on the private lane Pattern & implementation References

The loop · two recovery mechanisms

Step 1 drains the consumer's own PEL on (re)start; step 2 reclaims a dead peer's idle entries each beat; step 3 parks for new entries on a private lane.

Drain the PEL on (re)start — recover SELF

A consumer that crashes and restarts with the same name has its own backlog waiting in the PEL, keyed to that name. So the very first thing the loop does, before reading anything new, is drain its own PEL to exhaustion: XREADGROUP GROUP g <self> … 0 reads the un-acked entries keyed to this consumer (cursor 0 means “my pending,” not “new”), settles each, and repeats until the read comes back empty. Then it switches to >.

This is why a restarted consumer recovers its held work the instant it restarts — it does not wait for an idle timeout, because the entries are its own. And a clean cold start costs nothing extra: a fresh consumer has an empty PEL, so the first 0 read returns nothing and the loop goes straight to >. One code path covers both the crash-restart and the cold-start case.

# echo_mq — EchoMQ.StreamConsumer
# Drain the consumer's OWN PEL to exhaustion (recover SELF on (re)start): XREADGROUP ... 0
# reads the un-acked backlog keyed to THIS consumer name; each entry is settled by the
# handler verdict, then the read repeats until the PEL returns empty. A clean cold start
# has an empty PEL -> one round-trip, then the steady loop. The `0` read is NON-blocking.
defp drain_pel(s) do
  check_control()

  case read_group(s, "0") do
    [] ->
      :ok

    entries ->
      settle_each(s, entries)
      drain_pel(s)
  end
end

The 0 read is non-blocking — the PEL is already known server-side, so there is nothing to wait for. The drain runs to empty, then control passes to the steady loop.

Which mechanism recovers it?

Two facts decide which mechanism owns a recovery: did the dead consumer restart with the same name? and how long have its entries been idle? Pick a scenario and see which mechanism recovers the held work — the self-drain (instant, no wait) or the peer-reclaim (idle past the threshold). The model is the real loop's two branches over fixed scenarios.

A scenario · self-drain or peer-reclaim

The split is sharp: a consumer that came back drains its own PEL instantly; a consumer that never did has its work reclaimed by a peer only after it goes idle past :min_idle_ms. A cold start finds nothing to drain.

The XAUTOCLAIM beat — recover dead PEERS

The PEL-drain only recovers a consumer that came back. But a consumer can die and never restart — the machine is gone, the deploy removed it, the name retired. Its entries sit in its PEL with no one to self-drain them. That is what XAUTOCLAIM is for. On each beat of the steady loop, a live consumer reclaims entries that have been idle past :min_idle_ms — XAUTOCLAIM <key> <group> <self> <min_idle_ms> 0 re-assigns those entries to itself and hands them back to the same settle path as any other delivery.

The idle threshold is evaluated server-side, against Valkey's own XPENDING idle time — there is no host clock in the decision. That matters: clocks on different machines disagree, so a host-side “how long has it been idle” comparison would race; deferring to the server's idle measurement makes the reclaim decision a single, coherent one. :min_idle_ms is the one tunable for how long before a dead peer's work is re-delivered.

# echo_mq — EchoMQ.StreamConsumer
# Reclaim a dead PEER's idle entries: XAUTOCLAIM <key> <group> <self> <min_idle_ms> 0
# re-assigns to THIS consumer every entry idle past the threshold (server-side idle, no
# host clock). The reply is the [next-cursor, claimed-entries, deleted-ids] triple; the
# claimed entries are settled like any other delivery. One pass per beat (cursor 0).
defp reclaim_peers(s) do
  case Connector.command(s.conn, [
         "XAUTOCLAIM", s.key, s.group, s.consumer,
         Integer.to_string(s.min_idle_ms), "0",
         "COUNT", Integer.to_string(s.count)
       ]) do
    {:ok, [_cursor, claimed, _deleted]} when is_list(claimed) ->
      settle_each(s, claimed)

    {:ok, [_cursor, claimed]} when is_list(claimed) ->
      # some server builds answer a two-element [cursor, claimed] reply
      settle_each(s, claimed)

    _other ->
      :ok
  end
end

One pass per beat (cursor 0 each time); a deep peer backlog drains over successive beats rather than in one greedy sweep, so a single reclaim never blocks the loop on a huge dead-peer queue.

Park on `>` — on the consumer's own private lane

With self drained and dead peers reclaimed, the loop parks for new entries: XREADGROUP GROUP g <self> BLOCK <beat_ms> COUNT <n> STREAMS key >. This is the only blocking verb in the consumer, and it rides the consumer's own private connector lane — the same BLPOP-on-its-own-lane precedent the job consumer set. The single-owner socket the rest of the system shares is never stalled by a parked reader; the block lives on a dedicated connection that exists only to serve this loop.

A BLOCK of :beat_ms sets the cadence: each beat, the loop wakes, reclaims dead peers, and re-parks. So :min_idle_ms is how long before a dead peer's work moves, and :beat_ms is how often the loop checks. The two compose into the recovery latency without any background timer.

# echo_mq — EchoMQ.StreamConsumer
# The steady loop: on each beat reclaim dead PEERS' idle entries (XAUTOCLAIM), then park
# on the blocking `>` read for new entries. Control is honored at the settle points
# (between entries, never inside one). The lane dying takes the loop with it, for the
# supervisor to restart.
defp loop(s) do
  check_control()
  reclaim_peers(s)

  case read_group_block(s) do
    [] -> :ok
    entries -> settle_each(s, entries)
  end

  loop(s)
end

Control is honored at the settle points. A stop or a supervisor shutdown is honored between entries, never inside one. The loop traps exits, so a control message lands in the mailbox and check_control reads it at each settle point: a stop drains to :normal, a :shutdown drains to :shutdown, and the entry in hand finishes first. So stop/2 is bounded by the beat plus the one entry being settled — a parked consumer notices the request when its BLOCK returns. No entry is ever abandoned half-handled to honor a stop.

# echo_mq — EchoMQ.StreamConsumer
# The loop traps exits, so control arrives as messages and is honored at the settle points
# — between entries, never inside one. A stop request drains to :normal; the supervisor's
# :shutdown drains to :shutdown; the dedicated lane dying takes the loop with it.
defp check_control do
  receive do
    {:emq_stop, _from, _ref} -> exit(:normal)
    {:EXIT, _from, :shutdown} -> exit(:shutdown)
    {:EXIT, _from, reason} -> exit(reason)
  after
    0 -> :ok
  end
end

The parked BLOCK is what bounds stop latency: a consumer asleep on its read wakes when the block returns (at most a beat), notices the stop at the next settle point, finishes the entry in hand, and exits clean.

Pattern & implementation

the pattern (Redis Patterns Applied)

A consumer group keeps a per-consumer PEL; a crashed reader's pending entries are reclaimed by another with XAUTOCLAIM after an idle period. Streams & Events teaches the recovery loop.

the implementation (echo_mq)

Two mechanisms, sharply separated: a self-drain on (re)start for a consumer that came back (instant, no idle wait), and an XAUTOCLAIM beat for a peer that never did (idle past :min_idle_ms, server-clock). The blocking > read parks on a private lane so the shared socket is never stalled.

The pattern names one recovery; the implementation splits it by who came back — instant for your own work, idle-gated for a dead peer's — and keeps the only blocking read off the shared socket.

Recap — self first, then peers

The loop runs three steps: drain the consumer's own PEL on (re)start (recover self, instant), reclaim a dead peer's idle entries each beat (recover peers, server-clock), then park on a blocking > read on a private lane. Control is honored between entries, so a stop drains the entry in hand and exits clean. The next dive opens what “settle” means — the handler, the ack, and the cost of delivering an entry more than once.

Next in the module: At-least-once & the handler — the exact-mirror handler, XACK on success, leave-un-acked on failure, and why a re-claimed entry forces idempotency.

References

Where the recovery mechanisms are grounded, and the stream semantics they rest on.

Sources

Valkey — XREADGROUP — > for new entries, 0 to replay the consumer's own PEL.
Valkey — XAUTOCLAIM — reclaiming idle pending entries from a dead peer, server-side idle.
Valkey — XPENDING — the idle time the reclaim threshold is measured against.
Valkey — Introduction to Streams — the PEL and consumer recovery.

Related in this course

The consumer group — the module this dive belongs to.
The group door — opening the group the loop reads through.
At-least-once & the handler — what settling an entry does, and the cost.
The stream log — the writer whose log the loop drains.
redis-patterns · Streams & Events — the streams pattern that doors here.