Phoenix PubSub adapter using :pg with at-least-once delivery.

To start it, list it in your supervision tree as:

{Phoenix.PubSub, name: MyApp.PubSub, adapter: EchoPubSub}

You will also need to add :echo_pubsub to your deps:

defp deps do
  [{:echo_pubsub, "~> 0.2.0"}]
end

Options

• :name - The required name to register the PubSub processes, ie: MyApp.PubSub
• :pool_size - The number of producers and workers to run on each node, allowing concurrent message delivery.
• :buffer_size - The amount of messages to maintain in memory
• :batch_interval - The interval in milliseconds to batch messages before sending (default: 200)
• :call_timeout - The timeout in milliseconds for GenServer calls to remote nodes (default: 5000)
• :capacity_warning_threshold - Buffer fill ratio (0.0-1.0) that triggers warning; above 1.0 never warns (default: 0.4)
• :capacity_warning_interval - Minimum seconds between warnings (default: 60)

Configuration

Options can also be set via Application config per PubSub name:

# config/config.exs
config :echo_pubsub, MyApp.PubSub,
  pool_size: 2,
  buffer_size: 10_000,
  batch_interval: 100,
  call_timeout: 5000

Options passed directly to the supervisor take precedence over config values.

Non-blocking fan-out

Each flush hands every remote node's batch to its own task and returns, so a flush costs the slowest round-trip instead of their sum and the producer keeps accepting writes while a slow or unreachable peer burns its :call_timeout. A node's read cursor advances only when its task reports an ack, and at most one batch per node is in flight, so batches still arrive in cursor order.

Loading state at startup

Load state only after the node is connected and subscribed:

1. connect the node to the cluster
2. subscribe to the topics
3. load state from the source of truth (e.g. the database)

A peer only sends this node the messages it flushes after registering the node, which happens shortly after the node joins. A change a peer flushes before that is never delivered here, so it reaches this node only if the state is loaded after the registration.

A message broadcast before the load can still be delivered after it (batching delays delivery, failed sends are retried), so its change may already be in the loaded state: applying a message must tolerate seeing its change twice.

Implementation

The in memory buffer is a ring buffer, meaning that a constant number of messages are maintained and once the buffer is full, new messages overwrite the oldest message in the buffer.

This means that if a node is unreachable long enough - a sustained network failure, or an actual disconnect - that its cursor points to a message that no longer exists, it will receive a special message over pubsub: {:cursor_expired, node@host}

Applications are encouraged to handle and act on this message to get to a valid state, such as reloading all state from a source of truth like the db or another node. While this technically means that we don't guarentee every node will receive every message, we can guerentee that there are no gaps in messages. Recipt of message 3 guarentees you've received messages 1 and 2.