R5.02.2 · Stream consumer patterns — dive

Consumer groups — read, ack, resume.

A blocking read gives you delivery, not reliability: crash before the work finishes and the entry is gone. A consumer group adds the missing piece — the server tracks, per consumer, which entries were delivered and not yet acknowledged, so a crash does not lose them. This is the reader law.

Grounding: echo/apps/echo_mq/lib/echo_mq/stream_consumer.ex — the loop (drain own PEL → read > → XAUTOCLAIM), the handler map carrying the XPENDING delivery count, the XACK / leave-un-acked settle. Consumer: codemojex.

The loop runs three phases, in order. Select a phase to read what it recovers. Recovery comes before new work: drain your own backlog, read the new, then reclaim what a dead peer left.

readout · Select a phase. The loop recovers before it reads anything new: drain your own un-acked backlog from cursor 0, read > for new entries, and reclaim a dead peer's idle entries with XAUTOCLAIM.

§1

At-least-once, with resume

streams-consumer-patterns.md.txt

XREADGROUP GROUP <group> <consumer> STREAMS <key> > delivers entries the group has not yet delivered to anyone, and records each as pending for the consumer that received it. The entry stays pending until that consumer calls XACK. Two facts follow. First, delivery is at-least-once: an entry is re-delivered until acknowledged, so a handler must be idempotent. Second, the group resumes: a consumer that restarts with the same name still owns its un-acked entries, and reading from cursor 0 hands them back.

EchoMQ's settle is that rule in two branches: on :ok the entry is XACKed and retires from the pending list; on {:error, reason} — or a handler that raises, which the loop converts to {:error, reason} and survives — it is left un-acked, surviving in the pending list to be re-delivered. The only way an entry leaves the group's pending list is a handler that returns :ok.

A consumer group is the server remembering what each consumer was handed and has not yet settled. Delivery is at-least-once; the group resumes; the handler must be idempotent.

§2

The handler map

stream_consumer.ex · settle/2

EchoMQ.StreamConsumer hands every entry to one handler with a fixed shape, byte-identical to the job consumer's handler so one portable handler serves both:

fun(%{id: branded, payload: payload, attempts: attempts, group: group}) :: :ok | {:error, reason}

id is the stored branded EVT record id — the writer's receipt, recovered from the entry's id field, not the wire position. payload is the entry's remaining fields as a map. group is the consumer-group name. And attempts carries the XPENDING per-entry delivery count — how many times this entry has been delivered to a consumer — specified, not assumed, so a poison threshold (attempts >= N) calibrates against the quantity that actually grows on each re-claim, never against a handler-failure count it would confuse for it.

One entry, two settle paths. Select a verdict to read where the entry goes. An ack retires it from the PEL; an error leaves it there to be re-delivered.

readout · Select a verdict. On :ok the entry is XACKed and retires from the PEL; on an error — or a raise the loop converts to an error — it is left un-acked and re-delivered.

echo/apps/echo_mq/lib/echo_mq/stream_consumer.ex · settle/2, the ack-or-leave branchverbatim

case verdict do
  :ok ->
    ← the genuinely-pending entry retires from the PEL
    _ = Connector.command(s.conn, ["XACK", s.key, s.group, xadd_id])
    :ok

  {:error, _reason} ->
    ← left un-acked: it survives in the PEL, re-delivered
    :ok
end

§3

Recover self, then recover peers

stream_consumer.ex · drain_pel / reclaim_peers

A consumer that only ever requests > is the critical anti-pattern: crash after a delivery but before the XACK and those entries become zombies, stuck in the PEL and never re-read. Recovery is two complementary mechanisms, both named, and the loop runs them in order before it blocks for anything new.

Drain own PEL — recover self. On the first pass the consumer reads its own pending list from cursor 0 to exhaustion, settling each, then switches to >. A crashed consumer that restarts with the same name recovers its own held work the instant it restarts; a clean cold start has an empty PEL, so the 0 read returns nothing and one code path covers both.
The XAUTOCLAIM beat — recover peers. A consumer that dies and never restarts has a pending list nobody drains. On each beat the loop reclaims entries idle past :min_idle_ms (default 30_000) — decentralised work-stealing, no coordinator. The idle threshold is evaluated server-side against XPENDING idle, never a host clock.

echo/apps/echo_mq/lib/echo_mq/stream_consumer.ex · drain_pel/1, recover self on (re)startverbatim

defp drain_pel(s) do
  check_control()

  case read_group(s, "0") do          ← cursor 0: this consumer's un-acked backlog
    [] ->
      :ok                             ← empty PEL → straight to the steady loop

    entries ->
      settle_each(s, entries)
      drain_pel(s)                    ← drain to exhaustion, then read >
  end
end

Notes on ValkeyThe group door is lazy: XGROUP CREATE … MKSTREAM on start swallows only the BUSYGROUP reply (the group already exists), so a restart storm never errors, while a WRONGTYPE is loud — valkey.io/commands/xgroup.

§4

The order theorem under a group

stream_consumer.ex · the PEL exception

New entries arrive in mint order — the writer mints monotone EVT ids, so XREADGROUP ... > preserves the writer's order theorem. A re-claimed entry does not: recovered by XAUTOCLAIM or a PEL drain after newer entries were already handled, it returns with a branded id older (lower) than entries already handled. This is the irreducible cost of at-least-once — exactly-once is not claimed — and it makes the idempotence rule non-negotiable. Handling the same entry twice, or an older entry after a newer one, must be safe; the branded id is the dedup key, the BCS newer-wins discipline.

the pattern

Read with XREADGROUP, acknowledge with XACK, drain the PEL on restart, and reclaim a dead peer's entries with XAUTOCLAIM — at-least-once, so handlers must be idempotent because a re-claim can arrive out of order.

its EchoMQ application

EchoMQ.StreamConsumer is that reader law: the loop drains its own PEL, reads >, and reclaims peers on every beat; the handler map carries attempts for a poison threshold, and the branded EVT id is the dedup key.

The writer's order survives for new entries and breaks for re-claims, by design. Exactly-once is not claimed; idempotence on the branded id is what makes at-least-once safe.

§5

The pattern, applied

echo/apps/codemojex

A codemojex activity feed (echo/apps/codemojex) joins a game's stream as a consumer group. Several feed renderers can share the group, each draining its own backlog on restart and reclaiming a crashed renderer's entries on the next beat — the timeline is rendered at-least-once across all of them, and because each event is keyed by its branded EVT id, a re-claimed event re-rendered after a newer one is a no-op rather than a double line.

/echomq/bus

References

sources · related

Sources

Valkey — XREADGROUP — the group read, the > cursor for new entries, and the 0 cursor that drains a consumer's pending list.
Valkey — XACK — the acknowledgement that retires an entry from the pending list.
Valkey — XAUTOCLAIM — the atomic reclaim of entries idle past a threshold, the dead-peer recovery.
Valkey — XPENDING — the per-entry delivery count the attempts field carries, and the idle time the reclaim reads.

Related in this course

R5.02 · Stream consumer patternsthe module hub
R5.02.1 · The blocking readthe blocking > read this loop parks on
R5.02.3 · MAXLEN trimmingretention on the log the group reads
/echomq/busthe EchoMQ Bus pillar, the reader law in depth
/bcs/busthe manuscript bus chapter the figures are drawn from