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

Class Method Summary collapse

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.



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/pgbus/process/notify_listener.rb', line 59

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

Class Method Details

.physical_for(channel) ⇒ Object

Inverse of #channel_for: map a NOTIFY channel back to the physical queue name. Class-level because the channel format is owned here — NotifyHub (wake routing + union refresh) and Worker (queue-set sync) both consume it.



49
50
51
 
# File 'lib/pgbus/process/notify_listener.rb', line 49

def self.physical_for(channel)
  channel.delete_prefix(CHANNEL_PREFIX).delete_suffix(CHANNEL_SUFFIX)
end

Instance Method Details

#add_queue(physical_queue) ⇒ Object 



131
132
133
 
# File 'lib/pgbus/process/notify_listener.rb', line 131

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

#close_inherited_socket! ⇒ Object

Called ONLY inside a just-forked child (issue #381 hub hygiene): drop this process's copy of the LISTEN socket fd WITHOUT PQfinish — #close would send a libpq Terminate over the socket shared with the parent, killing the parent's LISTEN session. Closing the IO wrapper just closes the child's fd. The listener thread does not exist in the child (fork copies only the calling thread), so there is no concurrent owner and the single-owner rule (#375) does not apply.



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/pgbus/process/notify_listener.rb', line 153

def close_inherited_socket!
  conn = @state_mutex.synchronize do
    c = @conn
    @conn = nil
    @running = false
    c
  end
  conn&.socket_io&.close
rescue StandardError => e
  # Best-effort (a lingering fd copy is benign until the parent dies),
  # but never silent: the child keeps booting either way.
  @logger.warn do
    "[Pgbus::NotifyListener] inherited socket cleanup failed: #{e.class}: #{e.message}"
  end
  nil
end

#connected? ⇒ Boolean

Whether a live PG connection is currently published. running? stays true during a reconnect (the thread is alive, looping in reconnect!), so this is the signal that distinguishes "parked in wait_for_notify" from "between connections". The supervisor NotifyHub (issue #381) consults it to broadcast degraded status to forks the moment the shared connection drops, and healthy again once it is rebuilt.

Returns:

  • (Boolean) 


87
88
89
 
# File 'lib/pgbus/process/notify_listener.rb', line 87

def connected?
  @state_mutex.synchronize { !@conn.nil? }
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) 


97
98
99
 
# File 'lib/pgbus/process/notify_listener.rb', line 97

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

#listening_to ⇒ Object 



77
78
79
 
# File 'lib/pgbus/process/notify_listener.rb', line 77

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

#remove_queue(physical_queue) ⇒ Object 



135
136
137
 
# File 'lib/pgbus/process/notify_listener.rb', line 135

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) 


142
143
144
 
# File 'lib/pgbus/process/notify_listener.rb', line 142

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

#start ⇒ Object 



101
102
103
104
105
106
107
108
109
110
 
# File 'lib/pgbus/process/notify_listener.rb', line 101

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 



112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/pgbus/process/notify_listener.rb', line 112

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