feat(streams): master-owned shared LISTEN — one streams connection per web host (#382)

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.

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

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|...]"

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("<turbo-stream>warmup</turbo-stream>")
+  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("<turbo-stream>#{i}</turbo-stream>")
+    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(
+    "%-32<label>s listener=%-16<kind>s census=%<census>d  n=%<n>d  " \
+    "p50=%<p50>.2fms  p95=%<p95>.2fms  max=%<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."

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" => [

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

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

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