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.3"}]
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)
• :retry_interval - The interval in milliseconds before retrying after a failed call (default: 200)
• :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)

Each peer registers this node on its own, shortly after the node joins, and only sends it the messages it flushes after that. A peer that had already flushed writes when it registers this node sends {:cursor_expired, peer}, so reload state on that message too: the load may have missed those writes.

A notice that arrives before step 2 reaches no subscriber, which is harmless since step 3 follows. A node that subscribed early can get one notice per peer right after startup, so it is worth collapsing reloads that arrive together.

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.