diff --git a/lib/microwaveprop/repo_listener.ex b/lib/microwaveprop/repo_listener.ex index 6c5b050f..83ca222a 100644 --- a/lib/microwaveprop/repo_listener.ex +++ b/lib/microwaveprop/repo_listener.ex @@ -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