Rheo Backends: ETS, Mongo, SQLite, PostgreSQL, Redis (v0.9.0)

Mix.install(
  [
    # From this repo: prefer the path dep. From HexDocs / standalone Livebook:
    # {:rheo, "~> 0.9.0"},
    {:rheo, path: Path.join(__DIR__, "..")},
    {:mongodb_driver, "~> 1.5"},
    {:redix, "~> 1.5"},
    {:ecto_sqlite3, "~> 0.17"},
    {:postgrex, "~> 0.19"},
    {:kino, "~> 0.14"}
  ],
  config: [
    rheo: [
      start_on_application: false,
      clock: Rheo.Clock.System,
      default_lease_ms: 5_000,
      default_max_attempts: 3
    ]
  ]
)

Intro

Same Rheo.append / fetch / ack / query / Consumer API. Backends differ in durability, distribution, and mechanisms — not in how you write handlers.

This notebook:

1. Compares capabilities side by side
2. Runs a tiny happy path on ETS, Mongo, SQLite, Postgres, and Redis (optional where services are required)
3. Proves two Rheo instances in one BEAM (ETS + SQLite) do not bleed data
4. Runs a light throughput comparison across whichever backends are up

Sibling notebooks: Quickstart · Concepts · Pipelines · Index

Capability snapshot

Guarantees are what Rheo promises through the backend. Mechanisms are how storage implements or optimizes those promises. Application code should branch on guarantees sparingly — never to weaken fencing.

cap_row = fn name, caps ->
  g = caps.guarantees
  m = caps.mechanisms

  %{
    backend: name,
    durable: g[:durable],
    distributed: g[:distributed],
    partitions: g[:partitions],
    contiguous_frontier: g[:contiguous_frontier],
    secondary_indexes: g[:secondary_indexes],
    notifications: m[:notifications],
    atomic_cas: m[:atomic_compare_and_set]
  }
end

rows =
  [{"ETS", Rheo.Backend.ETS.capabilities()}]
  |> then(fn acc ->
    if Code.ensure_loaded?(Rheo.Backend.Mongo) do
      acc ++ [{"Mongo", Rheo.Backend.Mongo.capabilities()}]
    else
      acc
    end
  end)
  |> then(fn acc ->
    if Code.ensure_loaded?(Rheo.Backend.Redis) do
      acc ++ [{"Redis", Rheo.Backend.Redis.capabilities()}]
    else
      acc
    end
  end)
  |> then(fn acc ->
    if Code.ensure_loaded?(Rheo.Backend.Ecto) do
      acc ++
        [
          {"Ecto SQLite", Rheo.Backend.Ecto.capabilities(:sqlite)},
          {"Ecto Postgres", Rheo.Backend.Ecto.capabilities(:postgres)}
        ]
    else
      acc
    end
  end)
  |> Enum.map(fn {name, caps} -> cap_row.(name, caps) end)

Kino.DataTable.new(rows)

1. ETS — always available

Named instance :ets_demo so we can run several Rheos later without fighting over the default Rheo name.

stop_rheo = fn name  ->
  case Process.whereis(name) do
    nil -> :ok
    pid -> GenServer.stop(pid)
  end
end

stop_rheo.(Rheo)
{:ok, _} = Rheo.start_link(name: :ets_demo, backend: Rheo.Backend.ETS)
:ok = Rheo.ensure_indexes(rheo: :ets_demo)

s = "ets-demo"
:ok = Rheo.create_stream(s, rheo: :ets_demo)
:ok = Rheo.create_group(s, "workers", rheo: :ets_demo)
{:ok, ev} = Rheo.append(s, %{type: "ping", n: 1}, rheo: :ets_demo)
{:ok, [lease]} = Rheo.fetch(s, "workers", limit: 1, rheo: :ets_demo)
:ok = Rheo.ack(lease, rheo: :ets_demo)

%{
  event_id: ev.id,
  receipt: lease.receipt,
  lag: elem(Rheo.lag(s, "workers", rheo: :ets_demo), 1).lag
}

Note receipt: on ETS it equals the fencing lease_id. Native-stream backends will put a different opaque token there without changing your handler.

2. MongoDB — optional

Needs a running MongoDB. From the repo root:

docker compose up -d

Override the URL with RHEO_MONGO_URL if needed. If Mongo is down, this cell returns a structured failure instead of crashing the notebook.

mongo_url = System.get_env("RHEO_MONGO_URL", "mongodb://localhost:27017/rheo_livebook")

mongo =
  cond do
    not Code.ensure_loaded?(Rheo.Backend.Mongo) ->
      %{ok?: false, hint: "add {:mongodb_driver, \"~> 1.5\"} to Mix.install and re-evaluate"}

    true ->
      case Rheo.start_link(
             name: :mongo_demo,
             backend: {Rheo.Backend.Mongo, url: mongo_url, name: MongoLivebook}
           ) do
        {:ok, _} ->
          :ok = Rheo.ensure_indexes(rheo: :mongo_demo)
          s = "mongo-demo-#{System.unique_integer([:positive])}"
          :ok = Rheo.create_stream(s, partition_count: 2, rheo: :mongo_demo)
          :ok = Rheo.create_group(s, "workers", rheo: :mongo_demo)
          {:ok, ev} = Rheo.append(s, %{type: "ping", venue: "mongo"}, rheo: :mongo_demo)
          {:ok, [lease]} = Rheo.fetch(s, "workers", limit: 1, rheo: :mongo_demo)
          :ok = Rheo.ack(lease, rheo: :mongo_demo)
          {:ok, found} = Rheo.query(s, type: "ping", limit: 5, rheo: :mongo_demo)

          %{
            ok?: true,
            url: mongo_url,
            event_id: ev.id,
            queried: length(found),
            capabilities: Rheo.Backend.Mongo.capabilities().guarantees
          }

        {:error, reason} ->
          %{ok?: false, reason: inspect(reason), hint: "docker compose up -d"}
      end
  end

Mongo shines when you want secondary indexes and rich queries on the same log you consume from.

3. Redis Streams — optional

Needs Redis 6.2+. From the repo: docker compose up -d redis.

redis_url = System.get_env("RHEO_REDIS_URL", "redis://localhost:6379")

redis =
  cond do
    not Code.ensure_loaded?(Rheo.Backend.Redis) ->
      %{ok?: false, hint: "add {:redix, \"~> 1.5\"} to Mix.install and re-evaluate"}

    true ->
      case Rheo.start_link(
             name: :redis_demo,
             backend: {Rheo.Backend.Redis, name: RedisLivebook, url: redis_url}
           ) do
        {:ok, _} ->
          :ok = Rheo.ensure_indexes(rheo: :redis_demo)
          s = "redis-demo-#{System.unique_integer([:positive])}"
          :ok = Rheo.create_stream(s, partition_count: 2, rheo: :redis_demo)
          :ok = Rheo.create_group(s, "workers", rheo: :redis_demo)
          {:ok, ev} = Rheo.append(s, %{type: "ping", venue: "redis"}, rheo: :redis_demo)
          {:ok, [lease]} = Rheo.fetch(s, "workers", limit: 1, rheo: :redis_demo)
          :ok = Rheo.ack(lease, rheo: :redis_demo)

          %{
            ok?: true,
            url: redis_url,
            event_id: ev.id,
            receipt: lease.receipt,
            receipt_differs_from_lease_id?: lease.receipt != lease.lease_id,
            capabilities: Rheo.Backend.Redis.capabilities().guarantees
          }

        {:error, reason} ->
          %{ok?: false, reason: inspect(reason), hint: "docker compose up -d redis"}
      end
  end

On Redis, receipt is the stream entry id — different from lease_id (fencing token). Handlers still ignore receipt contents.

4. SQLite via Ecto — durable, no Docker

Rheo borrows a host-owned Ecto.Repo. Your app supervises the Repo; Rheo does not own the connection pool. Rheo.ensure_indexes/1 creates the Rheo tables idempotently (or use mix rheo.ecto.gen_migration in real apps).

defmodule Backends.SqliteRepo do
  use Ecto.Repo, otp_app: :rheo, adapter: Ecto.Adapters.SQLite3
end

db = Path.join(System.tmp_dir!(), "rheo_livebook_#{System.unique_integer([:positive])}.db")

case Process.whereis(Backends.SqliteRepo) do
  nil -> :ok
  pid -> GenServer.stop(pid)
end

stop_rheo.(Rheo)

{:ok, _} =
  Backends.SqliteRepo.start_link(
    database: db,
    pool_size: 5,
    name: Backends.SqliteRepo
  )

{:ok, _} =
  Rheo.start_link(
    name: :sqlite_demo,
    backend: {Rheo.Backend.Ecto, repo: Backends.SqliteRepo}
  )

:ok = Rheo.ensure_indexes(rheo: :sqlite_demo)

s = "sqlite-demo"
:ok = Rheo.create_stream(s, partition_count: 2, rheo: :sqlite_demo)
:ok = Rheo.create_group(s, "risk", rheo: :sqlite_demo)

{:ok, ev} =
  Rheo.append(s, %{type: "curve_update", key: "EUR-1", price: 1.0}, rheo: :sqlite_demo)

{:ok, [lease]} = Rheo.fetch(s, "risk", limit: 1, rheo: :sqlite_demo)
:ok = Rheo.ack(lease, rheo: :sqlite_demo)
{:ok, lag} = Rheo.lag(s, "risk", rheo: :sqlite_demo)

%{
  guarantees: Rheo.Backend.Ecto.capabilities(:sqlite).guarantees,
  event_partition: ev.partition,
  frontier: lag.partitions[ev.partition].frontier,
  db: db
}

SQLite declares distributed: false — one writer node. Ideal for appliances and local durable demos; prefer Postgres when multiple BEAM nodes claim the same group.

5. PostgreSQL via Ecto — optional

Set an Ecto URL, for example:

export RHEO_POSTGRES_URL=ecto://postgres:postgres@localhost:5432/rheo_livebook

Same {Rheo.Backend.Ecto, repo: ...} tuple as SQLite — only the Repo adapter changes.

defmodule Backends.PostgresRepo do
  use Ecto.Repo, otp_app: :rheo, adapter: Ecto.Adapters.Postgres
end

postgres =
  case System.get_env("RHEO_POSTGRES_URL") || System.get_env("DATABASE_URL") do
    nil ->
      %{
        ok?: false,
        hint: "export RHEO_POSTGRES_URL=ecto://postgres:postgres@localhost:5432/rheo_livebook"
      }

    url ->
      case Process.whereis(Backends.PostgresRepo) do
        nil -> :ok
        pid -> GenServer.stop(pid)
      end

      stop_rheo.(Rheo)

      {:ok, _} =
        Backends.PostgresRepo.start_link(
          url: url,
          pool_size: 5,
          name: Backends.PostgresRepo
        )

      {:ok, _} =
        Rheo.start_link(
          name: :pg_demo,
          backend: {Rheo.Backend.Ecto, repo: Backends.PostgresRepo}
        )

      :ok = Rheo.ensure_indexes(rheo: :pg_demo)

      s = "pg-demo-#{System.unique_integer([:positive])}"
      :ok = Rheo.create_stream(s, partition_count: 2, rheo: :pg_demo)
      :ok = Rheo.create_group(s, "risk", rheo: :pg_demo)
      {:ok, ev} = Rheo.append(s, %{type: "ping", venue: "postgres"}, rheo: :pg_demo)
      {:ok, [lease]} = Rheo.fetch(s, "risk", limit: 1, rheo: :pg_demo)
      :ok = Rheo.ack(lease, rheo: :pg_demo)

      %{
        ok?: true,
        event_id: ev.id,
        guarantees: Rheo.Backend.Ecto.capabilities(:postgres).guarantees,
        mechanisms:
          Rheo.Backend.Ecto.capabilities(:postgres).mechanisms
          |> Enum.filter(fn {_k, v} -> v end)
          |> Map.new()
      }
  end

Postgres is the multi-node SQL path. Wakeup via NOTIFY is an optional hint — polling remains the universal fallback (ADR 025).

6. Multi-instance: ETS + SQLite side by side

Two Rheo names in one BEAM node is a supported pattern (e.g. hot market feed in ETS, durable audit log in SQLite). Same stream name must not bleed across handles.

stop_rheo.(Rheo)
stop_rheo.(:ets_demo)
stop_rheo.(:sqlite_demo)
stop_rheo.(:mongo_demo)
stop_rheo.(:pg_demo)
stop_rheo.(:redis_demo)

case Process.whereis(Backends.SqliteRepo) do
  nil -> :ok
  pid -> GenServer.stop(pid)
end

db2 = Path.join(System.tmp_dir!(), "rheo_audit_#{System.unique_integer([:positive])}.db")

{:ok, _} =
  Backends.SqliteRepo.start_link(database: db2, pool_size: 5, name: Backends.SqliteRepo)

{:ok, _} = Rheo.start_link(name: :market, backend: Rheo.Backend.ETS)
{:ok, _} =
  Rheo.start_link(name: :audit, backend: {Rheo.Backend.Ecto, repo: Backends.SqliteRepo})

:ok = Rheo.ensure_indexes(rheo: :audit)

stream = "shared-name"
:ok = Rheo.create_stream(stream, rheo: :market)
:ok = Rheo.create_stream(stream, rheo: :audit)
:ok = Rheo.create_group(stream, "traders", rheo: :market)
:ok = Rheo.create_group(stream, "compliance", rheo: :audit)

{:ok, m} = Rheo.append(stream, %{type: "fill", venue: "ets"}, rheo: :market)
{:ok, a} = Rheo.append(stream, %{type: "fill", venue: "sqlite"}, rheo: :audit)

{:ok, only_m} = Rheo.query(stream, limit: 10, rheo: :market)
{:ok, only_a} = Rheo.query(stream, limit: 10, rheo: :audit)

%{
  market_ids: Enum.map(only_m, & &1.id),
  audit_ids: Enum.map(only_a, & &1.id),
  bleed?: m.id in Enum.map(only_a, & &1.id) or a.id in Enum.map(only_m, & &1.id)
}

bleed? must be false. Groups (traders vs compliance) are also isolated per instance — creating a group on :market does not create it on :audit.

7. Throughput comparison (informal)

Same portable API, different storage cost. This is a Livebook microbench, not a formal load test: single partition, sequential calls from one BEAM, N events append then drain with fetch/ack. Results skip backends that are not running or cannot connect.

Tune bench_n if you want a longer run. Re-evaluate after starting Docker services for Mongo / Redis / Postgres.

bench_n = 200
bench_payload = %{type: "bench", i: 0}

stop_rheo = fn name ->
  case Process.whereis(name) do
    nil -> :ok
    pid -> GenServer.stop(pid)
  end
end

# Repos may already exist from sections 4–5; define them if this cell is run alone.
unless Code.ensure_loaded?(Backends.SqliteRepo) do
  defmodule Backends.SqliteRepo do
    use Ecto.Repo, otp_app: :rheo, adapter: Ecto.Adapters.SQLite3
  end
end

unless Code.ensure_loaded?(Backends.PostgresRepo) do
  defmodule Backends.PostgresRepo do
    use Ecto.Repo, otp_app: :rheo, adapter: Ecto.Adapters.Postgres
  end
end

ensure_sqlite_repo = fn ->
  case Process.whereis(Backends.SqliteRepo) do
    nil ->
      db = Path.join(System.tmp_dir!(), "rheo_bench_#{System.unique_integer([:positive])}.db")

      {:ok, _} =
        Backends.SqliteRepo.start_link(
          database: db,
          pool_size: 5,
          name: Backends.SqliteRepo
        )

      :ok

    _pid ->
      :ok
  end
end

timed = fn label, fun ->
  t0 = System.monotonic_time(:microsecond)

  try do
    case fun.() do
      :ok ->
        us = System.monotonic_time(:microsecond) - t0
        ms = Float.round(us / 1000, 1)
        ops = Float.round(bench_n * 1_000_000 / us, 1)
        %{backend: label, ok?: true, n: bench_n, ms: ms, ops_per_sec: ops}

      {:error, reason} ->
        %{backend: label, ok?: false, reason: inspect(reason), n: bench_n}
    end
  rescue
    e ->
      %{backend: label, ok?: false, reason: Exception.message(e), n: bench_n}
  end
end

drain = fn rheo, stream, group ->
  drain_loop = fn drain_loop ->
    case Rheo.fetch(stream, group, limit: 50, rheo: rheo) do
      {:ok, []} ->
        :ok

      {:ok, leases} ->
        Enum.each(leases, fn lease -> :ok = Rheo.ack(lease, rheo: rheo) end)
        drain_loop.(drain_loop)

      {:error, _} = err ->
        err
    end
  end

  drain_loop.(drain_loop)
end

run_cycle = fn rheo, stream ->
  group = "bench"
  :ok = Rheo.create_stream(stream, partition_count: 1, rheo: rheo)
  :ok = Rheo.create_group(stream, group, rheo: rheo)

  Enum.each(1..bench_n, fn i ->
    {:ok, _} = Rheo.append(stream, Map.put(bench_payload, :i, i), rheo: rheo)
  end)

  drain.(rheo, stream, group)
end

candidates = [
  {"ETS",
   fn ->
     stop_rheo.(:ets_bench)
     {:ok, _} = Rheo.start_link(name: :ets_bench, backend: Rheo.Backend.ETS)
     :ok = Rheo.ensure_indexes(rheo: :ets_bench)
     run_cycle.(:ets_bench, "bench-ets")
   end},
  {"SQLite",
   fn ->
     stop_rheo.(:sqlite_bench)
     ensure_sqlite_repo.()

     {:ok, _} =
       Rheo.start_link(
         name: :sqlite_bench,
         backend: {Rheo.Backend.Ecto, repo: Backends.SqliteRepo}
       )

     :ok = Rheo.ensure_indexes(rheo: :sqlite_bench)
     run_cycle.(:sqlite_bench, "bench-sqlite-#{System.unique_integer([:positive])}")
   end},
  {"Mongo",
   fn ->
     if not Code.ensure_loaded?(Rheo.Backend.Mongo) do
       {:error, :mongo_not_loaded}
     else
       stop_rheo.(:mongo_bench)
       url = System.get_env("RHEO_MONGO_URL", "mongodb://localhost:27017/rheo_livebook")

       case Rheo.start_link(
              name: :mongo_bench,
              backend: {Rheo.Backend.Mongo, url: url, name: MongoBench}
            ) do
         {:ok, _} ->
           :ok = Rheo.ensure_indexes(rheo: :mongo_bench)
           run_cycle.(:mongo_bench, "bench-mongo-#{System.unique_integer([:positive])}")

         {:error, reason} ->
           {:error, reason}
       end
     end
   end},
  {"Redis",
   fn ->
     if not Code.ensure_loaded?(Rheo.Backend.Redis) do
       {:error, :redis_not_loaded}
     else
       stop_rheo.(:redis_bench)
       url = System.get_env("RHEO_REDIS_URL", "redis://localhost:6379")

       case Rheo.start_link(
              name: :redis_bench,
              backend: {Rheo.Backend.Redis, name: RedisBench, url: url}
            ) do
         {:ok, _} ->
           :ok = Rheo.ensure_indexes(rheo: :redis_bench)
           run_cycle.(:redis_bench, "bench-redis-#{System.unique_integer([:positive])}")

         {:error, reason} ->
           {:error, reason}
       end
     end
   end},
  {"Postgres",
   fn ->
     case System.get_env("RHEO_POSTGRES_URL") || System.get_env("DATABASE_URL") do
       nil ->
         {:error, :postgres_url_unset}

       url ->
         stop_rheo.(:pg_bench)

         case Process.whereis(Backends.PostgresRepo) do
           nil ->
             {:ok, _} =
               Backends.PostgresRepo.start_link(
                 url: url,
                 pool_size: 5,
                 name: Backends.PostgresRepo
               )

           _pid ->
             :ok
         end

         {:ok, _} =
           Rheo.start_link(
             name: :pg_bench,
             backend: {Rheo.Backend.Ecto, repo: Backends.PostgresRepo}
           )

         :ok = Rheo.ensure_indexes(rheo: :pg_bench)
         run_cycle.(:pg_bench, "bench-pg-#{System.unique_integer([:positive])}")
     end
   end}
]

bench_rows =
  Enum.map(candidates, fn {label, fun} ->
    timed.(label, fun)
  end)

Enum.each([:ets_bench, :sqlite_bench, :mongo_bench, :redis_bench, :pg_bench], stop_rheo)

Kino.DataTable.new(bench_rows)

Expect ETS to lead (in-memory), then roughly Redis / SQLite / Postgres / Mongo depending on your machine and Docker latency. Numbers are for relative ordering in this notebook — not published SLOs.

Takeaways

• Swap backends with {Rheo, backend: …} — keep Consumer / Query / Event meaning
• Read capabilities for ops decisions; do not weaken fencing in app code
• Optional packages / deps: Mongo driver, Redix, Ecto adapters, Broadway — core stays lean
• Multiple named Rheo instances in one node are fine when handles stay distinct
• Throughput here is informal; durable / remote backends trade speed for guarantees

HexDocs: ETS · Mongo · Redis · Using Ecto · Article 13 · Article 16