tutorial

How to Monitor Phoenix.PubSub with Vigilmon

Track Phoenix.PubSub broadcast health, subscriber counts, and cross-node delivery in your Elixir cluster, and use Vigilmon to alert when real-time message delivery degrades or nodes go dark.

How to Monitor Phoenix.PubSub with Vigilmon

Phoenix.PubSub is the distributed publish-subscribe system that powers real-time features in Phoenix applications. Every Phoenix Channel subscription, every LiveView update, and every broadcast!/3 call flows through PubSub. In a clustered deployment, PubSub connects nodes so that a message broadcast on node A is delivered to subscribers on node B within milliseconds — this is what makes Phoenix Channels scale horizontally without sticky sessions.

When PubSub fails, real-time features stop working. LiveView components freeze mid-update. Chat messages are delivered to only a subset of users. Presence lists become inconsistent across nodes. These failures are invisible to standard HTTP monitoring — your /health endpoint returns 200 while your users sit on a stale UI waiting for updates that will never arrive.

Vigilmon heartbeat monitors let you verify that your PubSub pipeline is delivering messages end-to-end, not just that the process is alive.

Why Monitor Phoenix.PubSub?

PubSub failures are invisible to traditional uptime monitoring:

  • Silent node partition — a network split between Erlang cluster nodes causes PubSub broadcasts to only reach local subscribers; remote subscribers receive nothing without any error
  • Subscriber leak — Phoenix Channel processes that crash without cleanup leave ghost subscriptions in the registry, wasting memory and causing phantom delivery attempts
  • PG2/Registry overload — on high-subscriber topics, the process group lookup at broadcast time can become a bottleneck, adding latency to every message
  • Adapter failover — when using Redis PubSub adapter, a Redis restart causes a gap in delivery while the adapter reconnects; messages broadcast during reconnection are lost
  • Topic explosion — dynamic topics keyed by entity ID (e.g., "user:#{id}") that are never cleaned up cause the PubSub registry to grow without bound
  • LiveView update stall — a PubSub message that triggers a LiveView re-render but whose payload causes a crash in handle_info/2 silently drops the update

Key Metrics to Track

| Metric | What it tells you | |--------|-------------------| | Broadcast delivery latency | End-to-end time from broadcast!/3 to subscriber receipt | | Subscriber count per topic | Pool size; spikes indicate leaked subscriptions | | Cross-node delivery rate | Whether broadcasts reach subscribers on remote nodes | | PubSub process memory | Registry growth from topic/subscriber accumulation | | Broadcast error rate | Whether broadcast!/3 raises or returns error tuples | | Adapter connection status | Whether the Redis/PG adapter is connected and healthy |

Step 1: Verify PubSub Is Running

Phoenix.PubSub starts automatically as part of your Phoenix application. Verify the configuration:

# config/config.exs
config :my_app, MyApp.PubSub,
  name: MyApp.PubSub

# lib/my_app/application.ex
children = [
  {Phoenix.PubSub, name: MyApp.PubSub},
  MyAppWeb.Endpoint,
  # rest of children
]

For a Redis-backed PubSub (for multi-node deployments without Erlang clustering):

# mix.exs
{:phoenix_pubsub_redis, "~> 3.0"}

# config/config.exs
config :my_app, MyApp.PubSub,
  name: MyApp.PubSub,
  adapter: Phoenix.PubSub.Redis,
  host: System.get_env("REDIS_HOST", "localhost"),
  port: String.to_integer(System.get_env("REDIS_PORT", "6379")),
  node_name: System.get_env("NODE_NAME", "myapp@localhost")

Step 2: Build an End-to-End Delivery Test

Checking that the PubSub process is alive is insufficient — you need to verify that messages are actually delivered. Build a self-test that broadcasts to a health topic and asserts delivery:

# lib/my_app/pubsub_health.ex
defmodule MyApp.PubSubHealth do
  @moduledoc """
  End-to-end PubSub delivery check. Subscribes to a health topic, broadcasts
  a probe message, and waits for receipt. Used by the health endpoint and
  the Vigilmon heartbeat reporter.
  """

  @topic "__health__:probe"
  @timeout 2_000

  def check do
    Phoenix.PubSub.subscribe(MyApp.PubSub, @topic)
    probe_id = :crypto.strong_rand_bytes(8) |> Base.encode16()

    start = System.monotonic_time(:millisecond)

    Phoenix.PubSub.broadcast!(MyApp.PubSub, @topic, {:probe, probe_id})

    result =
      receive do
        {:probe, ^probe_id} ->
          latency = System.monotonic_time(:millisecond) - start
          {:ok, latency}
      after
        @timeout -> {:error, :timeout}
      end

    Phoenix.PubSub.unsubscribe(MyApp.PubSub, @topic)
    result
  rescue
    e -> {:error, e}
  end
end

Step 3: Add a PubSub Health Endpoint

# lib/my_app_web/controllers/health_controller.ex
defmodule MyAppWeb.HealthController do
  use MyAppWeb, :controller

  def index(conn, _params) do
    pubsub_check = MyApp.PubSubHealth.check()
    db_check = check_db()

    checks = %{
      pubsub: format_check(pubsub_check),
      database: db_check
    }

    status = if all_ok?(pubsub_check, db_check), do: 200, else: 503

    conn
    |> put_status(status)
    |> json(%{
      status: if(status == 200, do: "ok", else: "degraded"),
      checks: checks,
      node: Node.self() |> to_string()
    })
  end

  defp format_check({:ok, latency_ms}), do: %{status: "ok", latency_ms: latency_ms}
  defp format_check({:error, reason}), do: %{status: "error", reason: inspect(reason)}

  defp check_db do
    case Ecto.Adapters.SQL.query(MyApp.Repo, "SELECT 1", []) do
      {:ok, _} -> "ok"
      _ -> "error"
    end
  rescue
    _ -> "error"
  end

  defp all_ok?({:ok, _}, "ok"), do: true
  defp all_ok?(_, _), do: false
end

# lib/my_app_web/router.ex
scope "/", MyAppWeb do
  get "/health", HealthController, :index
end

Step 4: Instrument Broadcasts with Telemetry

Wrap your application's broadcast calls to capture delivery telemetry:

# lib/my_app/broadcast.ex
defmodule MyApp.Broadcast do
  require Logger

  def broadcast(topic, event, payload) do
    start = System.monotonic_time()

    result = Phoenix.PubSub.broadcast(MyApp.PubSub, topic, {event, payload})

    duration = System.monotonic_time() - start

    :telemetry.execute(
      [:my_app, :pubsub, :broadcast],
      %{duration: duration},
      %{
        topic: topic_class(topic),
        event: event,
        result: result_tag(result)
      }
    )

    result
  end

  def broadcast!(topic, event, payload) do
    case broadcast(topic, event, payload) do
      :ok -> :ok
      {:error, reason} -> raise "PubSub broadcast failed: #{inspect(reason)}"
    end
  end

  defp topic_class(topic) do
    topic
    |> String.split(":")
    |> List.first()
  end

  defp result_tag(:ok), do: :ok
  defp result_tag({:error, _}), do: :error
end

Define Telemetry.Metrics entries:

# lib/my_app/telemetry.ex
[
  counter("my_app.pubsub.broadcast.count",
    tags: [:topic, :event, :result]
  ),
  distribution("my_app.pubsub.broadcast.duration",
    unit: {:native, :millisecond},
    reporter_options: [buckets: [1, 5, 10, 25, 50, 100]]
  )
]

Step 5: Monitor Subscriber Count and Memory

Track subscription growth to catch topic leaks:

# lib/my_app/pubsub_poller.ex
defmodule MyApp.PubSubPoller do
  use GenServer
  require Logger

  @poll_interval :timer.seconds(30)
  @max_topic_subscribers 10_000

  def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)

  def init(state) do
    schedule()
    {:ok, state}
  end

  def handle_info(:poll, state) do
    report_metrics()
    schedule()
    {:noreply, state}
  end

  defp report_metrics do
    try do
      # Get PubSub registry stats
      {_name, registry_pid, _, _} =
        Supervisor.which_children(MyApp.PubSub)
        |> Enum.find(fn {id, _, _, _} -> id == :registry end)

      memory = Process.info(registry_pid, :memory) |> elem(1)

      :telemetry.execute(
        [:my_app, :pubsub, :registry],
        %{memory_bytes: memory},
        %{}
      )

      if memory > 100 * 1024 * 1024 do
        Logger.warning("PubSub registry memory high: #{div(memory, 1024 * 1024)}MB")
      end
    rescue
      e -> Logger.warning("PubSubPoller error: #{inspect(e)}")
    end
  end

  defp schedule, do: Process.send_after(self(), :poll, @poll_interval)
end

Step 6: Create Monitors in Vigilmon

HTTP monitor for the health endpoint:

  1. Sign in at vigilmon.online
  2. Click New Monitor → HTTP
  3. Set URL to https://your-app.example.com/health
  4. Set interval to 60 seconds
  5. Alert condition: status 200 and body contains "pubsub":{"status":"ok"}

This monitor verifies real end-to-end PubSub delivery on every check — not just process liveness.

Heartbeat monitor for broadcast liveness:

# lib/my_app/pubsub_heartbeat.ex
defmodule MyApp.PubSubHeartbeat do
  use GenServer

  @heartbeat_url System.get_env("VIGILMON_PUBSUB_HEARTBEAT_URL")
  @topic "heartbeat:vigilmon"
  @interval :timer.minutes(1)

  def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)

  def init(state) do
    Phoenix.PubSub.subscribe(MyApp.PubSub, @topic)
    schedule()
    {:ok, state}
  end

  def handle_info({:heartbeat, probe_id}, state) do
    if state[:pending_probe] == probe_id do
      ping_vigilmon()
    end

    {:noreply, Map.delete(state, :pending_probe)}
  end

  def handle_info(:send_probe, state) do
    probe_id = :crypto.strong_rand_bytes(4) |> Base.encode16()
    Phoenix.PubSub.broadcast(MyApp.PubSub, @topic, {:heartbeat, probe_id})
    schedule()
    {:noreply, Map.put(state, :pending_probe, probe_id)}
  end

  def handle_info(_, state), do: {:noreply, state}

  defp ping_vigilmon do
    if @heartbeat_url do
      :httpc.request(:get, {String.to_charlist(@heartbeat_url), []}, [{:timeout, 3000}], [])
    end
  end

  defp schedule, do: Process.send_after(self(), :send_probe, @interval)
end

This heartbeat only pings Vigilmon when a message successfully travels from broadcast to subscriber — the minimum viable PubSub proof.

Create a Heartbeat monitor in Vigilmon with a 3-minute expected interval. A missed heartbeat means PubSub delivery is broken.

Step 7: Monitor Cross-Node Delivery in a Cluster

For clustered deployments, verify that messages cross node boundaries:

# lib/my_app/cluster_health.ex
defmodule MyApp.ClusterHealth do
  @topic "__health__:cluster"
  @timeout 3_000

  def check_cross_node do
    connected_nodes = Node.list()

    if connected_nodes == [] do
      {:ok, :single_node}
    else
      probe_id = :crypto.strong_rand_bytes(8) |> Base.encode16()
      Phoenix.PubSub.subscribe(MyApp.PubSub, @topic)

      # Ask a remote node to broadcast to us
      :rpc.cast(hd(connected_nodes), Phoenix.PubSub, :broadcast, [
        MyApp.PubSub,
        @topic,
        {:cross_node_probe, probe_id, Node.self()}
      ])

      result =
        receive do
          {:cross_node_probe, ^probe_id, _origin_node} -> {:ok, :delivered}
        after
          @timeout -> {:error, :cross_node_timeout}
        end

      Phoenix.PubSub.unsubscribe(MyApp.PubSub, @topic)
      result
    end
  end
end

Include this in your health endpoint response when running in a cluster:

# In HealthController
cluster_check = MyApp.ClusterHealth.check_cross_node()

checks = %{
  pubsub: format_check(pubsub_check),
  cluster: format_check(cluster_check),
  database: db_check
}

Step 8: Alerting

In Vigilmon, configure Notifications → New Channel:

Slack for PubSub delivery failure:

🔴 DOWN: Phoenix.PubSub Delivery — MyApp
Monitor: /health returning pubsub.status = "error"
Node: myapp@10.0.0.1
Action: Check PubSub process, Erlang cluster connectivity, and Redis adapter (if used)

PagerDuty for heartbeat miss:

Configure a high-priority alert on the broadcast heartbeat — a missed heartbeat means real-time features are broken for all users, which is an incident.

Common Phoenix.PubSub Issues and Fixes

Erlang cluster partition — messages not crossing nodes:

# Verify cluster connectivity
iex> Node.list()
[:"myapp@10.0.0.2", :"myapp@10.0.0.3"]  # ok
[]  # partitioned — PubSub won't cross nodes

Check Erlang distribution port (default 4369/EPMD) and node cookie match across nodes.

Subscriber leak — topic with thousands of dead subscriptions:

# When a Channel or LiveView process using a dynamic topic crashes without
# cleanup, use Phoenix.Tracker or explicit unsubscribe in terminate/2

@impl Phoenix.Channel
def terminate(_reason, socket) do
  Phoenix.PubSub.unsubscribe(MyApp.PubSub, "user:#{socket.assigns.user_id}")
  :ok
end

Redis adapter message gap on reconnect:

# Add reconnect logic and log gaps
config :my_app, MyApp.PubSub,
  adapter: Phoenix.PubSub.Redis,
  host: System.get_env("REDIS_HOST"),
  reconnect_after_ms: [100, 500, 1000, 5000]

Local-only delivery when you expect cluster-wide:

# broadcast/3 sends to all nodes
Phoenix.PubSub.broadcast(MyApp.PubSub, topic, message)

# local_broadcast/3 sends only to local node subscribers — easy to confuse
Phoenix.PubSub.local_broadcast(MyApp.PubSub, topic, message)

What You've Built

| What | How | |------|-----| | End-to-end delivery check | Self-probe: subscribe → broadcast → receive, with latency | | Structured health endpoint | JSON health response with pubsub and db checks | | Broadcast instrumentation | Telemetry events on every broadcast/3 call | | PubSub registry monitoring | Memory and subscriber count via telemetry_poller | | Liveness heartbeat | GenServer self-test pinging Vigilmon on successful delivery | | Cross-node cluster check | RPC probe verifying Erlang distribution is working | | Real-time alerting | Vigilmon HTTP monitor + PagerDuty on heartbeat miss |

Phoenix.PubSub makes real-time features possible at scale. Vigilmon makes sure those features keep working when it matters most.

Next Steps

  • Add a Grafana panel for broadcast latency by topic class to identify hot topics
  • Set up Phoenix Presence monitoring to track connected users as a business metric
  • Configure a separate heartbeat for each critical PubSub topic in your application
  • Use Vigilmon's downtime history to measure real-time feature availability for SLA reporting

Get started free at vigilmon.online.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles