The consumer group · dive 01

The group door

Ensure the group on every start — swallow the one reply that means “already done,” raise on everything else.

A consumer group has to exist before anyone reads through it, and the same group is shared by every consumer that joins. So creating it can't be a one-time setup step — every consumer that starts (or restarts) has to ensure the group is there without caring whether it already is. That is the group door: a single XGROUP CREATE … MKSTREAM on start, with exactly one reply swallowed (BUSYGROUP — the idempotent outcome) and everything else loud. Pick a reply on the right to see how the door classifies it, then read why :group_start is declared with no default.

One verb, group and stream Classify the reply The declared start position Pattern & implementation References

The door · one create, three outcomes

Pick a reply below to classify it.

One verb — the group and the stream

XGROUP CREATE <key> <group> <start> MKSTREAM creates the consumer group, and — because of MKSTREAM — the stream key too, if it does not yet exist. A consumer can start before a single entry has ever been appended; the door does not depend on the writer having run first. That decoupling matters in a real deployment, where a reader and a writer come up in no guaranteed order.

The one reply the door swallows is BUSYGROUP — Valkey's way of saying “that group already exists.” That is not an error here: it is the idempotent outcome. The second consumer to start, and the same consumer after a restart, both hit BUSYGROUP, and both treat it as success. A restart-storm — every consumer in a group bouncing at once — never produces an error from the door.

# echo_mq — EchoMQ.StreamConsumer
# The lazy-ensure group door: XGROUP CREATE <key> <group> <start> MKSTREAM creates the
# group (and the stream, if absent). The ONLY swallowed reply is BUSYGROUP (the group
# already exists — an idempotent start). A WRONGTYPE (a non-stream key collision) or ANY
# other error_reply is LOUD: the consumer fails to start (the gate-liveness discipline).
defp ensure_group!(s, start) do
  case Connector.command(s.conn, ["XGROUP", "CREATE", s.key, s.group, start, "MKSTREAM"]) do
    {:ok, "OK"} ->
      :ok

    {:ok, {:error_reply, <<@busygroup_prefix, _::binary>>}} ->
      :ok

    {:ok, {:error_reply, msg}} ->
      raise RuntimeError, "EchoMQ.StreamConsumer: XGROUP CREATE refused (not BUSYGROUP): #{msg}"

    {:error, reason} ->
      raise RuntimeError, "EchoMQ.StreamConsumer: XGROUP CREATE failed: #{inspect(reason)}"
  end
end

The matched @busygroup_prefix is the literal "BUSYGROUP"; the connector surfaces a Valkey error as an {:error_reply, msg} value, so the door pattern-matches on the prefix rather than catching an exception.

Classify the reply

The door has exactly two passing outcomes and one failing one. Pick a reply the connector might return and see which branch of ensure_group!/2 it lands in — and whether the consumer starts. The model is the real pattern-match above, evaluated over fixed replies.

A reply · pass or fail-to-start

Only two replies are a passing start: the group was created, or it already exists. A WRONGTYPE means the key is something other than a stream — a collision the door must not paper over — so it raises, and the consumer never comes up reading nothing.

The start position is declared, never defaulted

When you create a group you make one correctness decision that can never be taken back: where does the group begin reading? Two answers — :new → $ (only entries appended after the group is created; all existing history is ignored), or :head → 0 (from the head of the stream; everything already there is replayed).

EchoMQ.StreamConsumer makes :group_start a required option with no default. A missing or malformed value raises at start — on the caller's process, before the loop ever spawns. The replay-vs-tail decision is forced into the open: there is no quiet default that silently picks one and surprises an operator later, and no second verb (XGROUP SETID) needed to fix a guess.

# echo_mq — EchoMQ.StreamConsumer
# The declared start position: :new -> "$" (only NEW entries), :head -> "0" (from the
# stream head). Any other value RAISES (never defaulted) — the replay-vs-tail correctness
# decision is forced into the open, without a second verb.
defp start_position!(:new), do: "$"
defp start_position!(:head), do: "0"

defp start_position!(other) do
  raise ArgumentError,
        "EchoMQ.StreamConsumer requires a declared :group_start of :new (-> $) or :head (-> 0); got: #{inspect(other)}"
end

start_position!/1 runs inside start_link, on the calling process — so a bad :group_start is a start-time crash the caller sees, not a silent mis-start the loop swallows.

The door creates; it never destroys. At this rung there is no destructive group-tear-down verb. The door creates and swallows-on-exists — it never deletes a group, never drops a stream. The at-rest removal surface (tearing a group down, trimming the stream out from under it) stays unfrozen for the retention and archive family, where removing data at rest is the whole point and gets the deliberate, declared treatment it deserves. A reader's door is a create door, full stop.

Pattern & implementation

the pattern (Redis Patterns Applied)

A consumer group is created once with XGROUP CREATE and shared by all consumers; the start id picks the replay point. Streams & Events teaches the group.

the implementation (echo_mq)

Every consumer ensures the group lazily on start: MKSTREAM so the stream need not pre-exist, BUSYGROUP swallowed so restarts are idempotent, everything else loud so a mis-start is never silent, and :group_start declared so the replay decision is explicit.

The pattern says: create the group once. The implementation makes “once” safe to call on every start — idempotent on the one reply that means already-done, loud on every other, and explicit about where reading begins.

Recap — ensure, don't assume

The group door is a single XGROUP CREATE … MKSTREAM on start: it creates the group and the stream, swallows only BUSYGROUP (so a restart-storm is a no-op), and raises on a WRONGTYPE or any other reply (so a mis-start is never silent). The :group_start is declared with no default, forcing the replay-vs-tail decision into the open. The next dive opens the loop the door makes possible — how a consumer recovers its own held work, then a dead peer's.

Next in the module: Recover self, then peers — the PEL-drain that recovers a restarted consumer, and the XAUTOCLAIM beat that recovers a dead one.

References

Where the group door and its declared start position are grounded, and the stream semantics they rest on.

Sources

Valkey — XGROUP — CREATE … MKSTREAM, the BUSYGROUP reply, and the start id.
Valkey — Introduction to Streams — consumer groups and where a group begins reading.
Valkey — XREADGROUP — the read the door makes possible once the group exists.
Valkey — Cluster specification — the {q} hashtag co-locates the stream key with its queue.

Related in this course

The consumer group — the module this dive belongs to.
Recover self, then peers — the loop the door opens.
The stream log — the writer whose stream the group reads.
The Protocol — the braced keyspace the stream key is born to.
redis-patterns · Streams & Events — the streams pattern that doors here.