fix: make RepoListener resilient, bypass pgbouncer for NOTIFY/LISTEN

RepoListener now connects directly to Postgres on port 5432 for
NOTIFY/LISTEN, bypassing pgbouncer (port 6432) which does not
support session-level LISTEN in transaction-pooling mode.

Additionally, connection and LISTEN are moved from init/1 to
handle_continue/2 so the GenServer starts immediately and the
pod boots healthy even when the DB is temporarily unreachable.
Connection failures are retried with exponential backoff
(200ms → 30s max). The notifications process is monitored and
reconnected if it dies.

Falls back to the configured port if port 5432 is unreachable
(e.g. dev/test environments without pgbouncer).

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Graham McInitre 2026-07-20 13:12:01 -05:00
parent b65d3227cd
commit af50631a2a

View file

@ -3,6 +3,11 @@ defmodule Microwaveprop.RepoListener do
Listens for PostgreSQL NOTIFY events on enrichment status changes
and Oban job state changes, broadcasting via PubSub.
Replaces polling with event-driven updates.
Connection failures (pgbouncer pool exhaustion, transient network
blips) are retried with exponential backoff instead of crashing the
supervisor the pod boots healthy and backfills the LISTEN
subscription when the DB becomes reachable.
"""
use GenServer
@ -10,6 +15,8 @@ defmodule Microwaveprop.RepoListener do
require Logger
@channels ["contact_status_changed", "oban_job_changed"]
@retry_base_ms 200
@retry_max_ms 30_000
@spec start_link(keyword()) :: GenServer.on_start()
def start_link(opts \\ []) do
@ -19,22 +26,37 @@ defmodule Microwaveprop.RepoListener do
@impl true
def init(_opts) do
config = Microwaveprop.Repo.config()
{:ok, %{config: config, notif_pid: nil, retry_ms: @retry_base_ms}, {:continue, :connect}}
end
{:ok, pid} =
Postgrex.Notifications.start_link(
hostname: config[:hostname] || "localhost",
port: config[:port] || 5432,
username: config[:username],
password: config[:password],
database: config[:database]
)
@impl true
def handle_continue(:connect, state) do
case start_notifications(state.config) do
{:ok, pid} ->
Process.monitor(pid)
{:noreply, %{state | notif_pid: pid, retry_ms: @retry_base_ms}, {:continue, :subscribe}}
for channel <- @channels do
Postgrex.Notifications.listen!(pid, channel)
{:error, reason} ->
Logger.warning("RepoListener: notifications connect failed: #{inspect(reason)}")
schedule_retry(state.retry_ms)
{:noreply, %{state | retry_ms: next_retry(state.retry_ms)}}
end
end
Logger.info("RepoListener: listening on #{Enum.join(@channels, ", ")}")
{:ok, %{pid: pid}}
def handle_continue(:subscribe, state) do
pid = state.notif_pid
failed = Enum.reject(@channels, &listen_ok?(pid, &1))
if failed == [] do
Logger.info("RepoListener: listening on #{Enum.join(@channels, ", ")}")
{:noreply, state}
else
Logger.warning("RepoListener: LISTEN failed for #{inspect(failed)}, retrying in #{state.retry_ms}ms")
# Close the broken connection and reconnect from scratch
Process.exit(pid, :normal)
schedule_retry(state.retry_ms)
{:noreply, %{state | notif_pid: nil, retry_ms: next_retry(state.retry_ms)}}
end
end
@impl true
@ -72,5 +94,71 @@ defmodule Microwaveprop.RepoListener do
{:noreply, state}
end
# Notifications process died — reconnect.
def handle_info({:DOWN, _ref, :process, pid, _reason}, %{notif_pid: pid} = state) do
Logger.warning("RepoListener: notifications connection lost, reconnecting")
{:noreply, %{state | notif_pid: nil}, {:continue, :connect}}
end
def handle_info(:retry, state) do
{:noreply, state, {:continue, :connect}}
end
def handle_info(_msg, state), do: {:noreply, state}
# ── helpers ──────────────────────────────────────────────────────
# Connect directly to Postgres (port 5432) for NOTIFY/LISTEN, bypassing
# pgbouncer (port 6432) which does not support session-level LISTEN in
# transaction-pooling mode. Falls back to the configured port if 5432 is
# unreachable (e.g. dev/test environments where pgbouncer is absent).
defp start_notifications(config) do
hostname = config[:hostname] || "localhost"
case do_start_notifications(hostname, 5432, config) do
{:ok, _pid} = ok ->
ok
{:error, :direct_postgres_unreachable} ->
Logger.info("RepoListener: port 5432 unreachable, falling back to configured port #{config[:port] || 5432}")
do_start_notifications(hostname, config[:port] || 5432, config)
{:error, reason} ->
{:error, reason}
end
end
defp do_start_notifications(hostname, port, config) do
Postgrex.Notifications.start_link(
hostname: hostname,
port: port,
username: config[:username],
password: config[:password],
database: config[:database],
sync_connect: true,
connect_timeout: 3_000
)
rescue
# If the configured port is unreachable, Postgrex raises on sync_connect.
# Trap this so callers can fall back.
e in [ArgumentError, DBConnection.ConnectionError] ->
if port == 5432 do
{:error, :direct_postgres_unreachable}
else
{:error, e}
end
end
defp listen_ok?(pid, channel) do
case Postgrex.Notifications.listen(pid, channel) do
{:ok, _ref} -> true
{:error, _reason} -> false
end
end
defp schedule_retry(delay_ms) do
Process.send_after(self(), :retry, delay_ms)
end
defp next_retry(current), do: min(current * 2, @retry_max_ms)
end