feat(bench): execution-mode connection benchmark + DB-pool sizing guidance (#323)

CHANGELOG.md

@@ -7,6 +7,7 @@
 
 ### Added
 
+- **Execution-mode connection benchmark + DB-pool sizing guidance (`rake bench:execution_modes`).** A fair, reusable harness (`benchmarks/support/execution_mode_harness.rb` + `benchmarks/execution_modes_bench.rb`) that runs the *same* offered load through the threads and async execution pools and measures how many real Postgres connections each actually holds (`peak_busy = pool size − available`, sampled through the real pool checkout), so operators can size the DB pool per `execution_mode`. It settles the "async saves connections" question with data: a job holds a pgmq connection only for the `read_batch` + `archive` round-trip (not the job body), so **async's connection-density win is real only for I/O-light work** (a few connections serve many fibers) — for **DB-bound work async is connection-bound like threads**, and a too-small pool *serialises* (throughput collapse, ~3.4× in the measured cell) rather than erroring, because `pool_timeout` (5 s) dwarfs a typical checkout. New `docs/performance.md` section "Execution mode and DB pool sizing" carries the measured numbers and a sizing table. The harness's fairness/struct/measurement logic is unit-tested with fakes (no DB, runs in CI); the numbers come from a manual `PGBUS_DATABASE_URL` run (run-and-report, never a CI gate). Groundwork for the elastic-pools research in #323; a `pgbus doctor` mode/pool-mismatch hint is deferred to a follow-up. Refs #323.
 - **Off-thread durable-stream fanout writer (`config.streams_writer_threads`, `config.streams_writer_buffer_limit`) — the full head-of-line cure the #315 deadline only bounded.** The #315 deadline caps each slow-client fanout stall at `streams_fanout_write_deadline_ms` (250ms), but the single dispatcher thread still writes to every client serially, so K slow clients still cost ~`K × 250ms`. Setting `streams_writer_threads > 0` moves the **durable** fanout socket write off the dispatcher into N writer threads (each connection pinned to one worker by `id.hash % N`, preserving per-connection frame order and its per-io mutex). The dispatcher hands off the write and moves on; the pump reports back the highest msg_id it actually committed via an ack queue, and the dispatcher — still the **sole owner** of the read cursor — advances it only on that ack, so a blocked or partial write never advances the cursor past a frame that didn't reach the socket. A failed write posts a `DisconnectMessage` so the dispatcher scrubs the connection deterministically on its own thread (even on an otherwise-quiet stream). Measured (`streams_writer_offload_bench.rb`, dispatcher `handle_durable_wake` wall time, 50 fast + K slow clients @ 50ms): inline scales linearly (K=20 → **1071 ms**) while offload stays flat (K=20 → **0.09 ms**) — fast clients no longer wait behind slow ones. This is a **system-level latency win, not a per-write speedup** (total bytes written are unchanged); the dispatcher does slightly more per-wake bookkeeping under offload, which is why it is **opt-in and default-off** (`streams_writer_threads = 0` = byte-for-byte the pre-#321 inline behavior). **Ephemeral frames always stay inline** regardless of this setting — they have no archive to replay, so they must not risk an async drop (the pump raises if one is ever routed to it; issue #321 B1). `streams_writer_buffer_limit` (default `0` = unbounded) caps a connection's outbound buffer, dropping its oldest durable frame on overflow — safe because durable frames are re-read from the archive on reconnect; it's an OOM guard for a pathologically slow-but-alive client, not a delivery guarantee. Both knobs are dispatcher-internal tuning (not part of the 1.0 public-config surface). Refs #321 (follow-up to #315).
 - **Bounded dispatcher fanout head-of-line blocking (`config.streams_fanout_write_deadline_ms`, `config.streams_dispatch_queue_limit`).** One dispatcher thread per web-server process fans out every broadcast to every connected SSE client **serially**, writing to each socket in turn. A slow client's write blocked that thread up to the full `streams_write_deadline_ms` (5s) before the client was marked dead — and every connection queued behind it in the same wake waited, so K slow clients could stack a `K × 5s` stall on all other browsers on the worker. Fanout writes now use a separate, shorter `streams_fanout_write_deadline_ms` (default **250 ms**), bounding the worst-case serial stall to `K × 250 ms` (~20× lower); a slow-but-alive client that misses the window is marked dead and reconnects, replaying the gap from the durable archive via its `Last-Event-ID` (or a fresh re-render for ephemeral streams, which have no archive). **Connect-replay** writes keep the full `streams_write_deadline_ms` so a new client catching up a large backlog isn't evicted. The happy path (all-fast clients) is unchanged — the `deadline_ms` threading is a zero-allocation keyword pass-through (bench: 585k vs 583k i/s, within noise; 0 retained/op). `streams_dispatch_queue_limit` (default `0` = unbounded) optionally caps distinct-stream wake backlog: at the cap the Listener drops durable wakes (never ephemeral or connect/disconnect), safe because the next durable wake for a stream re-reads from the min cursor. Both knobs are dispatcher-internal tuning (not part of the 1.0 public-config surface). A per-connection writer offload — moving the socket write off the dispatcher thread entirely — is deferred to a follow-up (it needs an ack/replay path for ephemeral frames first). Refs #315.
 - **Dedicated streams DB connection pool isolates durable-stream publish + replay from the job pool (`config.streams_pool_size`, `config.streams_pool_timeout`).** Both stream hot paths used to ride on the single job pool sized for worker threads: `Client#send_stream_message` (the broadcast INSERT) drew from `@pgmq`, so under a saturated job pool a broadcast blocked on pool checkout up to `pool_timeout` (~5s) — the browser waited on jobs at the *publish* step; and the dispatcher's per-wake replay read (`read_after` / `stream_current_msg_id` / `stream_oldest_msg_id`) went through `with_raw_connection`, which on the String/Hash config does a **full `PG.connect` + close per call** (TCP + auth + TLS) on the single dispatcher thread inside the fanout loop. Both now use a dedicated `@streams_pgmq` pool (own `PGMQ::Client` → own `connection_pool`), sized independently of worker thread counts via `streams_pool_size` (default 5) / `streams_pool_timeout` (default 5). A new `Client#with_streams_connection` checks out a persistent pooled connection for the replay reads; `Client#streams_pool_stats` exposes its counters alongside `pool_stats`. On the shared-ActiveRecord (Proc) connection path no second pool is created — libpq isn't thread-safe, so `@streams_pgmq` aliases `@pgmq` and streams keep using the single serialized connection. Because every forked process opens its own streams pool on the dedicated path, budget Postgres/PgBouncer `max_connections` for `resolved_pool_size + streams_pool_size` per process (both pools are lazy). Measured (real DB, 50-row stream, `read_after`): pooled replay reads run **~23× faster** than the old fresh-connect-per-call path (5.2k i/s / 191 μs vs 226 i/s / 4.42 ms), removing the per-wake connection-setup cost; this is a client-layer connection-setup win, not end-to-end broadcast-to-browser latency (which is dominated by the PGMQ round-trip + LISTEN/NOTIFY + socket write). Refs #315.

Rakefile

@@ -23,7 +23,8 @@ namespace :bench do
   bench_dir = "benchmarks"
   # Benches that need a real PostgreSQL/PGMQ (or boot Puma) — excluded from the
   # no-DB unit suite that bench:all runs in CI.
-  db_benches = %w[connection_pool_bench integration_bench streams_bench streams_read_pool_bench].freeze
+  db_benches = %w[connection_pool_bench integration_bench streams_bench streams_read_pool_bench
+                  execution_modes_bench].freeze
   # The unit suite is every *_bench.rb that doesn't need a database, derived
   # from the directory so a new unit bench is picked up automatically (kept in
   # sync with bench:one, which globs the same files).
@@ -63,6 +64,11 @@ namespace :bench do
     ruby "benchmarks/streams_bench.rb"
   end
 
+  desc "Run execution-mode connection benchmark (threads vs async pool usage; requires PGBUS_DATABASE_URL)"
+  task :execution_modes do
+    ruby "benchmarks/execution_modes_bench.rb"
+  end
+
   desc "Run a single benchmark: rake bench:one[client_bench]"
   task :one, [:name] do |_t, args|
     name = args[:name] or abort "Usage: rake bench:one[serialization_bench|client_bench|...]"

benchmarks/connection_pool_bench.rb

@@ -4,6 +4,14 @@
 # Measures peak active Postgres connections under varying concurrency
 # for ThreadPool vs AsyncPool execution modes.
 #
+# NOTE: this bench opens a RAW PG.connect per task and offers UNEQUAL load per
+# mode, so its "peak connections" reflects raw connect behavior, not how a
+# shared pool is consumed. For the fair, pool-aware threads-vs-async
+# connection-consumption comparison (equal offered load, real pool checkout,
+# sizing guidance), use `benchmarks/execution_modes_bench.rb`
+# (`rake bench:execution_modes`) — see docs/performance.md "Execution mode and
+# DB pool sizing". This bench is kept for its raw peak-connection matrix.
+#
 # REQUIRES: PGBUS_DATABASE_URL environment variable
 #
 # Usage:

benchmarks/execution_modes_bench.rb

@@ -0,0 +1,163 @@
+# frozen_string_literal: true
+
+# Execution-mode connection benchmark (issue #323).
+#
+# Answers, with fair measurement, "for the same offered concurrency, how many
+# real Postgres connections does threads mode vs async mode actually hold, and
+# what throughput/latency does each deliver?" — so we can guide users on DB-pool
+# sizing per execution_mode.
+#
+# THE HONEST FINDING it exists to prove/disprove: pgbus holds a pgmq connection
+# only for the read/archive SQL round-trip, NOT the job body, so async does not
+# "save connections" by yielding mid-checkout. Async's connection advantage is
+# only the fraction of wall time spent OUTSIDE with_connection (io_light). For
+# DB-bound work async is connection-bound like threads and a tiny pool STARVES.
+# See ExecutionModeHarness for the mechanism.
+#
+# Requires PGBUS_DATABASE_URL. Run-and-report — never a CI gate.
+#   PGBUS_DATABASE_URL=postgres://user@host/db bundle exec rake bench:execution_modes
+
+require "logger"
+require "concurrent"
+require "pgbus"
+require "pg"
+
+require_relative "support/execution_mode_harness"
+
+DATABASE_URL = ENV.fetch("PGBUS_DATABASE_URL") do
+  warn "PGBUS_DATABASE_URL not set. This benchmark requires a real PostgreSQL database."
+  warn "Example: PGBUS_DATABASE_URL=postgres://user@localhost:5432/pgbus_test bundle exec rake bench:execution_modes"
+  exit 1
+end
+
+JOB_COUNT   = Integer(ENV.fetch("EM_BENCH_JOBS", "240"))
+CONCURRENCY = Integer(ENV.fetch("EM_BENCH_CONCURRENCY", "12"))
+MIN_FREE_BACKENDS = 20
+
+IO_LIGHT = ExecutionModeHarness::IoProfile.io_light
+DB_BOUND = ExecutionModeHarness::IoProfile.db_bound
+
+# mode, pool_size, io_profile — concurrency + job_count are fixed so offered
+# load is identical by construction across every row.
+MATRIX = [
+  [:threads, 12, IO_LIGHT], # 1 baseline: ~1 conn per busy thread's checkout window
+  [:async,   12, IO_LIGHT], # 2 async, generously provisioned
+  [:async,    3, IO_LIGHT], # 3 THE claim: can 3 conns serve 12 fibers? (io-light)
+  [:threads,  3, IO_LIGHT], # 4 under-provisioned threads -> expect pool_timeouts
+  [:threads, 12, DB_BOUND], # 5 disproof baseline
+  [:async,   12, DB_BOUND], # 6 disproof: async peak_busy should ~= threads
+  [:async,    3, DB_BOUND], # 7 honesty: does 3 starve when DB-bound?
+  [:async,    6, IO_LIGHT]  # 8 the interesting middle
+].freeze
+
+def build_client(pool_size)
+  config = Pgbus::Configuration.new.tap do |c|
+    c.database_url = DATABASE_URL
+    c.queue_prefix = "pgbus_execmode"
+    c.default_queue = "default"
+    c.logger = Logger.new(IO::NULL)
+    c.pool_size = pool_size
+    c.pool_timeout = 5
+    c.stats_enabled = false
+  end
+  Pgbus::Client.new(config, schema_ensured: true)
+end
+
+def free_backends
+  conn = PG.connect(DATABASE_URL)
+  total = conn.exec("SHOW max_connections").first["max_connections"].to_i
+  used  = conn.exec("SELECT count(*) AS n FROM pg_stat_activity").first["n"].to_i
+  conn.close
+  total - used
+end
+
+def preflight!
+  free = free_backends
+  return if free >= MIN_FREE_BACKENDS
+
+  warn "Only #{free} Postgres backends free (need >= #{MIN_FREE_BACKENDS}). " \
+       "The local box is shared; free some connections and retry."
+  exit 1
+rescue PG::Error => e
+  warn "Preflight connection failed: #{e.message}"
+  exit 1
+end
+
+def run_row(mode, pool_size, io_profile)
+  return skip_async_if_missing(mode) if mode == :async && !async_available?
+
+  client = build_client(pool_size)
+  begin
+    ExecutionModeHarness.run_cell(
+      mode: mode, pool_size: pool_size, concurrency: CONCURRENCY,
+      io_profile: io_profile, job_count: JOB_COUNT, client: client
+    )
+  ensure
+    client.close
+  end
+rescue PG::ConnectionBad, PGMQ::Errors::ConnectionError => e
+  warn "  skipped #{mode}/pool=#{pool_size}/#{io_profile.label}: #{e.message.lines.first&.strip}"
+  nil
+end
+
+def async_available?
+  require "async"
+  true
+rescue LoadError
+  false
+end
+
+def skip_async_if_missing(_mode)
+  warn "  skipped async row: the `async` gem is not installed (add gem \"async\")"
+  nil
+end
+
+def print_table(results)
+  headers = ExecutionModeHarness::Result.headers
+  widths = headers.map.with_index do |h, i|
+    [h.length, *results.map { |r| r.to_row[i].to_s.length }].max
+  end
+  puts headers.each_with_index.map { |h, i| h.ljust(widths[i]) }.join("  ")
+  puts widths.map { |w| "-" * w }.join("  ")
+  results.each do |r|
+    puts r.to_row.each_with_index.map { |v, i| v.to_s.ljust(widths[i]) }.join("  ")
+  end
+end
+
+puts "=" * 78
+puts "Execution-mode connection benchmark (issue #323)"
+puts "Database: #{DATABASE_URL.sub(%r{//[^@]+@}, "//***@")}"
+puts "Offered load per row: #{JOB_COUNT} jobs at concurrency #{CONCURRENCY} (identical for every row)"
+puts "=" * 78
+
+preflight!
+
+results = []
+MATRIX.each do |mode, pool_size, io_profile|
+  print "running #{mode}/pool=#{pool_size}/#{io_profile.label} ... "
+  $stdout.flush
+  result = run_row(mode, pool_size, io_profile)
+  puts result ? "done" : "skipped"
+  results << result if result
+  sleep 1 # let connections drain between cells (shared box)
+end
+
+puts
+print_table(results)
+
+puts
+puts "Reading this table:"
+puts "  peak_busy = max(size - available) = live pool checkouts. This is the"
+puts "  connection cost of the offered load — NOT a throughput number."
+puts "  pool_timeouts > 0 means the pool was too small for the offered concurrency."
+puts
+puts "Honest framing:"
+puts "  * A job holds a pgmq connection only for the read/archive round-trip, not"
+puts "    the job body. Async does NOT save connections by yielding mid-checkout."
+puts "  * io_light (work mostly OUTSIDE the checkout): async can share a tiny pool."
+puts "  * db_bound (work mostly INSIDE the checkout): async is connection-bound"
+puts "    like threads; a tiny pool starves (pool_timeouts)."
+puts "  * async DB concurrency is reactor-bound unless a fiber-aware PG driver is"
+puts "    used — SELECT pg_sleep in with_connection is a blocking libpq call."
+puts
+puts "Done."