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.

Constant Summary collapse

CHANNEL_PREFIX = 
"pgmq.q_"
CHANNEL_SUFFIX = 
".INSERT"
RECONNECT_BACKOFF_SECONDS = 
0.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.



39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/pgbus/process/notify_listener.rb', line 39

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 



103
104
105
 
# File 'lib/pgbus/process/notify_listener.rb', line 103

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) 


67
68
69
 
# File 'lib/pgbus/process/notify_listener.rb', line 67

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

#listening_to ⇒ Object 



57
58
59
 
# File 'lib/pgbus/process/notify_listener.rb', line 57

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

#remove_queue(physical_queue) ⇒ Object 



107
108
109
 
# File 'lib/pgbus/process/notify_listener.rb', line 107

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) 


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

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

#start ⇒ Object 



71
72
73
74
75
76
77
78
79
80
 
# File 'lib/pgbus/process/notify_listener.rb', line 71

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 



82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/pgbus/process/notify_listener.rb', line 82

def stop
  conn_to_close = nil
  @state_mutex.synchronize do
    return self unless @running

    @running = false
    conn_to_close = @conn
  end
  @commands << [:stop]
  # Interrupt the blocking wait by closing the socket; the rescue in
  # wait_once sees @running == false and exits cleanly.
  begin
    conn_to_close&.close if conn_to_close.respond_to?(:close)
  rescue StandardError
    nil
  end
  @thread&.join(5)
  @thread = nil
  self
end