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

@@ -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.

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=<high-water>`, 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) |
 
 ---
 

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

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.

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.

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! }
 

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

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

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:<Class>" 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