Module: Pgbus::Process::NotifyProbe

Defined in:
lib/pgbus/process/notify_probe.rb

Overview

One-shot self-probe that verifies a freshly built LISTEN connection can actually receive a NOTIFY it sends to itself. Two real-world failure modes break LISTEN/NOTIFY silently, and both are caught here:

1. Transaction-mode PgBouncer — LISTEN registrations do not survive the
 transaction boundaries the pooler inserts, so a self-NOTIFY is never
 delivered (the wait times out).
2. Read-only replica — `pg_notify()` runs inside a read-only transaction
 and raises, so the probe never even reaches the wait.

In both cases the caller "connected successfully" but would never wake on a real INSERT, degrading to slow fallback polling with no explanation. The probe surfaces the condition with an actionable error naming the direct- connection overrides, then returns false so the caller can degrade gracefully (workers keep polling; the streamer starts anyway).

These listeners own raw PG::Connections by design — the documented exception to the "all PGMQ access through Pgbus::Client" rule — so the probe operates directly on a PG::Connection.

Constant Summary collapse

PROBE_TIMEOUT_SECONDS = 
2

Class Method Summary collapse

Class Method Details

.probe_notify_delivery!(conn, logger: Pgbus.logger) ⇒ Object

Runs the probe on an already-built LISTEN connection. Returns true when the self-NOTIFY is delivered within PROBE_TIMEOUT_SECONDS, false on timeout or when pg_notify raises. Never raises: a probe is diagnostic, so a failure must not take down listener startup.



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/pgbus/process/notify_probe.rb', line 32

def probe_notify_delivery!(conn, logger: Pgbus.logger)
  channel = probe_channel
  begin
    conn.exec(%(LISTEN "#{channel}"))
    conn.exec_params("SELECT pg_notify($1, '')", [channel])
    delivered = wait_for_probe(conn, channel)
    log_failure(logger) unless delivered
    delivered
  rescue PG::Error => e
    # Read-only replica: pg_notify raised inside a read-only transaction.
    log_failure(logger, error: e)
    false
  ensure
    safe_unlisten(conn, channel, logger: logger)
  end
end