PgFlow.Signal.Notify (PgFlow v0.7.0)

Manages PostgreSQL LISTEN/NOTIFY for pgmq queue notifications.

Uses pgmq's built-in enable_notify_insert (pgmq 1.8.0+) to receive instant wake-up signals when messages are inserted into queue tables. Dispatches :poll_now messages to registered worker processes.

Built on Postgrex.Notifications which is the purpose-built solution for PostgreSQL LISTEN/NOTIFY. Notifications are delivered asynchronously via messages to the process that called listen/2.

pgmq keeps pgmq.notify_insert_throttle UNLOGGED, so crash recovery empties it, and so does Neon whenever a compute restarts. A queue whose throttle row is missing sends no NOTIFY, and its workers fall back to their slow poll. Every :throttle_reconcile_interval_ms this process runs PgFlow.Queries.Pgmq.restore_notify_throttle_rows/3 for the queues its workers listen on: one statement that restores missing rows, writes nothing when every row is present, and never changes a trigger or locks a queue table.

Summary

Types

state()

worker_entry()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

register_worker(server \\ __MODULE__, flow_slug, worker_pid)

Registers a worker for notifications on a queue.

register_worker_async(server \\ __MODULE__, flow_slug, worker_pid)

Registers a replacement worker without blocking its supervisor restart.

start_link(opts)

Starts the Signal.Notify process.

unregister_worker(server \\ __MODULE__, flow_slug)

Unregisters a worker from notifications.

Types

state()

@type state() :: %{
  repo: module(),
  notify_throttle_ms: non_neg_integer(),
  throttle_reconcile_interval_ms: pos_integer(),
  conn: pid() | nil,
  workers: %{required(String.t()) => worker_entry()},
  channels: %{required(String.t()) => String.t()}
}

worker_entry()

@type worker_entry() :: %{
  worker_pid: pid(),
  monitor_ref: reference(),
  listen_ref: reference() | nil
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

register_worker(server \\ MODULE, flow_slug, worker_pid)

@spec register_worker(GenServer.server(), String.t(), pid()) :: :ok | {:error, term()}

Registers a worker for notifications on a queue.

Starts listening on the queue's channel. Note that pgmq.enable_notify_insert must be called separately (done by PgFlow.Supervisor) before registration.

register_worker_async(server \\ MODULE, flow_slug, worker_pid)

@spec register_worker_async(GenServer.server(), String.t(), pid()) :: :ok

Registers a replacement worker without blocking its supervisor restart.

Transient LISTEN failures retry with capped exponential backoff while the replacement remains alive. Initial registration remains synchronous through register_worker/3 so PgFlow.FlowStarter can report startup failures.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Starts the Signal.Notify process.

Options

:repo - (required) The Ecto repository module
:notify_throttle_ms - (optional) Throttle interval for pgmq notifications (default: 250)
:throttle_reconcile_interval_ms - (optional) Milliseconds between checks that restore missing pgmq.notify_insert_throttle rows (default: 30000)

unregister_worker(server \\ MODULE, flow_slug)

@spec unregister_worker(GenServer.server(), String.t()) :: :ok

Unregisters a worker from notifications.