Streams: consolidate SSE LISTEN into the preforking web master (1 per host)

[mhenrixon]

Problem

Split out of #381 (item 2). On preforking web servers, each Puma worker lazily builds its own Pgbus::Web::Streamer::Instance — and with it one dedicated direct LISTEN connection (lib/pgbus/web/streamer/listener.rb, created per worker on the first SSE subscriber via Pgbus::Web::Streamer.current). A host with N Puma workers pins N direct connections; on transaction-pool PgBouncer platforms those come out of the scarce direct slice of max_connections.

Goal: one streams LISTEN connection per host — the listener lives in the Puma master, workers are woken over IPC.

Why this is not a trivial port of the supervisor-side change (#381 item 1)

The job-side consolidation only has to deliver a payload-free wake byte for a mostly-static queue set. The streamer Listener's contract is much richer, and each piece has to survive the master→worker process boundary:

1. Synchronous ensure_listening ack (listener.rb:134-138): Dispatcher#handle_connect blocks until LISTEN is actually active before issuing read_after, otherwise a broadcast committed in the gap is lost (neither in the read_after result nor delivered as a wake). Cross-process, this becomes a worker→master request/ack round trip with a bounded timeout, per new stream subscription.
2. Per-subscriber dynamic channel sets: the LISTEN set follows SSE subscriber registration/GC per worker (ensure_listening / lazy remove_listening). The master must maintain a refcounted union across workers and GC channels only when no worker holds a subscriber.
3. Ephemeral wakes carry payloads (listener.rb:253-278): WakeMessage.payload is the only copy of the broadcast HTML — it cannot be dropped and must be framed and fanned out to the right workers over the pipe/socket, not just a wake byte. Durable wakes (payload nil) are droppable under backpressure; ephemeral are not — the IPC channel needs that distinction.
4. Lazy instantiation: nothing starts at worker boot today (lib/puma/plugin/pgbus_streams.rb only registers teardown hooks); a master-owned listener needs a before_fork-time start plus per-worker inherited channels, and a story for single-mode / non-preforking servers (keep the current per-process listener behind a scope config, mirroring worker_notify_scope).

Sketch

• streams_listen_scope = :master | :process config (default TBD after Connection efficiency: host-level shared LISTEN (supervisor-owned NotifyListener, one direct connection per host) + explicit multi-queue read priority contract #381 item 1 burn-in).
• Puma plugin starts a master-side hub (socketpair per worker, created in before_fork): workers send subscribe(queue)/unsubscribe(queue) control frames and receive wake(queue, payload?) frames; master hub owns the single Listener and a refcounted channel union.
• Framed protocol (length-prefixed), not single bytes — payloads must cross.
• Health: master rebroadcasts listener status; workers degrade to their current per-process listener (or polling read_after) when the master hub is absent/dead.

Acceptance

• Preforking web server with N workers: exactly 1 streams LISTEN connection per host.
• No lost broadcasts across the subscribe/read_after gap (the ack contract holds cross-process).
• Ephemeral wakes are never dropped by the IPC layer; durable-wake backpressure semantics preserved (dispatch_queue_limit).
• Single-mode Puma / non-preforking servers keep working with the per-process listener.

Depends on

• Connection efficiency: host-level shared LISTEN (supervisor-owned NotifyListener, one direct connection per host) + explicit multi-queue read priority contract #381 item 1 (supervisor-owned NotifyListener) — establishes the scope-config + wake-pipe patterns this builds on.

[mhenrixon]

Implementation plan (updated with #381 learnings)

Supersedes the original sketch. #383 (job-side hub) is merged; its patterns and infrastructure are on main and this plan builds on them.

What #381 taught us that changes this design

1. The single-byte W/H/P pipe protocol does NOT transfer. write_nonblock-and-skip is correct for level-triggered wakes; payload frames (ephemeral broadcasts carry their only copy of the HTML) can never be dropped-and-repaired by a rebroadcast. This plan uses a lossless framed transport with per-worker outbound queues and an explicit wedged-worker eviction semantics instead.
2. Fallback discipline must be explicit (the poll-only review round on feat(process): supervisor-owned shared LISTEN — one direct connection per host (#381) #383). Streams have no safe polling equivalent for ephemeral broadcasts — no LISTEN anywhere means loss, not latency. Settled in interview: when the master hub is dead/absent, workers fall back to their own per-worker Listener (temporary connection balloon, zero semantic loss, visible in the census) — the deliberate inverse of the job-side poll-only rule, because the payoff differs.
3. Fork hygiene is solved and reusable: NotifyListener#close_inherited_socket! (IO-level fd close, never PQfinish over a shared socket) and consolidated child cleanup. The transport below avoids inherited FDs entirely (workers connect to a Unix socket lazily; nothing to close at fork), and the master's LISTEN socket still needs the child-side close in Puma workers only if preload inherits it — see step 6.
4. Census + doctor are ready: application_name=pgbus-listen already tags streams listeners; acceptance is measurable via pg_stat_activity, and the doctor's Connection budget "+1 per web-server process (streams)" clause becomes scope-aware.
5. Measurement protocol: baseline on main BEFORE code, three-column after (main / :process / :master), chaos scenarios with numbers in the PR — same as Connection efficiency: host-level shared LISTEN (supervisor-owned NotifyListener, one direct connection per host) + explicit multi-queue read priority contract #381 steps 0/12.
6. Web::Streamer::Listener should adopt NotifyListener.physical_for / shared channel constants instead of its duplicated pair.

Decision

Master-owned streams hub over a lazily-connected Unix domain socket.

• Pgbus::Web::Streamer::MasterHub runs in the Puma master (started by the pgbus_streams plugin): it owns ONE Web::Streamer::Listener (reused as-is — its sync ensure_listening ack, payload-carrying handle_notify, reconnect loop, and SEGV during shutdown: NotifyListener#safe_unlisten_all execs UNLISTEN on a connection #stop concurrently closed #375 shutdown discipline are battle-tested) plus a UNIXServer at a per-master socket path. It maintains refcounted channel subscriptions across workers: first subscriber → listener.ensure_listening, last unsubscribe → remove_listening.
• Workers do NOT inherit sockets. On first SSE use (exactly as lazy as today), a worker's streamer connects to the master's Unix socket. No fork hooks, no FD hygiene for the transport, and non-preforking/single-mode servers simply find no socket and keep today's per-process listener automatically.
• Framed protocol (4-byte big-endian length + JSON):
  • worker → master: {t:"sub", q:} (synchronous — awaits ack), {t:"unsub", q:} (fire-and-forget)
  • master → worker: {t:"ack", q:}, {t:"wake", q:, p: <payload-or-null>}, {t:"status", healthy: <bool>} (on change + periodic rebroadcast)
• Cross-process ack contract preserved: the worker's ensure_listening(q) sends sub and blocks (bounded, mirroring today's ack_timeout) until the master has actually executed LISTEN and replied ack — so Dispatcher#handle_connect's no-lost-broadcast gap guarantee holds across the process boundary. Ack timeout or any transport failure → immediate fallback (below).
• Backpressure: per-worker outbound queue + writer thread in the master. Durable wakes (p:null) are droppable at a queue cap (they self-heal via the next read_after, mirroring dispatch_queue_limit); ephemeral wakes are never dropped. A worker whose queue exceeds the hard cap is evicted (socket severed) — which by design triggers that worker's fallback listener, so a wedged worker degrades itself, never its siblings (the Connection efficiency: host-level shared LISTEN (supervisor-owned NotifyListener, one direct connection per host) + explicit multi-queue read priority contract #381 wedged-fork lesson, adapted for payloads).
• Fallback = failover decorator: Streamer::FailoverListener wraps the hub client and records the subscription set; on connect failure, ack timeout, EOF (master died), or eviction it builds a real per-worker Listener, re-LISTENs the recorded set, and swaps — Dispatcher and Instance are unchanged consumers of the same interface. Once fallen back, a worker stays local until it recycles (Puma phased restart replaces workers on deploy; no flap-back complexity).
• Config: streams_listen_scope — :master (default) | :process. Settled in interview: :master ships as the default immediately (matching Connection efficiency: host-level shared LISTEN (supervisor-owned NotifyListener, one direct connection per host) + explicit multi-queue read priority contract #381's stance); the graceful listener-fallback is what makes that safe — the failure mode is today's behavior, not loss. :process is the escape hatch and the automatic behavior wherever the master socket is absent.
• Master requirements: the hub needs Pgbus.configuration in the master → preload_app! (or an explicit database_url via env). A hub that cannot start logs and creates no socket; every worker falls back. Socket path exported to workers via ENV (PGBUS_STREAMS_HUB_SOCKET), unique per master PID so phased restarts get a fresh hub.

Alternatives rejected

• Inherited socketpairs per worker (the original sketch): requires Puma fork-hook wiring, per-fork FD hygiene, and breaks lazy start. The connect-to-path model needs none of it and yields the single/non-preforking fallback for free.
• Reusing Process::NotifyHub/WakePipe: their protocol is deliberately lossy (1-byte, skip-on-full) — wrong for payloads. The concepts (status rebroadcast, tick self-heal, refcount diffing) carry over; the transport does not.
• Poll-only hub-failure fallback: rejected in interview — ephemeral loss is user-visible; the budget balloon is temporary and census-visible.

Implementation steps (specs before code; deviation log from the first edit)

0. Baseline (BEFORE any code): on main, with the existing Puma test harness, record the census (pgbus-listen backends) for a 2-worker SSE deployment (expect 2) and the streams_bench single-broadcast p50/p95 + burst numbers. Same machine as the after-run.
1. Streamer::HubProtocol (lib/pgbus/web/streamer/hub_protocol.rb): frame encode/decode (length-prefix + JSON), max-frame guard. Pure unit specs including partial-read reassembly and oversized-frame rejection.
2. Streamer::MasterHub (lib/pgbus/web/streamer/master_hub.rb): UNIXServer accept loop, per-worker connection state (sub set, outbound queue, writer thread, caps), refcounted union → injected listener's ensure/remove, wake fanout from the listener's dispatch queue, durable-drop/ephemeral-never-drop caps, eviction, status broadcast (change + rebroadcast interval, injected clock), teardown. Specs with an instance_double listener + real UNIXSocket pairs.
3. Streamer::HubClient (lib/pgbus/web/streamer/hub_client.rb): connect to path, sync ensure_listening with ack deadline, remove_listening, wake frames → dispatch_queue (WakeMessage), status frames, EOF/timeout surfaced as failover triggers. Specs against a scripted fake master.
4. Streamer::FailoverListener (lib/pgbus/web/streamer/failover_listener.rb): decorator — records subscriptions, delegates to hub client, on failure builds the real per-worker Listener via the existing factory, re-LISTENs the set, swaps under a mutex. Specs for every trigger (connect refused, ack timeout, mid-stream EOF, eviction).
5. Config (streams_listen_scope, default :master, validated writer) + docs ConfigReference::GROUPS + drift spec.
6. Instance wiring: listener selection by scope + socket presence (FailoverListener when :master, plain Listener otherwise); Web::Streamer::Listener adopts the shared channel helpers (NotifyListener.physical_for); Puma-worker copy of a preloaded master LISTEN socket closed child-side if present.
7. Puma plugin: start the MasterHub in start(launcher) (rescue → log + no socket), export the socket path, extend the teardown hooks to stop the hub; single-mode detection (no hub).
8. Doctor: Connection budget streams clause becomes scope-aware ("1 per host (streams master hub)" vs "+1 per web-server process"); README table + drift spec.
9. Integration spec (real PG, no Puma): MasterHub + real Listener + two fake worker clients over the Unix socket — sub/ack, payload wake routing, refcount unlisten at zero, census == 1, master-kill → client EOF → FailoverListener serves wakes with its own connection.
10. End-to-end (Puma harness): 2 workers under :master — SSE broadcast delivery, census delta == 1; hub torn down mid-stream → SSE keeps working via fallback (census == 2, the accepted balloon).
11. After-measurement + chaos (same machine as step 0): three-column census + broadcast latency (main / :process / :master); chaos numbers: LISTEN-backend kill (recovery, zero durable loss), wedged worker (eviction → self-fallback, siblings unaffected), master death (EOF → fallback continuity), FD/connection-count flat over worker churn.
12. Docs + CHANGELOG: prominent default-change callout with streams_listen_scope = :process rollback; performance_tuning + running-workers-adjacent docs page section; docs/performance.md numbers.

Verification gates

• bundle exec rspec spec/pgbus — no new failures (fixed seed compare).
• Integration: steps 9–10 files + existing streams integration suite green under PGBUS_DATABASE_URL.
• bundle exec rake rubocop; docs rake lint + config drift spec; README doctor drift spec.
• Step 0 baseline committed before implementation; step 11 three-column numbers in the PR body. Honest framing: connection-footprint win on web hosts; broadcast latency must be within noise.

Out of scope

• Job-side (Process::NotifyHub) changes — done in feat(process): supervisor-owned shared LISTEN — one direct connection per host (#381) #383.
• Flap-back from fallback to a recovered hub without a worker recycle.
• Non-Puma preforking servers (the socket-path contract is server-agnostic, but only the Puma plugin is wired).

Settled in interview (do not re-litigate)

• Hub-failure fallback: per-worker listener (semantics over budget), not poll-only.
• Default: :master immediately; :process is the escape hatch.
• Scope: full implementation in one run/PR.