feat(client): streams-pool hot-swap primitive — de-risk elastic pools (#323 spike)#325
Merged
Conversation
…#323 spike) ## Summary A SPIKE proving that the dedicated streams DB pool can be resized under live load by hot-swapping the PGMQ::Client — the feasibility gate before building any elastic control loop. The #324 benchmark showed async workers collapse ~3.4x under a DB-bound burst on a small pool; this proves we can grow/shrink the pool to fix that, correctly and cheaply. connection_pool 2.x cannot resize in place, so growing = build a new PGMQ::Client + atomically swap the reference + drain and close the old one. This ships that swap primitive (opt-in, default-off, no auto-call — the control loop is deferred). ## The correctness spine - **Atomic swap:** all four streams-pool readers (send_stream_message, with_streams_connection, streams_pool_stats, close) go through a guarded accessor backed by Concurrent::AtomicReference — the dedicated path has no mutex, so a bare ivar swap is a data race. Swaps/close serialize on a mutex; reads stay lock-free (one volatile load). - **Drain on an in-flight COUNTER, not available==size (the load-bearing fix the design review caught):** pgmq's with_connection retries a lost connection (checkout->checkin->checkout), so available momentarily == size while a produce is mid-flight. Closing there would raise PoolShuttingDownError on the retry = a genuinely LOST durable broadcast (the PgBouncer/failover mid-INSERT case). Instead ResizablePool brackets each reader op with an AtomicFixnum on the captured pool and drains on THAT reaching zero — immune to the retry, and independent of the connection_pool version. - **Durable broadcasts loss-safe by construction:** produce captures the pool once, runs its whole retrying INSERT on it; both pools connect to the same DB; close waits for inflight==0 first, so a pool is never shut under a live produce. - **Retire = PGMQ::Client#close** (pgmq's own safe shutdown drain: idle conns close now, in-flight self-close on checkin) — never Thread#kill, bounded log-and-abandon like OutboundPump #321 B3. - **Shared-AR path:** hard no-op (streams pool aliases the job pool, pool_size 1). ## Measured (real DB, live load: 4 producers + 1 reader, grow 5->12, shrink 12->5) swap cost : ~12 ms total, ALL drain+close, build ~= 0 (lazy pool) zero loss : 11,790 produced == 11,790 landed zero leak : app-scoped backends 9 -> 9 zero races : 0 errors, 0 pool-closed final size : restored to 5 => ALL GATES PASS — elastic streams pools are FEASIBLE. Bench: benchmarks/pool_swap_bench.rb (rake bench:pool_swap). ## Test Coverage - resizable_pool_spec (13 unit, no DB): ref swap, inflight bracketing, drain waits on inflight NOT available==size (the data-loss regression test), drain-timeout log-and-abandon, telemetry, concurrent-swap serialization, close_current alias guard. - client_swap_spec (8, mocked PGMQ::Client): resize builds+swaps+routes to new pool, closes old, no-op on unchanged size / shared-AR path, invalid size, concurrent reader safety, close-after-swap. - streams_pool_swap_spec (DB-gated): zero lost broadcasts under concurrent grow, zero leaked connections across grow+shrink. - Existing client_spec (301) + streamer (307) pass unchanged — swap-off is byte-identical (one atomic read + counter bump per op). ## Deferred (out of scope) The control loop — saturation sampling, hysteresis, grow/shrink policy, fleet-wide connection budget, any automatic resize call. This spike decides the swap is cheap+correct; the loop is a follow-up. ## Verification - [x] bundle exec rubocop passes (7 files) - [x] bundle exec rspec spec/pgbus — 3058 examples, 0 failures - [x] DB-gated swap spec + bench pass against real Postgres Refs #323
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 17 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Repository UI (base), Organization UI (inherited) Review profile: ASSERTIVE Plan: Pro Plus Run ID: 📒 Files selected for processing (7)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
mhenrixon
added
performance
7 tasks
mhenrixon
added a commit
that referenced
this pull request
Jul 6, 2026
) * feat(streams): autoscale config + P1/P2 prerequisites (#323) Groundwork for self-tuning streams-pool autoscaling: - Config (opt-in, default off): streams_pool_autoscale (bool), streams_pool_max (optional hard cap, nil = dynamic fair-share ceiling), streams_pool_autoscale_interval (1.0s), streams_application_name ("pgbus_streams"). streams_pool_size stays the baseline/shrink-floor. Validation rejects an inverted ceiling (max < size) at boot. - P1: tag the streams pool connection with a per-process application_name ("pgbus_streams_<pid>") so the autoscaler can count peer processes via pg_stat_activity — DISTINCT application_name is then an exact process count. - P2: build the streams pool from streams_connection_options (not the job connection_options) so a separate/direct streams DB (streams_database_url/ host/port) is honored — fixes a latent wrong-DB bug in the existing streams pool that only surfaces with a separate streams database. Config-drift + full client/config specs green; no behavior change with autoscale off. * feat(streams): self-tuning PoolAutoscaler + streamer wiring (#323) The control loop on top of the #325 hot-swap primitive. One per web process (a sibling of Heartbeat in the Streamer Instance), opt-in and default-off. Decision function (per tick, priority order): 1. EMERGENCY SHRINK — if live DB free connections are critically low (free < max(5, 0.05·maxc)), resize to baseline NOW, overriding busy_ratio AND cooldown. Keys off `free` (an external DB fact) so it is immune to the post-swap lazy-pool transient (BUG-0) and may fire during cooldown. 2. GROW — busy_ratio ≥ 0.85 sustained 3 samples AND free > grow reserve (max(20, 0.20·maxc)): grow into a bounded FAIR SHARE of headroom, free/(peers·1.5), claiming ¼, capped by STEP_MAX(4) and floor(free/2). Peers = DISTINCT application_name LIKE 'pgbus_streams_%' — the exact live process count (zero-config). 3. SHRINK — busy_ratio < 0.30 sustained 20 samples: step toward baseline. 4. else HOLD. Proven safe (see spec): no multi-process exhaustion even in a synchronized cold-boot herd (four stacked grow guards bound the worst case; grow collapses next tick as pg_stat_activity reflects reality); no grow↔emergency limit cycle (GROW_RESERVE ≥ 4× EMERGENCY_MARGIN → ~15% dead-zone); BUG-0 immunity via the cooldown gate. Fail-soft: a probe timeout → HOLD (never blocks the loop). DB I/O is behind an injected HeadroomProbe seam, so the whole policy is DB-free unit-tested (17 cases incl. convergence-under-concurrent-growth and emergency-shrink-during-cooldown) with a scripted probe + injected clock. Wiring: composed only when streams_pool_autoscale AND !shared_connection? (nil — no thread — otherwise). Started after the heartbeat; stopped FIRST in shutdown! so the actuator quiesces before teardown. Client#with_streams_connection made public so the probe can read pg_stat_activity through the streams pool. Default-off = @Autoscaler nil, zero new threads, byte-identical. Streamer + client specs green (464 examples). * feat(streams): dedicated headroom probe + DB-gated autoscale proof (#323) Two fixes the DB-gated integration test forced, plus the test itself: - FIX (P1 URL encoding): tag_application_name space-appended `application_name=` onto the conn string, which is invalid for postgres:// URI form (it already carries ?options=... from apply_connection_bounds — libpq rejects the second bare key=value in the query). Now mirrors append_connection_bounds: URI form uses ?/& separators, keyword form uses a space. The mocked unit test couldn't catch this (fake PGMQ::Client never parses the string); the DB test did, at construction. - FIX (probe starvation): the HeadroomProbe read through the streams pool. At busy_ratio ~= 1.0 — exactly when the loop wants to grow — its own checkout times out (0 slots free), returns nil, and the loop can NEVER grow. Proven with a churn harness. The probe now uses its OWN dedicated single connection (Client#build_streams_probe_connection: lazy, reconnect-on-failure, tagged with pgbus_streams_<pid> so it counts as this process's peer). One extra backend per web process — the honest cost of a headroom monitor that must work under saturation. This reverses the earlier "reuse the pool connection" choice, which assumed the probe could get a slot; it provably can't at 100%. - DB-gated spec (auto-skips without PGBUS_DATABASE_URL): drives real #tick against a real Client + real HeadroomProbe reading live pg_stat_activity. Proves the pool GROWS under sustained saturation (respecting the hard cap), SHRINKS back to baseline when idle, and raises no PoolTimeout. Injects the clock to fast-forward past the 15s cooldown while all DB I/O stays real. Unit + streamer + client + config specs green (779). * feat(streams): autoscaler bench + docs-kit guidance + CHANGELOG (#323) - benchmarks/pool_autoscale_bench.rb (rake bench:pool_autoscale): measures the tick cost (~0.3ms — negligible at a 1s interval) and grow-under-load against a real DB. Honest framing: it proves the loop is cheap and reacts, NOT a per-op speedup — whether autoscaling beats a well-sized static pool is a burstiness question (#324). - docs-kit Phlex page (performance_tuning.rb): user-facing "Streams pool autoscaling" section — when to enable (bursty SSE), the zero-connection-config model, self-protection (emergency shrink), and the PgBouncer/direct-connection caveat. The contributor markdown runbook (docs/performance.md) gets only the two new bench names, per the docs split (user guidance lives in Phlex). - CHANGELOG: the full feature entry under Unreleased/Added. * refactor(streams): autoscaler as periodic maintenance on the LISTEN connection (#323) Redesign from PR review feedback: no extra connection, no extra thread, a pghero-style periodic check instead of a 1s control loop. Before: a per-process autoscaler thread with its OWN dedicated probe connection, polling every 1s with sustain/cooldown hysteresis. After: - PoolAutoscaler is a PURE DECISION OBJECT — evaluate(headroom) reads streams_pool_stats, decides emergency/grow/shrink/hold, resizes. No thread, no connection, no cooldown, no sustain counters. - The headroom query runs as a throttled MAINTENANCE task on the streamer's EXISTING idle LISTEN connection (Listener gains `maintenance:` + `clock:`, runs it in its health-check window on the listener thread — the only thread that may touch that non-thread-safe connection). ZERO extra connections, zero extra threads. - Cadence: streams_pool_autoscale_interval 1.0s -> 300s (5 min). The slow interval is itself the debounce, so per-sample hysteresis and the 15s cooldown are gone: one grow/shrink step per check; a sustained burst converges over a few checks. This also DISSOLVES the probe-starvation problem #327 had to work around: the query runs on the LISTEN connection (not the streams pool) and evaluate reads pool stats without a checkout, so a 100%-saturated pool no longer blocks the decision — no dedicated probe connection needed. Client#with_streams_connection reverts to private; build_streams_probe_connection is deleted. The fair-share + emergency-shrink SAFETY math and the P1/P2 prerequisites are unchanged. Pure-publisher workers (no streamer) aren't autoscaled — a publisher-triggered check is a planned follow-up; documented, mitigate with a larger static streams_pool_size. Bench: check cost ~0.34ms; grow 3->7->10 (capped). Unit 3091, streamer 255, docs 57, DB-gated 4 — all green. * fix: address PR review feedback (#327) - emergency_shrink: when the DB is critically low on connections but the shrink resize is a no-op (unchanged/shared), log a warning and return :hold instead of silently reporting :emergency_shrink — a failed emergency shrink is the one case where visibility matters most. Adds a covering unit test. - pool_autoscale_bench: capture streams_pool_stats[:size] once per loop iteration instead of querying it twice.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
A spike that de-risks elastic pools (#323) by proving the hard part: the dedicated streams DB pool can be hot-swapped to a new size under live load without losing a broadcast or leaking a connection. This is the feasibility gate — the grow/shrink control loop is deliberately out of scope and comes later, only because this spike shows the swap is cheap and correct.
Why now: the #324 benchmark showed async workers collapse ~3.4× under a DB-bound burst on a small streams pool (silent serialization, no timeouts). Elastic sizing fixes that — if the pool can be resized safely.
connection_pool2.x can't resize in place, so growing means building a newPGMQ::Client+ atomically swapping the reference + retiring the old pool. This ships that primitive, opt-in and default-off (no automatic caller).The correctness spine
Concurrent::AtomicReference(the dedicated path has no mutex, so a bare ivar swap is a data race). Swaps + close serialize on a mutex; reads stay lock-free.available == size— the load-bearing fix the design review caught. pgmq'swith_connectionretries a lost connection (checkout → checkin → checkout), soavailablemomentarily equalssizewhile a produce is mid-flight; closing there raisesPoolShuttingDownErroron the retry = a genuinely lost durable broadcast (the PgBouncer/failover mid-INSERT case).ResizablePoolinstead brackets each reader op with anAtomicFixnumand drains on that reaching zero — immune to the retry, and independent of the connection_pool version (which also answers the "what about connection_pool 3+?" question — we never touch its internals).inflight == 0first, so a pool is never shut under a live produce.PGMQ::Client#close(pgmq's own safe shutdown drain: idle conns close now, in-flight self-close on checkin) — neverThread#kill, bounded log-and-abandon like OutboundPump Streams: offload dispatcher fanout socket writes off the single dispatcher thread (follow-up to #315 item 3) #321 B3.pool_size 1).Measured (real DB, live load: 4 producers + 1 reader, grow 5→12, shrink 12→5)
Bench:
benchmarks/pool_swap_bench.rb(rake bench:pool_swap, requiresPGBUS_DATABASE_URL).Test plan
bundle exec rspec spec/pgbus— 3058 examples, 0 failuresresizable_pool_spec(13 unit, no DB) — incl. the data-loss regression test (drain waits on the in-flight counter, notavailable == size), drain-timeout log-and-abandon, concurrent-swap serialization, alias guardclient_swap_spec(8, mockedPGMQ::Client) — swap routing, no-ops (unchanged size / shared-AR), invalid size, concurrent reader safety, close-after-swapstreams_pool_swap_spec(DB-gated) — zero lost broadcasts under concurrent grow, zero leaked connections across grow+shrinkclient_spec(301) + streamer (307) pass unchanged — swap-off is byte-identical (one atomic read + counter bump per op)bundle exec rubocopclean (7 files)Deferred (out of scope for the spike)
The control loop: saturation sampling, hysteresis, grow/shrink policy, fleet-wide connection budget, and any automatic
resize_streams_poolcall. This spike decides the swap is safe; the loop is the follow-up, now on solid ground.Deviations & judgment calls
connection_pool'savailable == sizeis wrong because pgmq retries mid-op, so the predicate reads "drained" while a produce is in flight → a raced close loses the broadcast. Fixed by the in-flight-counter drain. This is exactly why the spike designed before coding.~> 2.4), and it's moot — the swap operates on wholePGMQ::Clientobjects via pgmq's ownclose/with_connection/stats, neverconnection_poolinternals. The in-flight-counter drain is version-independent as a direct result.PoolShuttingDownErrorwrapped (doesn't retry), so loss-safety rests on the drain-before-close ordering rather than a pgbus-level retry — the counter guarantees close can't happen under a live produce, which is stronger.ResizablePoolclass (160 lines) rather than inline —client.rbwas already 1361 lines (over the 800 cap), so the coding-style "extract complex logic" rule made this mandatory, not optional.resize_streams_pool, drain-timeoutstreams_pool_timeout + 1.0shardcoded for the spike.application_nameand counting only those. Real no-leak evidence:conns_closed = 0(drained to idle before close) + pool shrank back to 5.Refs #323.