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 (issues #381 / #437): release this process's copy of the LISTEN socket 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 is not enough: pg builds socket_io with autoclose=false, so IO#close leaves the fd open, and when GC frees the inherited PG::Connection its PQfinish still writes Terminate on that fd — i.e. on the PARENT's connection (issue #437, one reconnect per fork). Repointing the fd at /dev/null (IO#reopen, the ActiveRecord PostgreSQLAdapter#discard! idiom) makes the eventual PQfinish harmless.

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.



161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/pgbus/process/notify_listener.rb', line 161

def close_inherited_socket!
  conn = @state_mutex.synchronize do
    c = @conn
    @conn = nil
    @running = false
    c
  end
  conn&.socket_io&.reopen(IO::NULL)
rescue StandardError => e
  # Best-effort but never silent: the child keeps booting either way.
  # If this fails the child's GC-time PQfinish will hit the parent's
  # session, which the parent's reconnect! survives.
  @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