EchoPubSub

A Phoenix.PubSub adapter that distributes messages between nodes using the erlang :pg module, like the default adapter, however with the additional guarentees of "at least once" delivery.

This means that nodes can disconnect temporarily from the cluster - even for a blip as short as ~1ms - and then "catch up" when they rejoin, thanks to a buffer of messages and read cursors.

See the Docs for more information.

The Problem: messages lost during temporary network problems

The default Phoenix.PubSub.PG2 adapter is fire-and-forget. When a node broadcasts, the message is delivered to the nodes that are connected at that moment. There is no buffer and no acknowledgement - if a node is unreachable when the broadcast happens, the message is simply gone for that node.

Even a momentary network problem - a blip lasting as little as 1ms - means silent data loss for any message broadcast during it:

        Node A (broadcaster)            Node B (disconnected)
        ────────────────────            ────────────────────
  t0    broadcast msg 1   ───────────▶  received msg 1
  t1    ┌─ network blip (e.g. ~1ms): B drops out ─┐
  t2    broadcast msg 2   ──────✗               (never arrives)
  t3    broadcast msg 3   ──────✗               (never arrives)
  t4    └─ B reconnects ────────────────────────┘
  t5    broadcast msg 4   ───────────▶  received msg 4

When B comes back at t4 it carries on as if nothing happened: it has no idea that msg 2 and msg 3 ever existed. There is no error, no gap detection - just a hole in the stream. For anything that relies on the message stream being complete (replicated caches, event logs, derived state), this quietly corrupts B's view of the world.

How EchoPubSub solves it

EchoPubSub upgrades delivery from fire-and-forget to at-least-once by having each broadcasting node remember what it has sent and to whom:

• Ring buffer of recent messages - every producer keeps the last :buffer_size messages in memory, each stamped with a monotonically increasing write cursor.
• Per-node read cursors - the producer tracks how far each remote node has acknowledged. Remote delivery is a synchronous, acked call; a node's cursor only advances once it confirms receipt.
• Replay on reconnect - when a briefly disconnected node rejoins, the producer sees its read cursor is behind the write cursor and replays exactly the messages it missed, in order, before resuming normal flow.

Applied to the scenario above, B's cursor stays at msg 1 while it is disconnected. On reconnect the producer replays msg 2 and msg 3, so B catches up with no gaps before msg 4 arrives:

        Node A (broadcaster)            Node B (disconnected)
        ────────────────────            ────────────────────
  t0    broadcast msg 1   ───────────▶  received msg 1   (B cursor → 1)
  t1    ┌─ network blip (e.g. ~1ms): B drops out ─┐
  t2    broadcast msg 2     buffered             (B cursor stuck at 1)
  t3    broadcast msg 3     buffered             (B cursor stuck at 1)
  t4    └─ B reconnects ────────────────────────┘
  t4'   replay msg 2, 3   ───────────▶  received msg 2, 3 (B cursor → 3)
  t5    broadcast msg 4   ───────────▶  received msg 4    (B cursor → 4)

When the buffer can't cover the gap

The buffer is bounded, so a node that stays gone long enough that its missed messages get overwritten by newer writes cannot be replayed without gaps. Rather than silently delivering a corrupted stream, EchoPubSub gives up on replay explicitly: it sends the subscribing process a {:cursor_expired, node_name} message (see Usage) so the application can recover to a valid state - typically by reloading from a source of truth such as the database or another node.

This is the core guarantee: either you receive every message in order, or you are told that you fell behind. There are never silent gaps - receipt of message 3 guarantees you have already received messages 1 and 2.

Usage

Note: I used LLM for typing - but ideas and decisions were mine

def deps do
  [
    {:echo_pubsub, "~> 0.1.0"}
  ]
end

# application.ex
children = [
  # ...,
  {Phoenix.PubSub, name: MyApp.PubSub, adapter: EchoPubSub}
]

Config Options

Subscribing processes should handle the message {:cursor_expired, node_name} which indicates that your client has been disconnected long enough that your position in the broadcaster's buffer has been overwritten. At this point it is the subscribing process's job to return to a valid state i.e. reloading state from source like database or another node.

Credits

EchoPubSub is a fork of phoenix_pubsub_buffered by Eric Newbury, who designed and built the original at-least-once buffered PubSub adapter. All credit for the core design goes to him - this fork builds on that foundation with batched inter-node delivery, automatic replay on failure, flush-path expiry detection, telemetry, capacity warnings, and additional configuration options.