jinghui/hermes-webui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(#5210): bound HTTP worker dispatch under reconnect storms

Browse Source
This commit is contained in:
Rod Boev authored and nesquena-hermes committed 2026-06-29 18:10:20 +00:00
1 parent 68b211e0d5
commit 55ad113ee7
2 files changed
+410 -133
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server.py
+90 -133
View File
@@ -1,8 +1,4 @@
"""
Hermes Web UI -- Main server entry point.
Thin routing shell: imports Handler, delegates to api/routes.py, runs server.
All business logic lives in api/*.
"""
"""Hermes Web UI server entry point."""
import logging
import os
import re
@@ -15,40 +11,12 @@ import time
import traceback
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
# ── SIGPIPE handling ────────────────────────────────────────────────────────
# Ignore SIGPIPE so a client closing the connection mid-response (browser tab
# close, network drop, mobile backgrounding, a dropped long-poll, an
# `/api/updates/check` timeout, etc.) does not terminate the whole server
# process. Python's default action for SIGPIPE is `Term`, so a single dropped
# `socket.send()` in any request thread could kill the entire WebUI silently —
# no exception, no log, no `/health` response. With SIG_IGN the kernel still
# returns EPIPE to the offending write (surfaced as `BrokenPipeError`); the
# per-request handler unwinds and the connection just closes, while the server
# keeps serving. Set at import time so it is in effect before any
# ThreadingHTTPServer worker thread writes its first response. (Salvaged from
# #3407 @PatrickNoFilter — reproduced in production 2026-06-02.)
#
# SIGPIPE is POSIX-only; it does not exist on Windows (where there is no
# broken-pipe signal and writes to a dead socket raise an OSError directly), so
# guard with getattr to keep native-Windows support (#1952) working.
# Ignore SIGPIPE so a dropped client only aborts that write, not the whole WebUI process.
_SIGPIPE = getattr(signal, "SIGPIPE", None)
if _SIGPIPE is not None:
signal.signal(_SIGPIPE, signal.SIG_IGN)
# ── Test-mode network isolation ─────────────────────────────────────────────
# When `HERMES_WEBUI_TEST_NETWORK_BLOCK=1` is set in the environment, refuse
# outbound socket connections to anything that is not loopback / RFC1918 /
# link-local / reserved-TLD. This catches accidental real outbound (forgotten
# mocks, leaked credentials triggering SDK init, new code paths bypassing an
# existing mock) so the test suite stays hermetic and fast.
#
# tests/conftest.py sets this env var on every test_server subprocess so the
# server.py-side network isolation matches the pytest-process-side isolation
# already installed there.
#
# A test that legitimately needs real outbound spawns the server with the env
# var unset (no current callers — every test_server-using test should be
# mockable).
# Test-mode network isolation keeps subprocess-backed tests hermetic.
if os.environ.get("HERMES_WEBUI_TEST_NETWORK_BLOCK", "").strip() in ("1", "true", "yes"):
_REAL_CREATE_CONN = socket.create_connection
_REAL_SOCK_CONNECT = socket.socket.connect
@@ -56,8 +24,7 @@ if os.environ.get("HERMES_WEBUI_TEST_NETWORK_BLOCK", "").strip() in ("1", "true"
import re as _re
def _re_match_unique_local_ipv6(h):
"""Match IPv6 fc00::/7 (canonical syntax). Tighter than startswith('fc')
so we don't mistakenly classify hostnames like 'food.example.com' as local."""
"""Match IPv6 fc00::/7 without catching similar-looking hostnames."""
return bool(_re.match(r"^f[cd][0-9a-f]{0,2}:", h))
def _addr_is_local(host):
@@ -66,8 +33,6 @@ if os.environ.get("HERMES_WEBUI_TEST_NETWORK_BLOCK", "").strip() in ("1", "true"
h = host.strip().lower()
if not h:
return False
# IPv6 unique-local fc00::/7: require hex pair + colon to avoid
# matching hostnames like "food.example.com" or "fdsa.test".
if h in ("::1", "0:0:0:0:0:0:0:1") or h.startswith("fe80:") or _re_match_unique_local_ipv6(h):
return True
if h == "localhost" or h.endswith(".localhost"):
@@ -153,6 +118,13 @@ class QuietHTTPServer(ThreadingHTTPServer):
"""Custom HTTP server that silently handles common network errors."""
daemon_threads = True
request_queue_size = 64
max_request_workers = 128
_OVERFLOW_RESPONSE = (
b"HTTP/1.1 503 Service Unavailable\r\n"
b"Connection: close\r\n"
b"Content-Length: 0\r\n"
b"\r\n"
)
def __init__(self, *args, **kwargs):
server_address = args[0] if args else kwargs.get('server_address', None)
@@ -160,6 +132,7 @@ class QuietHTTPServer(ThreadingHTTPServer):
self.address_family = socket.AF_INET6
self.ssl_context: object | None = None
super().__init__(*args, **kwargs)
self._request_worker_slots = threading.BoundedSemaphore(self.max_request_workers)
self.accept_loop_requests_total = 0
self.accept_loop_last_request_at = 0.0
@@ -188,15 +161,7 @@ class QuietHTTPServer(ThreadingHTTPServer):
super().server_bind()
def get_request(self):
"""Accept a connection without letting TLS handshakes block the loop.
``ssl.wrap_socket(listening_socket)`` performs the TLS handshake inside
``accept()``. A browser/network probe that opens TCP but never sends a
ClientHello can then freeze the one accept loop, leaving the process
alive but unable to serve any later clients. Accept the raw socket here
and wrap the accepted connection with ``do_handshake_on_connect=False``
so any slow/broken handshake times out in its own request thread.
"""
"""Accept raw sockets and defer TLS handshake work to request threads."""
request, client_address = self.socket.accept()
ssl_context = getattr(self, "ssl_context", None)
if ssl_context is None:
@@ -213,68 +178,113 @@ class QuietHTTPServer(ThreadingHTTPServer):
return tls_request, client_address
def _handle_request_noblock(self):
"""Record accept-loop progress before dispatching a request handler.
A process can be alive and still stop accepting/dispatching requests.
Exposing this heartbeat on /health gives supervisors and watchdogs a
cheap signal that the accept loop is still moving.
Note: this method is called only from the single ``serve_forever()``
thread in CPython socketserver, so the un-locked ``+=`` increment is
safe — there is no other thread mutating these counters. The /health
readers may see a stale value momentarily but never an inconsistent
one (Python int reads are atomic). Per Opus advisor on stage-297.
"""
"""Record accept-loop progress before dispatching a request handler."""
self.accept_loop_requests_total += 1
self.accept_loop_last_request_at = time.time()
return super()._handle_request_noblock()
def _close_request_quietly(self, request) -> None:
try:
request.close()
except Exception:
pass
def _drain_request_input_nonblocking(self, request) -> None:
# Read through the current header block before replying so Windows
# doesn't reset the socket when we close with unread input.
deadline = time.monotonic() + 0.05
buffered = bytearray()
header_terminator = b"\r\n\r\n"
max_bytes = 65536
try:
timeout = request.gettimeout()
except Exception:
timeout = None
try:
while len(buffered) < max_bytes and header_terminator not in buffered:
wait = deadline - time.monotonic()
if wait <= 0:
break
try:
request.settimeout(wait)
chunk = request.recv(min(4096, max_bytes - len(buffered)))
except (BlockingIOError, InterruptedError, TimeoutError, socket.timeout):
break
except OSError:
break
if not chunk:
break
buffered.extend(chunk)
try:
request.shutdown(socket.SHUT_RD)
except OSError:
pass
finally:
try:
request.settimeout(timeout)
except Exception:
pass
def _reject_overflow_request(self, request) -> None:
if getattr(self, "ssl_context", None) is None:
self._drain_request_input_nonblocking(request)
try:
request.sendall(self._OVERFLOW_RESPONSE)
try:
request.shutdown(socket.SHUT_WR)
except Exception:
pass
except Exception:
pass
self._close_request_quietly(request)
def process_request(self, request, client_address):
if not self._request_worker_slots.acquire(blocking=False):
self._reject_overflow_request(request)
return
try:
return super().process_request(request, client_address)
except Exception:
self._request_worker_slots.release()
self._close_request_quietly(request)
raise
def process_request_thread(self, request, client_address):
try:
return super().process_request_thread(request, client_address)
finally:
self._request_worker_slots.release()
def handle_error(self, request, client_address):
"""Override to suppress logging for common client disconnect errors."""
"""Suppress logging for common client disconnect errors."""
exc_type, exc_value, _ = sys.exc_info()
# Silently ignore common connection errors caused by client disconnects
if exc_type in (
ConnectionResetError, BrokenPipeError, ConnectionAbortedError,
TimeoutError, ssl.SSLError, ssl.SSLEOFError,
):
return
# Also handle socket errors that indicate client disconnect
if issubclass(exc_type, OSError):
# errno 54 is Connection reset by peer on macOS/BSD
# errno 104 is Connection reset by peer on Linux
if getattr(exc_value, 'errno', None) in (32, 54, 104, 110): # EPIPE, ECONNRESET, ETIMEDOUT
return
# For other errors, use default logging
super().handle_error(request, client_address)
class Handler(BaseHTTPRequestHandler):
# HTTP/1.1 enables keep-alive connection reuse — major latency win on
# high-RTT links where every saved TCP handshake is 2×RTT. Each response
# MUST declare framing (Content-Length, Transfer-Encoding: chunked, or
# Connection: close) so the client knows where the message ends. Helpers
# j()/t() emit Content-Length; SSE/streaming endpoints emit
# Connection: close because the body has no terminator. See PR notes.
# HTTP/1.1 keep-alive stays on, so every response must declare framing.
protocol_version = "HTTP/1.1"
timeout = 30 # seconds — kills idle/incomplete connections to prevent thread exhaustion
def setup(self):
"""Set socket options for each accepted connection."""
super().setup()
# TCP_NODELAY — universal, disables Nagle for HTTP latency
try:
self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
except OSError:
pass
# SO_KEEPALIVE — universal master switch (must be set before timing params)
try:
self.connection.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
except OSError:
pass
# Per-platform timing parameters
if hasattr(socket, 'TCP_KEEPIDLE'): # Linux
try:
self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 10)
@@ -310,9 +320,6 @@ class Handler(BaseHTTPRequestHandler):
try:
print(message, flush=True)
except Exception:
# Agent/tool code can redirect or close process-wide stdout/stderr
# in another thread. HTTP response handling must not depend on
# those global streams remaining writable.
pass
def log_request(self, code: str='-', size: str='-') -> None:
@@ -345,7 +352,6 @@ class Handler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
self._req_t0 = time.time()
# Per-request profile context from cookie (issue #798)
cookie_profile = get_profile_cookie(self)
if cookie_profile:
set_request_profile(cookie_profile)
@@ -356,37 +362,25 @@ class Handler(BaseHTTPRequestHandler):
if result is False:
return j(self, {'error': 'not found'}, status=404)
except _CLIENT_DISCONNECT_ERRORS:
# The browser/client closed the socket while we were writing the
# response. This is expected for probes, tab closes, and SSE
# reconnect races; do not convert it into a misleading server 500.
return
except Exception:
self._safe_webui_print(f'[webui] ERROR {self.command} {self.path}\n' + traceback.format_exc())
try:
j(self, {'error': 'Internal server error'}, status=500)
except _CLIENT_DISCONNECT_ERRORS:
# Client disconnected while we were sending the 500 — nothing to do.
pass
except Exception:
# Unexpected failure while sending the error response itself.
# Log it so we know something is wrong with our error handler.
self._safe_webui_print(traceback.format_exc())
finally:
clear_request_profile()
def _handle_write(self, route_func) -> None:
self._req_t0 = time.time()
# Per-request profile context from cookie (issue #798)
cookie_profile = get_profile_cookie(self)
if cookie_profile:
set_request_profile(cookie_profile)
try:
parsed = urlparse(self.path)
# Stage-346 Opus SHOULD-FIX defense-in-depth: scope the CSP-report
# auth carve-out to POST only. The endpoint is intentionally
# unauthenticated (browsers omit cookies on CSP reports), but the
# carve-out should not extend to PATCH/DELETE on that path even
# though they currently fail through CSRF/routing fallthrough.
_is_csp_report_post = (
parsed.path == "/api/csp-report" and self.command == "POST"
)
@@ -395,20 +389,14 @@ class Handler(BaseHTTPRequestHandler):
if result is False:
return j(self, {'error': 'not found'}, status=404)
except _CLIENT_DISCONNECT_ERRORS:
# The browser/client closed the socket while we were writing the
# response. This is expected for probes, tab closes, and SSE
# reconnect races; do not convert it into a misleading server 500.
return
except Exception:
self._safe_webui_print(f'[webui] ERROR {self.command} {self.path}\n' + traceback.format_exc())
try:
j(self, {'error': 'Internal server error'}, status=500)
except _CLIENT_DISCONNECT_ERRORS:
# Client disconnected while we were sending the 500 — nothing to do.
pass
except Exception:
# Unexpected failure while sending the error response itself.
# Log it so we know something is wrong with our error handler.
self._safe_webui_print(traceback.format_exc())
finally:
clear_request_profile()
@@ -436,13 +424,7 @@ class Handler(BaseHTTPRequestHandler):
def _raise_fd_soft_limit(target: int = 4096) -> dict:
"""Best-effort raise of RLIMIT_NOFILE for persistent WebUI hosts.
macOS launchd jobs often start with a 256 soft limit. If a future FD leak
regresses, that low ceiling turns a leak into a hard HTTP wedge quickly.
Raising the soft limit does not hide leaks; it buys enough headroom for
diagnostics and watchdog recovery.
"""
"""Best-effort raise of RLIMIT_NOFILE for persistent WebUI hosts."""
if resource is None:
return {"status": "unsupported"}
try:
@@ -450,8 +432,6 @@ def _raise_fd_soft_limit(target: int = 4096) -> dict:
except Exception as exc:
return {"status": "error", "error": str(exc)}
# On Unix, RLIM_INFINITY is commonly a large int; keep the logic explicit
# so tests can use ordinary integers without depending on platform values.
desired = int(target)
if hard not in (-1, getattr(resource, "RLIM_INFINITY", object())):
desired = min(desired, int(hard))
@@ -555,13 +535,8 @@ def main() -> None:
elif fd_limit.get("status") == "error":
print(f"[!!] WARNING: Could not raise file descriptor limit: {fd_limit.get('error')}", flush=True)
# Fix sensitive file permissions before doing anything else
fix_credential_permissions()
# ── #1558 startup self-heal ─────────────────────────────────────────
# If a previous process wrote a session JSON with fewer messages than
# its .bak (the data-loss shape #1558 produced), restore from the .bak.
# Safe to run unconditionally — a clean install is a no-op.
try:
from api.models import _active_state_db_path
from api.session_recovery import recover_all_sessions_on_startup
@@ -577,7 +552,6 @@ def main() -> None:
print(f"[recovery] startup recovery failed: {exc}", flush=True)
within_container = False
# Check for the "/.within_container" file to determine if we're running inside a container; this file is created in the Dockerfile
try:
with open('/.within_container', 'r') as f:
within_container = True
@@ -587,7 +561,6 @@ def main() -> None:
if within_container:
print('[ok] Running within container.', flush=True)
# Security: warn if binding non-loopback without authentication
from api.auth import is_auth_enabled
if HOST not in ('127.0.0.1', '::1', 'localhost') and not is_auth_enabled():
print(f'[!!] WARNING: Binding to {HOST} with NO PASSWORD SET.', flush=True)
@@ -621,7 +594,6 @@ def main() -> None:
SESSION_DIR.mkdir(parents=True, exist_ok=True)
DEFAULT_WORKSPACE.mkdir(parents=True, exist_ok=True)
# Start the gateway session watcher for real-time SSE updates
try:
from api.gateway_watcher import start_watcher
@@ -639,10 +611,6 @@ def main() -> None:
except Exception as e:
print(f'[!!] WARNING: Gateway watcher failed to start: {e}', flush=True)
# Start the bg_task_complete drain thread for terminal(notify_on_complete=true)
# agent wakeup. Reads tools.process_registry.completion_queue and emits SSE
# bg_task_complete events (canonical name; legacy process_complete alias is
# still emitted for back-compat) to the matching session's stream.
try:
from api.background_process import start_drain_thread
if start_drain_thread():
@@ -650,9 +618,6 @@ def main() -> None:
except Exception as e:
print(f'[!!] WARNING: bg_task_complete drain failed to start: {e}', flush=True)
# Start the SessionChannel reaper for the persistent per-session SSE
# endpoint (/api/session/stream). Runs every 60s, collects channels with
# no subscribers past the grace period or past the idle TTL cap.
try:
from api.background_process import start_session_channel_reaper
if start_session_channel_reaper():
@@ -660,7 +625,6 @@ def main() -> None:
except Exception as e:
print(f'[!!] WARNING: SessionChannel reaper failed to start: {e}', flush=True)
# Load WebUI dashboard plugins
try:
from api.plugins import load_plugins
load_plugins()
@@ -670,7 +634,6 @@ def main() -> None:
_abort_if_already_serving(HOST, PORT)
httpd = QuietHTTPServer((HOST, PORT), Handler)
# ── TLS/HTTPS setup (optional) ─────────────────────────────────────────
from api.config import TLS_ENABLED, TLS_CERT, TLS_KEY
scheme = 'https' if TLS_ENABLED else 'http'
if TLS_ENABLED:
@@ -695,22 +658,16 @@ def main() -> None:
finally:
httpd.server_close()
_log_shutdown_audit()
# Stop the gateway watcher on shutdown
try:
from api.gateway_watcher import stop_watcher
stop_watcher()
except Exception:
logger.debug("Failed to stop gateway watcher during shutdown")
# Drain pending memory-provider lifecycle commits before exit
try:
from api.session_lifecycle import drain_all_on_shutdown
drain_all_on_shutdown()
except Exception:
logger.debug("Failed to drain lifecycle on shutdown", exc_info=True)
# Stop bg_task_complete drain + SessionChannel reaper (ours-original).
# The drain thread emits the canonical ``bg_task_complete`` event
# (with ``process_complete`` kept as a temporary backward-compat
# alias for older clients — see start_drain_thread comment above).
try:
from api.background_process import stop_drain_thread
stop_drain_thread()
tests/test_issue5210_http_worker_bound.py
+320
View File
@@ -0,0 +1,320 @@
"""Regression coverage for issue #5210 worker-bound request dispatch."""
from __future__ import annotations
import http.client
import socket
import ssl
import subprocess
import threading
import time
from http.server import BaseHTTPRequestHandler
from pathlib import Path
import pytest
from server import QuietHTTPServer
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
def _worker_count(server: QuietHTTPServer) -> int:
with server.worker_lock:
return server.worker_starts
def _wait_for_worker_count(server: QuietHTTPServer, expected: int, *, timeout: float = 2.0) -> None:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
current = _worker_count(server)
if current == expected:
return
if current > expected:
raise AssertionError(f"expected worker count {expected}, got {current}")
time.sleep(0.01)
raise AssertionError(f"timed out waiting for worker count {expected}, got {_worker_count(server)}")
def _assert_worker_count_stays(server: QuietHTTPServer, expected: int, *, duration: float = 0.35) -> None:
deadline = time.monotonic() + duration
while time.monotonic() < deadline:
current = _worker_count(server)
assert current == expected, f"expected worker count {expected}, got {current}"
time.sleep(0.01)
def _wait_for_worker_slot_release(server: QuietHTTPServer, *, timeout: float = 2.0) -> None:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if server._request_worker_slots.acquire(blocking=False):
server._request_worker_slots.release()
return
time.sleep(0.01)
raise AssertionError("timed out waiting for worker slot release")
def _request(port: int, path: str, *, use_ssl: bool = False, timeout: float = 2.0) -> tuple[int, dict[str, str], bytes]:
if use_ssl:
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
conn = http.client.HTTPSConnection("127.0.0.1", port, timeout=timeout, context=context)
else:
conn = http.client.HTTPConnection("127.0.0.1", port, timeout=timeout)
try:
conn.request("GET", path)
resp = conn.getresponse()
body = resp.read()
return resp.status, {k: v for k, v in resp.getheaders()}, body
finally:
conn.close()
def _start_request_thread(port: int, path: str, *, use_ssl: bool = False) -> tuple[threading.Thread, dict[str, object]]:
result: dict[str, object] = {}
def runner() -> None:
try:
result["value"] = _request(port, path, use_ssl=use_ssl, timeout=5.0)
except Exception as exc: # pragma: no cover - only used for failure diagnostics
result["error"] = exc
thread = threading.Thread(target=runner, daemon=True)
thread.start()
return thread, result
def _generate_self_signed_cert(tmp_path: Path) -> tuple[str, str]:
cert = tmp_path / "cert.pem"
key = tmp_path / "key.pem"
subprocess.run(
[
"openssl",
"req",
"-x509",
"-newkey",
"rsa:2048",
"-keyout",
str(key),
"-out",
str(cert),
"-days",
"1",
"-nodes",
"-subj",
"/CN=localhost",
],
check=True,
capture_output=True,
)
return str(cert), str(key)
class _ObservedQuietHTTPServer(QuietHTTPServer):
def __init__(self, *args, **kwargs):
self.worker_lock = threading.Lock()
self.worker_starts = 0
self.worker_started = threading.Event()
self.release_event = threading.Event()
self.hold_entered = threading.Event()
self.error_entered = threading.Event()
self.disconnect_entered = threading.Event()
super().__init__(*args, **kwargs)
def process_request_thread(self, request, client_address):
with self.worker_lock:
self.worker_starts += 1
self.worker_started.set()
return super().process_request_thread(request, client_address)
class _GateHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def log_message(self, *args) -> None: # pragma: no cover - noise suppression
pass
def _send_ok(self) -> None:
body = b"ok"
self.send_response(200)
self.send_header("Connection", "close")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self): # noqa: N802
if self.path == "/hold":
self.server.hold_entered.set()
if not self.server.release_event.wait(timeout=5):
raise TimeoutError("release_event not set")
self._send_ok()
return
if self.path == "/boom":
self.server.error_entered.set()
if not self.server.release_event.wait(timeout=5):
raise TimeoutError("release_event not set")
raise RuntimeError("boom")
if self.path == "/disconnect":
self.server.disconnect_entered.set()
if not self.server.release_event.wait(timeout=5):
raise TimeoutError("release_event not set")
self._send_ok()
return
if self.path == "/fast":
self._send_ok()
return
self.send_error(404)
class _ServerRunner:
def __init__(self, handler_cls, *, ssl_context=None):
self.port = _free_port()
self.httpd = _ObservedQuietHTTPServer(("127.0.0.1", self.port), handler_cls)
self.httpd.ssl_context = ssl_context
self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
def __enter__(self):
self.thread.start()
time.sleep(0.1)
return self
def __exit__(self, exc_type, exc, tb):
self.httpd.release_event.set()
self.httpd.shutdown()
self.httpd.server_close()
self.thread.join(timeout=5)
def test_over_capacity_plain_http_gets_fast_503_without_starting_handler(monkeypatch):
monkeypatch.setattr(QuietHTTPServer, "max_request_workers", 1, raising=False)
with _ServerRunner(_GateHandler) as srv:
hold_thread, hold_result = _start_request_thread(srv.port, "/hold")
assert srv.httpd.hold_entered.wait(timeout=5)
_wait_for_worker_count(srv.httpd, 1)
status, headers, body = _request(srv.port, "/fast")
assert status == 503
assert headers["Connection"].lower() == "close"
assert body == b""
_assert_worker_count_stays(srv.httpd, 1)
srv.httpd.release_event.set()
hold_thread.join(timeout=5)
assert "error" not in hold_result, hold_result.get("error")
assert hold_result["value"][0] == 200
def test_worker_slot_releases_after_request_finishes(monkeypatch):
monkeypatch.setattr(QuietHTTPServer, "max_request_workers", 1, raising=False)
with _ServerRunner(_GateHandler) as srv:
hold_thread, hold_result = _start_request_thread(srv.port, "/hold")
assert srv.httpd.hold_entered.wait(timeout=5)
_wait_for_worker_count(srv.httpd, 1)
status, headers, body = _request(srv.port, "/fast")
assert status == 503
assert headers["Connection"].lower() == "close"
assert body == b""
srv.httpd.release_event.set()
hold_thread.join(timeout=5)
assert "error" not in hold_result, hold_result.get("error")
assert hold_result["value"][0] == 200
status, headers, body = _request(srv.port, "/fast")
assert status == 200
assert headers["Connection"].lower() == "close"
assert body == b"ok"
@pytest.mark.parametrize("mode", ["boom", "disconnect"])
def test_worker_slot_releases_after_handler_error_or_disconnect(monkeypatch, mode):
monkeypatch.setattr(QuietHTTPServer, "max_request_workers", 1, raising=False)
with _ServerRunner(_GateHandler) as srv:
if mode == "boom":
held_thread, _ = _start_request_thread(srv.port, "/boom")
assert srv.httpd.error_entered.wait(timeout=5)
else:
sock = socket.create_connection(("127.0.0.1", srv.port), timeout=2)
sock.sendall(
b"GET /disconnect HTTP/1.1\r\n"
b"Host: localhost\r\n"
b"Connection: close\r\n"
b"\r\n"
)
held_thread = None
assert srv.httpd.disconnect_entered.wait(timeout=5)
_wait_for_worker_count(srv.httpd, 1)
status, headers, body = _request(srv.port, "/fast")
assert status == 503
assert headers["Connection"].lower() == "close"
assert body == b""
if mode == "boom":
srv.httpd.release_event.set()
held_thread.join(timeout=5)
assert not held_thread.is_alive()
else:
sock.close()
srv.httpd.release_event.set()
_wait_for_worker_slot_release(srv.httpd)
status, headers, body = _request(srv.port, "/fast")
assert status == 200
assert headers["Connection"].lower() == "close"
assert body == b"ok"
def test_long_running_request_does_not_reject_healthy_in_cap_request(monkeypatch):
monkeypatch.setattr(QuietHTTPServer, "max_request_workers", 2, raising=False)
with _ServerRunner(_GateHandler) as srv:
hold_thread, hold_result = _start_request_thread(srv.port, "/hold")
assert srv.httpd.hold_entered.wait(timeout=5)
_wait_for_worker_count(srv.httpd, 1)
status, headers, body = _request(srv.port, "/fast")
assert status == 200
assert headers["Connection"].lower() == "close"
assert body == b"ok"
_wait_for_worker_count(srv.httpd, 2)
srv.httpd.release_event.set()
hold_thread.join(timeout=5)
assert "error" not in hold_result, hold_result.get("error")
assert hold_result["value"][0] == 200
def test_tls_overflow_does_not_force_accept_loop_handshake(monkeypatch, tmp_path):
monkeypatch.setattr(QuietHTTPServer, "max_request_workers", 1, raising=False)
cert, key = _generate_self_signed_cert(tmp_path)
server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
server_context.load_cert_chain(cert, key)
with _ServerRunner(_GateHandler, ssl_context=server_context) as srv:
hold_thread, hold_result = _start_request_thread(srv.port, "/hold", use_ssl=True)
assert srv.httpd.hold_entered.wait(timeout=5)
_wait_for_worker_count(srv.httpd, 1)
raw = socket.create_connection(("127.0.0.1", srv.port), timeout=2)
try:
raw.settimeout(2)
assert raw.recv(1) == b""
finally:
raw.close()
_assert_worker_count_stays(srv.httpd, 1)
srv.httpd.release_event.set()
hold_thread.join(timeout=5)
assert "error" not in hold_result, hold_result.get("error")
assert hold_result["value"][0] == 200