The consumer group · dive 03

At-least-once & the handler

Ack on success, leave it on failure — and accept that some entries arrive twice, so make the handler idempotent.

A consumer group makes one delivery guarantee and refuses to overstate it: at-least-once. Every entry is delivered at least once; under crash and reclaim, some are delivered more than once. Exactly-once is not claimed. That single honest choice shapes the handler, the ack, and the one rule the caller must obey — be idempotent. The handler is the exact mirror of the job consumer's: :ok acks, {:error, reason} or a raise leaves the entry un-acked to be redelivered. Pick a verdict on the right to see what the consumer does to the entry.

The exact-mirror handler Ack, or leave it The order-theorem exception Pattern & implementation References

The handler verdict · ack or leave

Pick a verdict below to see what the consumer does to the entry.

The handler is the exact mirror of the job consumer's

A stream consumer's handler has the same shape as the job Consumer's handler — by design, so one portable handler works across both job and stream: fun(%{id, payload, attempts, group}) :: :ok | {:error, reason}.

The fields carry the whole delivery: id is the branded record id — the EchoMQ.Stream.append/4 receipt from the stream-log module, recovered from the entry's reserved id field, so the same 14-byte id the writer minted is the one the handler sees; payload is the remaining fields as a map; group is the consumer-group name; and attempts is the XPENDING per-entry delivery-count — how many times this entry has been delivered to a consumer. A fresh > delivery is 1; a re-claim increments it.

The attempts distinction is load-bearing and specced, not assumed: it counts deliveries, not handler failures. A poison entry — one that crashes every handler that touches it — has its delivery-count climb each time it is re-claimed, so a poison threshold attempts >= N calibrates correctly against how many times we've tried to deliver this, exactly the quantity you want to threshold on.

# echo_mq — EchoMQ.StreamConsumer
# Settle ONE entry: the branded id is the stored "id" field (the Stream.append/4 receipt);
# the payload is the remaining fields; `attempts` is the XPENDING per-entry delivery-count
# (NOT a handler-failure count). A raising handler converts to {:error, reason} and the
# loop SURVIVES. On :ok -> XACK; on {:error, _} -> LEAVE un-acked (it survives in the PEL).
defp settle(s, [xadd_id, kv]) when is_list(kv) do
  {branded, payload} = parse_fields(kv)
  attempts = delivery_count(s, xadd_id)

  verdict =
    try do
      s.handler.(%{id: branded, payload: payload, attempts: attempts, group: s.group})
    rescue
      e -> {:error, Exception.message(e)}
    catch
      :exit, reason -> {:error, "exit: " <> inspect(reason)}
      :throw, value -> {:error, "throw: " <> inspect(value)}
    end

  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 by the XAUTOCLAIM
      # beat or the next PEL-drain (the at-least-once posture).
      :ok
  end
end

One portable handler across job and stream: the same :ok | {:error, reason} contract, the same branded id, the same delivery-count — the only difference is the claim path that fed it and the settle that follows.

Ack, or leave it

The verdict drives one decision: ack, or don't. On :ok the entry is XACKed, retiring it from the PEL — it is done, no consumer will see it again. On {:error, reason} or a raise it is left un-acked, surviving in the PEL to be redelivered. Pick a verdict and see what happens to the entry, and to the loop. The model is the real settle/2 case above.

A verdict · the entry's fate

A bug in one handler call costs that entry a re-delivery, not the whole consumer: the try/rescue/catch converts any raise, exit, or throw into a typed {:error, reason}, and the loop survives to handle the next entry. The delivery_count read is itself defensive — if the XPENDING row is absent (a peer already acked it), it defaults to 1 rather than minting on nil.

# echo_mq — EchoMQ.StreamConsumer
# The XPENDING per-entry delivery-count: XPENDING <key> <group> <id> <id> 1 answers
# [[id, consumer, idle, delivery_count]]; the 4th field is how many times THIS entry has
# been delivered. A fresh `>` delivery is 1; a re-claim increments it. Defaults to 1 if
# the row is absent (a race — the entry already acked by a peer), never on nil.
defp delivery_count(s, xadd_id) do
  case Connector.command(s.conn, ["XPENDING", s.key, s.group, xadd_id, xadd_id, "1"]) do
    {:ok, [[^xadd_id, _consumer, _idle, count] | _]} -> to_int(count)
    _ -> 1
  end
end

The order-theorem PEL exception

The stream-log module proved a theorem: stream order == id sort == mint order. Under a group, that theorem holds for new entries — XREADGROUP … > hands them in mint order, the writer's guarantee untouched. But a re-claimed entry breaks real-time delivery order: it was minted before entries already handled, so when XAUTOCLAIM (or a PEL-drain) returns it, its branded id is older — lower — than ids the handler has already seen. A newer entry was delivered; then an older one arrives.

This is the irreducible cost of at-least-once. There is no way to have crash-recovery and perfect real-time ordering on a distributed read — recovering a dead peer's old work necessarily replays it after newer work moved on. The writer's theorem is not violated (the log is still id-ordered); what a re-claim gives up is the delivery being in that order. Step the timeline below to watch a re-claimed entry arrive out of order, and read why the branded id resolves it.

A delivery timeline · step the re-claim

So the rule the handler must obey: be idempotent. 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 applied at the read side: a handler that records “highest id I've applied” can recognize and skip a stale replay, the same way EchoStore.Coherence decides a cache write by comparing branded ids. The id that was the writer's receipt becomes the reader's dedup key.

This module reads the live tail. Two surfaces extend it, named here and built later: time-travel (EchoMQ.Stream.read_window/6 / read_since/5 read the log by a mint instant — a DateTime becomes an id range bound), and retention & the archive (EchoMQ.Stream.trim/4 bounds the log, and EchoStore.StreamArchive.fold/3 folds what it trims into the durable Graft floor — deep history without resident memory). The archive is the door to Echo Persistence.

Pattern & implementation

the pattern (Redis Patterns Applied)

A consumer group delivers at-least-once; XACK retires a handled entry; an unacked entry is re-delivered; consumers must be idempotent. Streams & Events teaches at-least-once delivery.

the implementation (echo_mq)

The exact-mirror handler returns :ok (ack) or {:error, _} (leave), a raise converts to an error and the loop survives, attempts is the delivery-count so a poison threshold calibrates, and a re-claimed entry returns out of real-time order — which is why the branded id is the idempotency dedup key.

The pattern says: deliver at-least-once, so be idempotent. The implementation hands you the dedup key in the delivery itself — the branded id that was the writer's receipt is the reader's newer-wins key.

Recap — honest delivery, idempotent handler

The handler is the exact mirror of the job consumer's: :ok acks, an error or a raise leaves the entry un-acked, and the loop survives a raise. attempts is the per-entry delivery-count, so a poison threshold calibrates. Delivery is at-least-once, so a re-claimed entry can arrive out of real-time order — the irreducible cost — which makes the handler's one obligation idempotency, with the branded id the dedup key. That completes the consumer group; the rest of the Bus reads the same log by a mint instant and folds what it trims to disk.

Next in the pillar: Time-travel — reading the log by a mint instant — and retention & the archive, the door to Echo Persistence.

References

Where the at-least-once handler and the order-theorem exception are grounded, and the delivery semantics they rest on.

Sources

Valkey — XACK — the acknowledgement that retires an entry from the PEL.
Valkey — XPENDING — the per-entry delivery-count attempts carries.
Valkey — Introduction to Streams — at-least-once delivery and the PEL.
Lamport — Time, Clocks, and the Ordering of Events — the total order a re-claim necessarily disturbs.

Related in this course

The consumer group — the module this dive belongs to.
Recover self, then peers — the loop that re-delivers an un-acked entry.
The order theorem — the writer's theorem the PEL exception bends.
Echo Persistence — the durable floor a trimmed stream history folds into.
The Branded Component System · the Bus — the manuscript chapter this module realizes.