R5.05 · Streams & Events — the chapter workshop
One log, a live feed.
The chapter capstone: a live codemojex activity feed on EchoMQ’s real shipped Stream Tier. Every game action is appended to one log, the feed is a fold of that log, a consumer group delivers without replaying, and what is trimmed folds to a durable floor.
Grounding: echo/apps/echo_mq (EchoMQ.Stream.append/4, read/6, read_window/6, trim/4 · EchoMQ.StreamConsumer · EchoMQ.StreamRetention) + echo/apps/echo_store (EchoStore.StreamArchive.fold/3) · echo/apps/codemojex (Codemojex.Guesses.submit/3, Codemojex.ScoreWorker, Codemojex.Settle.close/1) — every surface verified on disk. The engine is Valkey.
EVT entry.EVT entry on the log.The feed is the log seen sideways: each codemojex action is one immutable
EVT-branded entry, appended in mint order, and the rendered feed is a fold of that log —
not a row a worker overwrites.
Stage 1 — append the game’s events to the log
codemojex/game.ex · echo_mq/stream.exA codemojex room is a sequence of moves, and each one is a real surface in
echo/apps/codemojex, every entity carrying its own branded namespace:
USR— a player account;RMMa room;RNDa round, one secret to guess.JOB— a guess job on the player’s lane;GESa scored guess written by the authority.EVT— a feed entry, one brand per stream, minted host-side by the writer.
The three actions the feed records: a guess submitted —
Codemojex.Guesses.submit/3 validates the guess, charges the room’s currency, and enqueues
a branded JOB on the player’s lane; a guess scored —
Codemojex.ScoreWorker drains that lane, scores with the pure
Codemojex.Scoring.score/2, writes a GES guess, and for a classic game publishes a
scored event via EchoMQ.Events.publish/5; and a round settled
— Codemojex.Settle.close/1 enqueues a settle JOB whose consumer runs
Codemojex.Rooms.close_game/1 and pays out.
The feed appends one entry per action. EchoMQ.Stream.append(conn, "cm", "events", fields)
mints an EVT-branded record id host-side, derives the explicit XADD id from it by
field correspondence, and issues XADD emq:{cm}:stream:events with the branded id stored under
the id field. It returns {:ok, branded} — the branded id is
the receipt. The writer owns the mint, so there is nothing to spoof, and because successive mints over one
writer’s monotone cell strictly increase, the next id always exceeds the stream top: an XADD
is never rejected on a single writer, and a wrong-kind id raises before any wire.
# Codemojex.Guesses.submit/3 — the play API (echo/apps/codemojex/lib/codemojex/game.ex) job = EchoData.BrandedId.generate!("JOB") Lanes.enqueue(Bus.conn(), "cm", player, job, payload) # a guess JOB on the player's lane # the feed appends each action to ONE log — EchoMQ.Stream.append/4 EchoMQ.Stream.append(conn, "cm", "events", [{"kind", "guess"}, {"room", "RMM0…"}, {"player", "alex"}]) #=> {:ok, "EVT0…"} the branded EVT id IS the receipt; XADD emq:{cm}:stream:events
codemojex already produces the three actions; the workshop appends each as one immutable
EVT entry. The append order is the truth order — nothing in the feed is a value held in
place.
Stage 2 — fold the log to render the feed
echo_mq/stream.ex · read/6 · read_window/6To show the feed, read the log and fold it. EchoMQ.Stream.read(conn, "cm", "events", "-", "+")
issues XRANGE emq:{cm}:stream:events - + and parses each entry back into
{branded, fields_map} tuples in mint order — the branded EVT
id recovered from the stored id field, the remaining pairs the payload. The feed is that list
reduced to rendered lines: a fold, not a row a worker overwrites, so the same log always renders the same
feed. A read moves a cursor, it does not remove an entry, so a reader at the tail and a reader replaying from
the start see the same entries.
A windowed view — “the activity of the last five minutes” — is a fold scoped to a
slice. Because a stream entry id carries the millisecond it was minted, a DateTime becomes a
range bound: read_window(conn, "cm", "events", t0, t1) computes
from = minid_floor(t0) and to = maxid_ceil(t1) host-side and reads
[t0, t1] inclusive both edges; read_since(conn, "cm", "events", t0) reads
[t0, ∞). Time as a range bound is the cursor of R5.01 reapplied — a t0
entry reads back, a t0 - 1ms entry does not.
# fold the whole log — EchoMQ.Stream.read/6 (XRANGE, mint order) EchoMQ.Stream.read(conn, "cm", "events", "-", "+") #=> {:ok, [{"EVT0…a", %{"kind" => "guess", "player" => "alex"}}, # {"EVT0…b", %{"kind" => "scored", "player" => "alex", "pct" => "80"}}, # {"EVT0…c", %{"kind" => "settled", "winner" => "alex"}}]} in mint order EchoMQ.Stream.read_window(conn, "cm", "events", five_min_ago, now) # last 5 min, a windowed fold
The visible feed is read/6 over the log, reduced to lines; a windowed feed is
read_window/6 over a mint-instant slice. The fold renders the present from the past, the same way
every time.
Stage 3 — consume reliably with a group
echo_mq/stream_consumer.ex · the reader lawThe visible feed is a fold of the log; the notification side is a consumer. When the feed posts
“alex cracked the code”, a notification reader must deliver that to the room — once, and not
again on a restart. That is the reader law: EchoMQ.StreamConsumer is a BEAM consumer group over
the events stream that reads new entries with XREADGROUP, acks each with XACK, and
resumes where it left off. A reader that crashes and restarts with the same name drains its own un-acked
backlog first (its PEL, XREADGROUP … 0), then switches to new entries (>),
so it recovers exactly what it held; a clean cold start has an empty backlog. A dead peer’s held work is
reclaimed by the XAUTOCLAIM beat after an idle threshold.
The handler shape is fun(%{id, payload, attempts, group}) — id the branded
EVT receipt, payload the entry’s fields, attempts the per-entry
delivery count. The posture is at-least-once: on :ok the entry is XACKed; on
{:error, reason} or a raise it is left un-acked and re-delivered. Exactly-once is not claimed, so
a re-claimed entry can arrive after a newer one — its branded id is older. The consequence is the one
rule the handler must keep: it must be idempotent, and the branded EVT id is
the dedup key. Render the line for an EVT id once, record it was sent, and a second delivery of
the same id is a no-op. Resume, not replay.
# the notification side — EchoMQ.StreamConsumer (XREADGROUP/XACK), idempotent on the EVT id EchoMQ.StreamConsumer.start_link( queue: "cm", stream: "events", group: "feed-notify", consumer: "node-1", group_start: :new, # :new -> $ (tail), declared, no default handler: fn %{id: evt_id, payload: p, group: _} -> if not already_sent?(evt_id) do # idempotent on the EVT id — the dedup key Codemojex.Notifier.notify(p["chat"], render(p)) mark_sent(evt_id) end :ok # :ok -> XACK; {:error,_} -> left un-acked, re-delivered end )
emq:{cm}:stream:events log.Static verdicts — 1 append → XADD (EVT receipt) · 2 fold
→ XRANGE (mint order) · 3 consume → XREADGROUP/XACK
(resume, idempotent) · 4 bound → XTRIM MAXLEN ~ then fold to Graft.
Four stream techniques over one log: the append-only log (R5.01), the fold and the windowed projection (R5.01 + R5.04), the consumer group and the reader law (R5.02), and retention with the archive (R5.02). Each is a different reading of one append-only structure.
EchoMQ.Stream.append/4 records each game action, read/6 folds the feed,
EchoMQ.StreamConsumer drives the notification side without replaying, and
trim/4 under EchoMQ.StreamRetention bounds the live log while
EchoStore.StreamArchive.fold/3 keeps the deep history — all over
emq:{cm}:stream:events.
The chapter’s four techniques are one log read four ways. The same entries are appended, folded, consumed, and bounded — the structure never changes, only what each stage asks of it.
Stage 4 — bound the live log, keep the deep history
echo_mq/stream_retention.ex · echo_store/stream_archive.exA log that only grows is a leak, so retention is a policy, not a default.
EchoMQ.Stream.trim(conn, "cm", "events", {:maxlen, 10_000, true}) issues
XTRIM emq:{cm}:stream:events MAXLEN ~ 10000 — keep the ten thousand newest feed entries,
remove the older. The ~ approximate form trims in whole macro-nodes: it may under-trim but can
never over-trim, so it never deletes inside the window — the safe default. A {:minid, dt, true}
window trims by age instead, removing entries minted before dt. The named, opt-in
EchoMQ.StreamRetention driver re-applies a declared :policy on a beat through that
same public verb; a stream nobody declares is never silently trimmed, and a manual trim/4 call
is the equally-supported cadence.
What is trimmed is not lost. EchoStore.StreamArchive.fold/3 folds a mint-ordered slice of the
trimmed segment into the native EchoStore.Graft engine’s CubDB — one page per
record, in a reserved high page range disjoint from business pages — and advances a watermark
W (the branded EVT id of the highest-folded record). The deep feed is then readable
beside the live tail: a merge-read splits on W, records at or below it coming from the archive,
records above it from the live stream. The fold runs before the trim, so no entry is dropped
before it is durable. With a remote configured, the engine replicates the CubDB pages off-box to Tigris
behind a create-only commit fence — the durability dial taught end to end, from the volatile bus to the
durable floor.
Notes on ValkeyEvery stream key the feed touches carries the
queue’s {cm} hash tag, so all of codemojex’s stream keys hash (CRC16 of the brace
bytes) to one of the 16384 cluster slots and co-locate — the stream and its archive cursor on one slot,
never a CROSSSLOT error, which is why the keyspace is braced by design —
valkey.io/topics/cluster-spec.
The retention-and-archive cycle, as built
Retention bounds the live log; the fold runs first so the deep history is durable before anything is removed. The shape, verbatim from the as-built Stream Tier:
append/4 XADD emq:{cm}:stream:events <id> id <EVT> <fields> ← the branded receipt
read/6 XRANGE emq:{cm}:stream:events - + ← the fold, mint order
archive EchoStore.StreamArchive.fold/3 → Graft CubDB ← fold BEFORE trim
trim/4 XTRIM emq:{cm}:stream:events MAXLEN ~ 10000 ← bound the live log
The Bus pillar in depth
The broadcast and the retained, replayable log: consumer groups, time-travel reads, and the archive taught rung by rung in the EchoMQ Bus pillar.
/echo-persistenceThe durable floor
The archive a trimmed feed segment reaches past the bus: the durability dial, the CubDB page tier, the Tigris remote behind a create-only commit fence.
/bcs/busB3 · The Bus
The manuscript bus chapter the Stream Tier figures are drawn from: the keyspace as the contract, jobs and lanes, and the Stream Tier beside the queue.
The activity feed is the log seen sideways: identity orders the entries, the boundary owns them, and what survives a trim is a choice the system makes — not an accident of volume.
References
sources · relatedSources
- Valkey — Introduction to Streams — the append-only log, entry ids, consumer groups, and the range read the whole feed is built on.
- Valkey — XADD — the append behind
EchoMQ.Stream.append/4; the entry id the feed reads back as its receipt. - Valkey — XRANGE — the mint-ordered read behind
read/6and the windowedread_window/6; the feed is a fold of this. - Valkey — XREADGROUP — the consumer-group read behind
EchoMQ.StreamConsumer; read, ack, and resume, never replay. - Valkey — XTRIM — the bounded trim behind
EchoMQ.Stream.trim/4; theMAXLEN ~form the retention driver applies. - Valkey — Cluster specification — the
{cm}hash tag forces the feed’s stream keys onto one of the 16384 slots so a multi-key operation stays legal. - Kreps, J. — The Log — the log as the shared abstraction beneath a stream and a feed.
Related in this course
- R5.01 · Event sourcing on Streamsthe append-only log, replay to rebuild — stages one and two
- R5.02 · Stream consumer patternsthe reader law the notification side uses
- R5.03 · Pub/Sub vs Streamswhy the feed is a stream, not a channel
- R5.04 · Custom events & projectionsthe windowed fold the feed reuses
- R5 · Streams & Eventsthe chapter this workshop closes
- R4 · Time, Delay & Prioritythe chapter before this one
- EchoMQ · the Busthe Stream Tier in depth