Class: Pgbus::Process::NotifyHub

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

Overview

Supervisor-owned LISTEN hub (issue #381, worker_notify_scope: :supervisor). Owns ONE NotifyListener — one direct PG connection for the whole host — on the union of every capsule's and consumer's queue channels, and fans wakes out to worker/consumer forks over per-fork pipes (see WakePipe for the fork side and the W/H/P byte protocol).

Union: explicit capsule queues from config.workers, wildcard capsules via the shared WildcardQueueResolver, consumer queues via Registry#queue_names_for_topics — each gated by config.role_enabled?. Refreshed every REFRESH_INTERVAL_SECONDS from the supervisor's monitor tick so wildcard churn and late registry subscriptions converge; a transient gap is bounded by the forks' NOTIFY poll ceiling (15s), so under-listening degrades latency, never liveness.

Routing: a NOTIFY for queue Q wakes every fork whose registered set contains Q, plus every wildcard fork. Over-waking costs one empty read; under-waking is the only failure mode that matters, so wildcard routing is unconditional.

Status: healthy? = listener running + connected + delivering. Broadcast to all pipes on change and every STATUS_REBROADCAST_SECONDS (idempotent 1-byte writes), so a lost byte or a freshly full pipe can't wedge a fork in the wrong mode.

Threading: register/deregister/tick/stop run on the supervisor's main loop; the wake callback runs on the listener's thread. The fork table is the only shared state — guarded by @table_mutex. Pipe writes are 1-byte write_nonblock calls (atomic well under PIPE_BUF) and never block: a full pipe already guarantees a pending readable byte, so the wake is skipped, and status repair rides the periodic rebroadcast.

Constant Summary collapse

REFRESH_INTERVAL_SECONDS = 
30
STATUS_REBROADCAST_SECONDS = 
5
RETRY_BASE_SECONDS = 
5
RETRY_MAX_SECONDS = 
300

Instance Method Summary collapse

Constructor Details

#initialize(config:, listener_factory: nil, clock: nil, logger: Pgbus.logger) ⇒ NotifyHub

Returns a new instance of NotifyHub.



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

def initialize(config:, listener_factory: nil, clock: nil, logger: Pgbus.logger)
  @config = config
  @logger = logger
  @clock = clock || -> { ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) }
  @listener_factory = listener_factory || default_listener_factory
  @table_mutex = Mutex.new
  @forks = {}
  @listener = nil
  @last_status = nil
  @last_broadcast_at = nil
  @last_refresh_at = nil
  @retry_at = 0.0
  @retry_backoff = RETRY_BASE_SECONDS
end

Instance Method Details

#close_inherited! ⇒ Object

Called ONLY inside a just-forked child: release the child's copies of the parent's hub resources — the LISTEN socket fd (without a libpq Terminate, see NotifyListener#close_inherited_socket!) and the fork table's pipe write ends. Never touches parent state (fork copied it).



98
99
100
101
102
103
104
105
106
 
# File 'lib/pgbus/process/notify_hub.rb', line 98

def close_inherited!
  @listener&.close_inherited_socket!
  @listener = nil
  @table_mutex.synchronize do
    @forks.each_value { |entry| close_quietly(entry[:pipe]) }
    @forks.clear
  end
  self
end

#deregister_fork(pid) ⇒ Object 



88
89
90
91
92
 
# File 'lib/pgbus/process/notify_hub.rb', line 88

def deregister_fork(pid)
  entry = @table_mutex.synchronize { @forks.delete(pid) }
  close_quietly(entry[:pipe]) if entry
  self
end

#healthy? ⇒ Boolean

Returns:

  • (Boolean) 


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

def healthy?
  listener = @listener
  !!(listener&.running? && listener.connected? && listener.delivering?)
end

#register_fork(pid:, queues:, pipe:, wildcard: false) ⇒ Object

Register a fork's wake pipe. queues are PHYSICAL queue names (already prefixed); wildcard forks pass wildcard: true and are woken for every channel. The fork's current status byte is sent immediately so a freshly forked worker starts in the right polling mode.



75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/pgbus/process/notify_hub.rb', line 75

def register_fork(pid:, queues:, pipe:, wildcard: false)
  entry = { pipe: pipe, queues: Set.new(queues), wildcard: wildcard }
  @table_mutex.synchronize { @forks[pid] = entry }
  byte = status_byte
  write_byte(entry, byte)
  # The registration write is the broadcast baseline: without stamping
  # it, the next tick would treat @last_status as unknown and re-send
  # the same byte to every fork immediately.
  @last_status = byte
  @last_broadcast_at = now
  self
end

#start ⇒ Object 



56
57
58
59
60
 
# File 'lib/pgbus/process/notify_hub.rb', line 56

def start
  build_listener
  @last_refresh_at = now
  self
end

#stop ⇒ Object 



62
63
64
65
66
67
68
69
 
# File 'lib/pgbus/process/notify_hub.rb', line 62

def stop
  stop_listener_quietly
  @table_mutex.synchronize do
    @forks.each_value { |entry| close_quietly(entry[:pipe]) }
    @forks.clear
  end
  self
end

#tick ⇒ Object

One supervisor monitor-loop beat: self-heal the listener, refresh the LISTEN union, keep fork status fresh.



110
111
112
113
114
115
 
# File 'lib/pgbus/process/notify_hub.rb', line 110

def tick
  ensure_listener
  refresh_union_if_due
  broadcast_status
  self
end