R5.02.1 · Stream consumer patterns — dive

The blocking read — wait once, on your own lane.

A reader of a stream has to wait for the next entry, and there are two ways to wait. A busy-poll spins on XREADGROUP with nothing to show for it; a blocking read parks once with BLOCK and wakes the instant an entry lands — one round trip, no spin.

Grounding: echo/apps/echo_mq/lib/echo_mq/stream_consumer.ex · read_group_block/1 — the XREADGROUP ... BLOCK beat_ms on the consumer's own private connector lane, the command timeout set longer than the block.

Two ways to wait for the next entry. Select a strategy to read what it costs between arrivals. The green dot is when the entry actually lands; the boxes are the reader's requests.

readout · Select a strategy. A busy-poll spends requests on empty reads; BLOCK parks server-side and returns the instant an entry lands, with round-trip latency and one parked connection.

§1

Busy-poll versus block

streams-consumer-patterns.md.txt

A busy-poll loop reads, gets an empty reply, sleeps a little, and reads again. Its latency is the sleep interval — shrink the sleep to lower latency and the request rate against an idle stream climbs; raise the sleep to lower the rate and latency climbs. There is no setting that is both responsive and cheap.

A blocking read removes the trade. XREADGROUP GROUP g <self> BLOCK <ms> COUNT <n> STREAMS <key> > parks server-side until either an entry is appended — and it returns immediately, sub-millisecond latency — or ms elapses with nothing, and it returns an empty reply so the caller can re-arm. The latency is the network round trip, not a poll interval, and an idle stream costs one parked connection rather than a stream of empty reads.

A busy-poll trades latency against request rate and loses both. A blocking read waits at the network round trip and spends nothing on an idle stream but one parked connection.

§2

The connection cost

the blocking-read pitfall

The blocking form has one hazard: a parked read holds its connection for the whole block. With a pool of ten connections and ten worker threads each issuing XREADGROUP ... BLOCK 5000, all ten connections are parked at once. Any other code that needs the store — even a simple GET — waits for a connection that will not free for five seconds. The blocking verb did not deadlock the stream; it starved everything else of connections.

The three standard mitigations all point the same way:

Dedicated connections — give each blocking consumer its own connection, separate from the shared pool.
Shorter block times — a two-to-three-second block re-arms often enough to notice a shutdown, rather than an infinite BLOCK 0 that never returns on its own.
Timeout coordination — the server-side BLOCK timeout must be shorter than the client socket timeout, or the socket gives up before the server replies.

Where the blocking read parks. Select a topology to read what happens to an unrelated GET while the consumers block. A shared pool starves; a dedicated lane does not.

readout · Select a topology. A blocking read holds its connection for the whole block — on the shared pool that starves an unrelated GET; on its own lane the shared pool stays free.

§3

The dedicated lane

stream_consumer.ex · read_group_block/1

EchoMQ takes the dedicated-connection mitigation as a law, not an option. EchoMQ.StreamConsumer holds a private connector lane and parks the blocking read on it. This is the only blocking verb in the consumer, and it rides the consumer's own lane — the same rule the job consumer follows for BLPOP — so the single-owner socket the rest of the system shares is never stalled.

echo/apps/echo_mq/lib/echo_mq/stream_consumer.ex · read_group_block/1, the blocking read on the private laneverbatim

defp read_group_block(s) do
  parts =
    ["XREADGROUP", "GROUP", s.group, s.consumer, "BLOCK", Integer.to_string(s.beat_ms),
     "COUNT", Integer.to_string(s.count), "STREAMS", s.key, ">"]

  case Connector.command(s.conn, parts, s.beat_ms + 5_000) do
    {:ok, reply} -> group_entries(reply, s.key)   ← new entries this beat
    {:error, _} -> []                              ← timeout: re-arm, nothing new
  end
end

The block time is the :beat_ms cadence (default 1_000), short enough that the loop re-arms once a second and notices a stop request between entries. The command timeout is beat_ms + 5_000 — deliberately longer than the block — so the server's BLOCK returns first with either entries or an empty reply, and the socket timeout is the backstop, never the thing that fires. A timeout returns an empty list, and the loop comes round again.

Notes on ValkeyThe > id in a group read means entries never delivered to any consumer in this group — so a blocking > read parks for genuinely new entries, while a 0 read of the same group is the non-blocking PEL replay — valkey.io/commands/xreadgroup.

§4

The pattern, applied

echo/apps/codemojex

A codemojex activity-feed reader (echo/apps/codemojex) is a long-lived consumer of a game's stream. It is idle most of the time — a round is appended only when a guess is scored or a round opens — so a busy-poll would spend almost all of its requests on empty reads. On its own lane it parks on XREADGROUP ... BLOCK 1000 and wakes the instant an entry lands, rendering the live timeline with round-trip latency, while a separate GET-the-board call elsewhere in the app never waits on the feed's parked connection.

the pattern

Wait for the next entry with a blocking XREADGROUP ... BLOCK, not a busy-poll — but give the blocking reader a dedicated connection, because a parked read holds its connection for the whole block.

its EchoMQ application

EchoMQ.StreamConsumer.read_group_block/1 parks BLOCK beat_ms on the consumer's private lane, the command timeout set to beat_ms + 5_000 so the block returns first — the shared single-owner socket is never stalled.

/echomq/bus

References

sources · related

Sources

Valkey — XREADGROUP — the BLOCK <ms> form that parks for a new entry and the > cursor that delivers only new entries.
Valkey — Introduction to Streams — blocking reads, the consumer group, and the connection a block holds.
Valkey — XREAD — the un-grouped blocking read the long-poll starts from.
Valkey — Cluster specification — the {q} hash tag that keeps a queue's stream key on one of 16384 hash slots.

Related in this course

R5.02 · Stream consumer patternsthe module hub
R5.02.2 · Consumer groupsthe group read this block rides
R5.02.3 · MAXLEN trimmingbounding the log the reader drains
/echomq/busthe EchoMQ Bus pillar, the retained log in depth