From c8d5a8a20f284dd01485eff3949ca9af895af626 Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Sat, 27 Jun 2026 11:09:57 +0200 Subject: [PATCH] web: fix pubsub thread budget to not block entire app (#828) * web: fix pubsub thread budget to not block entire app * web: address review comments --- ACCEPTANCE.md | 59 ++++++++ SPEC.md | 1 + web/lib/potato_mesh/application.rb | 18 ++- web/lib/potato_mesh/application/pubsub.rb | 33 ++++- web/lib/potato_mesh/config.rb | 50 +++++++ web/spec/app_spec.rb | 10 ++ web/spec/config_spec.rb | 101 +++++++++++++ web/spec/pubsub_spec.rb | 43 ++++++ web/spec/sse_thread_budget_spec.rb | 172 ++++++++++++++++++++++ 9 files changed, 475 insertions(+), 12 deletions(-) create mode 100644 web/spec/sse_thread_budget_spec.rb diff --git a/ACCEPTANCE.md b/ACCEPTANCE.md index cf8f2a4..324c013 100644 --- a/ACCEPTANCE.md +++ b/ACCEPTANCE.md @@ -1894,3 +1894,62 @@ a subscriber-closed exit), **PS-A4 / LV-A6** (publish + 1s settle window are unchanged), **FH-A3** (federation reaps in seconds -- now actually reachable on Ctrl+C because the SSE no longer blocks Puma), and **B1** (all suites). No POST/GET/event contract change. + +## Bugfix: SSE streams must not starve the request-thread pool + +The live production instance went unresponsive: every request 502'd, including the +instance's own federation self-fetch of `/api/nodes`, and at shutdown exactly five +`/api/events` connections closed (durations 45-160s). Root cause: a `GET +/api/events` SSE stream pins one Puma worker thread for its whole lifetime (the +`pump` loop runs synchronously on the request thread; SD-A1), but the subscriber +cap (`MAX_SUBSCRIBERS` = 64) sat far above Puma's pool. With no thread config the +app ran on Puma's MRI default of **5** threads, so ~5 dashboard clients holding an +`EventSource` occupied every worker thread and no other request -- API read, +ingest POST, or federation self-fetch -- could be served. The cap never tripped +before the pool starved; live updates became load-bearing, violating **PS8**. +Pre-existing since the SSE pub/sub feature (#821). Fix (web-only): (1) size Puma's +thread pool in code via `server_settings[:Threads]` (`Config.puma_threads_setting`, +default `16:96`, env `MIN_THREADS`/`MAX_THREADS`); (2) clamp the SSE subscriber cap +to `puma_max_threads - sse_thread_reserve` (env `SSE_THREAD_RESERVE`, default 32) so +at least the reserve always remains for non-SSE traffic -- the defaults reconcile to +the original 64 (`96 - 32`). New decision **PS9** names the budget invariant +(`max_threads > MAX_SUBSCRIBERS + reserve`). The apex (I), privacy (II), and parity +(IV) invariants are untouched; no POST/GET/event contract changes. + +### TS-A1 -- SSE can never consume the whole request-thread pool +```bash +( cd web && bundle exec rspec spec/sse_thread_budget_spec.rb ) +``` +**Expected:** pass. Boots a real Puma with a small fixed pool (`Threads "6:6"`, +`SSE_THREAD_RESERVE=4`) and opens `pool`-many `/api/events` connections: at most +`pool - reserve` are accepted (the rest get `503` and fall back to the safety poll, +PS8), and a plain `GET /version` is still served promptly while SSE clients are +connected. Against the unfixed code all six connections are accepted and the +ordinary request times out (the outage). + +### TS-A2 -- thread budget exceeds the SSE subscriber cap by the reserve +```bash +( cd web && bundle exec rspec spec/config_spec.rb -e "puma thread budget" ) +( cd web && bundle exec rspec spec/pubsub_spec.rb -e "effective subscriber cap" ) +( cd web && bundle exec rspec spec/app_spec.rb -e "request-thread budget" ) +``` +**Expected:** pass. `Config.puma_max_threads` (default 96, env `MAX_THREADS`), +`Config.puma_min_threads` (default 16, env `MIN_THREADS`), and +`Config.sse_thread_reserve` (default 32, env `SSE_THREAD_RESERVE`) resolve and +clamp sanely (`min <= max`); `Config.puma_threads_setting` returns `"min:max"`; +`PubSub.effective_max_subscribers` equals `min(MAX_SUBSCRIBERS, max_threads - +reserve)` (= 64 at defaults) and shrinks when the pool shrinks; and the application +`server_settings[:Threads]` is present with `max > MAX_SUBSCRIBERS` (the invariant +that was silently false before, when no `:Threads` was set at all). + +### TS-R1 -- Regression: prior acceptance still holds +```bash +( cd web && bundle exec rspec ) && ( cd web && npm test ) +( . .venv/bin/activate && pytest -q tests/ ) +``` +**Expected:** all green. At risk and required to remain green: **PS-A2 / PS-A5** +(the `/api/events` SSE stream + reconnect-resync still work), **PS-A3** (the +subscriber cap still returns `503` at capacity -- now at the clamped value), +**SD-A1 / SD-A2** (shutdown still reaps SSE; `server_settings` still carries +`force_shutdown_after` alongside the new `Threads`), and **B1** (all suites). No +POST/GET/event contract change. diff --git a/SPEC.md b/SPEC.md index f8dddaa..efeb536 100644 --- a/SPEC.md +++ b/SPEC.md @@ -436,6 +436,7 @@ untouched. | **PS6** | **Privacy: no `messages` events when PRIVATE (Invariant II).** Under `PRIVATE=1` the server emits **no `messages` change events** and the `GET /api/events` stream carries none, mirroring the `/api/messages` 404 (A2a). Because events are thin (no row data) and the client re-fetches through the already-filtered `/api` endpoints, opt-out / `CLIENT_HIDDEN` rows never traverse the push. Defense-in-depth; Invariant II is preserved. | interview + Invariant II | | **PS7** | **Amends FC2/FC-A2/FC-R1 cadence wording (named, not silent).** The seed-then-delta cache *mechanism* is unchanged (still `since=`, only-misses-fetch, merge-by-id). Only the **trigger** changes: from a fixed 60 s cadence to event-driven (SSE ping / reconnect resync / slow safety poll). FC-A2 and FC-R1's "auto-refresh cadence is unchanged" clause is **explicitly amended** to "the fetch trigger is event-driven with a slow safety-poll fallback; the delta/merge/cache contract is unchanged." | interview (FC2 amendment) | | **PS8** | **Graceful degradation & engineering bar (D9).** If `EventSource` is unavailable, the stream errors, or a config flag disables it, the client silently falls back to the safety poll (today's network-only behavior) — the push is **never load-bearing**. All new code (the pub/sub registry, the `GET /api/events` route, the frontend SSE client) ships with 100% unit tests, full API docs (RDoc/JSDoc), the exact Apache header, and `rufo`-clean formatting; all existing suites stay green. | D9 + proposed | +| **PS9** | **Request-thread budget reserves capacity for non-SSE traffic (realizes PS8 server-side).** Each open `GET /api/events` stream pins **one Puma request thread** for its lifetime (the `pump` runs synchronously on it). So the Puma pool **must** stay larger than the SSE subscriber cap by a reserve: `puma_max_threads ≥ MAX_SUBSCRIBERS + sse_thread_reserve`, so that **at least `sse_thread_reserve` threads** (32 by default) always remain for API reads, ingest `POST`s, and the federation self-fetch even when every SSE slot is occupied. Enforced two ways: (a) the pool is sized **in code** via `server_settings[:Threads]` (default `16:96`, env `MIN_THREADS`/`MAX_THREADS`) — never left to Puma's MRI default of 5; and (b) `PubSub.subscribe`'s effective cap is **clamped** to `min(MAX_SUBSCRIBERS, puma_max_threads − sse_thread_reserve)` (env `SSE_THREAD_RESERVE`, default 32; defaults reconcile to the original 64), so a subscription flood gets `503` → safety poll (PS8) instead of starving the server. Without this a handful of dashboard clients took the whole instance down (502 everywhere). | bugfix (production incident) | --- diff --git a/web/lib/potato_mesh/application.rb b/web/lib/potato_mesh/application.rb index a931db8..aab0ff4 100644 --- a/web/lib/potato_mesh/application.rb +++ b/web/lib/potato_mesh/application.rb @@ -198,11 +198,19 @@ module PotatoMesh set :retention_shutdown_hook_installed, false set :port, resolve_port set :bind, DEFAULT_BIND_ADDRESS - # Bound Puma's graceful-shutdown wait so a long-lived /api/events SSE - # stream cannot block Ctrl+C; the signal handler closes subscribers for a - # graceful exit, and this force-terminates anything still in flight after - # the window. - set :server_settings, { force_shutdown_after: PotatoMesh::Config.puma_force_shutdown_seconds } + # Size the Puma worker-thread pool in code (default 16:96, env + # MIN_THREADS/MAX_THREADS). Each /api/events SSE stream pins one request + # thread for its lifetime, so the pool is kept well above the SSE + # subscriber cap (PubSub::MAX_SUBSCRIBERS) plus a reserve, ensuring live + # updates can never starve API/ingest/federation traffic (SPEC PS9); + # Puma's MRI default of 5 is far too small for that. force_shutdown_after + # bounds Puma's graceful-shutdown wait so a long-lived SSE stream cannot + # block Ctrl+C (the signal handler closes subscribers; this backstops + # anything still in flight after the window). + set :server_settings, { + force_shutdown_after: PotatoMesh::Config.puma_force_shutdown_seconds, + Threads: PotatoMesh::Config.puma_threads_setting, + } app_logger = PotatoMesh::Logging.build_logger($stdout) set :logger, app_logger diff --git a/web/lib/potato_mesh/application/pubsub.rb b/web/lib/potato_mesh/application/pubsub.rb index 45857f9..26b45ab 100644 --- a/web/lib/potato_mesh/application/pubsub.rb +++ b/web/lib/potato_mesh/application/pubsub.rb @@ -46,10 +46,13 @@ module PotatoMesh # {publish} so a future caller typo can never crash ingest. COLLECTIONS = %w[nodes messages positions telemetry neighbors traces].freeze - # Hard cap on concurrently registered subscribers. Each open SSE stream - # occupies one subscriber (and, under the threaded server, one request - # thread), so this bounds resource use against a subscription flood - # (+PS1+ / +PS8+). The route maps {CapacityError} to an HTTP 503. + # Upper bound on concurrently registered subscribers. Each open SSE stream + # occupies one subscriber **and, under the threaded server, one request + # thread for its whole lifetime**, so the effective ceiling is clamped to + # the request-thread budget by {effective_max_subscribers}: a flood of + # +/api/events+ connections can never consume the threads reserved for + # non-SSE traffic (+PS8+ / +PS9+). The route maps {CapacityError} to an + # HTTP 503, after which the client falls back to its safety poll. MAX_SUBSCRIBERS = 64 # Raised by {PubSub.subscribe} when {MAX_SUBSCRIBERS} is already reached. @@ -167,11 +170,12 @@ module PotatoMesh # Register a new subscriber and return its mailbox. # # @return [Subscriber] the freshly registered subscriber. - # @raise [CapacityError] when {MAX_SUBSCRIBERS} is already reached. + # @raise [CapacityError] when {effective_max_subscribers} is reached. def subscribe @mutex.synchronize do - if @subscribers.size >= MAX_SUBSCRIBERS - raise CapacityError, "subscriber limit (#{MAX_SUBSCRIBERS}) reached" + cap = effective_max_subscribers + if @subscribers.size >= cap + raise CapacityError, "subscriber limit (#{cap}) reached" end subscriber = Subscriber.new @@ -180,6 +184,21 @@ module PotatoMesh end end + # Effective concurrent-subscriber ceiling, clamped so SSE never consumes + # the request threads reserved for non-SSE traffic. Equal to + # +min(MAX_SUBSCRIBERS, Config.puma_max_threads - Config.sse_thread_reserve)+, + # which reconciles to {MAX_SUBSCRIBERS} at the default pool size + # (96 - 32 = 64) and shrinks automatically when the pool is configured + # smaller (SPEC PS9). Never negative: a pool at or below the reserve + # yields 0, so every {subscribe} is refused (503) and the client falls + # back to its safety poll (+PS8+) rather than the server starving. + # + # @return [Integer] the effective subscriber limit (>= 0). + def effective_max_subscribers + budget = PotatoMesh::Config.puma_max_threads - PotatoMesh::Config.sse_thread_reserve + [MAX_SUBSCRIBERS, [budget, 0].max].min + end + # Remove and close a subscriber. Safe to call more than once. # # @param subscriber [Subscriber] the subscriber to remove. diff --git a/web/lib/potato_mesh/config.rb b/web/lib/potato_mesh/config.rb index 7f68400..5e53c72 100644 --- a/web/lib/potato_mesh/config.rb +++ b/web/lib/potato_mesh/config.rb @@ -48,6 +48,20 @@ module PotatoMesh # instead of blocking on a long-lived /api/events connection. Backstop to # the graceful subscriber-close performed by the shutdown signal handler. DEFAULT_PUMA_FORCE_SHUTDOWN_SECONDS = 3 + + # Default Puma worker-thread pool bounds. The pool must stay larger than the + # SSE subscriber cap ({PotatoMesh::App::PubSub::MAX_SUBSCRIBERS}) by at least + # {DEFAULT_SSE_THREAD_RESERVE}, so a flood of long-lived +/api/events+ + # streams (each pins one request thread for its lifetime) can never occupy + # every worker thread and starve API/ingest/federation traffic (SPEC + # PS8/PS9). Puma's own MRI default is only 5, which is why the floor is + # raised here in code rather than left to the deployment environment. + DEFAULT_PUMA_MIN_THREADS = 16 + DEFAULT_PUMA_MAX_THREADS = 96 + # Request threads kept in reserve for non-SSE traffic even when every SSE + # subscriber slot is occupied. With the defaults this reconciles the SSE cap + # back to its historical value (96 - 32 = 64). + DEFAULT_SSE_THREAD_RESERVE = 32 DEFAULT_TILE_FILTER_LIGHT = "grayscale(1) saturate(0) brightness(0.92) contrast(1.05)" DEFAULT_TILE_FILTER_DARK = "grayscale(1) invert(1) brightness(0.9) contrast(1.08)" DEFAULT_MAP_CENTER_LAT = 38.761944 @@ -372,6 +386,42 @@ module PotatoMesh fetch_positive_integer("PUMA_FORCE_SHUTDOWN", DEFAULT_PUMA_FORCE_SHUTDOWN_SECONDS) end + # Minimum Puma worker threads kept warm. + # + # @return [Integer] positive minimum, overridable via +MIN_THREADS+. + def puma_min_threads + fetch_positive_integer("MIN_THREADS", DEFAULT_PUMA_MIN_THREADS) + end + + # Maximum Puma worker threads. Sized above the SSE subscriber cap plus + # {#sse_thread_reserve} so long-lived +/api/events+ streams cannot occupy + # every request thread and starve other traffic (SPEC PS9). + # + # @return [Integer] positive maximum, overridable via +MAX_THREADS+. + def puma_max_threads + fetch_positive_integer("MAX_THREADS", DEFAULT_PUMA_MAX_THREADS) + end + + # Request threads reserved for non-SSE traffic even when every SSE + # subscriber slot is taken, so live updates are never load-bearing + # (SPEC PS8/PS9). + # + # @return [Integer] positive reserve, overridable via +SSE_THREAD_RESERVE+. + def sse_thread_reserve + fetch_positive_integer("SSE_THREAD_RESERVE", DEFAULT_SSE_THREAD_RESERVE) + end + + # Build Puma's +Threads+ "min:max" pool spec. The minimum is clamped to the + # maximum so an out-of-order override (+MIN_THREADS+ > +MAX_THREADS+) cannot + # break the boot. + # + # @return [String] the +"min:max"+ thread-pool spec passed to Puma. + def puma_threads_setting + max = puma_max_threads + min = [puma_min_threads, max].min + "#{min}:#{max}" + end + # Retrieve the CSS filter used for light themed maps. # # @return [String] CSS filter string. diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb index 7d0ae43..9086881 100644 --- a/web/spec/app_spec.rb +++ b/web/spec/app_spec.rb @@ -8197,6 +8197,16 @@ RSpec.describe "Potato Mesh Sinatra app" do end end + describe "request-thread budget" do + it "sizes the Puma thread pool above the SSE subscriber cap (PS9)" do + threads = PotatoMesh::Application.settings.server_settings[:Threads] + expect(threads).to eq(PotatoMesh::Config.puma_threads_setting) + _min, max = threads.split(":").map(&:to_i) + expect(max).to be > PotatoMesh::App::PubSub::MAX_SUBSCRIBERS + expect(max - PotatoMesh::App::PubSub::MAX_SUBSCRIBERS).to be >= PotatoMesh::Config.sse_thread_reserve + end + end + describe "live-update shutdown handling" do before { PotatoMesh::App::PubSub.reset! } diff --git a/web/spec/config_spec.rb b/web/spec/config_spec.rb index 3f116b9..bd720c4 100644 --- a/web/spec/config_spec.rb +++ b/web/spec/config_spec.rb @@ -552,6 +552,107 @@ RSpec.describe PotatoMesh::Config do end end + describe "puma thread budget" do + describe ".puma_min_threads" do + it "returns the baked-in default when unset" do + within_env("MIN_THREADS" => nil) do + expect(described_class.puma_min_threads).to eq( + PotatoMesh::Config::DEFAULT_PUMA_MIN_THREADS, + ) + end + end + + it "accepts a positive override" do + within_env("MIN_THREADS" => "8") do + expect(described_class.puma_min_threads).to eq(8) + end + end + + it "falls back to the default for non-positive or invalid values" do + ["0", "-3", "abc", ""].each do |raw| + within_env("MIN_THREADS" => raw) do + expect(described_class.puma_min_threads).to eq( + PotatoMesh::Config::DEFAULT_PUMA_MIN_THREADS, + ) + end + end + end + end + + describe ".puma_max_threads" do + it "returns the baked-in default when unset" do + within_env("MAX_THREADS" => nil) do + expect(described_class.puma_max_threads).to eq( + PotatoMesh::Config::DEFAULT_PUMA_MAX_THREADS, + ) + end + end + + it "accepts a positive override" do + within_env("MAX_THREADS" => "120") do + expect(described_class.puma_max_threads).to eq(120) + end + end + + it "falls back to the default for non-positive or invalid values" do + ["0", "-1", "nope", ""].each do |raw| + within_env("MAX_THREADS" => raw) do + expect(described_class.puma_max_threads).to eq( + PotatoMesh::Config::DEFAULT_PUMA_MAX_THREADS, + ) + end + end + end + end + + describe ".sse_thread_reserve" do + it "returns the baked-in default when unset" do + within_env("SSE_THREAD_RESERVE" => nil) do + expect(described_class.sse_thread_reserve).to eq( + PotatoMesh::Config::DEFAULT_SSE_THREAD_RESERVE, + ) + end + end + + it "accepts a positive override" do + within_env("SSE_THREAD_RESERVE" => "12") do + expect(described_class.sse_thread_reserve).to eq(12) + end + end + + it "falls back to the default for non-positive or invalid values" do + ["0", "-2", "xyz", ""].each do |raw| + within_env("SSE_THREAD_RESERVE" => raw) do + expect(described_class.sse_thread_reserve).to eq( + PotatoMesh::Config::DEFAULT_SSE_THREAD_RESERVE, + ) + end + end + end + end + + describe ".puma_threads_setting" do + it "formats the configured min and max as a Puma min:max spec" do + within_env("MIN_THREADS" => "16", "MAX_THREADS" => "96") do + expect(described_class.puma_threads_setting).to eq("16:96") + end + end + + it "defaults to a pool larger than the SSE subscriber cap (PS9)" do + within_env("MIN_THREADS" => nil, "MAX_THREADS" => nil, "SSE_THREAD_RESERVE" => nil) do + _min, max = described_class.puma_threads_setting.split(":").map(&:to_i) + expect(max).to be > PotatoMesh::App::PubSub::MAX_SUBSCRIBERS + end + end + + it "clamps the minimum to the maximum when misconfigured (min > max)" do + within_env("MIN_THREADS" => "200", "MAX_THREADS" => "32") do + expect(described_class.puma_threads_setting).to eq("32:32") + end + end + end + end + describe ".sse_publish_cooldown_seconds" do it "returns the baked-in default when unset" do within_env("SSE_PUBLISH_COOLDOWN" => nil) do diff --git a/web/spec/pubsub_spec.rb b/web/spec/pubsub_spec.rb index 7595915..759a7c9 100644 --- a/web/spec/pubsub_spec.rb +++ b/web/spec/pubsub_spec.rb @@ -43,6 +43,34 @@ RSpec.describe PotatoMesh::App::PubSub do end end + describe ".effective_max_subscribers" do + it "equals MAX_SUBSCRIBERS at the default pool size (96 - 32 = 64)" do + within_env("MAX_THREADS" => nil, "SSE_THREAD_RESERVE" => nil) do + expect(described_class.effective_max_subscribers).to eq(described_class::MAX_SUBSCRIBERS) + end + end + + it "shrinks to the thread budget when the pool is smaller than the cap" do + within_env("MAX_THREADS" => "20", "SSE_THREAD_RESERVE" => "4") do + expect(described_class.effective_max_subscribers).to eq(16) + end + end + + it "never goes negative when the pool is at or below the reserve" do + within_env("MAX_THREADS" => "4", "SSE_THREAD_RESERVE" => "8") do + expect(described_class.effective_max_subscribers).to eq(0) + end + end + + it "caps subscribe at the clamped limit, then raises naming that limit" do + within_env("MAX_THREADS" => "10", "SSE_THREAD_RESERVE" => "8") do + expect(described_class.effective_max_subscribers).to eq(2) + 2.times { described_class.subscribe } + expect { described_class.subscribe }.to raise_error(described_class::CapacityError, /subscriber limit \(2\)/) + end + end + end + describe ".unsubscribe" do it "removes and closes the subscriber" do subscriber = described_class.subscribe @@ -348,4 +376,19 @@ RSpec.describe PotatoMesh::App::PubSub do expect(subscriber.drain(timeout: 0.1)).to eq([{ collection: "neighbors", hint: nil }]) end end + + # Temporarily set/clear ENV keys for the duration of the block, restoring the + # prior values (including "previously unset") afterwards. + def within_env(values) + original = values.transform_values { |_| :__unset__ } + values.each do |key, value| + original[key] = ENV.key?(key) ? ENV[key] : :__unset__ + value.nil? ? ENV.delete(key) : ENV[key] = value + end + yield + ensure + original.each do |key, value| + value == :__unset__ ? ENV.delete(key) : ENV[key] = value + end + end end diff --git a/web/spec/sse_thread_budget_spec.rb b/web/spec/sse_thread_budget_spec.rb new file mode 100644 index 0000000..cb2069f --- /dev/null +++ b/web/spec/sse_thread_budget_spec.rb @@ -0,0 +1,172 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# frozen_string_literal: true + +require "spec_helper" +require "socket" +require "net/http" +require "timeout" +require "rack/handler/puma" + +# Integration regression guard for the SSE thread-pool starvation outage. +# +# A +GET /api/events+ stream pins one Puma request thread for its whole lifetime +# (the +pump+ loop runs synchronously on that thread). Before the fix the +# subscriber cap (+MAX_SUBSCRIBERS+ = 64) sat far above Puma's pool (MRI default +# 5), so a handful of dashboard clients could occupy every worker thread and the +# instance returned nothing (502 to every other request, including its own +# federation self-fetch). This boots a real Puma with a small, fixed pool and +# proves SSE can never consume the whole pool — at least +sse_thread_reserve+ +# threads always remain for non-SSE traffic (SPEC PS8/PS9). +RSpec.describe "SSE request-thread budget" do + # A small pool keeps the test fast; the reserve is set so the clamped SSE cap + # (max_threads - reserve) is a small positive number we can saturate quickly. + POOL = 6 + RESERVE = 4 + HOST = "127.0.0.1" + + # Read from +sock+ until +needle+ (String or Regexp) appears or +timeout+ + # elapses, returning whatever was read. Uses IO.select so a stalled socket + # cannot hang the suite. + def read_until(sock, needle, timeout) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + buf = +"" + until (needle.is_a?(Regexp) ? buf.match?(needle) : buf.include?(needle)) + remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC) + break if remaining <= 0 + break unless IO.select([sock], nil, nil, remaining) + begin + chunk = sock.read_nonblock(512) + rescue IO::WaitReadable + next + rescue EOFError, Errno::ECONNRESET + break + end + buf << chunk + end + buf + end + + # Issue a plain (non-SSE) request and return its status code as a string, or + # "ERROR:" when it cannot be served (the starvation symptom). + def status_of(path, timeout:) + http = Net::HTTP.new(HOST, @port) + http.open_timeout = timeout + http.read_timeout = timeout + http.max_retries = 0 + http.start + http.get(path).code + rescue => e + "ERROR:#{e.class}" + ensure + http&.finish if http&.started? + end + + # Env keys this spec mutates; saved and restored verbatim around the run. + ENV_KEYS = %w[MIN_THREADS MAX_THREADS SSE_THREAD_RESERVE EVENTS PRIVATE SSE_HEARTBEAT_SECONDS].freeze + + before(:all) do + @saved_env = ENV_KEYS.to_h { |k| [k, ENV[k]] } + ENV["MIN_THREADS"] = POOL.to_s + ENV["MAX_THREADS"] = POOL.to_s + ENV["SSE_THREAD_RESERVE"] = RESERVE.to_s + ENV["EVENTS"] = "1" + ENV.delete("PRIVATE") + ENV["SSE_HEARTBEAT_SECONDS"] = "1" # snappy detection + teardown + + server = TCPServer.new(HOST, 0) + @port = server.addr[1] + server.close + + @launcher = nil + handler = if Object.const_defined?(:Rackup) && defined?(Rackup::Handler::Puma) + Rackup::Handler::Puma + else + Rack::Handler::Puma + end + # Size the pool through the real production helper (the same value + # application.rb feeds Puma via server_settings[:Threads]) rather than a + # hardcoded string, so this guard also exercises Config.puma_threads_setting. + pool_spec = PotatoMesh::Config.puma_threads_setting + @server_thread = Thread.new do + handler.run(PotatoMesh::Application, + Host: HOST, Port: @port, + Threads: pool_spec, Silent: true) do |l| + @launcher = l + end + end + + Timeout.timeout(20) do + loop do + begin + TCPSocket.new(HOST, @port).close + break + rescue Errno::ECONNREFUSED + sleep 0.05 + end + end + end + # Let the launcher finish wiring its thread pool before probing. + sleep 0.3 + end + + after(:all) do + @launcher&.stop + @server_thread&.join(5) + @saved_env.each { |k, v| v.nil? ? ENV.delete(k) : ENV[k] = v } + end + + # Closing subscribers server-side unwinds every parked pump (the #827 + # subscriber-closed exit), freeing the worker threads between examples. + after do + PotatoMesh::App::PubSub.reset! + sleep 0.2 + end + + it "never lets SSE subscribers consume the whole request-thread pool" do + held = [] + accepted = 0 + + POOL.times do + sock = TCPSocket.new(HOST, @port) + sock.write("GET /api/events HTTP/1.1\r\nHost: #{HOST}\r\nConnection: keep-alive\r\n\r\n") + sock.flush + resp = read_until(sock, /\r\n\r\n/, 3) + if resp.start_with?("HTTP/1.1 200") + # Confirm the stream is committed: the worker thread is now parked in + # Events.pump and will not service any other request. The ": connected" + # comment is usually flushed with the headers; read on only if not. + resp = read_until(sock, ": connected", 3) unless resp.include?(": connected") + accepted += 1 if resp.include?(": connected") + held << sock + else + # 503 capacity: the client falls back to its safety poll (PS8). Good — + # the server refused rather than cannibalising a reserved thread. + sock.close + end + end + + # The clamp keeps concurrent SSE strictly below the pool, preserving a + # reserve for non-SSE traffic. (Unfixed: all POOL connections are accepted.) + expect(accepted).to be <= (POOL - RESERVE) + + # With the reserve intact, an ordinary request is still served promptly — + # the instance is responsive even while live-update clients are connected. + # (Unfixed: every thread is held by SSE, so this times out.) + expect(status_of("/version", timeout: 4)).to eq("200") + ensure + held.each { |s| s.close rescue nil } + end +end