From 016c0ebe34647467bb827ab43d6bdb8bd5a28a33 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Sun, 2 Aug 2026 17:48:25 +0200 Subject: [PATCH 01/16] =?UTF-8?q?feat(streams):=20HubProtocol=20=E2=80=94?= =?UTF-8?q?=20length-prefixed=20JSON=20framing=20for=20the=20master=20hub?= =?UTF-8?q?=20(#382=20step=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lossless framed transport (unlike the deliberately lossy 1-byte job-side wake pipes): 4-byte big-endian length + UTF-8 JSON, blocking reads with short-read-as-EOF semantics, oversize and malformed-frame guards. --- lib/pgbus/web/streamer/hub_protocol.rb | 74 ++++++++++++++++ spec/pgbus/web/streamer/hub_protocol_spec.rb | 88 ++++++++++++++++++++ 2 files changed, 162 insertions(+) create mode 100644 lib/pgbus/web/streamer/hub_protocol.rb create mode 100644 spec/pgbus/web/streamer/hub_protocol_spec.rb diff --git a/lib/pgbus/web/streamer/hub_protocol.rb b/lib/pgbus/web/streamer/hub_protocol.rb new file mode 100644 index 00000000..f06d52eb --- /dev/null +++ b/lib/pgbus/web/streamer/hub_protocol.rb @@ -0,0 +1,74 @@ +# frozen_string_literal: true + +require "json" + +module Pgbus + module Web + module Streamer + # Framing for the master-hub Unix socket (issue #382): 4-byte big-endian + # payload length + UTF-8 JSON. Unlike the job-side wake pipes (1-byte, + # lossy-by-design — Process::WakePipe), stream frames can carry an + # ephemeral broadcast's ONLY copy of its HTML, so the transport is + # length-prefixed and lossless; drop decisions are made per-message by + # the MasterHub, never by the wire format. + # + # Message shapes (JSON objects; "t" is the discriminator): + # worker → master: {t:"sub", q:} subscribe, synchronous — master acks + # {t:"unsub", q:} unsubscribe, fire-and-forget + # master → worker: {t:"ack", q:} sub acknowledged (LISTEN active) + # {t:"wake", q:, p: } durable (p:nil) or ephemeral wake + # {t:"status", healthy: } listener health broadcast + # + # Reads are blocking (each side owns a dedicated reader thread); a short + # read means the peer died mid-frame and is reported as EOF (nil), never + # as a truncated message. + module HubProtocol + class ProtocolError < StandardError; end + + HEADER_BYTES = 4 + # Generous ceiling for ephemeral HTML payloads; a frame announcing + # more than this is a corrupt stream or a runaway producer — sever + # rather than allocate. + MAX_FRAME_BYTES = 4 * 1024 * 1024 + + module_function + + def encode(message) + json = JSON.generate(message) + bytes = json.b + raise ProtocolError, "frame too large: #{bytes.bytesize} bytes (max #{MAX_FRAME_BYTES})" if + bytes.bytesize > MAX_FRAME_BYTES + + [bytes.bytesize].pack("N") + bytes + end + + # Returns the decoded Hash, or nil on EOF (clean close or peer death + # mid-frame). Raises ProtocolError on an oversized announcement or + # malformed JSON. + def read_frame(io) + header = read_exactly(io, HEADER_BYTES) + return nil unless header + + length = header.unpack1("N") + raise ProtocolError, "frame too large: #{length} bytes (max #{MAX_FRAME_BYTES})" if length > MAX_FRAME_BYTES + + body = read_exactly(io, length) + return nil unless body + + JSON.parse(body.force_encoding(Encoding::UTF_8)) + rescue JSON::ParserError => e + raise ProtocolError, "malformed frame: #{e.message}" + end + + # Blocking read of exactly +count+ bytes; nil on EOF (including EOF + # partway through — IO#read returns the short tail once, then nil). + def read_exactly(io, count) + data = io.read(count) + return nil if data.nil? || data.bytesize < count + + data + end + end + end + end +end diff --git a/spec/pgbus/web/streamer/hub_protocol_spec.rb b/spec/pgbus/web/streamer/hub_protocol_spec.rb new file mode 100644 index 00000000..438ad779 --- /dev/null +++ b/spec/pgbus/web/streamer/hub_protocol_spec.rb @@ -0,0 +1,88 @@ +# frozen_string_literal: true + +require "spec_helper" +require "socket" + +RSpec.describe Pgbus::Web::Streamer::HubProtocol do + let(:sockets) { UNIXSocket.pair } + let(:reader) { sockets[0] } + let(:writer) { sockets[1] } + + after { sockets.each { |s| s.close unless s.closed? } } + + describe ".encode / .read_frame round trip" do + it "round-trips a message hash" do + writer.write(described_class.encode({ "t" => "sub", "q" => "pgbus_stream_chat" })) + + expect(described_class.read_frame(reader)).to eq({ "t" => "sub", "q" => "pgbus_stream_chat" }) + end + + it "keeps multiple back-to-back frames separate" do + writer.write(described_class.encode({ "t" => "ack", "q" => "a" })) + writer.write(described_class.encode({ "t" => "wake", "q" => "b", "p" => "
hi
" })) + + expect(described_class.read_frame(reader)).to eq({ "t" => "ack", "q" => "a" }) + expect(described_class.read_frame(reader)).to eq({ "t" => "wake", "q" => "b", "p" => "
hi
" }) + end + + it "round-trips multibyte payloads (ephemeral HTML is arbitrary UTF-8)" do + payload = { "t" => "wake", "q" => "chat", "p" => "
héllo — ünïcode 🎉
" } + writer.write(described_class.encode(payload)) + + expect(described_class.read_frame(reader)).to eq(payload) + end + + it "reassembles a frame delivered in partial writes" do + frame = described_class.encode({ "t" => "wake", "q" => "chat", "p" => "x" * 512 }) + t = Thread.new do + frame.each_char.each_slice(7) do |chunk| + writer.write(chunk.join) + sleep 0.001 + end + end + + expect(described_class.read_frame(reader)).to include("t" => "wake", "q" => "chat") + t.join + end + end + + describe "EOF handling" do + it "returns nil on a cleanly closed peer" do + writer.close + + expect(described_class.read_frame(reader)).to be_nil + end + + it "returns nil on EOF mid-frame (peer died mid-write)" do + frame = described_class.encode({ "t" => "wake", "q" => "chat", "p" => "x" * 100 }) + writer.write(frame[0, 10]) + writer.close + + expect(described_class.read_frame(reader)).to be_nil + end + end + + describe "guards" do + it "rejects an oversized frame announcement without reading it" do + writer.write([described_class::MAX_FRAME_BYTES + 1].pack("N")) + + expect { described_class.read_frame(reader) } + .to raise_error(described_class::ProtocolError, /frame too large/i) + end + + it "rejects an unencodable oversize payload at encode time" do + huge = { "t" => "wake", "p" => "x" * (described_class::MAX_FRAME_BYTES + 1) } + + expect { described_class.encode(huge) } + .to raise_error(described_class::ProtocolError, /frame too large/i) + end + + it "wraps malformed JSON in a ProtocolError" do + garbage = "not json".b + writer.write([garbage.bytesize].pack("N") + garbage) + + expect { described_class.read_frame(reader) } + .to raise_error(described_class::ProtocolError, /malformed/i) + end + end +end From 8a17e03d724694b9fd5379145ef110c99b670151 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Sun, 2 Aug 2026 17:54:22 +0200 Subject: [PATCH 02/16] =?UTF-8?q?feat(streams):=20MasterHub=20=E2=80=94=20?= =?UTF-8?q?one=20LISTEN=20connection=20per=20web=20host=20(#382=20step=202?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Owns a single Web::Streamer::Listener on the refcounted union of every worker's channels; workers connect lazily over a Unix socket and receive HubProtocol frames. Register-before-LISTEN + ack-after-ensure preserves the no-lost-wake contract cross-process (every sub round-trips the listener's idempotent ack; only UNLISTEN is refcounted). Backpressure: per-worker outbox + writer thread, durable wakes droppable at a cap, ephemeral never dropped, hard-cap eviction severs a non-draining worker so it self-degrades to its own listener. Listener gains alive?/connected? health readers for the status broadcasts. --- lib/pgbus/web/streamer/listener.rb | 12 + lib/pgbus/web/streamer/master_hub.rb | 332 +++++++++++++++++++++ spec/pgbus/web/streamer/master_hub_spec.rb | 281 +++++++++++++++++ 3 files changed, 625 insertions(+) create mode 100644 lib/pgbus/web/streamer/master_hub.rb create mode 100644 spec/pgbus/web/streamer/master_hub_spec.rb diff --git a/lib/pgbus/web/streamer/listener.rb b/lib/pgbus/web/streamer/listener.rb index 1d5a4cd7..bd653b4c 100644 --- a/lib/pgbus/web/streamer/listener.rb +++ b/lib/pgbus/web/streamer/listener.rb @@ -102,6 +102,18 @@ def start self end + # Health signals for the MasterHub's status broadcasts (issue #382). + # Read cross-thread without synchronization: ivar assignment is atomic + # in MRI and a momentarily stale value only delays one status tick — + # these must never touch the connection itself (single-owner, #375). + def alive? + !!@thread&.alive? + end + + def connected? + !@conn.nil? + end + def stop return unless @running diff --git a/lib/pgbus/web/streamer/master_hub.rb b/lib/pgbus/web/streamer/master_hub.rb new file mode 100644 index 00000000..438b714e --- /dev/null +++ b/lib/pgbus/web/streamer/master_hub.rb @@ -0,0 +1,332 @@ +# frozen_string_literal: true + +require "socket" +require "fileutils" + +module Pgbus + module Web + module Streamer + # 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). + class MasterHub + DEFAULT_DURABLE_QUEUE_LIMIT = 256 + DEFAULT_HARD_QUEUE_LIMIT = 1024 + # Status is rebroadcast every REBROADCAST_TICKS status intervals even + # unchanged, so a worker that connected mid-outage converges. + REBROADCAST_TICKS = 5 + + attr_reader :socket_path + + 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 + @table_mutex = Mutex.new + @workers = {} + @queue_refs = Hash.new { |h, k| h[k] = Set.new } + @next_id = 0 + @dropped_durable_wakes = 0 + @evicted_workers = 0 + @running = false + end + + def dropped_durable_wakes + @table_mutex.synchronize { @dropped_durable_wakes } + end + + def evicted_workers + @table_mutex.synchronize { @evicted_workers } + end + + # The factory must return a STARTED listener wired to +dispatch_queue+. + def start + @running = true + @listener = @listener_factory.call(dispatch_queue: @dispatch_queue) + FileUtils.rm_f(@socket_path) + @server = UNIXServer.new(@socket_path) + @accept_thread = Thread.new { accept_loop } + @fanout_thread = Thread.new { fanout_loop } + @status_thread = Thread.new { status_loop } + self + end + + def stop + return self unless @running + + @running = false + close_quietly(@server) + @dispatch_queue << :stop + worker_ids = @table_mutex.synchronize { @workers.keys } + worker_ids.each { |id| cleanup_worker(id) } + [@accept_thread, @fanout_thread, @status_thread].each { |t| t&.join(2) } + @listener&.stop + FileUtils.rm_f(@socket_path) + self + end + + private + + def default_listener_factory + lambda do |dispatch_queue:| + build_connection = -> { Pgbus::DedicatedConnection.connect(@config.streams_connection_options) } + conn = build_connection.call + Pgbus::Process::PrimaryValidator.validate_primary!(conn) + Listener.new( + pg_connection: conn, + dispatch_queue: dispatch_queue, + health_check_ms: @config.streams_listen_health_check_ms, + connection_factory: build_connection, + dispatch_queue_limit: @config.streams_dispatch_queue_limit, + logger: @logger + ).tap(&:start) + end + end + + def accept_loop + loop do + sock = @server.accept + register_worker(sock) + end + rescue IOError, Errno::EBADF, Errno::EINVAL + # server closed during stop + end + + def register_worker(sock) + entry = { + sock: sock, subs: Set.new, outbox: [], durable_count: 0, open: true, + outbox_mutex: Mutex.new, outbox_cond: ConditionVariable.new + } + id = @table_mutex.synchronize do + @next_id += 1 + @workers[@next_id] = entry + @next_id + end + entry[:writer] = Thread.new { writer_loop(id, entry) } + entry[:reader] = Thread.new { reader_loop(id, entry) } + id + end + + def reader_loop(id, entry) + loop do + frame = HubProtocol.read_frame(entry[:sock]) + break if frame.nil? + + handle_frame(id, entry, frame) + end + rescue HubProtocol::ProtocolError => e + @logger.warn { "[Pgbus::Streamer::MasterHub] worker #{id} protocol error: #{e.message}" } + rescue IOError, Errno::EBADF, Errno::ECONNRESET + # severed by eviction or stop + ensure + cleanup_worker(id) + end + + def handle_frame(id, entry, frame) + case frame["t"] + when "sub" then handle_sub(id, entry, frame["q"]) + when "unsub" then handle_unsub(id, frame["q"]) + else + @logger.warn { "[Pgbus::Streamer::MasterHub] worker #{id} sent unknown frame: #{frame["t"].inspect}" } + end + end + + # Register FIRST, LISTEN second, ack LAST — the ordering the no-lost- + # wake contract rests on (see class comment). Runs on this worker's + # reader thread; ensure_listening blocks bounded by the listener's own + # ack budget. + def handle_sub(id, entry, queue) + @table_mutex.synchronize do + entry[:subs].add(queue) + @queue_refs[queue].add(id) + end + @listener.ensure_listening(queue) + enqueue_frame(id, entry, { "t" => "ack", "q" => queue }, droppable: false) + end + + def handle_unsub(id, queue) + release_queue_refs(id, [queue]) + @table_mutex.synchronize { @workers[id]&.[](:subs)&.delete(queue) } + end + + def fanout_loop + loop do + message = @dispatch_queue.pop + break if message == :stop + + deliver(message) + end + rescue StandardError => e + @logger.error { "[Pgbus::Streamer::MasterHub] fanout loop died: #{e.class}: #{e.message}" } + end + + def deliver(message) + frame = { "t" => "wake", "q" => message.queue_name, "p" => message.payload } + droppable = message.payload.nil? + targets = @table_mutex.synchronize do + @queue_refs[message.queue_name].filter_map { |id| [id, @workers[id]] if @workers[id] } + end + targets.each { |id, entry| enqueue_frame(id, entry, frame, droppable: droppable) } + end + + # Non-blocking enqueue with the drop/evict policy. Never blocks the + # fanout thread on one slow worker (the head-of-line lesson from + # issue #315 item 3, applied cross-process). + def enqueue_frame(id, entry, frame, droppable:) + evict = false + entry[:outbox_mutex].synchronize do + return unless entry[:open] + + if droppable && entry[:durable_count] >= @durable_queue_limit + @table_mutex.synchronize { @dropped_durable_wakes += 1 } + return + end + + entry[:outbox] << [frame, droppable] + entry[:durable_count] += 1 if droppable + evict = entry[:outbox].size > @hard_queue_limit + entry[:outbox_cond].signal + end + evict_worker(id, entry) if evict + end + + # Sever a worker that stopped draining. Closing the socket unblocks + # its writer (IOError) and its reader (EOF on the client side makes + # the worker's HubClient fail over to a local listener) — the wedged + # worker degrades itself, never its siblings. + def evict_worker(id, entry) + already = false + entry[:outbox_mutex].synchronize do + already = !entry[:open] + entry[:open] = false + entry[:outbox_cond].broadcast + end + return if already + + @table_mutex.synchronize { @evicted_workers += 1 } + @logger.warn do + "[Pgbus::Streamer::MasterHub] evicting worker #{id}: outbound queue exceeded " \ + "#{@hard_queue_limit} frames (worker not draining) — it falls back to its own listener" + end + close_quietly(entry[:sock]) + end + + def writer_loop(_id, entry) + loop do + frame = nil + entry[:outbox_mutex].synchronize do + entry[:outbox_cond].wait(entry[:outbox_mutex]) while entry[:outbox].empty? && entry[:open] + return unless entry[:open] + + frame, droppable = entry[:outbox].shift + entry[:durable_count] -= 1 if droppable + end + entry[:sock].write(HubProtocol.encode(frame)) + end + rescue IOError, Errno::EPIPE, Errno::ECONNRESET, Errno::EBADF + # severed / worker died; reader-side cleanup handles bookkeeping + end + + def status_loop + last_status = nil + ticks_since_broadcast = 0 + loop do + sleep @status_interval + break unless @running + + healthy = listener_healthy? + ticks_since_broadcast += 1 + next unless healthy != last_status || ticks_since_broadcast >= REBROADCAST_TICKS + + broadcast_status(healthy) + last_status = healthy + ticks_since_broadcast = 0 + end + end + + def listener_healthy? + listener = @listener + !!(listener&.alive? && listener.connected?) + end + + def broadcast_status(healthy) + frame = { "t" => "status", "healthy" => healthy } + entries = @table_mutex.synchronize { @workers.to_a } + entries.each { |id, entry| enqueue_frame(id, entry, frame, droppable: false) } + end + + # Idempotent teardown for one worker — reachable from its reader's + # ensure, an eviction, and stop. + def cleanup_worker(id) + entry = @table_mutex.synchronize { @workers.delete(id) } + return unless entry + + entry[:outbox_mutex].synchronize do + entry[:open] = false + entry[:outbox_cond].broadcast + end + close_quietly(entry[:sock]) + release_queue_refs(id, entry[:subs].to_a) + end + + # Decrement refcounts; UNLISTEN queues that hit zero (async — no + # correctness path waits on unlisten, mirroring remove_listening). + def release_queue_refs(id, queues) + released = @table_mutex.synchronize do + queues.select do |q| + refs = @queue_refs[q] + refs.delete(id) + @queue_refs.delete(q) if refs.empty? + end + end + released.each { |q| @listener.remove_listening(q) } + end + + def close_quietly(io) + io.close if io && !io.closed? + rescue IOError, Errno::EBADF + nil + end + end + end + end +end diff --git a/spec/pgbus/web/streamer/master_hub_spec.rb b/spec/pgbus/web/streamer/master_hub_spec.rb new file mode 100644 index 00000000..aeeef732 --- /dev/null +++ b/spec/pgbus/web/streamer/master_hub_spec.rb @@ -0,0 +1,281 @@ +# frozen_string_literal: true + +require "spec_helper" +require "socket" +require "tmpdir" + +RSpec.describe Pgbus::Web::Streamer::MasterHub do + subject(:hub) do + described_class.new( + config: config, + socket_path: socket_path, + listener_factory: listener_factory, + status_interval: 0.05, + logger: logger + ) + end + + let(:config) do + Pgbus::Configuration.new.tap do |c| + c.queue_prefix = "pgbus_test" + c.database_url = "postgres://fake@localhost/fake" + end + end + let(:tmpdir) { Dir.mktmpdir("pgbus-hub-spec") } + let(:socket_path) { File.join(tmpdir, "hub.sock") } + let(:logger) { Logger.new(IO::NULL) } + + let(:fake_listener) do + instance_double( + Pgbus::Web::Streamer::Listener, + ensure_listening: :done, remove_listening: nil, stop: nil, + alive?: true, connected?: true + ) + end + # Captures the dispatch queue the hub hands its listener, so specs can + # inject WakeMessages as if NOTIFY fired. + let(:captured) { {} } + let(:listener_factory) do + lambda do |dispatch_queue:| + captured[:dispatch_queue] = dispatch_queue + fake_listener + end + end + + after do + hub.stop + FileUtils.remove_entry(tmpdir) if File.directory?(tmpdir) + end + + def connect_worker + UNIXSocket.new(socket_path) + end + + def send_frame(sock, message) + sock.write(Pgbus::Web::Streamer::HubProtocol.encode(message)) + end + + def read_frame(sock, timeout: 2) + raise "no frame within #{timeout}s" unless sock.wait_readable(timeout) + + Pgbus::Web::Streamer::HubProtocol.read_frame(sock) + end + + # Reads frames until one matches the type (status rebroadcasts interleave). + def read_frame_of_type(sock, type, timeout: 2) + deadline = Time.now + timeout + while Time.now < deadline + frame = read_frame(sock, timeout: timeout) + return frame if frame && frame["t"] == type + end + raise "no #{type} frame within #{timeout}s" + end + + def wake(queue, payload = nil) + captured[:dispatch_queue] << Pgbus::Web::Streamer::Listener::WakeMessage.new( + queue_name: queue, payload: payload + ) + end + + describe "subscription lifecycle" do + it "acks a sub after the listener actually LISTENs" do + hub.start + worker = connect_worker + send_frame(worker, { "t" => "sub", "q" => "pgbus_test_chat" }) + + expect(read_frame_of_type(worker, "ack")).to include("q" => "pgbus_test_chat") + expect(fake_listener).to have_received(:ensure_listening).with("pgbus_test_chat") + worker.close + end + + it "registers the subscription BEFORE the LISTEN completes (no lost-wake gap)" do + # A wake that fires between LISTEN-active and sub-registration would be + # lost. Pin the ordering: block ensure_listening, inject a wake while + # blocked, then release — the worker must still receive that wake. + gate = Queue.new + allow(fake_listener).to receive(:ensure_listening) do |_q| + gate.pop + :done + end + hub.start + worker = connect_worker + send_frame(worker, { "t" => "sub", "q" => "pgbus_test_chat" }) + sleep 0.1 # let the sub reach the blocked ensure_listening + wake("pgbus_test_chat", nil) + gate << :go + + # The wake delivered while LISTEN was still in flight precedes the ack + # in the outbox FIFO — collect both (status rebroadcasts interleave). + seen = {} + until seen.key?("ack") && seen.key?("wake") + frame = read_frame(worker) + seen[frame["t"]] = frame unless frame["t"] == "status" + end + expect(seen["ack"]).to include("q" => "pgbus_test_chat") + expect(seen["wake"]).to include("q" => "pgbus_test_chat") + worker.close + end + + it "acks every subscriber through the listener (idempotent) but UNLISTENs only at zero refs" do + # Each sub must round-trip ensure_listening so ITS ack carries the + # LISTEN-active guarantee (a refcount shortcut would ack subscriber B + # while subscriber A's LISTEN was still in flight — reopening the + # lost-wake gap). ensure_listening is cheap when already listening. + hub.start + worker_a = connect_worker + worker_b = connect_worker + send_frame(worker_a, { "t" => "sub", "q" => "pgbus_test_chat" }) + read_frame_of_type(worker_a, "ack") + send_frame(worker_b, { "t" => "sub", "q" => "pgbus_test_chat" }) + read_frame_of_type(worker_b, "ack") + + expect(fake_listener).to have_received(:ensure_listening).with("pgbus_test_chat").twice + + send_frame(worker_a, { "t" => "unsub", "q" => "pgbus_test_chat" }) + sleep 0.1 + expect(fake_listener).not_to have_received(:remove_listening) + + send_frame(worker_b, { "t" => "unsub", "q" => "pgbus_test_chat" }) + sleep 0.1 + expect(fake_listener).to have_received(:remove_listening).with("pgbus_test_chat") + [worker_a, worker_b].each(&:close) + end + + it "releases a dead worker's subscriptions on EOF" do + hub.start + worker = connect_worker + send_frame(worker, { "t" => "sub", "q" => "pgbus_test_chat" }) + read_frame_of_type(worker, "ack") + + worker.close + sleep 0.2 + + expect(fake_listener).to have_received(:remove_listening).with("pgbus_test_chat") + end + end + + describe "wake fanout" do + it "routes wakes only to subscribed workers, payload intact" do + hub.start + worker_a = connect_worker + worker_b = connect_worker + send_frame(worker_a, { "t" => "sub", "q" => "pgbus_test_chat" }) + read_frame_of_type(worker_a, "ack") + send_frame(worker_b, { "t" => "sub", "q" => "pgbus_test_other" }) + read_frame_of_type(worker_b, "ack") + + wake("pgbus_test_chat", "
ephemeral
") + + frame = read_frame_of_type(worker_a, "wake") + expect(frame).to include("q" => "pgbus_test_chat", "p" => "
ephemeral
") + expect(worker_b.wait_readable(0.3)).to be_falsey.or(satisfy do |r| + # Only status frames may arrive on B; never a wake for chat. + r && Pgbus::Web::Streamer::HubProtocol.read_frame(worker_b)["t"] != "wake" + end) + [worker_a, worker_b].each(&:close) + end + + it "delivers durable wakes with a null payload" do + hub.start + worker = connect_worker + send_frame(worker, { "t" => "sub", "q" => "pgbus_test_chat" }) + read_frame_of_type(worker, "ack") + + wake("pgbus_test_chat", nil) + + expect(read_frame_of_type(worker, "wake")).to include("q" => "pgbus_test_chat", "p" => nil) + worker.close + end + end + + describe "backpressure" do + subject(:hub) do + described_class.new( + config: config, socket_path: socket_path, listener_factory: listener_factory, + status_interval: 60, durable_queue_limit: 3, hard_queue_limit: 8, logger: logger + ) + end + + it "drops excess durable wakes for a non-draining worker but keeps ephemeral" do + hub.start + worker = connect_worker + send_frame(worker, { "t" => "sub", "q" => "pgbus_test_chat" }) + sleep 0.1 + # Wedge the writer below the hard cap: a few LARGE ephemeral frames + # fill the kernel socket buffer (worker never reads), blocking the + # writer mid-write with the outbox well under hard_queue_limit(8). + 3.times { wake("pgbus_test_chat", "x" * 262_144) } + sleep 0.2 + # Durable wakes now pile into the outbox: droppable beyond limit 3. + 20.times { wake("pgbus_test_chat", nil) } + sleep 0.2 + + expect(hub.dropped_durable_wakes).to be > 0 + expect(hub.evicted_workers).to eq(0) + worker.close + end + + it "evicts a worker whose queue exceeds the hard cap (its fallback takes over)" do + hub.start + worker = connect_worker + send_frame(worker, { "t" => "sub", "q" => "pgbus_test_chat" }) + sleep 0.1 + # Ephemeral frames are never dropped, so they push past the hard cap → + # eviction severs the socket. + 200.times { wake("pgbus_test_chat", "x" * 65_536) } + + deadline = Time.now + 5 + severed = false + while Time.now < deadline + begin + worker.read_nonblock(1_048_576) + rescue IO::WaitReadable + sleep 0.05 + rescue EOFError, Errno::ECONNRESET + severed = true + break + end + end + expect(severed).to be true + expect(hub.evicted_workers).to eq(1) + end + end + + describe "status broadcast" do + it "broadcasts degraded and healthy transitions" do + hub.start + worker = connect_worker + send_frame(worker, { "t" => "sub", "q" => "pgbus_test_chat" }) + read_frame_of_type(worker, "ack") + + allow(fake_listener).to receive(:connected?).and_return(false) + frame = read_frame_of_type(worker, "status", timeout: 3) + expect(frame).to include("healthy" => false) + + allow(fake_listener).to receive(:connected?).and_return(true) + frame = read_frame_of_type(worker, "status", timeout: 3) + expect(frame).to include("healthy" => true) + worker.close + end + end + + describe "#stop" do + it "stops the listener, closes clients, and unlinks the socket" do + hub.start + worker = connect_worker + + hub.stop + + expect(fake_listener).to have_received(:stop) + expect(File.exist?(socket_path)).to be false + expect(worker.wait_readable(1) && Pgbus::Web::Streamer::HubProtocol.read_frame(worker)).to be_nil + worker.close + end + + it "replaces a stale socket file on start" do + File.write(socket_path, "stale") + expect { hub.start }.not_to raise_error + expect(File.socket?(socket_path)).to be true + end + end +end From e8452ebeea3ea14fb24837181cc0e86b291fa30a Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Sun, 2 Aug 2026 17:55:53 +0200 Subject: [PATCH 03/16] =?UTF-8?q?feat(streams):=20HubClient=20=E2=80=94=20?= =?UTF-8?q?worker-side=20master=20hub=20transport=20(#382=20step=203)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Listener-shaped surface (sync ensure_listening ack contract preserved cross-process, async remove_listening); wakes re-materialize into the worker's dispatch queue as WakeMessages. Never retries: connect refusal, ack deadline, or EOF marks the transport dead, fails pending subs, and fires on_failure exactly once — the FailoverListener's swap cue. --- lib/pgbus/web/streamer/hub_client.rb | 172 +++++++++++++++++++++ spec/pgbus/web/streamer/hub_client_spec.rb | 147 ++++++++++++++++++ 2 files changed, 319 insertions(+) create mode 100644 lib/pgbus/web/streamer/hub_client.rb create mode 100644 spec/pgbus/web/streamer/hub_client_spec.rb diff --git a/lib/pgbus/web/streamer/hub_client.rb b/lib/pgbus/web/streamer/hub_client.rb new file mode 100644 index 00000000..752d70ca --- /dev/null +++ b/lib/pgbus/web/streamer/hub_client.rb @@ -0,0 +1,172 @@ +# frozen_string_literal: true + +require "socket" + +module Pgbus + module Web + module Streamer + # Worker-side client for the MasterHub (issue #382). Presents the same + # surface the Dispatcher consumes from a Listener — synchronous + # `ensure_listening` (the no-lost-broadcast ack contract, now crossing + # the process boundary), async `remove_listening` — while wakes arrive + # as HubProtocol frames and are re-materialized into the worker's + # dispatch queue as WakeMessages. + # + # Failure model: this class never retries. Connect refusal, an ack + # deadline, or transport EOF (master died / eviction) marks the client + # dead, fails every pending sub, and fires +on_failure+ exactly once — + # the FailoverListener's cue to swap in a per-worker Listener. One-way: + # once a worker has fallen back it stays local until it recycles + # (settled on #382 — no flap-back complexity). + class HubClient + class HubUnavailableError < StandardError; end + + # Optimistic before the first status broadcast, mirroring WakePipe / + # NotifyListener: a just-connected worker isn't treated as degraded + # before the hub has said anything. + def initialize(socket_path:, dispatch_queue:, ack_timeout: 2.0, + on_failure: nil, logger: Pgbus.logger) + @socket_path = socket_path + @dispatch_queue = dispatch_queue + @ack_timeout = ack_timeout + @on_failure = on_failure + @logger = logger + @write_mutex = Mutex.new + @ack_mutex = Mutex.new + @pending_acks = Hash.new { |h, k| h[k] = [] } + @hub_healthy = true + @dead = false + @stopping = false + @sock = nil + @reader = nil + end + + def connect + @sock = UNIXSocket.new(@socket_path) + @reader = Thread.new { reader_loop } + self + rescue SystemCallError => e + raise HubUnavailableError, "cannot reach master hub at #{@socket_path}: #{e.class}: #{e.message}" + end + + def hub_healthy? + @hub_healthy + end + + def dead? + @dead + end + + # Synchronous, bounded: returns :done once the master has confirmed + # LISTEN is active for +queue+. Raises HubUnavailableError on a dead + # transport or an expired ack deadline (which also kills the + # transport — a hub that can't ack in time can't be trusted with the + # no-lost-broadcast contract either). + def ensure_listening(queue) + raise HubUnavailableError, "master hub transport is dead" if @dead + + waiter = Queue.new + @ack_mutex.synchronize { @pending_acks[queue] << waiter } + write_frame({ "t" => "sub", "q" => queue }) + + result = waiter.pop(timeout: @ack_timeout) + if result.nil? + discard_waiter(queue, waiter) + mark_dead("sub ack for #{queue} not received within #{@ack_timeout}s") + raise HubUnavailableError, "master hub ack timeout for #{queue}" + end + raise HubUnavailableError, "master hub died while awaiting ack for #{queue}" if result == :dead + + :done + end + + # Lazy GC, fire-and-forget — no correctness path waits on UNLISTEN + # (mirrors Listener#remove_listening). A dead transport is a no-op: + # the master's EOF cleanup already released this worker's refs. + def remove_listening(queue) + return if @dead + + write_frame({ "t" => "unsub", "q" => queue }) + rescue HubUnavailableError + nil + end + + def stop + @stopping = true + close_quietly(@sock) + @reader&.join(2) + @reader = nil + self + end + + private + + def reader_loop + loop do + frame = HubProtocol.read_frame(@sock) + break if frame.nil? + + handle_frame(frame) + end + mark_dead("master hub closed the transport") unless @stopping + rescue HubProtocol::ProtocolError => e + mark_dead("master hub protocol error: #{e.message}") unless @stopping + rescue IOError, Errno::EBADF, Errno::ECONNRESET + mark_dead("master hub transport error") unless @stopping + end + + def handle_frame(frame) + case frame["t"] + when "wake" + @dispatch_queue << Listener::WakeMessage.new(queue_name: frame["q"], payload: frame["p"]) + when "ack" + @ack_mutex.synchronize { @pending_acks[frame["q"]].shift }&.push(:ack) + when "status" + @hub_healthy = frame["healthy"] + else + @logger.warn { "[Pgbus::Streamer::HubClient] unknown frame from master: #{frame["t"].inspect}" } + end + end + + # Frames must never interleave — all writes go through one mutex + # (writers: dispatcher thread via ensure/remove; no writer thread + # needed client-side, sub/unsub frames are tiny). + def write_frame(message) + @write_mutex.synchronize { @sock.write(HubProtocol.encode(message)) } + rescue IOError, Errno::EPIPE, Errno::EBADF, Errno::ECONNRESET => e + mark_dead("write to master hub failed: #{e.class}") + raise HubUnavailableError, "master hub transport is dead" + end + + # Idempotent: first caller flips @dead, fails every waiter, fires + # on_failure once. Reachable from the reader (EOF/protocol error) and + # from ack timeouts / failed writes on caller threads. + def mark_dead(reason) + waiters = @ack_mutex.synchronize do + return if @dead + + @dead = true + drained = @pending_acks.values.flatten + @pending_acks.clear + drained + end + @hub_healthy = false + waiters.each { |w| w << :dead } + close_quietly(@sock) + @logger.warn { "[Pgbus::Streamer::HubClient] #{reason} — falling back to a per-worker listener" } + @on_failure&.call + end + + def discard_waiter(queue, waiter) + @ack_mutex.synchronize { @pending_acks[queue].delete(waiter) } + end + + def close_quietly(io) + io.close if io && !io.closed? + rescue IOError, Errno::EBADF + nil + end + end + end + end +end diff --git a/spec/pgbus/web/streamer/hub_client_spec.rb b/spec/pgbus/web/streamer/hub_client_spec.rb new file mode 100644 index 00000000..58e48672 --- /dev/null +++ b/spec/pgbus/web/streamer/hub_client_spec.rb @@ -0,0 +1,147 @@ +# frozen_string_literal: true + +require "spec_helper" +require "socket" +require "tmpdir" + +RSpec.describe Pgbus::Web::Streamer::HubClient do + subject(:client) do + described_class.new( + socket_path: socket_path, + dispatch_queue: dispatch_queue, + ack_timeout: 0.5, + on_failure: -> { failures << :failed }, + logger: logger + ) + end + + let(:tmpdir) { Dir.mktmpdir("pgbus-hub-client-spec") } + let(:socket_path) { File.join(tmpdir, "hub.sock") } + let(:dispatch_queue) { Queue.new } + let(:failures) { [] } + let(:logger) { Logger.new(IO::NULL) } + + let(:server) { UNIXServer.new(socket_path) } + let(:master_side) { [] } + + after do + client.stop + master_side.each { |s| s.close unless s.closed? } + server.close unless server.closed? + FileUtils.remove_entry(tmpdir) if File.directory?(tmpdir) + end + + def accept_master + server # bind first + thread = Thread.new { server.accept } + yield if block_given? + sock = thread.value + master_side << sock + sock + end + + def master_read(sock) + Pgbus::Web::Streamer::HubProtocol.read_frame(sock) + end + + def master_send(sock, message) + sock.write(Pgbus::Web::Streamer::HubProtocol.encode(message)) + end + + describe "subscription round trip" do + it "ensure_listening blocks until the master acks" do + master = accept_master { client.connect } + acker = Thread.new do + frame = master_read(master) + master_send(master, { "t" => "ack", "q" => frame["q"] }) if frame["t"] == "sub" + end + + expect(client.ensure_listening("pgbus_stream_chat")).to eq(:done) + acker.join + end + + it "raises HubUnavailableError when the ack never arrives (and marks the transport dead)" do + accept_master { client.connect } # master reads nothing, acks nothing + + expect { client.ensure_listening("pgbus_stream_chat") } + .to raise_error(described_class::HubUnavailableError, /ack/i) + expect { client.ensure_listening("pgbus_stream_other") } + .to raise_error(described_class::HubUnavailableError) + end + + it "remove_listening sends an unsub frame without waiting" do + master = accept_master { client.connect } + + client.remove_listening("pgbus_stream_chat") + + expect(master_read(master)).to eq({ "t" => "unsub", "q" => "pgbus_stream_chat" }) + end + end + + describe "wake delivery" do + it "pushes wake frames into the dispatch queue as WakeMessages, payload intact" do + master = accept_master { client.connect } + + master_send(master, { "t" => "wake", "q" => "pgbus_stream_chat", "p" => "
hi
" }) + + message = dispatch_queue.pop + expect(message).to be_a(Pgbus::Web::Streamer::Listener::WakeMessage) + expect(message.queue_name).to eq("pgbus_stream_chat") + expect(message.payload).to eq("
hi
") + end + + it "delivers durable wakes with a nil payload" do + master = accept_master { client.connect } + + master_send(master, { "t" => "wake", "q" => "pgbus_stream_chat", "p" => nil }) + + expect(dispatch_queue.pop.payload).to be_nil + end + end + + describe "status tracking" do + it "tracks the hub's health broadcasts" do + master = accept_master { client.connect } + expect(client.hub_healthy?).to be true # optimistic before first status + + master_send(master, { "t" => "status", "healthy" => false }) + sleep 0.1 + expect(client.hub_healthy?).to be false + + master_send(master, { "t" => "status", "healthy" => true }) + sleep 0.1 + expect(client.hub_healthy?).to be true + end + end + + describe "transport failure" do + it "fires on_failure and fails pending subs when the master dies (EOF)" do + master = accept_master { client.connect } + + waiter = Thread.new do + client.ensure_listening("pgbus_stream_chat") + rescue described_class::HubUnavailableError + :raised + end + sleep 0.1 + master.close + + expect(waiter.value).to eq(:raised) + sleep 0.1 + expect(failures).to eq([:failed]) + end + + it "raises HubUnavailableError from connect when no socket exists" do + expect { client.connect }.to raise_error(described_class::HubUnavailableError) + end + + it "does not fire on_failure for a clean stop" do + accept_master { client.connect } + + client.stop + sleep 0.1 + + expect(failures).to be_empty + end + end +end From 27d976c7c808ef9219e3e3b1b3d646636f301e43 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Sun, 2 Aug 2026 17:57:41 +0200 Subject: [PATCH 04/16] =?UTF-8?q?feat(streams):=20FailoverListener=20?= =?UTF-8?q?=E2=80=94=20one-way=20hub=20=E2=86=92=20per-worker=20listener?= =?UTF-8?q?=20seam=20(#382=20step=204)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Records the subscription set; on hub transport death (async on_failure or a synchronous ensure failure) builds the local Listener once, re-LISTENs the recorded set, and swaps. Never raises to the dispatcher: a double failure (hub dead + local build failing) degrades to the Listener's existing nil-on-timeout contract until the worker recycles. --- lib/pgbus/web/streamer/failover_listener.rb | 92 +++++++++++++++ .../web/streamer/failover_listener_spec.rb | 107 ++++++++++++++++++ 2 files changed, 199 insertions(+) create mode 100644 lib/pgbus/web/streamer/failover_listener.rb create mode 100644 spec/pgbus/web/streamer/failover_listener_spec.rb diff --git a/lib/pgbus/web/streamer/failover_listener.rb b/lib/pgbus/web/streamer/failover_listener.rb new file mode 100644 index 00000000..abb8ad06 --- /dev/null +++ b/lib/pgbus/web/streamer/failover_listener.rb @@ -0,0 +1,92 @@ +# frozen_string_literal: true + +module Pgbus + module Web + module Streamer + # The worker-side seam between the two listening modes (issue #382): + # starts on the master hub (HubClient) and fails over — once, one-way — + # to a per-worker Listener when the hub transport dies (master gone, + # ack deadline, eviction). The Dispatcher/Instance consume the same + # ensure_listening/remove_listening/stop surface either way and never + # learn which mode is active. + # + # Fallback direction is settled on #382: per-worker listener, not + # poll-only — ephemeral broadcasts have no polling equivalent (their + # payload exists only in the NOTIFY), so an outage trades connections + # for unchanged semantics. Once fallen back, the worker stays local + # until it recycles; no flap-back. + # + # The subscription set is recorded here so failover can rebuild the + # exact LISTEN set on the fresh local connection before anything else + # relies on it. ensure_listening NEVER raises to the dispatcher: on a + # double failure (hub dead AND local build failing — e.g. DB down) it + # logs and returns nil, matching the Listener's own ack-timeout + # contract, which the dispatcher already tolerates. + class FailoverListener + def initialize(hub_client:, local_listener_factory:, logger: Pgbus.logger) + @hub_client = hub_client + @local_listener_factory = local_listener_factory + @logger = logger + @mutex = Mutex.new + @subscriptions = Set.new + @impl = hub_client + @failed_over = false + end + + def ensure_listening(queue) + @mutex.synchronize { @subscriptions.add(queue) } + current_impl.ensure_listening(queue) + rescue HubClient::HubUnavailableError + fail_over! + begin + current_impl.ensure_listening(queue) + rescue HubClient::HubUnavailableError + # fail_over! itself failed (factory raised) and @impl is still the + # dead client — reported there; honor the nil-on-timeout contract. + nil + end + end + + def remove_listening(queue) + @mutex.synchronize { @subscriptions.delete(queue) } + current_impl.remove_listening(queue) + rescue HubClient::HubUnavailableError + nil + end + + # Idempotent, callable from the client's on_failure (reader thread) + # and from a synchronous ensure failure (dispatcher thread) — the + # mutex serializes them; the second caller finds the swap done. + def fail_over! + @mutex.synchronize do + return if @failed_over + + @failed_over = true + local = @local_listener_factory.call + @subscriptions.each { |q| local.ensure_listening(q) } + @impl = local + end + rescue StandardError => e + # Hub dead AND the local listener can't be built (DB down, config + # broken). Leave @impl on the dead client — every ensure_listening + # resolves nil and the dispatcher rides its existing timeout + # tolerance until the worker recycles. + @logger.error do + "[Pgbus::Streamer::FailoverListener] fallback listener failed to build " \ + "(#{e.class}: #{e.message}) — streams degraded until this worker recycles" + end + end + + def stop + current_impl.stop + end + + private + + def current_impl + @mutex.synchronize { @impl } + end + end + end + end +end diff --git a/spec/pgbus/web/streamer/failover_listener_spec.rb b/spec/pgbus/web/streamer/failover_listener_spec.rb new file mode 100644 index 00000000..4c75bf27 --- /dev/null +++ b/spec/pgbus/web/streamer/failover_listener_spec.rb @@ -0,0 +1,107 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe Pgbus::Web::Streamer::FailoverListener do + subject(:failover) do + described_class.new( + hub_client: hub_client, + local_listener_factory: local_listener_factory, + logger: logger + ) + end + + let(:hub_client) do + instance_double(Pgbus::Web::Streamer::HubClient, + ensure_listening: :done, remove_listening: nil, stop: nil) + end + let(:local_listener) do + instance_double(Pgbus::Web::Streamer::Listener, + ensure_listening: :done, remove_listening: nil, stop: nil) + end + let(:factory_calls) { [] } + let(:local_listener_factory) do + lambda do + factory_calls << :built + local_listener + end + end + let(:logger) { Logger.new(IO::NULL) } + + describe "hub mode (healthy)" do + it "delegates ensure_listening to the hub client and records the subscription" do + expect(failover.ensure_listening("pgbus_stream_chat")).to eq(:done) + expect(hub_client).to have_received(:ensure_listening).with("pgbus_stream_chat") + expect(factory_calls).to be_empty + end + + it "delegates remove_listening and forgets the subscription" do + failover.ensure_listening("pgbus_stream_chat") + failover.remove_listening("pgbus_stream_chat") + + expect(hub_client).to have_received(:remove_listening).with("pgbus_stream_chat") + end + end + + describe "failover on asynchronous transport death (on_failure)" do + it "builds ONE local listener and re-subscribes every recorded subscription" do + failover.ensure_listening("pgbus_stream_a") + failover.ensure_listening("pgbus_stream_b") + failover.remove_listening("pgbus_stream_a") + + failover.fail_over! + failover.fail_over! # idempotent — e.g. on_failure raced with an ensure error + + expect(factory_calls).to eq([:built]) + expect(local_listener).to have_received(:ensure_listening).with("pgbus_stream_b") + expect(local_listener).not_to have_received(:ensure_listening).with("pgbus_stream_a") + end + + it "routes subsequent calls to the local listener" do + failover.fail_over! + failover.ensure_listening("pgbus_stream_chat") + + expect(local_listener).to have_received(:ensure_listening).with("pgbus_stream_chat") + expect(hub_client).not_to have_received(:ensure_listening) + end + end + + describe "failover on a synchronous ensure failure" do + it "falls over and completes the sub on the local listener (ack contract preserved)" do + allow(hub_client).to receive(:ensure_listening) + .and_raise(Pgbus::Web::Streamer::HubClient::HubUnavailableError, "dead") + + expect(failover.ensure_listening("pgbus_stream_chat")).to eq(:done) + # Twice: once rebuilding the recorded set inside fail_over!, once for + # the retried call itself — both land on the local listener. + expect(local_listener).to have_received(:ensure_listening).with("pgbus_stream_chat").twice + end + end + + describe "double failure (local listener factory raises)" do + let(:local_listener_factory) { -> { raise StandardError, "db down" } } + + it "never raises to the dispatcher — logs and returns nil (Listener's timeout contract)" do + allow(hub_client).to receive(:ensure_listening) + .and_raise(Pgbus::Web::Streamer::HubClient::HubUnavailableError, "dead") + allow(logger).to receive(:error) + + expect(failover.ensure_listening("pgbus_stream_chat")).to be_nil + expect(logger).to have_received(:error) + end + end + + describe "#stop" do + it "stops the hub client in hub mode" do + failover.stop + expect(hub_client).to have_received(:stop) + end + + it "stops the local listener after failover" do + failover.fail_over! + failover.stop + + expect(local_listener).to have_received(:stop) + end + end +end From e24d294b809c93bb19ead9ab9a7a36beb71a3307 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Sun, 2 Aug 2026 17:58:21 +0200 Subject: [PATCH 05/16] =?UTF-8?q?feat(config):=20streams=5Flisten=5Fscope?= =?UTF-8?q?=20=E2=80=94=20:master=20(default)=20|=20:process=20(#382=20ste?= =?UTF-8?q?p=205)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/app/models/config_reference.rb | 4 ++++ lib/pgbus/configuration.rb | 29 +++++++++++++++++++++++++++++ spec/pgbus/configuration_spec.rb | 25 +++++++++++++++++++++++++ 3 files changed, 58 insertions(+) diff --git a/docs/app/models/config_reference.rb b/docs/app/models/config_reference.rb index 6ef852f4..8cdf47ed 100644 --- a/docs/app/models/config_reference.rb +++ b/docs/app/models/config_reference.rb @@ -51,6 +51,10 @@ module ConfigReference { name: "worker_notify_scope", type: "Symbol", default: ":supervisor", desc: "Where the LISTEN connection lives: :supervisor shares ONE direct connection per host " \ "(forks woken over pipes); :fork keeps one dedicated connection per worker/consumer fork." }, + { name: "streams_listen_scope", type: "Symbol", default: ":master", + desc: "Where the streams LISTEN connection lives: :master shares ONE connection per web host " \ + "(Puma workers connect to a master hub, with automatic per-worker fallback); " \ + ":process keeps one per web process." }, { name: "zombie_detection", type: "Boolean", default: "true", desc: "Detect and reclaim work from crashed workers." } ], "Dispatcher & maintenance" => [ diff --git a/lib/pgbus/configuration.rb b/lib/pgbus/configuration.rb index dc754069..69f5ce45 100644 --- a/lib/pgbus/configuration.rb +++ b/lib/pgbus/configuration.rb @@ -270,6 +270,7 @@ def initialize @worker_notify_wakeup = nil @worker_notify_scope = :supervisor + @streams_listen_scope = :master @worker_notify_host = nil @worker_notify_port = nil @worker_notify_database_url = nil @@ -632,6 +633,34 @@ def doctor_on_boot=(mode) @doctor_on_boot = coerced end + # Where the streams LISTEN connection lives (issue #382): + # :master (default) — ONE shared listener in the preforking web master + # (MasterHub); workers connect lazily over a Unix socket and fall back + # to a per-worker listener whenever the hub is absent or dies. + # :process — one listener per web process: the pre-0.13 behavior, and + # the automatic behavior on single-mode / non-preforking servers. + attr_reader :streams_listen_scope + + VALID_STREAMS_LISTEN_SCOPES = %i[master process].freeze + + def streams_listen_scope=(scope) + coerced = case scope + when Symbol then scope + when String then scope.to_sym + else + raise Pgbus::ConfigurationError, + "Invalid streams_listen_scope type: #{scope.class}. " \ + "Must be :master (one shared LISTEN connection per web host) or :process (one per worker)" + end + unless VALID_STREAMS_LISTEN_SCOPES.include?(coerced) + raise Pgbus::ConfigurationError, + "Invalid streams_listen_scope: #{coerced.inspect}. " \ + "Must be :master (one shared LISTEN connection per web host) or :process (one per worker)" + end + + @streams_listen_scope = coerced + end + VALID_WORKER_NOTIFY_SCOPES = %i[supervisor fork].freeze # Validated at assignment time like the other enum options. A String is diff --git a/spec/pgbus/configuration_spec.rb b/spec/pgbus/configuration_spec.rb index ac2ccc19..2f8886f6 100644 --- a/spec/pgbus/configuration_spec.rb +++ b/spec/pgbus/configuration_spec.rb @@ -1648,6 +1648,31 @@ end end + describe "#streams_listen_scope" do + # Where the streams LISTEN connection lives (issue #382): :master (default) + # runs ONE shared listener in the preforking master (workers connect over + # a Unix socket); :process keeps one listener per web process. + + it "defaults to :master" do + expect(config.streams_listen_scope).to eq(:master) + end + + it "accepts :process" do + config.streams_listen_scope = :process + expect(config.streams_listen_scope).to eq(:process) + end + + it "coerces a String" do + config.streams_listen_scope = "process" + expect(config.streams_listen_scope).to eq(:process) + end + + it "rejects an unknown scope with an actionable error" do + expect { config.streams_listen_scope = :hosted } + .to raise_error(Pgbus::ConfigurationError, /streams_listen_scope.*:master.*:process/m) + end + end + describe "#worker_notify_connection_options" do # Mirrors streams_connection_options: defaults to connection_options, # overridable so the listener's persistent LISTEN connection can be From 30c6ba65b89f9ba3d99db48a6342a954b8cc53b9 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Sun, 2 Aug 2026 18:01:07 +0200 Subject: [PATCH 06/16] feat(streams): Instance selects hub vs per-worker listener by scope (#382 step 6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit :master + reachable hub socket (PGBUS_STREAMS_HUB_SOCKET) → FailoverListener over a HubClient with NO per-worker LISTEN connection opened; scope :process, an absent/dead socket (single mode, hub failed to start), or a refused connect keeps today's per-worker Listener. Streamer Listener's channel constants are now single-sourced from NotifyListener. --- lib/pgbus/web/streamer/failover_listener.rb | 6 ++ lib/pgbus/web/streamer/instance.rb | 89 ++++++++++++++++----- lib/pgbus/web/streamer/listener.rb | 7 +- spec/pgbus/web/streamer/instance_spec.rb | 63 +++++++++++++++ 4 files changed, 143 insertions(+), 22 deletions(-) diff --git a/lib/pgbus/web/streamer/failover_listener.rb b/lib/pgbus/web/streamer/failover_listener.rb index abb8ad06..4307fe97 100644 --- a/lib/pgbus/web/streamer/failover_listener.rb +++ b/lib/pgbus/web/streamer/failover_listener.rb @@ -33,6 +33,12 @@ def initialize(hub_client:, local_listener_factory:, logger: Pgbus.logger) @failed_over = false end + # Interface parity with Listener for Instance#start: the hub client + # connected at construction and the fallback starts itself on swap. + def start + self + end + def ensure_listening(queue) @mutex.synchronize { @subscriptions.add(queue) } current_impl.ensure_listening(queue) diff --git a/lib/pgbus/web/streamer/instance.rb b/lib/pgbus/web/streamer/instance.rb index 6850f429..67adfac4 100644 --- a/lib/pgbus/web/streamer/instance.rb +++ b/lib/pgbus/web/streamer/instance.rb @@ -39,7 +39,6 @@ def initialize( @dispatch_queue = dispatch_queue || Queue.new @stream_counter = StreamCounter.new - @pg_connection = pg_connection || build_pg_connection # Self-tuning streams-pool autoscaler (issue #323). Opt-in; nil unless # enabled AND on the dedicated connection path (the shared-AR streams # pool aliases the non-thread-safe job pool and resize is a no-op there). @@ -50,25 +49,7 @@ def initialize( if @config.streams_pool_autoscale && !@client.shared_connection? Pgbus::Streams::PoolAutoscaler.new(client: @client, config: @config, logger: @logger) end - @listener = Listener.new( - pg_connection: @pg_connection, - dispatch_queue: @dispatch_queue, - health_check_ms: @config.streams_listen_health_check_ms, - # Opt-in dispatch-queue backpressure (issue #315 item 3). 0 = - # unbounded (default). The queue itself stays an unbounded - # Queue.new so the request-thread Connect push and the dispatcher's - # own prune_dead self-post never block. - dispatch_queue_limit: @config.streams_dispatch_queue_limit, - maintenance: build_autoscale_maintenance, - logger: @logger, - # On reconnect the Listener rebuilds its OWN connection via this - # factory (fresh connect re-resolves DNS, converges on the promoted - # primary after a failover) instead of resetting a possibly-dead - # socket. Always provided — even when an initial pg_connection: is - # injected, the reconnect path builds a fresh raw connection. A test - # can inject its own factory to avoid touching real configuration. - connection_factory: connection_factory || -> { build_raw_pg_connection } - ) + @listener = build_listener(pg_connection, connection_factory) # Off-thread durable fanout writer (issue #321). Built only when # streams_writer_threads > 0; nil means fanout writes stay inline on # the dispatcher thread (the default, pre-#321 behavior). The pump @@ -169,6 +150,74 @@ def shutdown! private + # Selects the wake source by streams_listen_scope (issue #382). + # :master with a reachable hub socket → FailoverListener over a + # HubClient (NO per-worker LISTEN connection is opened). Anything + # else — scope :process, no socket exported (single mode, + # non-preforking server, hub failed to start), or a refused connect — + # keeps today's per-worker Listener. + def build_listener(pg_connection, connection_factory) + hub = build_hub_listener(connection_factory) + return hub if hub + + build_local_listener(pg_connection || build_pg_connection, connection_factory) + end + + def build_hub_listener(connection_factory) + return nil unless @config.streams_listen_scope == :master + + socket_path = ENV.fetch("PGBUS_STREAMS_HUB_SOCKET", nil) + return nil if socket_path.nil? || socket_path.empty? + + # The worker's ack deadline must exceed the master's own internal + # ensure_listening budget (its listener's health-check cycle + 1s). + failover = nil + client = HubClient.new( + socket_path: socket_path, + dispatch_queue: @dispatch_queue, + ack_timeout: (@config.streams_listen_health_check_ms / 1000.0) + 2.0, + # failover is assigned right below; a transport death in the gap + # is caught by the FailoverListener's synchronous ensure path. + on_failure: -> { failover&.fail_over! }, + logger: @logger + ) + client.connect + failover = FailoverListener.new( + hub_client: client, + local_listener_factory: lambda do + build_local_listener(build_pg_connection, connection_factory).tap(&:start) + end, + logger: @logger + ) + rescue HubClient::HubUnavailableError => e + @logger.info do + "[Pgbus::Streamer] master hub not reachable (#{e.message}) — using a per-worker listener" + end + nil + end + + def build_local_listener(pg_connection, connection_factory) + Listener.new( + pg_connection: pg_connection, + dispatch_queue: @dispatch_queue, + health_check_ms: @config.streams_listen_health_check_ms, + # Opt-in dispatch-queue backpressure (issue #315 item 3). 0 = + # unbounded (default). The queue itself stays an unbounded + # Queue.new so the request-thread Connect push and the dispatcher's + # own prune_dead self-post never block. + dispatch_queue_limit: @config.streams_dispatch_queue_limit, + maintenance: build_autoscale_maintenance, + logger: @logger, + # On reconnect the Listener rebuilds its OWN connection via this + # factory (fresh connect re-resolves DNS, converges on the promoted + # primary after a failover) instead of resetting a possibly-dead + # socket. Always provided — even when an initial pg_connection: is + # injected, the reconnect path builds a fresh raw connection. A test + # can inject its own factory to avoid touching real configuration. + connection_factory: connection_factory || -> { build_raw_pg_connection } + ) + end + def safely yield rescue StandardError => e diff --git a/lib/pgbus/web/streamer/listener.rb b/lib/pgbus/web/streamer/listener.rb index bd653b4c..bf018271 100644 --- a/lib/pgbus/web/streamer/listener.rb +++ b/lib/pgbus/web/streamer/listener.rb @@ -41,8 +41,11 @@ def initialize(queue_name:, payload: nil) end end - CHANNEL_PREFIX = "pgmq.q_" - CHANNEL_SUFFIX = ".INSERT" + # Single-sourced from NotifyListener, which owns the pgmq channel + # format (issue #381 review — the two copies had already drifted apart + # once in spirit if not in bytes). + CHANNEL_PREFIX = Pgbus::Process::NotifyListener::CHANNEL_PREFIX + CHANNEL_SUFFIX = Pgbus::Process::NotifyListener::CHANNEL_SUFFIX RECONNECT_BACKOFF_SECONDS = 0.5 diff --git a/spec/pgbus/web/streamer/instance_spec.rb b/spec/pgbus/web/streamer/instance_spec.rb index debb76e2..c1a564d3 100644 --- a/spec/pgbus/web/streamer/instance_spec.rb +++ b/spec/pgbus/web/streamer/instance_spec.rb @@ -549,4 +549,67 @@ def build_instance end end end + + describe "listener selection by streams_listen_scope (issue #382)" do + require "tmpdir" + + let(:tmpdir) { Dir.mktmpdir("pgbus-instance-hub") } + let(:socket_path) { File.join(tmpdir, "hub.sock") } + let(:hub_server) { UNIXServer.new(socket_path) } + + after do + hub_server.close if File.socket?(socket_path) && !hub_server.closed? + FileUtils.remove_entry(tmpdir) if File.directory?(tmpdir) + end + + around do |example| + original = ENV.fetch("PGBUS_STREAMS_HUB_SOCKET", nil) + example.run + ensure + original ? ENV["PGBUS_STREAMS_HUB_SOCKET"] = original : ENV.delete("PGBUS_STREAMS_HUB_SOCKET") + end + + it "uses a FailoverListener (no per-worker LISTEN connection) when the hub socket is reachable" do + hub_server # bind before the instance connects + ENV["PGBUS_STREAMS_HUB_SOCKET"] = socket_path + config.streams_listen_scope = :master + allow(PG).to receive(:connect) # must NOT be called — that's the whole point + + instance = described_class.new(client: client, config: config, logger: Logger.new(IO::NULL)) + + expect(instance.listener).to be_a(Pgbus::Web::Streamer::FailoverListener) + expect(PG).not_to have_received(:connect) + instance.listener.stop + end + + it "falls back to a per-worker Listener when the socket path is exported but dead" do + ENV["PGBUS_STREAMS_HUB_SOCKET"] = socket_path # nothing bound there + + instance = described_class.new( + client: client, config: config, pg_connection: fake_pg, logger: Logger.new(IO::NULL) + ) + + expect(instance.listener).to be_a(Pgbus::Web::Streamer::Listener) + end + + it "uses a per-worker Listener under scope :process even with a live hub socket" do + hub_server + ENV["PGBUS_STREAMS_HUB_SOCKET"] = socket_path + config.streams_listen_scope = :process + + instance = described_class.new( + client: client, config: config, pg_connection: fake_pg, logger: Logger.new(IO::NULL) + ) + + expect(instance.listener).to be_a(Pgbus::Web::Streamer::Listener) + end + + it "uses a per-worker Listener when no socket path is exported (single mode)" do + instance = described_class.new( + client: client, config: config, pg_connection: fake_pg, logger: Logger.new(IO::NULL) + ) + + expect(instance.listener).to be_a(Pgbus::Web::Streamer::Listener) + end + end end From 7e8b66ecde49e5ac85baff6b314380047b56eeba Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Sun, 2 Aug 2026 18:03:51 +0200 Subject: [PATCH 07/16] feat(streams): Puma plugin boots the master hub (#382 step 7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MasterHubBoot exports the socket path pre-fork (workers inherit ENV) and defers the actual hub start behind a config-readiness poller — with preload_app! the initializer lands before the first fork; without it the deadline expires quietly and workers keep per-worker listeners (:master effectively requires preload_app!, documented). Cluster mode only; every failure path degrades to no-socket fallback. --- lib/pgbus/web/streamer/master_hub_boot.rb | 112 +++++++++++++++ lib/puma/plugin/pgbus_streams.rb | 36 +++++ .../web/streamer/master_hub_boot_spec.rb | 132 ++++++++++++++++++ 3 files changed, 280 insertions(+) create mode 100644 lib/pgbus/web/streamer/master_hub_boot.rb create mode 100644 spec/pgbus/web/streamer/master_hub_boot_spec.rb diff --git a/lib/pgbus/web/streamer/master_hub_boot.rb b/lib/pgbus/web/streamer/master_hub_boot.rb new file mode 100644 index 00000000..9cf7cba8 --- /dev/null +++ b/lib/pgbus/web/streamer/master_hub_boot.rb @@ -0,0 +1,112 @@ +# frozen_string_literal: true + +module Pgbus + module Web + module Streamer + # Deferred MasterHub startup for the Puma master (issue #382). The + # pgbus_streams plugin's `start` runs BEFORE `preload_app!` loads the + # Rails app (and with it the pgbus initializer), so the hub cannot be + # built eagerly. This class splits the two halves: + # + # 1. The socket path is exported to ENV IMMEDIATELY — workers inherit + # it across fork and connect lazily on first SSE use. + # 2. A poller thread waits for Pgbus.configuration to become ready + # (the initializer has run — with preload_app!, before the first + # fork), then builds and starts the MasterHub. Workers that race a + # still-booting hub simply fail to connect and fall back to their + # own listener until they recycle — degraded footprint, never + # degraded semantics. + # + # Without preload_app! the master never loads the app, the deadline + # expires quietly, no socket is ever bound, and every worker keeps + # today's per-worker listener — :master scope effectively requires + # preload_app!, documented on the docs site. + class MasterHubBoot + def self.default_socket_path + File.join(Dir.tmpdir, "pgbus-streams-hub-#{::Process.pid}.sock") + end + + def initialize(socket_path: self.class.default_socket_path, hub_factory: nil, + poll_interval: 1.0, deadline: 120, logger: nil) + @socket_path = socket_path + @hub_factory = hub_factory || lambda do |socket_path:| + MasterHub.new(config: Pgbus.configuration, socket_path: socket_path) + end + @poll_interval = poll_interval + @deadline = deadline + @logger = logger + @hub = nil + @running = false + @thread = nil + end + + def start + ENV["PGBUS_STREAMS_HUB_SOCKET"] = @socket_path + @running = true + @thread = Thread.new { wait_and_start } + self + end + + def stop + @running = false + @thread&.join(2) + @thread = nil + @hub&.stop + @hub = nil + self + end + + private + + def wait_and_start + waited = 0.0 + until configuration_ready? + return unless @running + return give_up if waited >= @deadline + + sleep @poll_interval + waited += @poll_interval + end + return unless @running && master_scope? + + @hub = @hub_factory.call(socket_path: @socket_path) + @hub.start + log(:info) { "[Pgbus::Streamer::MasterHubBoot] master hub listening at #{@socket_path}" } + rescue StandardError => e + @hub = nil + log(:error) do + "[Pgbus::Streamer::MasterHubBoot] master hub failed to start " \ + "(#{e.class}: #{e.message}) — workers fall back to per-worker listeners" + end + end + + # Ready once the app's initializer has produced connection options a + # dedicated LISTEN connection can be built from (String URL or libpq + # Hash; the Proc fallback means "nothing configured yet"). + def configuration_ready? + return false unless defined?(Pgbus) && Pgbus.configuration.streams_enabled + + options = Pgbus.configuration.streams_connection_options + options.is_a?(String) || options.is_a?(Hash) + rescue StandardError + false + end + + def master_scope? + Pgbus.configuration.streams_listen_scope == :master + end + + def give_up + log(:info) do + "[Pgbus::Streamer::MasterHubBoot] configuration never became ready within #{@deadline}s " \ + "(no preload_app!?) — no master hub; workers use per-worker listeners" + end + end + + def log(level, &) + (@logger || Pgbus.logger).public_send(level, &) + end + end + end + end +end diff --git a/lib/puma/plugin/pgbus_streams.rb b/lib/puma/plugin/pgbus_streams.rb index 71d59196..edd33462 100644 --- a/lib/puma/plugin/pgbus_streams.rb +++ b/lib/puma/plugin/pgbus_streams.rb @@ -22,15 +22,51 @@ # and a non-Rails use case). Explicit opt-in is safer. Puma::Plugin.create do def start(launcher) + # Master-side streams hub (issue #382): in cluster mode, ONE LISTEN + # connection in the master serves every worker over a Unix socket. The + # socket path is exported to ENV here (pre-fork, so workers inherit it); + # the hub itself starts once the preloaded app has configured Pgbus (see + # MasterHubBoot). Any failure means no socket — workers keep their own + # per-worker listeners, trading connections for unchanged semantics. + boot_master_hub(launcher) + launcher.events.register(:after_stopped) do + teardown_master_hub(launcher) teardown_streamer(launcher) end launcher.events.register(:before_restart) do + teardown_master_hub(launcher) teardown_streamer(launcher) end end + def boot_master_hub(launcher) + return unless defined?(Pgbus::Web::Streamer::MasterHubBoot) + # Single mode: the master IS the (only) serving process — one listener + # per host already; a hub would just add a socket hop. + return unless cluster_mode?(launcher) + + @master_hub_boot = Pgbus::Web::Streamer::MasterHubBoot.new + @master_hub_boot.start + rescue StandardError => e + @master_hub_boot = nil + log_error(launcher, e) + end + + def cluster_mode?(launcher) + launcher.respond_to?(:options) && launcher.options[:workers].to_i.positive? + rescue StandardError + false + end + + def teardown_master_hub(launcher) + @master_hub_boot&.stop + @master_hub_boot = nil + rescue StandardError => e + log_error(launcher, e) + end + def teardown_streamer(launcher) return unless defined?(Pgbus::Web::Streamer) diff --git a/spec/pgbus/web/streamer/master_hub_boot_spec.rb b/spec/pgbus/web/streamer/master_hub_boot_spec.rb new file mode 100644 index 00000000..305f52be --- /dev/null +++ b/spec/pgbus/web/streamer/master_hub_boot_spec.rb @@ -0,0 +1,132 @@ +# frozen_string_literal: true + +require "spec_helper" +require "tmpdir" + +RSpec.describe Pgbus::Web::Streamer::MasterHubBoot do + subject(:boot) do + described_class.new( + socket_path: socket_path, + hub_factory: hub_factory, + poll_interval: 0.02, + deadline: 0.5, + logger: logger + ) + end + + let(:tmpdir) { Dir.mktmpdir("pgbus-hub-boot") } + let(:socket_path) { File.join(tmpdir, "hub.sock") } + let(:logger) { Logger.new(IO::NULL) } + let(:hub) { instance_double(Pgbus::Web::Streamer::MasterHub, start: nil, stop: nil) } + let(:factory_calls) { [] } + let(:hub_factory) do + lambda do |socket_path:| + factory_calls << socket_path + hub + end + end + + after do + boot.stop + FileUtils.remove_entry(tmpdir) if File.directory?(tmpdir) + end + + def wait_until(timeout: 2) + deadline = Time.now + timeout + until yield + raise "timed out waiting for condition" if Time.now > deadline + + sleep 0.01 + end + end + + around do |example| + original = ENV.fetch("PGBUS_STREAMS_HUB_SOCKET", nil) + example.run + ensure + original ? ENV["PGBUS_STREAMS_HUB_SOCKET"] = original : ENV.delete("PGBUS_STREAMS_HUB_SOCKET") + end + + describe "#start" do + it "exports the socket path immediately (workers must inherit it across fork)" do + allow(boot).to receive(:configuration_ready?).and_return(false) + + boot.start + + expect(ENV.fetch("PGBUS_STREAMS_HUB_SOCKET", nil)).to eq(socket_path) + end + + it "starts the hub once the configuration becomes ready (post-preload)" do + ready = false + allow(boot).to receive(:configuration_ready?) { ready } + allow(boot).to receive(:master_scope?).and_return(true) + + boot.start + sleep 0.05 + expect(factory_calls).to be_empty + + ready = true + wait_until { factory_calls.any? } + + expect(factory_calls).to eq([socket_path]) + expect(hub).to have_received(:start) + end + + it "gives up quietly after the deadline when configuration never appears" do + allow(boot).to receive(:configuration_ready?).and_return(false) + + boot.start + sleep 0.7 + + expect(factory_calls).to be_empty + expect(ENV.fetch("PGBUS_STREAMS_HUB_SOCKET", nil)).to eq(socket_path) + end + + it "does not start the hub when the resolved scope is :process" do + allow(boot).to receive_messages(configuration_ready?: true, master_scope?: false) + + boot.start + sleep 0.1 + + expect(factory_calls).to be_empty + end + + it "logs and survives a hub factory failure (workers fall back)" do + allow(logger).to receive(:error) + failing = described_class.new( + socket_path: socket_path, + hub_factory: ->(socket_path:) { raise StandardError, "no db" }, # rubocop:disable Lint/UnusedBlockArgument + poll_interval: 0.02, deadline: 0.5, logger: logger + ) + allow(failing).to receive_messages(configuration_ready?: true, master_scope?: true) + + failing.start + sleep 0.2 + + expect(logger).to have_received(:error) + failing.stop + end + end + + describe "#stop" do + it "stops a started hub" do + allow(boot).to receive_messages(configuration_ready?: true, master_scope?: true) + boot.start + wait_until { factory_calls.any? } + + boot.stop + + expect(hub).to have_received(:stop) + end + + it "cancels a still-waiting poller" do + allow(boot).to receive(:configuration_ready?).and_return(false) + boot.start + + boot.stop + sleep 0.1 + + expect(factory_calls).to be_empty + end + end +end From 5e46ad9410d79fe048235948680d6269d03a0431 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Sun, 2 Aug 2026 18:07:52 +0200 Subject: [PATCH 08/16] feat(doctor)+test(integration): scope-aware streams budget clause + hub acceptance (#382 steps 8-10) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Doctor's Connection budget prints '1 per web host (streams master hub)' under :master, per-process under :process. Integration: real PG proves one census-tagged connection serving multiple workers, ephemeral payload fidelity, LISTEN-backend-kill recovery, and master-death failover with wake continuity. E2E: two full streamer Instances (FailoverListener → HubClient → Dispatcher → hijacked SSE sockets) deliver through ONE shared connection, then keep delivering after the hub dies — census 1 → 2, the accepted fallback balloon. --- README.md | 2 +- lib/pgbus/doctor.rb | 13 +- .../streams/master_hub_e2e_spec.rb | 136 +++++++++++++++ spec/integration/streams/master_hub_spec.rb | 159 ++++++++++++++++++ spec/pgbus/doctor_spec.rb | 9 +- 5 files changed, 316 insertions(+), 3 deletions(-) create mode 100644 spec/integration/streams/master_hub_e2e_spec.rb create mode 100644 spec/integration/streams/master_hub_spec.rb diff --git a/README.md b/README.md index a732e967..d7b892e4 100644 --- a/README.md +++ b/README.md @@ -1876,7 +1876,7 @@ A single preflight command that answers "is this environment healthy enough to r | Broadcast queue | — | Turbo broadcasts share the default queue in production, or `streams_broadcast_queue` is set but no worker capsule drains it | | Primary affinity | — | Job connection is on a read-only replica (`pg_is_in_recovery`) — a read/write-splitting pooler may be stalling jobs | | Dedicated connections | Streamer LISTEN and/or worker notify dedicated path cannot connect | — | -| Connection budget | — (informational: prints how many direct LISTEN connections the current config pins — 1 per host under `worker_notify_scope: :supervisor`, one per fork under `:fork`, plus 1 per web process when streams are enabled) | — | +| Connection budget | — (informational: prints how many direct LISTEN connections the current config pins — 1 per host under `worker_notify_scope: :supervisor`, one per fork under `:fork`; streams add 1 per web host under `streams_listen_scope: :master` or 1 per web process under `:process`) | — | ```bash pgbus doctor # prints the report; exit 1 unless every check passed diff --git a/lib/pgbus/doctor.rb b/lib/pgbus/doctor.rb index 327f247f..2465daac 100644 --- a/lib/pgbus/doctor.rb +++ b/lib/pgbus/doctor.rb @@ -394,12 +394,23 @@ def check_connection_budget consumers: consumers, con_plural: consumers == 1 ? "" : "s", share: count == 1 && @config.worker_notify_scope == :supervisor ? " share it" : "" ) - detail += " + 1 per web-server process (streams)" if @config.streams_enabled + detail += streams_budget_clause if @config.streams_enabled Check.new(name: "Connection budget", status: :ok, detail: detail) rescue StandardError => e Check.new(name: "Connection budget", status: :warn, detail: "#{e.class}: #{e.message}") end + # Streams add their own LISTEN footprint on web hosts: one per host with + # the master hub (#382, the default — workers fall back per-worker only + # during a hub outage), one per web process under :process scope. + def streams_budget_clause + if @config.streams_listen_scope == :master + " + 1 per web host (streams master hub)" + else + " + 1 per web-server process (streams)" + end + end + # Open one dedicated connection the way the runtime does, verify it # answers, close it. Returns nil on success, "label: error" on failure. def probe_dedicated_connection(label, opts) diff --git a/spec/integration/streams/master_hub_e2e_spec.rb b/spec/integration/streams/master_hub_e2e_spec.rb new file mode 100644 index 00000000..3927acc2 --- /dev/null +++ b/spec/integration/streams/master_hub_e2e_spec.rb @@ -0,0 +1,136 @@ +# frozen_string_literal: true + +require_relative "../../integration_helper" +require_relative "../../support/puma_test_harness" +require_relative "../../support/sse_test_client" +require "tmpdir" + +# End-to-end for issue #382: the full SSE path under :master scope. +# +# MasterHub (one LISTEN connection) +# ← Unix socket → Instance A (FailoverListener → HubClient) → SSE client A +# ← Unix socket → Instance B (FailoverListener → HubClient) → SSE client B +# +# Two streamer Instances stand in for two Puma workers (the process boundary +# itself is proven in master_hub_spec.rb; this spec proves the full +# Instance → HubClient → Dispatcher → hijacked-socket delivery chain), then +# the hub DIES mid-test and both instances keep delivering via their +# fallback listeners — the settled outage semantics: connections over loss. +RSpec.describe "Streams master hub end-to-end (issue #382)", :integration do + before(:all) do + @saved_listen_notify = Pgbus.configuration.listen_notify + Pgbus.configuration.listen_notify = true + Pgbus.configuration.streams_signed_name_secret = "a" * 64 + Pgbus.configuration.streams_listen_health_check_ms = 100 + Pgbus.configuration.streams_heartbeat_interval = 30 + Pgbus.configuration.streams_write_deadline_ms = 5_000 + Pgbus.reset_client! + end + + after(:all) do + Pgbus.configuration.listen_notify = @saved_listen_notify + Pgbus.configuration.streams_signed_name_secret = nil + Pgbus.reset_client! + end + + let(:tmpdir) { Dir.mktmpdir("pgbus-hub-e2e") } + let(:socket_path) { File.join(tmpdir, "hub.sock") } + let(:stream_name) { "hube2e_#{SecureRandom.hex(4)}" } + let(:hub) do + Pgbus::Web::Streamer::MasterHub.new( + config: Pgbus.configuration, socket_path: socket_path, + status_interval: 0.5, logger: Logger.new(IO::NULL) + ) + end + + def build_worker_instance + Pgbus::Web::Streamer::Instance.new( + client: Pgbus.client, + config: Pgbus.configuration, + logger: Logger.new(IO::NULL) + ) + end + + def build_app(streamer) + Pgbus::Web::StreamApp.new( + streamer: streamer, + config: Pgbus.configuration, + logger: Logger.new(IO::NULL) + ) + end + + def listen_backend_pids + ActiveRecord::Base.connection.select_values(<<~SQL) + SELECT pid FROM pg_stat_activity + WHERE application_name = 'pgbus-listen' AND datname = current_database() + SQL + end + + around do |example| + original = ENV.fetch("PGBUS_STREAMS_HUB_SOCKET", nil) + ENV["PGBUS_STREAMS_HUB_SOCKET"] = socket_path + example.run + ensure + original ? ENV["PGBUS_STREAMS_HUB_SOCKET"] = original : ENV.delete("PGBUS_STREAMS_HUB_SOCKET") + FileUtils.remove_entry(tmpdir) if File.directory?(tmpdir) + end + + before { Pgbus.client.ensure_stream_queue(stream_name) } + + def signed(name) + Pgbus::Streams::SignedName.sign(name) + end + + it "delivers SSE through one shared connection and keeps delivering after the hub dies" do + baseline_pids = listen_backend_pids + hub.start + + worker_a = build_worker_instance + worker_b = build_worker_instance + expect(worker_a.listener).to be_a(Pgbus::Web::Streamer::FailoverListener) + expect(worker_b.listener).to be_a(Pgbus::Web::Streamer::FailoverListener) + worker_a.start + worker_b.start + + harness_a = SseTestSupport::PumaTestHarness.boot(rack_app: build_app(worker_a)) + harness_b = SseTestSupport::PumaTestHarness.boot(rack_app: build_app(worker_b)) + + stream = Pgbus.stream(stream_name) + watermark = stream.current_msg_id + client_a = SseTestSupport::SseTestClient.connect( + url: "#{harness_a.url("/#{signed(stream_name)}")}?since=#{watermark}", timeout: 5 + ) + client_b = SseTestSupport::SseTestClient.connect( + url: "#{harness_b.url("/#{signed(stream_name)}")}?since=#{watermark}", timeout: 5 + ) + + # Both workers served SSE — yet the whole "host" pins ONE connection. + expect((listen_backend_pids - baseline_pids).size).to eq(1) + + stream.broadcast("via hub") + expect(client_a.wait_for_events(count: 1, timeout: 5).map(&:data)) + .to eq(["via hub"]) + expect(client_b.wait_for_events(count: 1, timeout: 5).map(&:data)) + .to eq(["via hub"]) + + # The hub dies. Both workers fail over to their own listeners (the + # accepted, census-visible balloon) and SSE delivery continues. + hub.stop + sleep 0.5 + + stream.broadcast("via fallback") + expect(client_a.wait_for_events(count: 2, timeout: 10).map(&:data).last) + .to eq("via fallback") + expect(client_b.wait_for_events(count: 2, timeout: 10).map(&:data).last) + .to eq("via fallback") + + expect((listen_backend_pids - baseline_pids).size).to eq(2) + ensure + client_a&.close + client_b&.close + worker_a&.shutdown! + worker_b&.shutdown! + harness_a&.shutdown + harness_b&.shutdown + end +end diff --git a/spec/integration/streams/master_hub_spec.rb b/spec/integration/streams/master_hub_spec.rb new file mode 100644 index 00000000..a491232c --- /dev/null +++ b/spec/integration/streams/master_hub_spec.rb @@ -0,0 +1,159 @@ +# frozen_string_literal: true + +require_relative "../../integration_helper" + +# Issue #382 acceptance against real PostgreSQL + real LISTEN/NOTIFY: +# - one census-tagged LISTEN connection (the MasterHub's) serves multiple +# "workers" connected over the Unix socket +# - real ephemeral NOTIFY payloads ride the frames intact +# - killing the shared LISTEN backend is survived (listener reconnect, +# wakes flow again) +# - a worker whose master DIES fails over to its own listener and keeps +# receiving wakes (the settled fallback: connections over loss) +RSpec.describe "Streams master hub (issue #382)", :integration do + let(:config) { Pgbus.configuration } + let(:logger) { Logger.new(IO::NULL) } + let(:tmpdir) { Dir.mktmpdir("pgbus-hub-int") } + let(:socket_path) { File.join(tmpdir, "hub.sock") } + let(:stream_name) { "hubint_#{SecureRandom.hex(4)}" } + let(:physical) { config.queue_name(stream_name) } + + around do |example| + saved = config.listen_notify + config.listen_notify = true + example.run + ensure + config.listen_notify = saved + FileUtils.remove_entry(tmpdir) if File.directory?(tmpdir) + end + + before { Pgbus.client.ensure_stream_queue(stream_name) } + + def wait_until(timeout: 10) + deadline = Time.now + timeout + until yield + raise "timed out waiting for condition" if Time.now > deadline + + sleep 0.05 + end + end + + def send_frame(sock, message) + sock.write(Pgbus::Web::Streamer::HubProtocol.encode(message)) + end + + def read_frame_of_type(sock, type, timeout: 5) + deadline = Time.now + timeout + while Time.now < deadline + raise "no #{type} frame within #{timeout}s" unless sock.wait_readable(timeout) + + frame = Pgbus::Web::Streamer::HubProtocol.read_frame(sock) + raise "peer closed while waiting for #{type}" if frame.nil? + return frame if frame["t"] == type + end + raise "no #{type} frame within #{timeout}s" + end + + def listen_backend_pids + ActiveRecord::Base.connection.select_values(<<~SQL) + SELECT pid FROM pg_stat_activity + WHERE application_name = 'pgbus-listen' AND datname = current_database() + SQL + end + + it "serves workers over ONE connection, carries ephemeral payloads, survives a backend kill" do + baseline_pids = listen_backend_pids + hub = Pgbus::Web::Streamer::MasterHub.new( + config: config, socket_path: socket_path, status_interval: 0.5, logger: logger + ) + worker = nil + begin + hub.start + wait_until { (listen_backend_pids - baseline_pids).size == 1 } + + worker = UNIXSocket.new(socket_path) + send_frame(worker, { "t" => "sub", "q" => physical }) + read_frame_of_type(worker, "ack") + + # Census: the whole host still pins exactly ONE streams connection. + expect((listen_backend_pids - baseline_pids).size).to eq(1) + + # A real ephemeral broadcast (pg_notify with payload) rides the frame. + Pgbus.client.notify_stream(stream_name, "
ephemeral hello
") + frame = read_frame_of_type(worker, "wake") + expect(frame["q"]).to eq(physical) + expect(frame["p"]).to include("ephemeral hello") + + # Chaos: kill the shared LISTEN backend; the listener reconnects and + # wakes flow again. + old_pids = listen_backend_pids - baseline_pids + ActiveRecord::Base.connection.execute(<<~SQL) + SELECT pg_terminate_backend(pid) + FROM pg_stat_activity + WHERE pid IN (#{old_pids.join(",")}) + SQL + wait_until do + fresh = listen_backend_pids - baseline_pids + !fresh.empty? && !fresh.intersect?(old_pids) + end + sleep 0.3 + Pgbus.client.notify_stream(stream_name, "
after recovery
") + frame = read_frame_of_type(worker, "wake", timeout: 10) + expect(frame["p"]).to include("after recovery") + ensure + worker&.close + hub.stop + end + end + + it "a worker fails over to its OWN listener when the master dies, without losing wakes" do + hub = Pgbus::Web::Streamer::MasterHub.new( + config: config, socket_path: socket_path, status_interval: 0.5, logger: logger + ) + dispatch_queue = Queue.new + failover = nil + begin + hub.start + wait_until { File.socket?(socket_path) } + + client = Pgbus::Web::Streamer::HubClient.new( + socket_path: socket_path, dispatch_queue: dispatch_queue, + ack_timeout: 5, on_failure: -> { failover&.fail_over! }, logger: logger + ) + client.connect + failover = Pgbus::Web::Streamer::FailoverListener.new( + hub_client: client, + local_listener_factory: lambda { + conn_factory = -> { Pgbus::DedicatedConnection.connect(config.streams_connection_options) } + Pgbus::Web::Streamer::Listener.new( + pg_connection: conn_factory.call, + dispatch_queue: dispatch_queue, + health_check_ms: 250, + connection_factory: conn_factory, + logger: logger + ).tap(&:start) + }, + logger: logger + ) + + failover.ensure_listening(physical) + Pgbus.client.notify_stream(stream_name, "
via hub
") + expect(dispatch_queue.pop(timeout: 5)&.payload).to include("via hub") + + # Master dies. The client EOFs, fail_over! builds a real per-worker + # listener and re-LISTENs the recorded set. + hub.stop + wait_until(timeout: 5) { client.dead? } + # ensure_listening after death exercises the sync failover path too. + failover.ensure_listening(physical) + + sleep 0.3 + Pgbus.client.notify_stream(stream_name, "
via fallback
") + message = dispatch_queue.pop(timeout: 10) + message = dispatch_queue.pop(timeout: 10) while message && !message.payload&.include?("via fallback") + expect(message&.payload).to include("via fallback") + ensure + failover&.stop + end + end +end diff --git a/spec/pgbus/doctor_spec.rb b/spec/pgbus/doctor_spec.rb index d787042b..58636022 100644 --- a/spec/pgbus/doctor_spec.rb +++ b/spec/pgbus/doctor_spec.rb @@ -507,9 +507,16 @@ def budget_check expect(budget_check[:detail]).not_to include("1 capsules") end - it "notes the per-web-process streams listener when streams are enabled" do + it "notes one streams connection per web host under :master scope (the default)" do allow(config).to receive(:streams_enabled).and_return(true) + expect(budget_check[:detail]).to include("+ 1 per web host (streams master hub)") + end + + it "notes the per-web-process streams listener under :process scope" do + allow(config).to receive(:streams_enabled).and_return(true) + config.streams_listen_scope = :process + expect(budget_check[:detail]).to include("+ 1 per web-server process (streams)") end From 787ab8590d0a397a23b72aa8fd27ca900acea308 Mon Sep 17 00:00:00 2001 From: mhenrixon Date: Sun, 2 Aug 2026 18:18:01 +0200 Subject: [PATCH 09/16] bench(streams): master-hub hop cost + census benchmark (#382 step 11) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Same single-broadcast SSE roundtrip under :process vs :master. Measured (local PG, n=50, durable mode): :process p50=16.93ms p95=26.67ms; :master p50=16.00ms p95=19.19ms — the socket hop is noise-level free. Durable mode + a warmup probe because the default :ephemeral mode races subscription setup and a lost first event wedges cumulative waits. --- Rakefile | 7 +- benchmarks/streams_hub_bench.rb | 134 ++++++++++++++++++++++++++++++++ 2 files changed, 140 insertions(+), 1 deletion(-) create mode 100644 benchmarks/streams_hub_bench.rb diff --git a/Rakefile b/Rakefile index 266e2e9d..b67321ef 100644 --- a/Rakefile +++ b/Rakefile @@ -25,7 +25,7 @@ namespace :bench do # no-DB unit suite that bench:all runs in CI. db_benches = %w[connection_pool_bench integration_bench streams_bench streams_read_pool_bench execution_modes_bench pool_swap_bench pool_autoscale_bench job_burst_bench - notify_wake_bench notify_chaos_bench].freeze + notify_wake_bench notify_chaos_bench streams_hub_bench].freeze # The unit suite is every *_bench.rb that doesn't need a database, derived # from the directory so a new unit bench is picked up automatically (kept in # sync with bench:one, which globs the same files). @@ -95,6 +95,11 @@ namespace :bench do ruby "benchmarks/notify_chaos_bench.rb" end + desc "Run streams master-hub latency benchmark (#382 hop cost + census; requires PGBUS_DATABASE_URL)" + task :streams_hub do + ruby "benchmarks/streams_hub_bench.rb" + end + desc "Run a single benchmark: rake bench:one[client_bench]" task :one, [:name] do |_t, args| name = args[:name] or abort "Usage: rake bench:one[serialization_bench|client_bench|...]" diff --git a/benchmarks/streams_hub_bench.rb b/benchmarks/streams_hub_bench.rb new file mode 100644 index 00000000..0b7e1f91 --- /dev/null +++ b/benchmarks/streams_hub_bench.rb @@ -0,0 +1,134 @@ +# frozen_string_literal: true + +# Streams master-hub latency benchmark (issue #382): measures the price of +# the master→worker socket hop by running the SAME single-broadcast SSE +# roundtrip twice — +# +# A. :process — the per-worker Listener path (pre-#382 architecture) +# B. :master — MasterHub in-process, the streamer on a FailoverListener +# over the Unix socket (one extra frame hop per wake) +# +# plus the LISTEN-connection census for each mode. Compare column A against +# main's streams_bench section 1 to isolate refactor noise from hop cost. +# +# Requires PGBUS_DATABASE_URL: +# PGBUS_DATABASE_URL=postgres://user@host/db bundle exec rake bench:streams_hub + +require "json" +require "logger" +require "tmpdir" +require "securerandom" +require "active_record" +require "pgbus" + +require_relative "../spec/support/puma_test_harness" +require_relative "../spec/support/sse_test_client" + +DATABASE_URL = ENV.fetch("PGBUS_DATABASE_URL") do + abort "PGBUS_DATABASE_URL not set. Example: postgres://user@host/db" +end + +SAMPLES = Integer(ENV.fetch("HUB_BENCH_SAMPLES", "50")) + +ActiveRecord::Base.establish_connection(DATABASE_URL) + +Pgbus.configure do |c| + c.database_url = DATABASE_URL + c.queue_prefix = "pgbus_hbench" + c.default_queue = "default" + c.logger = Logger.new(IO::NULL) + c.pgmq_schema_mode = :embedded + c.listen_notify = true + c.streams_signed_name_secret = "a" * 64 + c.streams_listen_health_check_ms = 100 + c.streams_heartbeat_interval = 30 + c.streams_write_deadline_ms = 5_000 + # Durable broadcasts: race-immune against subscription setup (a broadcast + # landing before LISTEN is active is still caught by the connect-time + # read_after) and the representative wake -> read_after -> fanout path. + c.streams_default_broadcast_mode = :durable + c.stats_enabled = false if c.respond_to?(:stats_enabled=) +end + +def percentile(sorted, pct) + sorted[[(sorted.size * pct / 100.0).ceil - 1, 0].max] +end + +def census + ActiveRecord::Base.connection.select_value(<<~SQL).to_i + SELECT count(*) FROM pg_stat_activity + WHERE application_name = 'pgbus-listen' AND datname = current_database() + SQL +end + +def measure_roundtrips(label) + stream_name = "hb_#{SecureRandom.hex(4)}" + Pgbus.client.ensure_stream_queue(stream_name) + streamer = Pgbus::Web::Streamer::Instance.new( + client: Pgbus.client, config: Pgbus.configuration, logger: Logger.new(IO::NULL) + ) + streamer.start + app = Pgbus::Web::StreamApp.new( + streamer: streamer, config: Pgbus.configuration, logger: Logger.new(IO::NULL) + ) + harness = SseTestSupport::PumaTestHarness.boot(rack_app: app) + stream = Pgbus.stream(stream_name) + signed = Pgbus::Streams::SignedName.sign(stream_name) + client = SseTestSupport::SseTestClient.connect( + url: "#{harness.url("/#{signed}")}?since=#{stream.current_msg_id}", timeout: 5 + ) + + listener_kind = streamer.listener.class.name.split("::").last + mode_census = census + # Warmup: proves the subscription is live before timing starts. + stream.broadcast("warmup") + abort "#{label}: warmup broadcast never delivered" if + client.wait_for_events(count: 1, timeout: 10).empty? + + samples = [] + SAMPLES.times do |i| + t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC) + stream.broadcast("#{i}") + client.wait_for_events(count: i + 2, timeout: 10) + samples << ((Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0) * 1000.0) + end + + sorted = samples.sort + puts format( + "%-32