diff --git a/CHANGELOG.md b/CHANGELOG.md
index d54b43b4..811a6661 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,8 @@
### Added
+- **Streams: one LISTEN connection per web host — `streams_listen_scope` (issue #382).** ⚠️ **Default behavior change.** Previously every Puma worker lazily opened its own dedicated streams LISTEN connection on first SSE use, so a web host pinned one direct connection per worker. Under the new default (`streams_listen_scope = :master`) the `pgbus_streams` Puma plugin runs a **MasterHub** in the preforking master: ONE `Web::Streamer::Listener` on the refcounted union of every worker's stream channels, fanning wakes — **including ephemeral payloads** — out to workers over a Unix domain socket with length-prefixed frames (`Streamer::HubProtocol`). Workers connect lazily (nothing is inherited across fork) and the synchronous `ensure_listening` ack contract is preserved cross-process: a sub is registered before LISTEN executes and acked only after, so the no-lost-broadcast guarantee holds. Backpressure follows the streams rules: durable wakes are droppable at a per-worker cap (they self-heal via `read_after`), **ephemeral wakes are never dropped** — a worker that stops draining is evicted, which triggers its own fallback. **Fallback is per-worker listeners, not loss**: whenever the hub is absent or dies (no `preload_app!`, single-mode Puma, crash, eviction) each worker's `FailoverListener` swaps in a real per-worker `Listener` and re-LISTENs its recorded subscriptions — connection footprint balloons back to pre-#382 levels (census-visible) but no broadcast semantics change; the worker stays local until it recycles. Measured (local PG, n=50): the master→worker hop is noise-level free — single-broadcast SSE roundtrip p50 16.00ms via the hub vs 16.93ms per-worker. **`:master` effectively requires `preload_app!`** (the hub waits for the app's pgbus initializer; without it the deadline expires quietly and workers stay per-worker). **Rollback:** `config.streams_listen_scope = :process`. Refs #382, builds on the #381 patterns.
+
- **Host-level shared LISTEN: `worker_notify_scope` — the supervisor now owns ONE direct LISTEN connection for the whole host (issue #381).** ⚠️ **Default behavior change.** Previously every worker fork and every consumer fork opened its own dedicated LISTEN connection (`NotifyListener`), so a host's direct-connection footprint scaled with fork count — on transaction-pool PgBouncer platforms those connections come out of the scarcest slice of `max_connections`, and a 5-capsule + 2-consumer host pinned 7. Under the new default (`config.worker_notify_scope = :supervisor`) the supervisor runs a single `NotifyHub`: one `NotifyListener` on the union of every capsule's and consumer's queue channels (wildcards via the shared resolver, consumer sets via the registry), fanning wakes out to forks over per-fork pipes (`W` wake / `H` healthy / `P` degraded bytes; a fork whose pipe reports degraded or reaches EOF falls back to fast polling exactly like a failed local listener). Footprint drops to **1 direct LISTEN connection per job host**, verified by integration test: routing is per-fork (an insert wakes only the forks reading that queue, wildcard capsules unconditionally), and `pg_terminate_backend` on the shared connection is survived — reconnect, re-LISTEN, wakes flow again. **Rollback:** `config.worker_notify_scope = :fork` restores the previous per-fork listeners byte-for-byte. Dedicated LISTEN connections are now census-tagged `application_name=pgbus-listen` so `pg_stat_activity` can count them. Refs #381.
- **`pgbus doctor`: new "Connection budget" check (issue #381).** Prints how many direct LISTEN connections the current config pins — 1 per host under `:supervisor` scope, capsules + consumers under `:fork` (honoring `config.roles`), plus a "+1 per web-server process (streams)" clause — so operators can do pooler capacity math from the doctor output alone. Informational, always `:ok`. Refs #381.
- **Benchmarks: `rake bench:notify_wake` and `rake bench:notify_chaos` (issue #381).** Wake-path latency (send → wake, p50/p95/p99, direct vs hub-mediated), empty-read cost, LISTEN connection census, and failure-mode measurements (killed LISTEN backend, wedged fork, FD churn, fan-out cost). Refs #381.
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/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..17c43d32
--- /dev/null
+++ b/benchmarks/streams_hub_bench.rb
@@ -0,0 +1,138 @@
+# 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"))
+abort "HUB_BENCH_SAMPLES must be a positive integer (got #{SAMPLES})" unless SAMPLES.positive?
+
+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} ")
+ events = client.wait_for_events(count: i + 2, timeout: 10)
+ # A silently dropped/late wake would otherwise record a ~10s sample
+ # straight into the reported percentiles.
+ abort "#{label}: sample #{i} never delivered (got #{events.size}, expected #{i + 2})" if events.size < i + 2
+ samples << ((Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0) * 1000.0)
+ end
+
+ sorted = samples.sort
+ puts format(
+ "%-32s listener=%-16s census=%d n=%d " \
+ "p50=%.2fms p95=%.2fms max=%.2fms",
+ label: label, kind: listener_kind, census: mode_census, n: sorted.size,
+ p50: percentile(sorted, 50), p95: percentile(sorted, 95), max: sorted.last
+ )
+ensure
+ client&.close
+ streamer&.shutdown!
+ harness&.shutdown
+end
+
+puts "=" * 70
+puts "Pgbus streams master-hub benchmark (issue #382) samples=#{SAMPLES}"
+puts "=" * 70
+
+# ─── A. :process (per-worker listener — the pre-#382 path) ───
+Pgbus.configuration.streams_listen_scope = :process
+measure_roundtrips("A. :process (per-worker)")
+
+# ─── B. :master (hub interposed) ───
+tmpdir = Dir.mktmpdir("pgbus-hub-bench")
+socket_path = File.join(tmpdir, "hub.sock")
+hub = Pgbus::Web::Streamer::MasterHub.new(
+ config: Pgbus.configuration, socket_path: socket_path, logger: Logger.new(IO::NULL)
+)
+hub.start
+ENV["PGBUS_STREAMS_HUB_SOCKET"] = socket_path
+Pgbus.configuration.streams_listen_scope = :master
+begin
+ measure_roundtrips("B. :master (hub -> socket hop)")
+ensure
+ ENV.delete("PGBUS_STREAMS_HUB_SOCKET")
+ hub.stop
+ FileUtils.remove_entry(tmpdir) if File.directory?(tmpdir)
+end
+
+puts "\nDone."
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/docs/app/views/docs/pages/performance_tuning.rb b/docs/app/views/docs/pages/performance_tuning.rb
index 7752646c..4ddc08b3 100644
--- a/docs/app/views/docs/pages/performance_tuning.rb
+++ b/docs/app/views/docs/pages/performance_tuning.rb
@@ -13,6 +13,7 @@ def content
autovacuum
archive
job_burst_tuning
+ streams_master_hub
streams_pool_autoscaling
fanout_throughput
health_metrics
@@ -93,6 +94,46 @@ def fanout_throughput
end
end
+ def streams_master_hub
+ DocsUI::Section("Streams master hub", description: "One streams LISTEN connection per web host.") do
+ md <<~'MD'
+ Each Puma worker used to open its own dedicated streams `LISTEN`
+ connection on first SSE use — one direct connection per worker, on the
+ same scarce direct-port budget the job-side supervisor scope protects.
+ By default (`streams_listen_scope = :master`) the `pgbus_streams`
+ plugin now runs **one shared listener in the Puma master**; workers
+ connect to it lazily over a Unix socket and receive every wake —
+ including ephemeral broadcast payloads — as framed messages.
+ MD
+ DocsUI::Code(<<~'RUBY', lexer: :ruby, filename: "config/puma.rb")
+ preload_app! # required for :master — the hub waits for the pgbus initializer
+ plugin :pgbus_streams
+ RUBY
+ md <<~'MD'
+ Delivery semantics are unchanged: the synchronous subscribe/ack
+ contract crosses the process boundary, durable wakes self-heal via
+ `read_after`, and ephemeral wakes are never dropped by the transport.
+ The measured cost of the extra hop is noise-level (single-broadcast
+ SSE roundtrip p50 16.0ms via the hub vs 16.9ms per-worker on the same
+ machine).
+ MD
+ DocsUI::Callout(:note) do
+ plain "Fail-safe in every direction: if the hub is absent or dies "
+ plain "(no "
+ code { "preload_app!" }
+ plain ", single-mode Puma, crash), each worker falls back to its own "
+ plain "listener — the connection footprint balloons back to one per "
+ plain "worker (visible in "
+ code { "pgbus doctor" }
+ plain "'s Connection budget and the "
+ code { "pgbus-listen" }
+ plain " census) but no broadcast is ever lost. Roll back with "
+ code { "streams_listen_scope = :process" }
+ plain "."
+ end
+ end
+ end
+
def streams_pool_autoscaling
DocsUI::Section("Streams pool autoscaling",
description: "Let the SSE streams pool grow into spare connections under a burst, and shrink back when it's over.") do
diff --git a/docs/performance.md b/docs/performance.md
index 9f8261cd..25ff0ed5 100644
--- a/docs/performance.md
+++ b/docs/performance.md
@@ -31,6 +31,7 @@ to guess.
| Fan-out writer throughput | does the writer pool scale with thread count? (issue #323 phase 1) | `writer_burst_bench.rb` |
| NOTIFY wake path | every job-insert wake-up (direct listener vs supervisor hub, issue #381) | `notify_wake_bench.rb` |
| NotifyHub failure modes | killed LISTEN backend, wedged fork, FD churn, fan-out cost (issue #381) | `notify_chaos_bench.rb` |
+| Streams master-hub hop | broadcast→SSE roundtrip, per-worker vs master hub (issue #382) | `streams_hub_bench.rb` |
## Measuring
@@ -47,6 +48,7 @@ rake bench:one[streams_read_pool_bench] # streamer replay-read pool (requires P
rake bench:execution_modes # threads vs async DB-connection consumption (requires PGBUS_DATABASE_URL)
rake bench:notify_wake # NOTIFY wake latency + LISTEN census (requires PGBUS_DATABASE_URL)
rake bench:notify_chaos # NotifyHub failure-mode measurements (requires PGBUS_DATABASE_URL)
+rake bench:streams_hub # streams master-hub hop cost + census (requires PGBUS_DATABASE_URL)
```
- **Unit benches** (`benchmarks/*_bench.rb`) isolate gem overhead with a mocked
@@ -88,6 +90,26 @@ register/deregister cycles leak **0** FDs; hub fan-out costs **10.5µs** per
NOTIFY to 10 forks (the pgmq trigger throttle caps real NOTIFY load at
4/s/queue, so the hub is never the bottleneck).
+### Streams master hub (issue #382)
+
+One `Web::Streamer::Listener` in the Puma master serves every worker over a
+Unix socket instead of one dedicated LISTEN connection per worker. Measured
+on the same machine (`streams_hub_bench.rb`, n=50, durable broadcasts,
+single-broadcast SSE roundtrip):
+
+| Mode | p50 | p95 | LISTEN connections (host) |
+|------|-----|-----|---------------------------|
+| `:process` (per-worker, pre-#382) | 16.93ms | 26.67ms | 1 per worker |
+| `:master` (hub → socket hop) | 16.00ms | 19.19ms | **1** |
+
+The master→worker frame hop is noise-level free — the DB round trips
+(`read_after` + NOTIFY) dominate. Like #381, this is a
+**connection-footprint win, not a latency win**. On hub outage every worker
+falls back to its own listener (census-visible balloon, unchanged
+semantics) — verified end-to-end by
+`spec/integration/streams/master_hub_e2e_spec.rb` (census 1 → 2 across a
+mid-stream hub death with zero missed broadcasts).
+
### Streamer connection model (issue #315)
The durable-stream publish and replay hot paths run on a **dedicated streams
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/lib/pgbus/doctor.rb b/lib/pgbus/doctor.rb
index 327f247f..a8b198e7 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; per-worker fallback during a hub outage costs 1 per web process)"
+ 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/lib/pgbus/web/streamer/failover_listener.rb b/lib/pgbus/web/streamer/failover_listener.rb
new file mode 100644
index 00000000..1899db21
--- /dev/null
+++ b/lib/pgbus/web/streamer/failover_listener.rb
@@ -0,0 +1,130 @@
+# 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
+ # @state_mutex guards the cheap shared state (@subscriptions, @impl,
+ # @failed_over) and is only ever held for constant-time work — the
+ # dispatcher's ensure/remove path must never wait behind a failover
+ # build. @failover_mutex serializes the (blocking) build + replay:
+ # a fresh PG connect + N re-LISTEN acks can stall for seconds when
+ # the trigger IS a database problem (review on #384).
+ @state_mutex = Mutex.new
+ @failover_mutex = Mutex.new
+ @subscriptions = Set.new
+ @impl = hub_client
+ @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)
+ @state_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)
+ @state_mutex.synchronize { @subscriptions.delete(queue) }
+ current_impl.remove_listening(queue)
+ rescue HubClient::HubUnavailableError => e
+ @logger.debug do
+ "[Pgbus::Streamer::FailoverListener] remove_listening on a dead hub client " \
+ "(#{e.message}) — ignoring, unlisten GC is best-effort"
+ end
+ nil
+ end
+
+ # Idempotent, callable from the client's on_failure (reader thread)
+ # and from a synchronous ensure failure (dispatcher thread).
+ # @failover_mutex serializes concurrent callers — the second blocks
+ # until the first finishes and then no-ops, so a synchronous retry
+ # after fail_over! always lands on the swapped-in local listener.
+ # The blocking build + replay runs OUTSIDE @state_mutex so concurrent
+ # ensure/remove/stop calls never stall behind it.
+ def fail_over!
+ local = nil
+ @failover_mutex.synchronize do
+ return if @state_mutex.synchronize { @failed_over }
+
+ local = @local_listener_factory.call
+ @state_mutex.synchronize { @subscriptions.dup }.each { |q| local.ensure_listening(q) }
+ # Subscriptions recorded between the snapshot and this swap arrive
+ # via their own retried ensure_listening call on the new impl.
+ @state_mutex.synchronize do
+ @impl = local
+ @failed_over = true
+ end
+ # Ownership transferred to @impl — the rescue must not stop it.
+ local = nil
+ end
+ rescue StandardError => e
+ # A listener the factory STARTED but that never swapped in (the
+ # replay raised) would otherwise leak its thread and LISTEN
+ # connection alongside the dead hub client.
+ begin
+ local&.stop
+ rescue StandardError
+ nil
+ end
+ # Hub dead AND the local listener can't be built (DB down, config
+ # broken). Mark failed-over so callers stop rebuilding; @impl stays
+ # on the dead client — every ensure_listening resolves nil and the
+ # dispatcher rides its existing timeout tolerance until the worker
+ # recycles.
+ @state_mutex.synchronize { @failed_over = true }
+ @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
+ @state_mutex.synchronize { @impl }
+ end
+ end
+ end
+ end
+end
diff --git a/lib/pgbus/web/streamer/hub_client.rb b/lib/pgbus/web/streamer/hub_client.rb
new file mode 100644
index 00000000..9d2e8268
--- /dev/null
+++ b/lib/pgbus/web/streamer/hub_client.rb
@@ -0,0 +1,199 @@
+# 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, IOError, ArgumentError, ThreadError => e
+ # ArgumentError: a socket path over the platform sun_path limit;
+ # IOError: a path that exists but is not a socket; ThreadError: the
+ # reader thread could not spawn. All must fall back exactly like a
+ # refused connect, never abort worker boot — and never leak the
+ # half-opened socket.
+ close_quietly(@sock)
+ @sock = nil
+ 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
+ rescue StandardError => e
+ # The reader thread is the ONLY detector of hub death — an
+ # unexpected error must not let it exit with the client still
+ # reporting healthy, or the worker goes silently deaf.
+ mark_dead("master hub reader crashed: #{e.class}: #{e.message}") 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). Bounded: a master
+ # that stopped draining its input would otherwise block this write
+ # forever, and the ack deadline only starts ticking AFTER the write
+ # returns — so a stalled write is itself a failover trigger.
+ def write_frame(message)
+ data = HubProtocol.encode(message)
+ deadline = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) + @ack_timeout
+ @write_mutex.synchronize do
+ until data.empty?
+ begin
+ written = @sock.write_nonblock(data)
+ data = data.byteslice(written..)
+ rescue IO::WaitWritable
+ remaining = deadline - ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
+ raise Errno::ETIMEDOUT, "write stalled" if remaining <= 0 || !@sock.wait_writable(remaining)
+ end
+ end
+ end
+ rescue IOError, Errno::EPIPE, Errno::EBADF, Errno::ECONNRESET, Errno::ETIMEDOUT => 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/lib/pgbus/web/streamer/hub_protocol.rb b/lib/pgbus/web/streamer/hub_protocol.rb
new file mode 100644
index 00000000..0c92fb47
--- /dev/null
+++ b/lib/pgbus/web/streamer/hub_protocol.rb
@@ -0,0 +1,85 @@
+# 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, peer death
+ # mid-frame, OR a connection reset: an abrupt close can surface as
+ # ECONNRESET instead of orderly EOF depending on unread data and
+ # platform (Ruby 4.0 reports it deterministically where 3.x saw EOF),
+ # and both mean the same thing here: the peer is gone. 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
+
+ body = body.force_encoding(Encoding::UTF_8)
+ raise ProtocolError, "malformed frame: invalid UTF-8" unless body.valid_encoding?
+
+ decoded = JSON.parse(body)
+ raise ProtocolError, "malformed frame: expected a JSON object, got #{decoded.class}" unless decoded.is_a?(Hash)
+
+ decoded
+ rescue JSON::ParserError => e
+ raise ProtocolError, "malformed frame: #{e.message}"
+ rescue Errno::ECONNRESET
+ nil
+ 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/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 1d5a4cd7..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
@@ -102,6 +105,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..0f97f513
--- /dev/null
+++ b/lib/pgbus/web/streamer/master_hub.rb
@@ -0,0 +1,414 @@
+# 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
+ # Serializes the FULL start and stop sequences: a stop racing an
+ # in-progress start (blocked in the listener factory's PG connect)
+ # must WAIT for it and then tear everything down — otherwise stop
+ # returns having cleaned nothing and start finishes building a live
+ # hub afterwards. Loop threads never take this mutex (they read
+ # @running via @table_mutex), so holding it across the blocking
+ # startup cannot deadlock them.
+ @lifecycle_mutex = Mutex.new
+ @table_mutex = Mutex.new
+ @workers = {}
+ # Plain Hash, entries created ONLY at subscribe time — a default
+ # proc here would leak one empty Set per wake that arrives for an
+ # already-unsubscribed channel (in-flight NOTIFYs after the last
+ # unsub, per-record stream names → unbounded, review on #384).
+ @queue_refs = {}
+ @stop_signal = Queue.new
+ @next_id = 0
+ @dropped_durable_wakes = 0
+ @evicted_workers = 0
+ @running = false
+ end
+
+ 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+.
+ # If any step after the listener exists fails (bad socket path, chmod,
+ # thread spawn), the listener — and its dedicated LISTEN connection,
+ # the exact resource this hub conserves — is stopped before the error
+ # propagates; MasterHubBoot's rescue never sees a leaked connection.
+ def start
+ @lifecycle_mutex.synchronize { locked_start }
+ end
+
+ def stop
+ @lifecycle_mutex.synchronize { locked_stop }
+ end
+
+ private
+
+ def locked_start
+ @table_mutex.synchronize { @running = true }
+ @listener = @listener_factory.call(dispatch_queue: @dispatch_queue)
+ FileUtils.rm_f(@socket_path)
+ # Owner-only: the socket carries every stream wake including
+ # ephemeral HTML payloads, and there is no peer authentication —
+ # the filesystem mode IS the access control. The umask covers the
+ # bind itself so there is no window in which the socket exists with
+ # wider permissions; the chmod stays as the second guarantee.
+ # File.umask is process-wide, but this runs once at hub start in
+ # the Puma master and is restored in the ensure.
+ old_umask = File.umask(0o177)
+ begin
+ @server = UNIXServer.new(@socket_path)
+ ensure
+ File.umask(old_umask)
+ end
+ File.chmod(0o600, @socket_path)
+ @accept_thread = Thread.new { accept_loop }
+ @fanout_thread = Thread.new { fanout_loop }
+ @status_thread = Thread.new { status_loop }
+ self
+ rescue StandardError
+ @table_mutex.synchronize { @running = false }
+ close_quietly(@server)
+ @listener&.stop
+ @listener = nil
+ raise
+ end
+
+ def locked_stop
+ @table_mutex.synchronize do
+ return self unless @running
+
+ @running = false
+ end
+ @stop_signal << :stop
+ 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
+
+ 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 running?
+ @table_mutex.synchronize { @running }
+ end
+
+ def accept_loop
+ loop do
+ begin
+ sock = @server.accept
+ rescue IOError, Errno::EBADF, Errno::EINVAL
+ # server closed during stop
+ break
+ end
+ begin
+ register_worker(sock)
+ rescue StandardError => e
+ # One bad connection must not stop the hub accepting others.
+ @logger.warn { "[Pgbus::Streamer::MasterHub] failed to register a worker: #{e.class}: #{e.message}" }
+ close_quietly(sock)
+ end
+ end
+ 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
+ rescue StandardError => e
+ # e.g. ensure_listening raising inside handle_sub — the ensure still
+ # severs this worker (its fallback takes over), but never silently.
+ @logger.warn { "[Pgbus::Streamer::MasterHub] reader for worker #{id} failed: #{e.class}: #{e.message}" }
+ 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] ||= Set.new).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
+
+ begin
+ deliver(message)
+ rescue StandardError => e
+ # One bad message must not stop wake delivery for the host.
+ @logger.error { "[Pgbus::Streamer::MasterHub] wake delivery failed: #{e.class}: #{e.message}" }
+ end
+ end
+ end
+
+ def deliver(message)
+ frame = { "t" => "wake", "q" => message.queue_name, "p" => message.payload }
+ droppable = message.payload.nil?
+ targets = @table_mutex.synchronize do
+ refs = @queue_refs[message.queue_name]
+ refs ? refs.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
+ rescue StandardError => e
+ # e.g. a ProtocolError from encode — never die silently; sever this
+ # worker so its fallback takes over.
+ @logger.warn { "[Pgbus::Streamer::MasterHub] writer for worker #{id} failed: #{e.class}: #{e.message}" }
+ ensure
+ cleanup_worker(id)
+ end
+
+ def status_loop
+ last_status = nil
+ ticks_since_broadcast = 0
+ loop do
+ # A stop-signal wait instead of sleep, so #stop wakes the thread
+ # immediately even with a long status_interval.
+ break if @stop_signal.pop(timeout: @status_interval)
+ break unless running?
+
+ begin
+ 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
+ rescue StandardError => e
+ @logger.warn { "[Pgbus::Streamer::MasterHub] status tick failed: #{e.class}: #{e.message}" }
+ end
+ 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]
+ next false unless refs
+
+ 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/lib/pgbus/web/streamer/master_hub_boot.rb b/lib/pgbus/web/streamer/master_hub_boot.rb
new file mode 100644
index 00000000..07d312d8
--- /dev/null
+++ b/lib/pgbus/web/streamer/master_hub_boot.rb
@@ -0,0 +1,149 @@
+# frozen_string_literal: true
+
+require "tmpdir"
+
+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
+ # Guards @hub and @running: written by the caller thread
+ # (start/stop) and the background poller. A hub whose start
+ # outlives stop's join budget is stopped by whichever side sees
+ # the flag last, so teardown can never leave a live hub behind.
+ @state_mutex = Mutex.new
+ @hub = nil
+ @running = false
+ @thread = nil
+ end
+
+ def start
+ ENV["PGBUS_STREAMS_HUB_SOCKET"] = @socket_path
+ @state_mutex.synchronize { @running = true }
+ @thread = Thread.new { wait_and_start }
+ self
+ end
+
+ def stop
+ to_stop = @state_mutex.synchronize do
+ @running = false
+ hub = @hub
+ @hub = nil
+ hub
+ end
+ @thread&.join(2)
+ @thread = nil
+ to_stop&.stop
+ self
+ end
+
+ private
+
+ def running?
+ @state_mutex.synchronize { @running }
+ end
+
+ 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
+ # Register-or-late-stop: if stop ran while the hub was building,
+ # this thread owns the teardown of the hub stop never saw.
+ late = @state_mutex.synchronize do
+ if @running
+ @hub = hub
+ nil
+ else
+ hub
+ end
+ end
+ late&.stop
+ return if late
+
+ log(:info) { "[Pgbus::Streamer::MasterHubBoot] master hub listening at #{@socket_path}" }
+ rescue StandardError => e
+ @state_mutex.synchronize { @hub = nil }
+ # The method-scoped hub may have STARTED before a later step raised
+ # (e.g. a failing logger after registration) — stop it here or its
+ # LISTEN connection outlives the boot failure. MasterHub#stop is
+ # idempotent and safe on a never-started hub.
+ begin
+ hub&.stop
+ rescue StandardError
+ nil
+ end
+ 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..c5a6bed5 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, "master hub boot")
+ 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, "master hub teardown")
+ end
+
def teardown_streamer(launcher)
return unless defined?(Pgbus::Web::Streamer)
@@ -43,8 +79,8 @@ def teardown_streamer(launcher)
log_error(launcher, e)
end
- def log_error(launcher, error)
- message = "[Pgbus::Puma::Plugin] streamer teardown raised: #{error.class}: #{error.message}"
+ def log_error(launcher, error, operation = "streamer teardown")
+ message = "[Pgbus::Puma::Plugin] #{operation} raised: #{error.class}: #{error.message}"
if launcher.respond_to?(:log_writer)
launcher.log_writer.log(message)
elsif defined?(Pgbus) && Pgbus.respond_to?(:logger)
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..4d19683a
--- /dev/null
+++ b/spec/integration/streams/master_hub_e2e_spec.rb
@@ -0,0 +1,144 @@
+# 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
+ @saved_signed_name_secret = Pgbus.configuration.streams_signed_name_secret
+ @saved_health_check_ms = Pgbus.configuration.streams_listen_health_check_ms
+ @saved_heartbeat_interval = Pgbus.configuration.streams_heartbeat_interval
+ @saved_write_deadline_ms = Pgbus.configuration.streams_write_deadline_ms
+ 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_listen_health_check_ms = @saved_health_check_ms
+ Pgbus.configuration.streams_heartbeat_interval = @saved_heartbeat_interval
+ Pgbus.configuration.streams_write_deadline_ms = @saved_write_deadline_ms
+ Pgbus.configuration.streams_signed_name_secret = @saved_signed_name_secret
+ 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
+ hub.stop # idempotent — a mid-test failure must not leak the hub into later examples
+ 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..3402106a
--- /dev/null
+++ b/spec/integration/streams/master_hub_spec.rb
@@ -0,0 +1,160 @@
+# 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
+ hub.stop # idempotent — the happy path already stopped it mid-test
+ end
+ end
+end
diff --git a/spec/pgbus/configuration_spec.rb b/spec/pgbus/configuration_spec.rb
index ac2ccc19..0869343b 100644
--- a/spec/pgbus/configuration_spec.rb
+++ b/spec/pgbus/configuration_spec.rb
@@ -1648,6 +1648,36 @@
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
+
+ it "rejects a non-symbolizable value" do
+ expect { config.streams_listen_scope = 42 }
+ .to raise_error(Pgbus::ConfigurationError, /streams_listen_scope/)
+ 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
diff --git a/spec/pgbus/doctor_spec.rb b/spec/pgbus/doctor_spec.rb
index d787042b..04741786 100644
--- a/spec/pgbus/doctor_spec.rb
+++ b/spec/pgbus/doctor_spec.rb
@@ -55,7 +55,7 @@
it "returns hashes with :name, :status, :detail keys" do
doctor.run.each do |check|
expect(check).to include(:name, :status, :detail)
- expect(check[:status]).to be_in(%i[ok warn fail])
+ expect(check[:status]).to(satisfy { |status| %i[ok warn fail].include?(status) })
end
end
@@ -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
@@ -617,7 +624,7 @@ def budget_check
it "returns the same hash shape as #run" do
doctor.boot_checks.each do |check|
expect(check).to include(:name, :status, :detail)
- expect(check[:status]).to be_in(%i[ok warn fail])
+ expect(check[:status]).to(satisfy { |status| %i[ok warn fail].include?(status) })
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..415d99df
--- /dev/null
+++ b/spec/pgbus/web/streamer/failover_listener_spec.rb
@@ -0,0 +1,120 @@
+# 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 "replay failure after the factory started the listener" do
+ it "stops the never-swapped listener so its LISTEN connection cannot leak" do
+ failover.ensure_listening("pgbus_stream_chat")
+ allow(local_listener).to receive(:ensure_listening).and_raise(StandardError, "replay boom")
+ allow(logger).to receive(:error)
+
+ failover.fail_over!
+
+ expect(local_listener).to have_received(:stop)
+ expect(logger).to have_received(:error)
+ 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
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..aa853d05
--- /dev/null
+++ b/spec/pgbus/web/streamer/hub_client_spec.rb
@@ -0,0 +1,152 @@
+# 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
+
+ def wait_until(timeout: 2.0)
+ deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
+ sleep 0.01 until yield || Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline
+ 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 })
+ wait_until { client.hub_healthy? == false }
+ expect(client.hub_healthy?).to be false
+
+ master_send(master, { "t" => "status", "healthy" => true })
+ wait_until { client.hub_healthy? == true }
+ 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
+ wait_until { client.instance_variable_get(:@pending_acks).values.flatten.any? }
+ master.close
+
+ expect(waiter.value).to eq(:raised)
+ wait_until { failures.any? }
+ 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
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..15d342f5
--- /dev/null
+++ b/spec/pgbus/web/streamer/hub_protocol_spec.rb
@@ -0,0 +1,112 @@
+# 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
+
+ it "reports a connection reset as EOF (abrupt peer close, Ruby-4.0-visible)" do
+ resetting_io = Class.new do
+ def read(_count) = raise Errno::ECONNRESET
+ end.new
+
+ expect(described_class.read_frame(resetting_io)).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
+
+ it "rejects a valid-JSON non-object frame (scalars/arrays would break dispatch)" do
+ body = "[1,2,3]".b
+ writer.write([body.bytesize].pack("N") + body)
+
+ expect { described_class.read_frame(reader) }
+ .to raise_error(described_class::ProtocolError, /expected a JSON object/i)
+ end
+
+ it "rejects invalid UTF-8 bytes" do
+ body = "\xff\xfe{}".b
+ writer.write([body.bytesize].pack("N") + body)
+
+ expect { described_class.read_frame(reader) }
+ .to raise_error(described_class::ProtocolError, /invalid UTF-8/i)
+ end
+ end
+end
diff --git a/spec/pgbus/web/streamer/instance_spec.rb b/spec/pgbus/web/streamer/instance_spec.rb
index debb76e2..ab101c3f 100644
--- a/spec/pgbus/web/streamer/instance_spec.rb
+++ b/spec/pgbus/web/streamer/instance_spec.rb
@@ -549,4 +549,68 @@ 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
+ config.streams_listen_scope = :master
+
+ 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
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..d4cef980
--- /dev/null
+++ b/spec/pgbus/web/streamer/master_hub_boot_spec.rb
@@ -0,0 +1,156 @@
+# 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 "stops a hub that finished building only after stop returned (register-or-late-stop)" do
+ gate = Queue.new
+ slow_factory = lambda do |socket_path:|
+ factory_calls << socket_path
+ gate.pop
+ hub
+ end
+ late = described_class.new(
+ socket_path: socket_path, hub_factory: slow_factory,
+ poll_interval: 0.02, deadline: 0.5, logger: logger
+ )
+ allow(late).to receive_messages(configuration_ready?: true, master_scope?: true)
+
+ late.start
+ poller = late.instance_variable_get(:@thread)
+ wait_until { factory_calls.any? } # the poller is now blocked in the factory
+ stopper = Thread.new { late.stop } # join budget expires while blocked
+ stopper.join
+ gate << :built # the build completes AFTER stop returned
+
+ poller.join(2)
+ expect(hub).to have_received(:stop) # the poller thread owned the teardown
+ 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
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..1cb9cfaa
--- /dev/null
+++ b/spec/pgbus/web/streamer/master_hub_spec.rb
@@ -0,0 +1,325 @@
+# 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
")
+ # Drain B for a bounded window: only status frames may arrive there —
+ # never a wake for chat (a single-frame peek could be satisfied by a
+ # status frame while a leaked wake sat behind it).
+ types = []
+ deadline = Time.now + 0.3
+ while Time.now < deadline && worker_b.wait_readable(0.05)
+ b_frame = Pgbus::Web::Streamer::HubProtocol.read_frame(worker_b)
+ break if b_frame.nil?
+
+ types << b_frame["t"]
+ end
+ expect(types).not_to include("wake")
+ [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
+
+ it "makes a racing stop WAIT for an in-progress start, then tear everything down" do
+ # Without the lifecycle mutex, stop would return before start had built
+ # anything, and start would then finish constructing a fully live hub —
+ # listener, socket, and threads surviving a completed stop.
+ gate = Queue.new
+ gated_factory = lambda do |dispatch_queue:|
+ captured[:dispatch_queue] = dispatch_queue
+ gate.pop
+ fake_listener
+ end
+ racy = described_class.new(
+ config: config, socket_path: socket_path,
+ listener_factory: gated_factory, status_interval: 0.05, logger: logger
+ )
+
+ starter = Thread.new { racy.start }
+ sleep 0.1 # starter is now blocked inside the factory
+ stopper = Thread.new { racy.stop }
+ sleep 0.1
+ expect(stopper).to be_alive # stop is waiting on the lifecycle mutex
+
+ gate << :go
+ starter.join(5)
+ stopper.join(5)
+
+ expect(fake_listener).to have_received(:stop)
+ expect(File.exist?(socket_path)).to be false
+ end
+
+ it "creates the socket with owner-only permissions" do
+ # The socket carries every stream wake including ephemeral HTML and has
+ # no peer authentication — the mode IS the access control.
+ hub.start
+ expect(File.stat(socket_path).mode & 0o777).to eq(0o600)
+ end
+ end
+end