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

[mhenrixon]

Summary

Implements #382: one streams LISTEN connection per web host — the web-side completion of #381's connection-footprint work, built on its shipped patterns (scope config, census tag, doctor budget, fail-safe fallback discipline).

• Streamer::MasterHub (lib/pgbus/web/streamer/master_hub.rb) runs in the Puma master: ONE Web::Streamer::Listener on the refcounted union of every worker's stream channels, fanning wakes — including ephemeral payloads — to workers over a Unix domain socket.
• Streamer::HubProtocol — length-prefixed JSON frames (sub/unsub/ack/wake/status). Deliberately lossless, unlike Connection efficiency: host-level shared LISTEN (supervisor-owned NotifyListener, one direct connection per host) + explicit multi-queue read priority contract #381's 1-byte wake pipes: ephemeral frames carry the only copy of their HTML.
• Streamer::HubClient + Streamer::FailoverListener — the worker side. The synchronous ensure_listening ack contract crosses the process boundary intact (register-before-LISTEN, ack-after-ensure); any transport failure (connect refused, ack deadline, EOF, eviction) swaps in a real per-worker Listener with the recorded subscription set, one-way, until the worker recycles.
• Backpressure: per-worker outbound queues; durable wakes droppable at a cap (self-heal via read_after), ephemeral never dropped — a non-draining worker is evicted and thereby self-degrades to its own listener without touching siblings.
• Streamer::MasterHubBoot + pgbus_streams plugin — exports the socket path pre-fork, defers hub start behind a config-readiness poller (preload_app! effectively required for :master; without it everything quietly stays per-worker).
• streams_listen_scope — :master (default) | :process. ⚠️ Default behavior change, CHANGELOG callout with one-line rollback.
• Doctor Connection budget clause is now scope-aware; docs site + docs/performance.md updated.

Refs #382 (close at your discretion — acceptance below). Refs #381.

Acceptance vs the plan on #382

Acceptance	Result
Preforking server, N workers → 1 streams LISTEN connection	✅ e2e: 2 Instances, census delta = 1
No lost broadcasts across the subscribe/read_after gap	✅ register-before-LISTEN + ack-after-ensure, pinned by a gated-listener unit spec
Ephemeral wakes never dropped by IPC; durable backpressure preserved	✅ unit (drop/evict caps) + integration (real pg_notify payload fidelity)
Single-mode / non-preforking servers keep working	✅ no socket → per-worker listener automatically (Instance selection specs)

Measurements (same machine, local PG)

Baseline captured on main before any code (streams_bench): single-broadcast SSE roundtrip median ~13–15ms, p95 ~20–24ms.

benchmarks/streams_hub_bench.rb (n=50, durable broadcasts):

Mode	p50	p95	LISTEN connections
:process (per-worker, pre-#382)	16.93ms	26.67ms	1 per worker
:master (hub → socket hop)	16.00ms	19.19ms	1 per host

The master→worker hop is noise-level free — DB round trips dominate. Honest framing: connection-footprint win, not a latency win (same as #381).

Failure modes, all measured/asserted against real PG:

• LISTEN backend killed (pg_terminate_backend): listener reconnects, wakes flow again — integration-asserted with real ephemeral payloads before/after.
• Master dies mid-stream: both e2e workers fail over to their own listeners and the next broadcast still reaches both SSE clients; census 1 → 2 (the accepted, census-visible balloon).
• Wedged worker: durable wakes dropped at the cap without eviction; ephemeral pushes past the hard cap → eviction severs only that worker (unit-pinned with real sockets).
• Worker EOF: hub releases its refcounts; last unsubscriber triggers UNLISTEN.

Test plan

• Unit: bundle exec rspec spec/pgbus — 3539 examples, 0 failures (seed 12345)
• Integration (real PG): master_hub_spec, master_hub_e2e_spec, plus pre-existing audience/replay/transactional streams flows and the Connection efficiency: host-level shared LISTEN (supervisor-owned NotifyListener, one direct connection per host) + explicit multi-queue read priority contract #381 specs — 14 examples, 0 failures on Rails 8.1
• bundle exec rake rubocop — 542 files, no offenses
• docs/: rake lint + 69 specs green (config-reference drift guard includes streams_listen_scope)
• README doctor drift spec green

Deviations & judgment calls

• Plan updated on Streams: consolidate SSE LISTEN into the preforking web master (1 per host) #382 before execution (user asked whether it was good to go; it wasn't). Connection efficiency: host-level shared LISTEN (supervisor-owned NotifyListener, one direct connection per host) + explicit multi-queue read priority contract #381 learnings changed the transport (lossless frames, not 1-byte pipes) and forced the fallback decision. Interview settled: per-worker listener fallback (semantics over budget), :master default immediately, one PR.
• Every sub round-trips the listener's idempotent ensure_listening; only UNLISTEN is refcounted. A refcount shortcut on subscribe would ack worker B while worker A's LISTEN was still in flight — reopening the lost-wake gap. My own first spec draft had this wrong; the corrected contract is pinned with a comment.
• Workers connect to a Unix socket path; nothing is inherited across fork (the original sketch used inherited socketpairs). No fork-hook wiring, no FD hygiene for the transport, and single-mode/non-preforking fallback comes free. The socket path travels via ENV exported pre-fork.
• Hub start is deferred behind a config-readiness poller — with preload_app! the plugin's start runs before the app loads, so the hub can't build eagerly. Without preload_app! the deadline expires quietly and workers stay per-worker: :master effectively requires preload_app! (documented on the tuning page).
• E2E ships as two full in-process streamer Instances against a real hub rather than the planned fork-based harness — the Unix socket already IS the process boundary (proven separately with raw client sockets); forking added orchestration without coverage. No puma CLI cluster driving.
• No separate main-worktree census run: the e2e's post-hub-death fallback end-state (census = 2 for 2 workers) is the per-worker architecture's census, measured in the same spec that shows hub mode = 1.
• Bench uses durable mode + a warmup probe: the default :ephemeral broadcast mode races subscription setup, and one lost first event wedges cumulative waits at their timeout — discovered when the first bench run recorded exactly 10010ms per sample.
• Under :master, listener-hosted autoscaler maintenance only runs on the fallback local listener (no idle per-worker LISTEN connection to ride); the publish-path autoscale trigger is unaffected.
• Local-env note: a transient root Gemfile.lock drift to Rails 7.1 made transactional_spec fail locally (Transaction#after_commit is 7.2+); verified pre-existing/environmental against main and green after re-resolving to 8.1. Not a branch defect; the lock is gitignored.

Summary by CodeRabbit

• New Features

  • Streams can share a single master-process LISTEN connection across Puma workers, reducing database connections.
  • Added streams_listen_scope with :master (default) and :process options.
  • Added backpressure handling, worker eviction, durable and ephemeral notification delivery, and automatic failover to independent listeners.
  • Added benchmarking support for comparing listener modes and measuring latency.

• Documentation

  • Added configuration, connection-budget, performance, startup, benchmarking, and rollback guidance.

• Tests

  • Added comprehensive coverage for shared delivery, failover, connection management, and backpressure.

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

📝 Walkthrough

Walkthrough

Streams now support a master-owned PostgreSQL LISTEN connection. Workers receive wake frames over Unix sockets. The change adds listener scopes, acknowledgements, backpressure, health reporting, failover, Puma integration, tests, benchmarks, and documentation.

Changes

Streams master hub

Layer / File(s)	Summary
Listener scope contracts lib/pgbus/configuration.rb, lib/pgbus/doctor.rb, README.md, docs/app/models/config_reference.rb, spec/pgbus/configuration_spec.rb, spec/pgbus/doctor_spec.rb	Adds validated :master and :process scopes. Updates connection-budget reporting and documentation.
Hub protocol and worker failover lib/pgbus/web/streamer/hub_protocol.rb, lib/pgbus/web/streamer/hub_client.rb, lib/pgbus/web/streamer/failover_listener.rb, lib/pgbus/web/streamer/instance.rb, lib/pgbus/web/streamer/listener.rb, spec/pgbus/web/streamer/*	Adds framed Unix-socket transport, subscription acknowledgements, wake delivery, health handling, and local-listener fallback.
Master hub lifecycle and delivery lib/pgbus/web/streamer/master_hub.rb, spec/pgbus/web/streamer/master_hub_spec.rb, spec/integration/streams/master_hub_spec.rb	Adds shared LISTEN ownership, worker subscriptions, wake fanout, backpressure, worker eviction, health broadcasts, cleanup, and recovery.
Puma startup and shutdown lib/pgbus/web/streamer/master_hub_boot.rb, lib/puma/plugin/pgbus_streams.rb, spec/pgbus/web/streamer/master_hub_boot_spec.rb, spec/integration/streams/master_hub_e2e_spec.rb	Starts the hub after configuration readiness in cluster mode. Stops the hub during shutdown and restart.
Benchmark and documentation support benchmarks/streams_hub_bench.rb, Rakefile, docs/performance.md, docs/app/views/docs/pages/performance_tuning.rb, CHANGELOG.md	Adds process-versus-master latency measurement. Documents connection counts, fallback behavior, configuration, and rollback settings.

Estimated code review effort: 5 (Critical) | ~90+ minutes

Sequence Diagram(s)

sequenceDiagram
  participant PumaPlugin
  participant MasterHubBoot
  participant MasterHub
  participant HubClient
  participant StreamerInstance
  PumaPlugin->>MasterHubBoot: start in cluster mode
  MasterHubBoot->>MasterHub: start after configuration readiness
  MasterHubBoot-->>StreamerInstance: export hub socket path
  StreamerInstance->>HubClient: connect and subscribe
  HubClient->>MasterHub: send subscription
  MasterHub-->>HubClient: acknowledge after LISTEN
  MasterHub-->>HubClient: forward wake frame
  HubClient-->>StreamerInstance: dispatch stream wake
  HubClient->>StreamerInstance: trigger local-listener fallback

Loading

Possibly related issues

• Streams: consolidate SSE LISTEN into the preforking web master (1 per host) #382 — The PR implements the issue’s master-owned Streams LISTEN hub and related delivery, backpressure, failover, and scope configuration.
• Connection efficiency: host-level shared LISTEN (supervisor-owned NotifyListener, one direct connection per host) + explicit multi-queue read priority contract #381 — The PR adds shared LISTEN scope configuration and connection-budget reporting.

Possibly related PRs

• mhenrixon/pgbus#80 — Modifies Streams listener construction and connection handling used by this change.
• mhenrixon/pgbus#151 — Relates to wake payload handling and durable or ephemeral event delivery.
• mhenrixon/pgbus#377 — Provides related listener shutdown and thread-safety changes used by hub-backed listener management.

Suggested labels: streaming, stability

Poem

A rabbit shares one listening ear,
Unix-socket wakes arrive clear.
Durable queues track each stream,
Workers fail over when needed.
One hub carries every wake.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 23.23% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly summarizes the main change: a master-owned shared LISTEN connection for Streams per web host.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches 💡 1

📝 Generate docstrings 💡

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch issue-382-streams-master-hub

---

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Actionable comments posted: 17

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@benchmarks/streams_hub_bench.rb`:
- Around line 88-94: Update the sampling loop around wait_for_events to capture
its returned event count and validate it before appending to samples. Require
the expected count of i + 2, and abort or report failure consistently with the
warmup validation if delivery is incomplete; only record timing samples after
successful delivery.

In `@lib/pgbus/web/streamer/failover_listener.rb`:
- Around line 66-84: Update fail_over! so `@failed_over` is set under `@mutex`, then
build the local listener and replay subscriptions outside the lock; reacquire
`@mutex` only to swap `@impl` to the completed listener. Preserve the existing
one-way no-retry behavior by keeping `@failed_over` true and leaving `@impl`
unchanged when the factory or replay raises, while retaining the current error
logging.

In `@lib/pgbus/web/streamer/hub_client.rb`:
- Around line 104-116: Update reader_loop to rescue unexpected StandardError
exceptions after the existing specific rescues and call mark_dead with an
appropriate error message unless `@stopping`, ensuring errors from handle_frame or
logging cannot terminate the reader thread silently. Preserve the existing
protocol and transport-specific handling.
- Around line 44-50: Widen the rescue in HubClient#connect to convert socket
setup failures such as ArgumentError and IOError, in addition to
SystemCallError, into HubUnavailableError with the existing contextual message,
allowing Instance#build_hub_listener to fall back to a per-worker listener.
- Around line 134-139: Update HubClient#write_frame to wait for socket
writability with a bounded timeout before writing, using the client’s configured
`@ack_timeout` or established write-timeout mechanism. If the writable wait
expires, mark the hub dead and raise HubUnavailableError so failover proceeds;
preserve the existing transport-error handling for write failures.

In `@lib/pgbus/web/streamer/master_hub_boot.rb`:
- Around line 50-57: Synchronize shared `@hub` and `@running` state in
MasterHubBoot#start, `#stop`, and `#wait_and_start` with a Mutex, ensuring hub
creation, assignment, startup, and teardown cannot race. Make `#stop` wait for the
background startup path to finish before clearing or stopping `@hub`, so it
returns only after the hub and its thread are fully stopped.

In `@lib/pgbus/web/streamer/master_hub.rb`:
- Around line 85-86: The hub socket currently inherits an unsafe mode; update
master_hub.rb at lines 85-86 to call File.chmod(0o600, `@socket_path`) immediately
after UNIXServer.new(`@socket_path`). Add a master_hub_spec.rb example at lines
262-280 asserting File.stat(socket_path).mode & 0o777 equals 0o600 after
hub.start.
- Around line 93-105: Synchronize the shared running-state in MasterHub#stop and
the corresponding status/accept loop checks using `@table_mutex` or
Concurrent::AtomicBoolean. Make the check-and-transition to stopped atomic so
concurrent stop calls allow only one caller to close the server, signal workers,
join threads, stop the listener, and remove the socket; subsequent calls should
return self without repeating cleanup.
- Around line 191-200: Ensure every long-lived MasterHub loop logs failures with
Pgbus.logger and continues running. In
lib/pgbus/web/streamer/master_hub.rb:191-200, move rescue handling inside
fanout_loop; at 125-132, rescue per iteration around `@server.accept` and
register_worker while `@running`, retaining the outer stop-path rescue; at
253-267, add StandardError handling after socket rescues, log the worker id, and
evict_worker; at 269-284, rescue each health/status iteration and replace sleep
`@status_interval` with a stop-signal wait so stop wakes the thread. Each site
requires direct changes.
- Around line 202-209: Update deliver and release_queue_refs to read `@queue_refs`
without invoking its default proc, so missing queue names do not create empty
Set entries. Preserve the existing target filtering and release behavior for
queues that already have references, while avoiding release/listener actions for
queues absent from the table.

In `@spec/integration/streams/master_hub_e2e_spec.rb`:
- Around line 128-135: Update the ensure block in the integration example to
call hub.stop, alongside the existing client, worker, and harness cleanup.
Ensure the hub is stopped even when expectations fail before the mid-test
hub.stop call, while preserving the existing cleanup operations.
- Around line 20-34: Update the setup and teardown hooks in the master hub
integration spec to save the original values of streams_listen_health_check_ms,
streams_heartbeat_interval, and streams_write_deadline_ms before overriding them
in before(:all), then restore each saved value in after(:all) alongside
listen_notify and streams_signed_name_secret.

In `@spec/integration/streams/master_hub_spec.rb`:
- Around line 155-157: Update the example teardown around MasterHub to call
hub.stop from the ensure block, ensuring cleanup runs even when earlier
expectations or wait_until raise. Track whether the happy-path stop already
occurred and only perform the ensure stop when needed, while preserving safe
cleanup if the hub was not stopped.

In `@spec/pgbus/web/streamer/hub_client_spec.rb`:
- Around line 102-115: Replace the fixed sleep waits in the hub client
status-tracking examples and the corresponding examples around the reader-thread
assertions with bounded polling that exits as soon as the expected state is
observed. Preserve the fast path and enforce a timeout so slow CI runs fail
clearly; for the negative failures assertion, use a short fixed wait or assert
after client.stop joins the reader thread.

In `@spec/pgbus/web/streamer/instance_spec.rb`:
- Around line 585-593: Update the “falls back to a per-worker Listener when the
socket path is exported but dead” example to set streams_listen_scope explicitly
to :master, matching the sibling examples and ensuring the test reaches the
dead-socket connection fallback in build_hub_listener.

In `@spec/pgbus/web/streamer/master_hub_spec.rb`:
- Around line 262-280: Add an example in the `#stop` specs that starts `hub`,
inspects `File.stat(socket_path).mode & 0o777`, and asserts the socket mode is
`0o600`, preserving the owner-only permission contract.
- Around line 171-174: Replace the compound matcher around
worker_b.wait_readable with a bounded drain of worker_b’s available frames,
decoding each via HubProtocol.read_frame and asserting directly that none has
“t” equal to “wake”. Keep the bounded timeout and avoid blocking reads inside
the matcher so any leaked wake frame is detected regardless of arrival order.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Repository UI (base), Organization UI (inherited)

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: b995f4b4-76b7-4101-9a41-6b2ea41e8afb

📥 Commits

Reviewing files that changed from the base of the PR and between 1bac78b and 3c2adb3.

📒 Files selected for processing (27)

• CHANGELOG.md
• README.md
• Rakefile
• benchmarks/streams_hub_bench.rb
• docs/app/models/config_reference.rb
• docs/app/views/docs/pages/performance_tuning.rb
• docs/performance.md
• lib/pgbus/configuration.rb
• lib/pgbus/doctor.rb
• lib/pgbus/web/streamer/failover_listener.rb
• lib/pgbus/web/streamer/hub_client.rb
• lib/pgbus/web/streamer/hub_protocol.rb
• lib/pgbus/web/streamer/instance.rb
• lib/pgbus/web/streamer/listener.rb
• lib/pgbus/web/streamer/master_hub.rb
• lib/pgbus/web/streamer/master_hub_boot.rb
• lib/puma/plugin/pgbus_streams.rb
• spec/integration/streams/master_hub_e2e_spec.rb
• spec/integration/streams/master_hub_spec.rb
• spec/pgbus/configuration_spec.rb
• spec/pgbus/doctor_spec.rb
• spec/pgbus/web/streamer/failover_listener_spec.rb
• spec/pgbus/web/streamer/hub_client_spec.rb
• spec/pgbus/web/streamer/hub_protocol_spec.rb
• spec/pgbus/web/streamer/instance_spec.rb
• spec/pgbus/web/streamer/master_hub_boot_spec.rb
• spec/pgbus/web/streamer/master_hub_spec.rb

[coderabbitai]

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (2)

lib/pgbus/web/streamer/master_hub_boot.rb (1)

85-107: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Rescue paths discard a started listener without stopping it. Both sites build an object that already owns a started PostgreSQL LISTEN connection and a thread, then drop the reference in a rescue StandardError without calling stop. The connection and thread stay alive for the life of the process, which is the resource the :master scope exists to save.

• lib/pgbus/web/streamer/master_hub_boot.rb#L85-L107: hold hub in a method-scoped local and call hub&.stop in the rescue, because MasterHub#start starts the shared Listener before it binds the Unix socket.
• lib/pgbus/web/streamer/failover_listener.rb#L77-L101: hold local in a method-scoped local, track whether the swap completed, and call local.stop in the rescue when the swap never happened.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@lib/pgbus/web/streamer/master_hub_boot.rb` around lines 85 - 107, Rescue
paths leak started listeners by discarding them without stopping. In
lib/pgbus/web/streamer/master_hub_boot.rb:85-107, keep hub method-scoped and
call hub&.stop in the rescue. In
lib/pgbus/web/streamer/failover_listener.rb:77-101, keep local method-scoped,
track whether the swap completed, and call local.stop in rescue when it did not.

benchmarks/streams_hub_bench.rb (1)

1-51: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Validate HUB_BENCH_SAMPLES before running the benchmark.

Integer(ENV.fetch("HUB_BENCH_SAMPLES", "50")) accepts zero and negative values. These values make SAMPLES.times produce no samples, so sorted.last is nil. Require SAMPLES.positive? and abort with a clear error when the value is not positive.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@benchmarks/streams_hub_bench.rb` around lines 1 - 51, Validate
HUB_BENCH_SAMPLES immediately after parsing it into SAMPLES, requiring
SAMPLES.positive? and aborting with a clear, actionable error when it is zero or
negative; preserve the existing default and benchmark flow for positive values.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@lib/pgbus/web/streamer/master_hub.rb`:
- Around line 91-95: Update the socket setup around UNIXServer.new in the master
hub to temporarily set a restrictive process umask while binding, restoring the
previous umask afterward with guaranteed cleanup; retain File.chmod as the
second permission guarantee. Ensure the umask scope covers socket creation and
does not leave the process umask changed.

---

Outside diff comments:
In `@benchmarks/streams_hub_bench.rb`:
- Around line 1-51: Validate HUB_BENCH_SAMPLES immediately after parsing it into
SAMPLES, requiring SAMPLES.positive? and aborting with a clear, actionable error
when it is zero or negative; preserve the existing default and benchmark flow
for positive values.

In `@lib/pgbus/web/streamer/master_hub_boot.rb`:
- Around line 85-107: Rescue paths leak started listeners by discarding them
without stopping. In lib/pgbus/web/streamer/master_hub_boot.rb:85-107, keep hub
method-scoped and call hub&.stop in the rescue. In
lib/pgbus/web/streamer/failover_listener.rb:77-101, keep local method-scoped,
track whether the swap completed, and call local.stop in rescue when it did not.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Repository UI (base), Organization UI (inherited)

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: cd650460-2d6c-4d44-9680-67271b028f3c

📥 Commits

Reviewing files that changed from the base of the PR and between 19cd776 and bae2e22.

📒 Files selected for processing (10)

• benchmarks/streams_hub_bench.rb
• lib/pgbus/web/streamer/failover_listener.rb
• lib/pgbus/web/streamer/hub_client.rb
• lib/pgbus/web/streamer/master_hub.rb
• lib/pgbus/web/streamer/master_hub_boot.rb
• spec/integration/streams/master_hub_e2e_spec.rb
• spec/integration/streams/master_hub_spec.rb
• spec/pgbus/web/streamer/hub_client_spec.rb
• spec/pgbus/web/streamer/instance_spec.rb
• spec/pgbus/web/streamer/master_hub_spec.rb

[coderabbitai]

Actionable comments posted: 11

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@lib/pgbus/doctor.rb`:
- Around line 397-413: Update streams_budget_clause for :master scope to append
a brief note that the reported per-web-host budget assumes preload_app! and a
running, reachable streams master hub. Keep the existing :process clause and
budget behavior unchanged.

In `@lib/pgbus/web/streamer/failover_listener.rb`:
- Around line 63-68: Update the rescue for HubClient::HubUnavailableError in
remove_listening to log the failure through Pgbus.logger before returning nil,
matching the file’s existing error-logging behavior and keeping other error
paths unchanged.

In `@lib/pgbus/web/streamer/hub_protocol.rb`:
- Around line 45-66: Update read_frame to reject invalid UTF-8 bodies before
JSON.parse by validating body.valid_encoding? and raising ProtocolError, then
validate the parsed JSON result is a Hash and raise ProtocolError otherwise.
Preserve the existing malformed-JSON, oversized-frame, and connection-reset
handling.

In `@lib/pgbus/web/streamer/instance.rb`:
- Around line 166-197: Update build_hub_listener to rescue ThreadError raised
during HubClient startup, close `@sock` before converting it to
HubClient::HubUnavailableError, and let the existing fallback logging and
per-worker listener path handle the converted error.

In `@lib/pgbus/web/streamer/master_hub_boot.rb`:
- Around line 25-27: Add require "tmpdir" at the top of master_hub_boot.rb so
Dir.tmpdir is available when default_socket_path is evaluated. Keep the existing
default_socket_path implementation unchanged.

In `@lib/pgbus/web/streamer/master_hub.rb`:
- Around line 183-196: Update reader_loop to rescue unexpected StandardError
exceptions after the existing protocol and socket-error handlers, logging the
failure through `@logger` with the worker id and exception details before
cleanup_worker(id) runs. Mirror the writer loop’s unexpected-error handling
without changing the existing specialized rescue behavior.
- Around line 87-109: Update MasterHub#start to clean up partial startup
failures: wrap all steps after `@listener` is created in failure handling that
stops the listener, resets `@running` to false under `@table_mutex`, and re-raises
the original exception. Preserve normal startup behavior and ensure cleanup also
covers failures from UNIXServer.new, File.chmod, or thread creation.

In `@lib/puma/plugin/pgbus_streams.rb`:
- Around line 52-55: Update log_error to accept an operation name with "streamer
teardown" as its default, and include that name in the logged message. In
boot_master_hub’s rescue path, pass a startup-specific operation such as "master
hub boot" when calling log_error, while preserving the existing teardown
behavior for other callers.

In `@spec/integration/streams/master_hub_e2e_spec.rb`:
- Around line 20-40: Update the before(:all) and after(:all) hooks to save and
restore Pgbus.configuration.streams_signed_name_secret using the same pattern as
the other settings. Replace the unconditional nil assignment with restoration of
the saved value, while preserving the existing client reset behavior.

In `@spec/pgbus/configuration_spec.rb`:
- Around line 1651-1675: Add an example to the `#streams_listen_scope` spec
covering a non-symbolizable value such as an Integer, and assert that assigning
it raises Pgbus::ConfigurationError. Match the existing `#worker_notify_scope`
rejection test’s behavior and style so the streams_listen_scope= type-check else
branch is covered.

In `@spec/pgbus/web/streamer/master_hub_boot_spec.rb`:
- Around line 111-131: Add a `#stop` example covering the register-or-late-stop
race: create a separate `described_class` instance with a hub factory blocked by
a `Queue`, start it with configuration and master scope ready, call `stop` while
construction is blocked, release the factory afterward, then wait for and assert
that the built `hub` received `stop`. Use the existing `factory_calls`,
`wait_until`, and hub-stop assertion helpers.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Repository UI (base), Organization UI (inherited)

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: d2b3f2db-ff04-44b8-92c0-75b175e971b5

📥 Commits

Reviewing files that changed from the base of the PR and between bae2e22 and 22996e2.

📒 Files selected for processing (27)

• CHANGELOG.md
• README.md
• Rakefile
• benchmarks/streams_hub_bench.rb
• docs/app/models/config_reference.rb
• docs/app/views/docs/pages/performance_tuning.rb
• docs/performance.md
• lib/pgbus/configuration.rb
• lib/pgbus/doctor.rb
• lib/pgbus/web/streamer/failover_listener.rb
• lib/pgbus/web/streamer/hub_client.rb
• lib/pgbus/web/streamer/hub_protocol.rb
• lib/pgbus/web/streamer/instance.rb
• lib/pgbus/web/streamer/listener.rb
• lib/pgbus/web/streamer/master_hub.rb
• lib/pgbus/web/streamer/master_hub_boot.rb
• lib/puma/plugin/pgbus_streams.rb
• spec/integration/streams/master_hub_e2e_spec.rb
• spec/integration/streams/master_hub_spec.rb
• spec/pgbus/configuration_spec.rb
• spec/pgbus/doctor_spec.rb
• spec/pgbus/web/streamer/failover_listener_spec.rb
• spec/pgbus/web/streamer/hub_client_spec.rb
• spec/pgbus/web/streamer/hub_protocol_spec.rb
• spec/pgbus/web/streamer/instance_spec.rb
• spec/pgbus/web/streamer/master_hub_boot_spec.rb
• spec/pgbus/web/streamer/master_hub_spec.rb

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

lib/pgbus/web/streamer/master_hub.rb (1)

91-136: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift

Serialize start and stop resource transitions.

start sets @running before @listener, @server, and the threads exist. If stop runs while @listener_factory.call blocks, it sets @running to false and returns after cleaning no listener or server. start can then create the listener, socket, and accept thread. Later calls to stop return early because @running is false.

Guard startup, startup-failure cleanup, and shutdown with one lifecycle mutex or equivalent state transition. Add a race spec that blocks listener_factory, calls stop, then releases the factory and verifies that no hub resources remain active.

As per coding guidelines: “Do not leave shared mutable state unsynchronized; use Mutex or Concurrent primitives for concurrent code.”

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@lib/pgbus/web/streamer/master_hub.rb` around lines 91 - 136, The start/stop
lifecycle is unsynchronized, allowing stop to finish before start has created
its resources. Add a dedicated lifecycle mutex/state transition around the full
startup sequence in start, including failure cleanup, and the shutdown sequence
in stop; ensure stop waits for an in-progress start and subsequent stop calls
can clean up all resources. Add a race spec that blocks listener_factory,
invokes stop, releases the factory, and verifies no listener, socket, or worker
threads remain active.

Source: Coding guidelines

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Outside diff comments:
In `@lib/pgbus/web/streamer/master_hub.rb`:
- Around line 91-136: The start/stop lifecycle is unsynchronized, allowing stop
to finish before start has created its resources. Add a dedicated lifecycle
mutex/state transition around the full startup sequence in start, including
failure cleanup, and the shutdown sequence in stop; ensure stop waits for an
in-progress start and subsequent stop calls can clean up all resources. Add a
race spec that blocks listener_factory, invokes stop, releases the factory, and
verifies no listener, socket, or worker threads remain active.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Repository UI (base), Organization UI (inherited)

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 23019484-c0b1-4093-91a2-0d7b4be243d5

📥 Commits

Reviewing files that changed from the base of the PR and between 22996e2 and bb6b248.

📒 Files selected for processing (12)

• lib/pgbus/doctor.rb
• lib/pgbus/web/streamer/failover_listener.rb
• lib/pgbus/web/streamer/hub_client.rb
• lib/pgbus/web/streamer/hub_protocol.rb
• lib/pgbus/web/streamer/master_hub.rb
• lib/pgbus/web/streamer/master_hub_boot.rb
• lib/puma/plugin/pgbus_streams.rb
• spec/integration/streams/master_hub_e2e_spec.rb
• spec/pgbus/configuration_spec.rb
• spec/pgbus/doctor_spec.rb
• spec/pgbus/web/streamer/hub_protocol_spec.rb
• spec/pgbus/web/streamer/master_hub_boot_spec.rb