Connection efficiency: host-level shared LISTEN (supervisor-owned NotifyListener, one direct connection per host) + explicit multi-queue read priority contract

[mhenrixon]

Problem

On transaction-pool PgBouncer platforms (PlanetScale Postgres and friends), every LISTEN connection must bypass the pooler and hit the direct port — and the direct slice of max_connections is the scarcest resource in the whole system (the pooler's server-side pools draw from the same ceiling, minus superuser-reserved slots).

pgbus's current LISTEN footprint scales with process count, not host count:

• Worker NotifyListener (lib/pgbus/process/notify_listener.rb): one dedicated direct connection per worker fork.
• Consumer NotifyListener (lib/pgbus/process/consumer.rb:366): one more per consumer fork.
• Streamer Listener (lib/pgbus/web/streamer/listener.rb): one per web-server worker process.

A deployment with 5 worker capsules + 2 consumers on a job host and 2 web hosts × 4 Puma workers pins 7 + 8 = 15 direct connections at steady state — before migrations, consoles, or monitoring tools take theirs. On a small cluster (max_connections = 30) this collides with the pooler's own server pools and produces FATAL: remaining connection slots are reserved for roles with the SUPERUSER attribute.

The waste is structural: a single PG connection can LISTEN on any number of channels. N forks on one host listening to overlapping channel sets need one connection, not N.

Proposal

1. Supervisor-owned NotifyListener (job role): 1 direct connection per host

Move the NotifyListener up from Worker/Consumer forks into the Supervisor process:

• Supervisor builds ONE NotifyListener on the union of all capsules' + consumers' physical queue channels (channel names are already deterministic: pgmq.q_<prefix>_<queue>.INSERT).
• On NOTIFY, the supervisor fans the wake signal out to the fork(s) whose queue list contains the notifying queue, via IPC — a self-pipe per fork (write end held by supervisor, read end inherited across fork) fits the existing WakeSignal shape (lib/pgbus/process/wake_signal.rb); the fork's wait loop selects on the pipe instead of (or in addition to) its own LISTEN connection.
• Fork-side behavior on missing/dead pipe: fall back to plain polling, exactly like today's NotifyListener failed to start path.
• Existing reconnect/self-probe/health-check logic (issues Self-probe LISTEN/NOTIFY delivery at listener start #191, Retry failed NotifyListener startup from the worker loop #193, Adopt NotifyListener's reconnect loop in the Streamer Listener #194, SEGV during shutdown: NotifyListener#safe_unlisten_all execs UNLISTEN on a connection #stop concurrently closed #375) stays — it just runs once per host instead of once per fork.
• Config: something like worker_notify_scope = :supervisor | :fork (default :fork for one release, flip default after burn-in), so the change is opt-in and reversible.

Result: a job host's LISTEN footprint drops from worker_forks + consumer_forks to 1.

2. Streams listener consolidation (web role): 1 per host instead of 1 per Puma worker

Puma workers are forked from a master; the same self-pipe pattern applies:

• A Puma plugin (or before_fork hook) starts the Streamer Listener in the master process; workers inherit read-pipes and get woken via IPC.
• Single-mode / non-preforking servers keep the current per-process listener (scope config again).

Result: web hosts go from puma_workers LISTEN connections to 1.

3. Make multi-queue read priority an explicit, documented, tested contract

capsule_dsl.rb documents "list order = strict priority", but the actual mechanism is incidental: read_multi builds pgmq.read(q1) UNION ALL pgmq.read(q2) ... LIMIT n (pgmq-ruby client/multi_queue.rb), and the ordering only holds because Postgres's Append node fills the LIMIT from subqueries in order. Nothing asserts it, and a future planner change or pgmq-ruby refactor could silently break it. Also note a side effect worth documenting: subqueries that execute claim messages (set vt) even when the outer LIMIT discards their rows — those messages go invisible for a full visibility timeout without being processed.

• Document the strict-priority semantics (and the vt-claim caveat) on read_multi and in the capsule DSL docs.
• Add an integration test: enqueue on q1/q2/q3, read with a limit smaller than total, assert earlier-listed queues win.
• Consider making the ordering first-class (e.g. per-queue reads in list order with a shared budget, like fetch_prioritized already does for priority sub-queues) so the contract doesn't depend on planner behavior.

4. Doctor: report the direct-connection budget

doctor already checks "Dedicated connections … connect OK". Extend it to count what the current config will pin (forks × listeners under :fork scope, 1 under :supervisor, streams scope, etc.) and print the number, so operators can do capacity math from the doctor output alone.

Why not "just turn worker_notify off"

Disabling NOTIFY wake-up reintroduces the empty-read polling storm the listener exists to prevent (tens of millions of pgmq.read calls/day at default polling intervals). The fix is to make wake-ups cheap, not to remove them.

Acceptance

• Job host with 5 capsules + 2 consumers: exactly 1 direct LISTEN connection (supervisor scope).
• Preforking web server with N workers: exactly 1 streams LISTEN connection per host.
• Kill -9 on the supervisor's listener connection: forks degrade to polling within one health-check cycle and recover when the listener reconnects.
• Priority contract test green against real PostgreSQL.
• Doctor prints the pinned direct-connection count.

[mhenrixon]

Implementation plan

Scope note: this plan covers items 1 (supervisor-owned NotifyListener), 3 (multi-queue priority contract), and 4 (doctor connection budget). Item 2 (streams listener consolidation into the preforking web master) is split into #382 — the streamer Listener has a synchronous ensure_listening ack contract the SSE dispatcher depends on for correctness, per-subscriber dynamic channel sets, and ephemeral wakes that carry HTML payloads; all of that would have to cross the master→worker boundary over IPC and deserves its own design.

Problem / Goal

See issue body above. Done looks like: a job host pins exactly one direct LISTEN connection regardless of capsule/consumer count, forks degrade to polling within one health-check cycle when that connection dies and recover when it reconnects, the strict-priority read contract is documented and pinned by an integration test, and pgbus doctor prints the pinned direct-connection count.

Context (read these first)

• lib/pgbus/process/notify_listener.rb — the per-fork LISTEN owner being hoisted. Single-owner connection discipline (issue SEGV during shutdown: NotifyListener#safe_unlisten_all execs UNLISTEN on a connection #stop concurrently closed #375 SEGV), delivering? self-probe (issue Pooler safety: primary-affinity on job read path, NOTIFY poll-floor fix, GUC forwarding #332), reconnect! loop, add_queue/remove_queue command queue. All of this is reused as-is by the supervisor-owned instance; only additions: a connected? accessor and passing the notifying channel to on_wake.
• lib/pgbus/process/supervisor.rb — fork bookkeeping (@forks), 1s monitor_loop tick, liveness-pipe pattern (fork_worker creates IO.pipe, child closes inherited sibling FDs via close_inherited_liveness_readers, parent drains in drain_liveness_pipes, reap closes FDs). The wake pipes mirror this pattern in the opposite direction.
• lib/pgbus/process/worker.rb — start_notify_listener / ensure_notify_listener / stop_dead_notify_listener (lines ~645-707), wake_timeout + listener_delivering? (~612-631), resolve_wildcard_queues (~404-460, stream/event-queue exclusions), physical_queue_names. The :supervisor scope replaces the listener lifecycle with a pipe watcher; wake_timeout semantics are preserved.
• lib/pgbus/process/consumer.rb — mirrors the worker (start_notify_listener ~363, wake_timeout ~353, setup_subscriptions ~152 which derives queue names from EventBus::Registry topic matching).
• lib/pgbus/process/wake_signal.rb — single-waiter event; the pipe watcher thread is just another notify! caller, which is safe by design.
• lib/pgbus/configuration.rb — worker_notify_wakeup + worker_notify_host/port/database_url accessors (~182-188), defaults in #initialize, custom-writer validation pattern (e.g. doctor_on_boot= at ~609), worker_notify_connection_options (~1269), validate! (~626).
• lib/pgbus/doctor.rb — CHECKS registry (33-44), Check struct, BOOT_SKIP/STRICT_FATAL, check_dedicated_connections (348-367). README doctor table drift is enforced by spec/pgbus/readme_doctor_checks_spec.rb (also the "runs N checks" prose).
• lib/pgbus/client.rb:429-449 — read_multi wrapper (the qty: per-queue vs limit: total caveat, issue bug(worker): fetch_multi overflows execution pool, crashes worker loop on multi-queue capsules #123).
• lib/pgbus/process/worker.rb:321-354 — fetch_prioritized (true ordered per-queue reads) vs fetch_multi (UNION ALL, ordering incidental); lib/pgbus/process/consumer.rb fetch_multi_consumer (~267).
• lib/pgbus/configuration/capsule_dsl.rb — documents "list order = strict priority".
• lib/pgbus/event_bus/registry.rb — subscribers, event_queue_names; apps subscribe in initializers (README ~212), so the supervisor process (Rails env already booted by the binstub) sees the registry without eager loading.
• spec/support/pgmq_doubles.rb (build_mock_client), spec/pgbus/process/notify_listener_spec.rb (scripted fake_pg pattern), spec/integration_helper.rb (PGBUS_DATABASE_URL gating, build_pg_listen_connection).
• docs/app/models/config_reference.rb + docs/spec/config_reference_spec.rb — every new config writer must be classified in GROUPS or INTERNAL_ONLY or docs CI fails.

Decision

Chosen approach: supervisor-owned NotifyHub with per-fork wake pipes and supervisor-side union computation.

A new Pgbus::Process::NotifyHub (supervisor process) owns ONE NotifyListener on the union of all capsules' + consumers' physical queue channels and fans wakes out to forks over one IO.pipe per worker/consumer fork (write end held by the supervisor, read end inherited across fork — the liveness-pipe pattern reversed). The supervisor computes the union itself from config.workers (explicit queues), a supervisor-side wildcard resolution (extracted from Worker#resolve_wildcard_queues), and EventBus::Registry topic matching (extracted from Consumer#setup_subscriptions), refreshed every 30s so wildcard churn and late registrations converge. Any transient union gap is bounded by the existing 15s NOTIFY-mode poll ceiling (NOTIFY_FALLBACK_POLL_SECONDS), so under-listening degrades latency, never liveness.

Pipe protocol (parent → child, one pipe per worker/consumer fork)

Single-byte messages:

• W — wake: a NOTIFY arrived for a queue in this fork's set. Child's watcher thread calls wake_signal.notify!.
• H — hub listener healthy (running, connected, delivering). Child treats the remote listener as delivering → wake_timeout may use the 15s NOTIFY ceiling.
• P — hub listener degraded (dead thread, mid-reconnect, or probe says not delivering). Child falls back to polling_interval.

Rules:

• Parent writes with write_nonblock; on IO::WaitWritable the write is skipped — a full pipe already guarantees a pending readable byte, and status is rebroadcast (see below).
• Hub rebroadcasts the current status byte on every state change AND every 5s (cheap; only wakes the watcher thread, not the main loop — the watcher only calls notify! for W). This makes lost/coalesced status bytes harmless.
• Child watcher: blocking readpartial loop; EOFError/IOError (supervisor gone) → mark not-delivering, exit thread; the fork keeps running on plain polling.
• FD hygiene mirrors the liveness pipes exactly, in reverse: at fork, the child closes every inherited sibling wake read-end AND all wake write-ends including its own copy (generalize close_inherited_liveness_readers into a close_inherited_fork_pipes that walks @forks entries for both pipe kinds); the parent closes its copy of the child's read end after fork; reap_children closes the write end when the fork leaves @forks.

Routing

Hub keeps {pid => {queues: Set[physical names] | :wildcard, pipe: IO}}, registered by fork_worker/fork_consumer and deregistered by reap_children. On NOTIFY for channel → physical queue Q: write W to every fork whose set contains Q, plus every :wildcard fork. Over-waking a wildcard fork for an event-bus queue insert costs one empty read_multi and is accepted (correctness only requires never under-waking a fork for a queue it reads).

Hub lifecycle in the supervisor

• Built in Supervisor#run after bootstrap_queues, before boot_processes, when config.worker_notify_wakeup? && scope is :supervisor && (:workers or :consumers role enabled).
• monitor_loop's 1s tick calls hub.tick: (a) ensure the listener is running — restart a dead listener thread with the existing NOTIFY_RETRY_BASE_SECONDS→NOTIFY_RETRY_MAX_SECONDS backoff pattern; (b) refresh the channel union if 30s elapsed (diff via add_queue/remove_queue); (c) compute status = listener.running? && listener.connected? && listener.delivering? and broadcast H/P on change or 5s elapsed.
• shutdown stops the hub (listener #stop — never touches the connection from the stopping thread, per SEGV during shutdown: NotifyListener#safe_unlisten_all execs UNLISTEN on a connection #stop concurrently closed #375) and closes all write ends.

NotifyListener gains: (1) connected? — mutex-guarded !@conn.nil?, so the hub can see "mid-reconnect" (during reconnect! the conn is closed before the new one is published); (2) on_wake is called with the notifying channel: @on_wake.call(channel) — the two existing fork-side lambdas change to ->(_channel) { @wake_signal.notify! }.

Fork side (:supervisor scope)

Worker and Consumer accept a wake_pipe: IO (like liveness_pipe:). When present:

• start_notify_listener/ensure_notify_listener are skipped entirely (no local listener, no dedicated connection).
• A new small class Pgbus::Process::WakePipe wraps the read end: owns the watcher thread, exposes delivering? (last status byte was H, thread alive, no EOF) and stop.
• wake_timeout consults wake_pipe.delivering? where it consults listener_delivering? today (Consumer's wake_timeout also gains the delivering? consult it currently lacks — under :fork scope it keeps consulting @notify_listener.running? as today).
• Missing/dead pipe (nil, EOF, watcher death) → plain polling, exactly like today's "NotifyListener failed to start" path.

Config

config.worker_notify_scope — :supervisor (default) | :fork. Custom writer validates the symbol; documented in ConfigReference::GROUPS. :fork preserves today's per-fork listeners byte-for-byte as the escape hatch. This ships as a default behavior change — CHANGELOG must state it prominently with the rollback (config.worker_notify_scope = :fork).

Alternatives rejected

• Fork→supervisor subscription reporting over IPC (forks tell the hub their queue sets): the general answer, but it needs a second child→parent channel with framing plus a sync protocol, and buys nothing — the supervisor already has config.workers, can resolve wildcards itself, and the 30s refresh + 15s poll ceiling cover dynamic sets. Rejected for complexity.
• Broadcast every NOTIFY to all forks (no routing table): N empty reads per insert across unrelated forks — reintroduces a read storm proportional to fork count on busy hosts. Rejected.
• Eager-loading the Rails app in the supervisor to guarantee registry completeness: a real behavior change (parent memory profile, fork-unsafe initializer risk) for a problem the refresh tick + poll ceiling already bound. Rejected.
• First-class per-queue ordered reads in fetch_multi (item 3): changes a hot path (one round trip per non-empty queue vs one UNION ALL). Deferred; document + pin with a test instead (settled in interview).

Settled in interview (do not re-litigate)

• Item 2 (streams/Puma consolidation) is OUT of this plan — tracked in the follow-up issue linked below.
• Item 3 is document + integration test only; no fetch_multi rewrite, no new config flag.
• worker_notify_scope defaults to :supervisor from day one; :fork is the escape hatch. No burn-in release with an opt-in default.

Implementation steps

Specs before code at every step (RED → GREEN). Suggested as three stacked PRs: (A) steps 0–4, (B) steps 5–9, (C) steps 10–13.

0. Baseline capture — BEFORE any code change (perf prime directive: no baseline, no delta):

   • New integration bench benchmarks/notify_wake_bench.rb (BenchSupport harness, gated on PGBUS_DATABASE_URL, wired into rake bench:integration): measures p50/p95/p99 wake latency (send_message → worker claim on an idle queue, NOTIFY path) and empty-read rate (pgbus.client.read_multi/read_batch calls per second at steady idle) for a supervisor with several capsules + consumers.
   • Run it on main under today's per-fork listeners and record: wake latency distribution, empty-read rate, sustained throughput (existing rake bench:integration suite), and the direct-connection census — count of dedicated LISTEN connections at steady state (SELECT count(*) FROM pg_stat_activity filtered on the dedicated connections' application_name; if DedicatedConnection doesn't set one yet, add fallback_application_name: "pgbus-listen" as part of this step so the census is queryable before AND after).
   • Commit the numbers in the PR body (and docs/performance.md if representative numbers move). /perf automates the same-machine before/after with a worktree.

1. Extract shared queue-set derivation (Layer 4 groundwork):

   • spec/pgbus/process/wildcard_queue_resolver_spec.rb, then lib/pgbus/process/wildcard_queue_resolver.rb: extract the body of Worker#resolve_wildcard_queues (pgmq.meta query, DLQ/stream/event exclusions, prefix stripping) into a call-able used by both Worker and the hub. Worker keeps its logging/fallback wrapper.
   • spec/pgbus/event_bus/registry_spec.rb addition, then Registry#queue_names_for_topics(topics): extract the pattern_overlaps? matching from Consumer#setup_subscriptions; Consumer delegates to it.

2. Configuration (Layer 0): spec/pgbus/configuration_spec.rb — worker_notify_scope default :supervisor, writer accepts :supervisor/:fork (string or symbol), raises ConfigurationError otherwise. Implement accessor + writer + default. Add to docs/app/models/config_reference.rb GROUPS (LISTEN/NOTIFY group); run docs/spec/config_reference_spec.rb.

3. NotifyListener additions: extend spec/pgbus/process/notify_listener_spec.rb (scripted fake_pg pattern) — connected? true after start, false during a scripted reconnect window and after stop; on_wake receives the channel string. Implement; update the two fork lambdas to ->(_channel) arity.

4. WakePipe (child side): spec/pgbus/process/wake_pipe_spec.rb using real IO.pipe — W byte notifies an injected wake signal; H/P flip delivering?; EOF marks not-delivering and ends the thread; stop joins. Implement lib/pgbus/process/wake_pipe.rb.

5. NotifyHub (supervisor side): spec/pgbus/process/notify_hub_spec.rb with an injected fake listener (instance_double) and real pipes — union computation from a config with explicit capsules, a wildcard capsule (resolver stubbed), and consumers (registry stubbed); routing (targeted W, wildcard fan-out); status broadcast on change + rebroadcast interval (injected clock); dead-listener restart backoff; refresh diffing calls add_queue/remove_queue; deregister closes the write end. Implement lib/pgbus/process/notify_hub.rb.

6. Supervisor wiring (Layer 4): extend spec/pgbus/process/supervisor_spec.rb — hub built only under :supervisor scope + worker_notify_wakeup? + relevant roles; fork_worker/fork_consumer create wake pipes and register with the hub; child-side FD hygiene (close_inherited_fork_pipes); reap deregisters; monitor_loop calls hub.tick; shutdown stops the hub. Add the scope to the boot banner (log_boot_banner: worker_notify_scope=...). Implement.

7. Worker under :supervisor scope: extend spec/pgbus/process/worker_spec.rb — with wake_pipe: present: no local NotifyListener is started or ensured; wake_timeout uses the NOTIFY ceiling when wake_pipe.delivering? and polling_interval otherwise; shutdown stops the pipe watcher. :fork scope behavior unchanged (regression assertions). Implement.

8. Consumer under :supervisor scope: same shape in spec/pgbus/process/consumer_spec.rb (mock client via build_mock_client; remember it is a strict double — stub any new client calls). Also give Consumer's wake_timeout the delivering? consult under :fork scope to match Worker (issue Pooler safety: primary-affinity on job read path, NOTIFY poll-floor fix, GUC forwarding #332 parity). Implement.

9. Integration spec (gated on PGBUS_DATABASE_URL): spec/integration/supervisor_notify_scope_spec.rb — run a real NotifyHub with a real NotifyListener against PG (reuse build_pg_listen_connection conventions): register two fake forks (real pipes) with disjoint queue sets, send_message to one queue, assert only that fork's pipe receives W; terminate the listener's backend via pg_terminate_backend, assert P is broadcast within ~2 health-check cycles and H after reconnect (acceptance: kill −9 degradation/recovery).

10. Doctor connection budget (item 4): extend spec/pgbus/doctor_spec.rb — new check "Connection budget" reporting the pinned direct-LISTEN count: under :fork scope, capsule count + consumer count (when worker_notify_wakeup?, honoring config.roles); under :supervisor, 1; plus a "+1 per web server process (streams)" clause when streams_enabled. Always :ok (informational). Register in CHECKS; not in STRICT_FATAL; runs at boot (config-only). Update the README doctor table + "runs N checks" prose (spec/pgbus/readme_doctor_checks_spec.rb enforces both).

11. Priority contract (item 3, document + test only): spec/integration/multi_queue_priority_spec.rb — create three prefixed queues, enqueue 3 messages each, Pgbus.client.read_multi([q1, q2, q3], qty: 5, limit: 5), assert all of q1's messages returned and none of q3's (earlier-listed queues win). Document on Client#read_multi (client.rb ~429) and in capsule_dsl.rb: the strict-priority semantics, that the mechanism is Postgres Append-order filling the LIMIT (this test is the canary), and the vt-claim caveat — discarded subquery rows are still claimed and go invisible for a full visibility timeout.

12. After-measurement + failure-mode measurement (same machine, same scripts as step 0):

    • Re-run benchmarks/notify_wake_bench.rb, rake bench:integration, and the connection census under :supervisor scope, AND under :fork scope (the escape hatch must be measurably unchanged from the step-0 baseline). Report all three columns (main / :fork / :supervisor) in the PR body.
    • Expected/honest framing for the PR: this is a connection-footprint win (N forks + M consumers → 1 direct LISTEN connection), not a throughput win. Wake latency should be ~unchanged (the pipe adds one supervisor hop, microseconds); throughput should be unchanged; any regression beyond noise blocks merge until explained.
    • Failure-mode measurements — each with a number in the PR body; automate as integration specs where the assertion is stable, script + document the rest as a repeatable chaos run:
      a. Shared LISTEN backend killed (pg_terminate_backend): time to P broadcast (target ≤ 2 health-check cycles ≈ 2s at defaults), empty-read rate during the degraded window (forks fast-poll at polling_interval — latency improves, DB read load spikes; both get measured), time to H after reconnect, and zero lost jobs (messages enqueued mid-outage all process).
      b. Worker fork SIGKILL under load: restart re-registers a fresh wake pipe; surviving forks' wake latency unaffected; no FD leak across 50 crash→restart cycles (supervisor open-FD count flat).
      c. Supervisor SIGKILL: orphaned forks observe pipe EOF → degrade to polling and keep processing (measure continued throughput until they're reaped/restarted by the platform).
      d. Wedged fork (stops draining its pipe): supervisor write_nonblock skips it; other forks' wake latency unaffected (measure), no supervisor loop stall.
      e. DB restart / failover: one reconnect from the hub (not a per-fork stampede — count reconnect attempts), full H recovery, jobs enqueued during the outage drain afterward.
      f. Sustained NOTIFY flood (~1k inserts/s across queues): supervisor CPU% and p95 wake latency — confirms the single hub is not a fan-out bottleneck vs per-fork listeners.

13. Docs + CHANGELOG: CHANGELOG entries — feat(process): supervisor-owned shared LISTEN (worker_notify_scope, default :supervisor) with the behavior-change callout + :fork rollback, feat(doctor): connection budget check, perf: note if representative numbers moved (with the before/after), docs: multi-queue strict-priority contract. Update docs/performance.md with the wake-latency/connection-census numbers and the new bench. Update the docs site page covering LISTEN/NOTIFY / PgBouncer deployment (find the page under docs/app/views/docs/pages/ that documents worker_notify_*; follow docs/ authoring conventions — rake lint inside docs/, config drift spec).

Verification gates

• bundle exec rspec spec/pgbus/process/wildcard_queue_resolver_spec.rb spec/pgbus/process/wake_pipe_spec.rb spec/pgbus/process/notify_hub_spec.rb spec/pgbus/process/notify_listener_spec.rb spec/pgbus/process/worker_spec.rb spec/pgbus/process/consumer_spec.rb spec/pgbus/process/supervisor_spec.rb spec/pgbus/configuration_spec.rb spec/pgbus/doctor_spec.rb spec/pgbus/readme_doctor_checks_spec.rb — all green.
• PGBUS_DATABASE_URL=postgres://localhost:5432/pgbus_test?gssencmode=disable bundle exec rspec spec/integration/supervisor_notify_scope_spec.rb spec/integration/multi_queue_priority_spec.rb — green against real PG.
• bundle exec rubocop — no offenses.
• Full suite before/after with the same --order seed — no NEW failures (the suite has known cross-spec ordering pollution; compare, don't expect zero).
• docs/: bundle exec rake lint and docs/spec/config_reference_spec.rb green after the config reference addition.

Performance gates (see steps 0 and 12 — measure before, measure after, report both honestly):

• Step-0 baseline captured on main BEFORE step 1 lands: wake latency (p50/p95/p99), empty-read rate, throughput, direct-LISTEN connection census.
• Step-12 after-numbers for :fork (must match baseline within noise) and :supervisor (connection census drops to 1; latency/throughput within noise of baseline).
• Failure-mode numbers (12a–f) in the PR body: degradation ≤ ~2s, recovery verified, zero lost jobs, no FD leaks, no fan-out bottleneck at ~1k inserts/s.
• fetch_multi/read_multi themselves are untouched (item 3 is document + test only), so no client-bench delta is expected there — rake bench:integration before/after confirms.

Out of scope

• Streams/Puma-master listener consolidation (item 2) — Streams: consolidate SSE LISTEN into the preforking web master (1 per host) #382. Do NOT touch lib/pgbus/web/streamer/ or lib/puma/plugin/pgbus_streams.rb.
• Rewriting fetch_multi to first-class ordered reads, or any strict_queue_priority config.
• Fork→supervisor subscription-reporting IPC.
• Eager-loading the Rails app in the supervisor process.
• Dispatcher/scheduler/outbox forks get no wake pipes (they don't LISTEN today).
• Changing worker_notify_* connection-override resolution or DedicatedConnection.

Execution

Execute with /lfg 381.

[mhenrixon]

Fully delivered: items 1, 3, 4 in #383 (supervisor-owned NotifyHub, priority-contract canary, doctor Connection budget) and item 2 in #384 (streams master hub). The example deployment's steady-state direct-LISTEN footprint drops from 15 connections to 2 — one per job host, one per web host — with fail-safe fallback in every direction.