Rheo

v0.7.1 — Durable consumer-group semantics over searchable databases. Backends today: MongoDB, PostgreSQL / SQLite (via a host-owned Ecto.Repo), and ETS (ephemeral, zero-infra). Rheo is an Elixir/OTP library you embed in your supervision tree, not a standalone messaging server. Consume with Rheo.Consumer or feed a Broadway pipeline with Rheo.Producer.

Delivery guarantee: at-least-once. Duplicates are possible after failures — use stable event IDs for idempotency. Ordering is guaranteed within a partition only (not globally across partitions).

When to use

Use Rheo when you want:

• An immutable, queryable event log in a database you already run
• Consumer groups with leases, ACK, retry, and dead-lettering
• Competing consumers and independent groups on the same stream
• Partitions with key routing, a contiguous ACK frontier, and Rheo.lag/3
• SQL backends via a host-owned Ecto.Repo (PostgreSQL or SQLite)
• A durable source for Broadway/GenStage instead of a second broker
• OTP-native demand, concurrency, and lease renewal — without standing up Kafka, RabbitMQ, or a separate broker cluster

Skip Rheo when you need a dedicated broker (massive fan-out, cross-language clients, exactly-once claims) or when a simple job queue is enough.

Compared with other approaches

Databases already store and search historical events well. Message brokers already coordinate consumers well. Rheo combines those strengths: immutable, queryable events in a database you run (MongoDB or PostgreSQL/SQLite), with leases, acknowledgement, retry, and competing consumers in OTP.

Installation

Add Rheo to your mix.exs dependencies:

def deps do
  [
    {:rheo, "~> 0.7.0"}
  ]
end

Then fetch deps:

mix deps.get

Choose a backend:

# Zero-infra (tests, Livebook, ephemeral apps)
{Rheo, name: MyRheo, backend: Rheo.Backend.ETS}

# Durable MongoDB
{Rheo, name: MyRheo, backend: {Rheo.Backend.Mongo, url: "mongodb://localhost:27017/rheo"}}

# Durable SQL on a repo your app already supervises (PostgreSQL or SQLite)
{Rheo, name: MyRheo, backend: {Rheo.Backend.Ecto, repo: MyApp.Repo}}

The Ecto backend needs the driver your repo uses — {:postgrex, "~> 0.19"} or {:ecto_sqlite3, "~> 0.17"} — since Rheo leaves that choice to you.

Quick start

ETS (no Docker)

children = [
  {Rheo, name: MyRheo, backend: Rheo.Backend.ETS},
  {MyApp.RiskConsumer, rheo: MyRheo, concurrency: 8, max_demand: 100}
]

Supervisor.start_link(children, strategy: :one_for_one)

MongoDB

children = [
  {Rheo, name: MyRheo, backend: {Rheo.Backend.Mongo, url: "mongodb://localhost:27017/rheo"}},
  {MyApp.RiskConsumer, rheo: MyRheo, concurrency: 8, max_demand: 100}
]

Supervisor.start_link(children, strategy: :one_for_one)

PostgreSQL or SQLite (Ecto)

Your app owns the repo; Rheo borrows it and never starts the pool:

children = [
  MyApp.Repo,
  {Rheo, name: MyRheo, backend: {Rheo.Backend.Ecto, repo: MyApp.Repo}},
  {MyApp.RiskConsumer, rheo: MyRheo, concurrency: 8, max_demand: 100}
]

Supervisor.start_link(children, strategy: :one_for_one)

Create the five rheo_* tables with a migration (preferred in production, so it runs once under your release's migration step):

mix rheo.ecto.gen_migration --repo MyApp.Repo
mix ecto.migrate

Rheo.ensure_indexes(rheo: MyRheo) also creates them idempotently on boot. Pass notify: true for a PostgreSQL NOTIFY rheo_events wakeup hint, or prefix: "rheo" to keep the tables in their own schema. On PostgreSQL, metadata and payload are jsonb, so the event log stays queryable in plain SQL. See ADR 017.

Define a consumer — handlers only implement handle_event/2; a local Rheo.Group owns fetch, concurrency, lease renewal, and settle:

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

  @impl true
  def handle_event(event, state) do
    case Risk.process(event) do
      :ok ->
        {:ack, state}

      {:temporary_error, reason} ->
        {:retry, reason, state}

      {:permanent_error, reason} ->
        {:reject, reason, state}
    end
  end
end

Create a stream, append events, and query history:

Rheo.create_stream("market-events")
Rheo.create_group("market-events", "risk")

Rheo.append("market-events", %{
  type: "curve_update",
  currency: "EUR",
  price: 2.913
})

Rheo.query("market-events", type: "curve_update", currency: "EUR")

Interactive walkthrough: open the Livebook demo in Livebook (or browse it on HexDocs). The notebook defaults to ETS (no Docker). CLI demo: mix rheo.demo (ETS) or RHEO_BACKEND=mongo mix rheo.demo.

Upgrading:

• 0.6 → 0.7 (additive — Broadway/GenStage interop)
• 0.5 → 0.6 (additive — Ecto SQL backend)
• 0.4 → 0.5 (partitions, frontier, lag)
• 0.3 → 0.4 (additive — search, replay, lineage)
• 0.1 → 0.2 (breaking Group / Query changes)

Documentation

Links below use HexDocs (and GitHub for the Livebook source). Relative docs/… paths break on hex.pm because those files are not in the Hex tarball.

Guides

• Introduction: Quick Start · Configuration · Consumer Groups · Enqueuing · Dequeuing
• Advanced: Replay · Querying · Partitions and lag · Broadway · GenStage · Building your own backend
• Cookbook: ETS · Mongo · Using Ecto
• Livebook demo (HexDocs)
• Changelog

Migrating from previous versions

• 0.6 → 0.7 · 0.5 → 0.6 · 0.4 → 0.5
• 0.3 → 0.4 · 0.1 → 0.2

Design

• Architecture: Architecture · Diagrams · Roadmap
• ADRs · Tutorials index
• Article 12: ACKs Are Not a Cursor
• Article 13: One Consumer API, PostgreSQL and SQLite
• Article 14: Rheo Is Not Broadway — It Feeds Broadway
• ADR 017: Ecto SQL backend
• ADR 018: GenStage / Broadway interop

More examples

Low-level fetch / ACK

{:ok, leases} = Rheo.fetch("market-events", "risk", limit: 10, rheo: MyRheo)

Enum.each(leases, fn lease ->
  # process lease.event
  Rheo.ack(lease, rheo: MyRheo)
end)

Portable queries

Rheo.query("market-events",
  type: "curve_update",
  after_sequence: 100,
  order_by: [sequence: :desc],
  limit: 50
)

{:ok, page} = Rheo.query_page("market-events", type: "curve_update", limit: 100)
Rheo.stream_query("market-events", type: "curve_update", limit: 100) |> Enum.take(250)

Replay without copying events

# Safest: new group from a cursor
Rheo.create_group("market-events", "risk-replay", start_after: 1_000)

# Or reopen an existing group (duplicates expected)
Rheo.replay("market-events", "risk", from_sequence: 1_000)
Rheo.reset_group("market-events", "risk", confirm: true)

Event lineage metadata

meta =
  Rheo.Event.Lineage.put(%{},
    correlation_id: "trade-42",
    causation_id: "cmd-9",
    producer: "pricing-v3",
    schema: "curve_update",
    schema_version: "1"
  )

Rheo.append("market-events", %{type: "curve_update", currency: "EUR", metadata: meta})
Rheo.query("market-events", correlation_id: "trade-42")

Partitions, frontier, and lag

Sequences are monotonic per partition. Append with a :key (or explicit :partition); ordering across partitions is undefined. Progress is a contiguous committed frontier — ACKs with holes do not advance lag (see Article 12).

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

Rheo.append("market-events", %{type: "curve_update", key: "EUR-EURIBOR-6M", price: 2.9})

{:ok, lag} = Rheo.lag("market-events", "risk")
# lag.partitions[p] => %{frontier: …, high_watermark: …, lag: …}
# lag.lag => sum of per-partition lags

# Static ownership (no automatic rebalance):
{MyApp.RiskConsumer, partitions: [0, 1], concurrency: 4}

Rheo.replay("market-events", "risk", from_sequence: 0, partition: 1)

Competing consumers and independent groups

Multiple processes can compete for the same durable group. Separate groups on the same stream process every event independently (e.g. "risk" and "surveillance").

Rheo.create_group("market-events", "risk")
Rheo.create_group("market-events", "surveillance")

Lease renewal and concurrency

Rheo.Group renews inflight leases (~half of lease_ms) and runs up to :concurrency handler tasks while bounding outstanding leases with :max_demand.

{MyApp.RiskConsumer,
 rheo: MyRheo,
 concurrency: 8,
 max_demand: 100,
 lease_ms: 30_000,
 poll_ms: 200}

Broadway pipeline (or plain GenStage)

Rheo.Producer is a GenStage producer that turns demand into Rheo.fetch/3 and emits %Rheo.Lease{}. Under Broadway, Rheo.Broadway.transform/2 wraps each lease into a %Broadway.Message{} whose acknowledger settles it — ACK on success, NACK (or reject) on failure.

defmodule MyApp.RiskBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module:
          {Rheo.Producer,
           rheo: MyRheo, stream: "market-events", group: "risk", max_demand: 50},
        transformer: {Rheo.Broadway, :transform, []},
        concurrency: 1
      ],
      processors: [default: [concurrency: 8]]
    )
  end

  @impl true
  def handle_message(_processor, message, _context) do
    case Risk.process(message.data) do
      :ok -> message
      {:error, reason} -> Broadway.Message.failed(message, reason)
    end
  end
end

message.data is the %Rheo.Event{}; message.metadata carries :lease, :stream, :group, :partition, and :attempt. :max_demand bounds unsettled leases, while Broadway's :concurrency bounds pipeline work — they are separate knobs.

Pick one surface per {rheo, stream, group}: Rheo.Consumer for the OTP handler API, Rheo.Producer when you want Broadway's batching, rate limiting, or fan-out. Plain GenStage consumers can handle leases directly, settling with Rheo.ack/2 and then Rheo.Producer.confirm/2. See Article 14 and ADR 018.

Named instances

Run more than one Rheo instance in the same BEAM:

children = [
  {Rheo, name: MyRheo, backend: {Rheo.Backend.Mongo, url: primary_url}},
  {Rheo, name: MyRheoAudit, backend: {Rheo.Backend.Mongo, url: audit_url}}
]

Pass rheo: MyRheo (or rheo: MyRheoAudit) on APIs and consumers.

Roadmap

Still out of scope through 1.0 unless demand forces it: standalone Rheo server, exactly-once claims, K8s operator, auth frameworks, multi-tenancy. Details in the roadmap.

License

MIT — see LICENSE.

Building and developing the library

For contributors working on Rheo itself (not application consumers):

mix deps.get
mix test
mix rheo.demo
# Mongo demo:
docker compose up -d && RHEO_BACKEND=mongo mix rheo.demo

Quality gates:

mix format --check-formatted
mix compile --warnings-as-errors
mix credo --strict
mix dialyzer
mix coveralls

Mongo-backed tests run by default when Mongo is available. Heavier end-to-end scenarios are tagged :integration and excluded unless enabled:

RHEO_INTEGRATION=1 mix test
# or
mix test.integration

Unit-only (excludes Mongo):

mix test.unit

The Ecto backend contract suite runs on SQLite on every mix test (a throwaway temp database, no service needed). PostgreSQL cases are tagged :ecto_postgres and run only when a URL is set:

docker compose up -d postgres
RHEO_POSTGRES_URL=ecto://postgres:postgres@localhost:5432/rheo_test mix test

Livebook from a clone (ETS by default — no Docker):

livebook server notebooks/rheo_demo.livemd

CI tests Erlang/OTP 27–29 × Elixir 1.17–1.20 (excluding unsupported pairs). Format, Credo, Dialyzer, and Coveralls run on Elixir 1.20.2 / OTP 29.

Coverage publishes to Coveralls from CI via mix coveralls.github. Hex releases publish from annotated tags (v*); set repository secrets COVERALLS_REPO_TOKEN and HEX_API_KEY.