Stream consumer patterns — block, resume, trim, archive.

Most stream adoption failures stem from treating a stream like a list. The difference is not cosmetic: a list read is destructive and a stream read moves a cursor, so the queue state — pending versus handled — lives in consumer-group metadata, not in the data itself. A stream is an immutable log, and that one shift is what the rest of this module is built on.

EchoMQ keeps the two shapes as separate code paths, never a mode on one. The job EchoMQ.Consumer claims a job and completes it once — the work leaves the queue. The EchoMQ.StreamConsumer reads a log many readers consume at their own pace: a different claim path (XREADGROUP group read, not a lease pop) and a different settle (XACK or leave-un-acked, not complete or retry). A queue is for work claimed and completed once; a stream is the append-only log beside it.

When a consumer reads via XREADGROUP, each delivered entry enters that consumer's pending entries list — the PEL. It stays there until the consumer acknowledges it with XACK. The PEL is the per-consumer record of delivered to me, not yet settled: it is what survives a crash, and it is what XAUTOCLAIM reads to recover a dead peer.

EchoMQ rides the PEL directly. On :ok the entry is XACKed and retires; on {:error, reason} — or a raising handler, which the loop converts to an error and survives — it is left un-acked, surviving in the PEL to be re-delivered. The only way an entry leaves the PEL is a handler that returns :ok.

A consumer that only ever requests > (new entries) is the critical anti-pattern: if it crashes after a delivery but before the XACK, 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 PEL nobody drains. On each beat the loop reclaims entries idle past :min_idle_ms (default 30_000) via XAUTOCLAIM — decentralised work-stealing, no coordinator. The idle threshold is evaluated server-side, never against a host clock.

New entries arrive in mint order, so XREADGROUP ... > preserves the writer's order theorem. But a re-claimed entry returns with a branded id older than entries already handled — the irreducible cost of at-least-once. The consequence is a hard rule: the handler must be idempotent, with the branded id as the dedup key.

A log that only grows is a leak, and acknowledgement does not shrink it — XACK marks an entry handled in group metadata, but the entry stays in the log. The right tool is XTRIM, not XDEL: XDEL marks an entry deleted but frees memory only when a whole macro-node empties, leaving fragmentation. EchoMQ.Stream.trim/4 bounds the log by length ({:maxlen, count, approx?}) or age ({:minid, dt, approx?}), and the approx? flag selects ~ — the safe default: it may under-trim but can never over-trim, so a trim can never delete inside the declared window.

Retention is decoupled from consumer liveness. EchoMQ.StreamRetention is a named, opt-in driver that beats on a tick and re-applies a declared policy: a stream nobody drains still trims, because bounded memory is a safety property and a consumer is up is a liveness fact — coupling the two is a silent-no-op the design refuses. A manual trim/4 call is the equally-supported cadence.

The arc is wait cheaply → read reliably → bound the log. Read them in order — the blocking read is how a reader waits, the consumer group is how it stays reliable, and trimming is how the log stays bounded without losing history.

codemojex (echo/apps/codemojex) is the worked consumer — a Telegram emoji-guessing game on the same stack. A game round appends an entry to its activity stream on the {q} queue — a guess scored, a round opened — and a feed reader joins the consumer group to render the live timeline. When the reader restarts, it drains its own PEL first, so the feed never double-renders an event it already showed. When the stream grows past the declared window, EchoMQ.StreamRetention trims it on its beat, and EchoStore.StreamArchive folds the trimmed segments into the Graft floor — the full round history stays readable for an audit without holding it all in memory.