Class: Pgbus::Web::Streamer::MasterHub

Inherits:
Object
  • Object
show all
Defined in:
lib/pgbus/web/streamer/master_hub.rb

Overview

Master-process streams hub (issue #382): ONE LISTEN connection per web host instead of one per Puma worker. Runs in the Puma master (started by the pgbus_streams plugin), owns a single Web::Streamer::Listener on the refcounted union of every worker's stream channels, and fans wakes (including ephemeral payloads) out to workers over a Unix domain socket using HubProtocol frames.

Workers are CLIENTS: they connect lazily to socket_path on first SSE use (HubClient). Nothing is inherited across fork, so there is no FD hygiene for this transport, and a server that never starts a hub (no preload_app!, single mode, hub crash) simply has no socket — every worker falls back to its own per-worker Listener (FailoverListener), trading connections for unchanged semantics (settled on #382).

The no-lost-wake ack contract, cross-process: a worker's sub is registered in the routing table BEFORE the hub executes LISTEN, and the ack is sent only AFTER ensure_listening returns — so from the moment LISTEN is active every wake reaches the subscribing worker. Over-delivery before the ack is harmless; under-delivery is the only failure mode that matters (same principle as Process::NotifyHub).

Backpressure (per-worker outbound queue + writer thread):

- durable wakes (payload nil) are droppable beyond durable_queue_limit
— the next durable wake re-reads from the min cursor, so they
self-heal (mirrors dispatch_queue_limit semantics);
- ephemeral wakes are NEVER dropped: they push past the durable cap,
and a worker whose queue exceeds hard_queue_limit is EVICTED
(socket severed) — which triggers that worker's own fallback
listener. A wedged worker degrades itself, never its siblings.

Threading: accept thread + fanout thread + status thread, plus one reader and one writer thread per connected worker. The routing table is guarded by @table_mutex; each worker's outbox by its own mutex. All socket WRITES go through that worker's writer thread (frames must never interleave).

Constant Summary collapse

DEFAULT_DURABLE_QUEUE_LIMIT = 
256
DEFAULT_HARD_QUEUE_LIMIT = 
1024
REBROADCAST_TICKS =

Status is rebroadcast every REBROADCAST_TICKS status intervals even unchanged, so a worker that connected mid-outage converges.

5

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config:, socket_path:, listener_factory: nil, status_interval: 1.0, durable_queue_limit: DEFAULT_DURABLE_QUEUE_LIMIT, hard_queue_limit: DEFAULT_HARD_QUEUE_LIMIT, logger: Pgbus.logger) ⇒ MasterHub

Returns a new instance of MasterHub.



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/pgbus/web/streamer/master_hub.rb', line 53

def initialize(config:, socket_path:, listener_factory: nil, status_interval: 1.0,
               durable_queue_limit: DEFAULT_DURABLE_QUEUE_LIMIT,
               hard_queue_limit: DEFAULT_HARD_QUEUE_LIMIT, logger: Pgbus.logger)
  @config = config
  @socket_path = socket_path
  @status_interval = status_interval
  @durable_queue_limit = durable_queue_limit
  @hard_queue_limit = hard_queue_limit
  @logger = logger
  @listener_factory = listener_factory || default_listener_factory
  @dispatch_queue = Queue.new
  # Serializes the FULL start and stop sequences: a stop racing an
  # in-progress start (blocked in the listener factory's PG connect)
  # must WAIT for it and then tear everything down — otherwise stop
  # returns having cleaned nothing and start finishes building a live
  # hub afterwards. Loop threads never take this mutex (they read
  # @running via @table_mutex), so holding it across the blocking
  # startup cannot deadlock them.
  @lifecycle_mutex = Mutex.new
  @table_mutex = Mutex.new
  @workers = {}
  # Plain Hash, entries created ONLY at subscribe time — a default
  # proc here would leak one empty Set per wake that arrives for an
  # already-unsubscribed channel (in-flight NOTIFYs after the last
  # unsub, per-record stream names → unbounded, review on #384).
  @queue_refs = {}
  @stop_signal = Queue.new
  @next_id = 0
  @dropped_durable_wakes = 0
  @evicted_workers = 0
  @running = false
end

Instance Attribute Details

#socket_path ⇒ Object (readonly)

Returns the value of attribute socket_path.



51
52
53
 
# File 'lib/pgbus/web/streamer/master_hub.rb', line 51

def socket_path
  @socket_path
end

Instance Method Details

#dropped_durable_wakes ⇒ Object 



86
87
88
 
# File 'lib/pgbus/web/streamer/master_hub.rb', line 86

def dropped_durable_wakes
  @table_mutex.synchronize { @dropped_durable_wakes }
end

#evicted_workers ⇒ Object 



90
91
92
 
# File 'lib/pgbus/web/streamer/master_hub.rb', line 90

def evicted_workers
  @table_mutex.synchronize { @evicted_workers }
end

#start ⇒ Object

The factory must return a STARTED listener wired to dispatch_queue. If any step after the listener exists fails (bad socket path, chmod, thread spawn), the listener — and its dedicated LISTEN connection, the exact resource this hub conserves — is stopped before the error propagates; MasterHubBoot's rescue never sees a leaked connection.



99
100
101
 
# File 'lib/pgbus/web/streamer/master_hub.rb', line 99

def start
  @lifecycle_mutex.synchronize { locked_start }
end

#stop ⇒ Object 



103
104
105
 
# File 'lib/pgbus/web/streamer/master_hub.rb', line 103

def stop
  @lifecycle_mutex.synchronize { locked_stop }
end