Consumer Groups

A consumer group is named progress over a stream: leases, retries, dead-letters, and a contiguous ACK frontier — separate from the immutable event log. Many groups can read the same events independently.

Create a group

:ok = Rheo.create_stream("market-events", partition_count: 4)
:ok = Rheo.create_group("market-events", "risk")
:ok = Rheo.create_group("market-events", "surveillance")

Optional start cursor (preferred replay path):

:ok = Rheo.create_group("market-events", "risk-replay", start_after: 1_000)

Competing consumers (same group)

Workers that share {stream, group} compete for leases. Each event is offered to one worker at a time (until lease expiry / nack).

{:ok, a} = Rheo.fetch("market-events", "risk", limit: 10, consumer_id: "risk-1")
{:ok, b} = Rheo.fetch("market-events", "risk", limit: 10, consumer_id: "risk-2")
# a and b claim disjoint available work

Independent groups

Different group names progress separately on the same stream:

{:ok, risk} = Rheo.fetch("market-events", "risk", limit: 5)
{:ok, surv} = Rheo.fetch("market-events", "surveillance", limit: 5)
# Same event ids can appear in both

OTP surface

Prefer Rheo.Consumer so a local Rheo.Group owns demand, concurrency, and lease renewal:

defmodule MyApp.RiskConsumer do
  use Rheo.Consumer, stream: "market-events", group: "risk"

  @impl true
  def handle_event(event, _context) do
    case Risk.process(event) do
      :ok -> :ack
      {:temporary, reason} -> {:retry, reason}
      {:permanent, reason} -> {:reject, reason}
    end
  end
end

Alternatively feed Broadway / GenStage with Rheo.Producer — still one group, different runtime. Do not run both a Consumer and a Producer on the same {rheo, stream, group} unless you want competing consumers by design.

Partitions and lag

With partition_count > 1, sequences and ordering are per partition. Group progress is a contiguous frontier; holes block lag advancement:

{:ok, lag} = Rheo.lag("market-events", "risk")
lag.lag
lag.partitions[0].frontier

See Dequeuing and the tutorial ACKs are not a cursor.