Class: Pgbus::Process::NotifyListener

Inherits:
Object
  • Object
show all
Defined in:
lib/pgbus/process/notify_listener.rb

Overview

Owns a single dedicated PG::Connection that LISTENs on the INSERT NOTIFY channel of every queue a Worker/Consumer reads, and fires a WakeSignal the moment any of them receives a row. This converts the worker/consumer loop from "blind-read every polling_interval" into "sleep until a real insert, poll only as a fallback" — eliminating the empty-read storm that dominates DB load on idle queues.

pgmq-ruby's wait_for_notify(queue, timeout:) is single-queue and wraps the wait in with_connection, which only watches one channel and holds the pooled connection for the whole wait. Neither fits a worker that reads N queues on a small shared pool. So we own ONE raw PG::Connection and hand-roll per-channel LISTEN on it.

A persistent LISTEN connection silently dies under a transaction-pool PgBouncer (LISTEN does not survive COMMIT boundaries). Point this connection at a DIRECT port via config.worker_notify_* overrides. The health-check-on-timeout catches a connection killed out from under us and re-LISTENs everything.

NOTIFY channel naming (pgmq trigger): PG_NOTIFY('pgmq.' || table || '.' || TG_OP). For queue pgbus_default the table is q_pgbus_default, so the channel is pgmq.q_pgbus_default.INSERT.

Thread safety: @running, @conn, and @listening_to are guarded by blocking IO call where the mutex MUST NOT be held), so wait_once reads the connection out of the mutex first and operates on a local. Reconnect publishes the new connection + channel set under the mutex.

The mutex only makes the ivar READ safe. PG::Connection itself is not thread-safe, so the connection is single-owner: the listener thread is the ONLY thread that may exec, wait, or close on it, from build through teardown. #stop signals by clearing @running and joining — it never touches the connection, because #close is PQfinish and freeing the PGconn under a concurrent libpq call is a process-killing SEGV, not a rescuable PG::Error (issue #375).

Constant Summary collapse

CHANNEL_PREFIX = 
"pgmq.q_"
CHANNEL_SUFFIX = 
".INSERT"
RECONNECT_BACKOFF_SECONDS = 
0.5
STOP_JOIN_GRACE_SECONDS =

Grace added to one health-check cycle when #stop joins the listener thread. See #stop_join_timeout.

5

Instance Method Summary collapse

Constructor Details

#initialize(physical_queues:, on_wake:, connection_options:, health_check_ms: 1000, logger: Pgbus.logger) ⇒ NotifyListener

Returns a new instance of NotifyListener.



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/pgbus/process/notify_listener.rb', line 51

def initialize(physical_queues:, on_wake:, connection_options:,
               health_check_ms: 1000, logger: Pgbus.logger)
  @physical_queues = Array(physical_queues)
  @on_wake = on_wake
  @connection_options = connection_options
  @health_check_ms = health_check_ms
  @logger = logger
  @state_mutex = Mutex.new
  @listening_to = Set.new
  @commands = Queue.new
  @running = false
  @thread = nil
  @conn = nil
  # Optimistic until the start-time self-probe runs: assume NOTIFY delivery
  # works so a not-yet-probed listener isn't mistaken for a pooler-deaf one.
  @delivering = true
end

Instance Method Details

#add_queue(physical_queue) ⇒ Object 



113
114
115
 
# File 'lib/pgbus/process/notify_listener.rb', line 113

def add_queue(physical_queue)
  @commands << [:listen, physical_queue]
end

#delivering? ⇒ Boolean

Whether the start-time self-probe confirmed this connection can actually receive a NOTIFY. False when a transaction-mode pooler or replica silently drops LISTEN: the thread is still alive (running? == true) but will never wake the loop. The Worker/Consumer consult this so a live-but-deaf listener is treated as absent for wake-timeout purposes — fast polling, not the 15s NOTIFY ceiling (issue #332).

Returns:

  • (Boolean) 


79
80
81
 
# File 'lib/pgbus/process/notify_listener.rb', line 79

def delivering?
  @state_mutex.synchronize { @delivering }
end

#listening_to ⇒ Object 



69
70
71
 
# File 'lib/pgbus/process/notify_listener.rb', line 69

def listening_to
  @state_mutex.synchronize { @listening_to.dup }
end

#remove_queue(physical_queue) ⇒ Object 



117
118
119
 
# File 'lib/pgbus/process/notify_listener.rb', line 117

def remove_queue(physical_queue)
  @commands << [:unlisten, physical_queue]
end

#running? ⇒ Boolean

Public so the owning worker can detect a listener whose thread died (run_loop hit a fatal error and cleared @running in its ensure) and restart it. Guarded by @state_mutex like every other @running access.

Returns:

  • (Boolean) 


124
125
126
 
# File 'lib/pgbus/process/notify_listener.rb', line 124

def running?
  @state_mutex.synchronize { @running }
end

#start ⇒ Object 



83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgbus/process/notify_listener.rb', line 83

def start
  @state_mutex.synchronize do
    return self if @running

    @running = true
  end
  @physical_queues.each { |q| @commands << [:listen, q] }
  @thread = Thread.new { run_loop }
  self
end

#stop ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/pgbus/process/notify_listener.rb', line 94

def stop
  @state_mutex.synchronize do
    return self unless @running

    @running = false
  end
  @commands << [:stop]
  # Deliberately does NOT touch @conn. PG::Connection is not thread-safe
  # and #close is PQfinish: it frees the PGconn and its OpenSSL objects
  # out from under whatever libpq call the listener thread is making on
  # the same connection. That is a use-after-free — a process-killing
  # SEGV, not a rescuable PG::Error (issue #375). The listener thread is
  # the sole owner of @conn for its entire life and closes it in
  # run_loop's ensure; clearing @running above is the whole stop signal.
  @thread&.join(stop_join_timeout)
  @thread = nil
  self
end