From c3a3293b3b9009a8b6149b92ced38faad932938f Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 01:44:47 -0500 Subject: [PATCH 01/14] feat: add bounded HTTP/2 prior-knowledge server --- pyproject.toml | 3 +- smallserver/__init__.py | 2 + smallserver/app.py | 188 +++++++++++++++- smallserver/http2.py | 484 ++++++++++++++++++++++++++++++++++++++++ smallserver/server.py | 19 +- tests/test_http2.py | 276 +++++++++++++++++++++++ 6 files changed, 965 insertions(+), 7 deletions(-) create mode 100644 smallserver/http2.py create mode 100644 tests/test_http2.py diff --git a/pyproject.toml b/pyproject.toml index 337debe..df79c26 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "smallserver" version = "0.1.0" -description = "A SmallOS-native HTTP/1.1 web framework" +description = "A SmallOS-native HTTP/1.1 and HTTP/2 web framework" readme = "README.md" requires-python = ">=3.10" license = "MIT" @@ -14,6 +14,7 @@ dependencies = [] [project.optional-dependencies] dev = ["build>=1.2"] +http2 = ["h2>=4,<5"] [tool.setuptools.packages.find] include = ["smallserver*"] diff --git a/smallserver/__init__.py b/smallserver/__init__.py index 6600810..b48c309 100644 --- a/smallserver/__init__.py +++ b/smallserver/__init__.py @@ -10,6 +10,7 @@ ServerStartupError, ) from .http import Headers, Request, Response +from .http2 import HTTP2Config from .server import ServerConfig, ServerHandle if TYPE_CHECKING: @@ -31,6 +32,7 @@ def __getattr__(name: str) -> Any: "AdapterShutdownError", "Headers", "HTTPError", + "HTTP2Config", "Request", "Response", "ServerConfig", diff --git a/smallserver/app.py b/smallserver/app.py index 6815502..d413545 100644 --- a/smallserver/app.py +++ b/smallserver/app.py @@ -24,10 +24,32 @@ _CleanupTransaction, ) from .http import Request, Response +from .http2 import HTTP2Config, H2Protocol, require_http2 from .server import HTTPParseError, HTTPRequestParser, ServerConfig, ServerHandle Handler = Callable[[Request], Awaitable[Response]] _METHODS = frozenset({"GET", "POST", "PUT", "PATCH", "DELETE"}) +_HTTP2_WRITER_SIGNAL = 30 + + +class _H2ConnectionState: + def __init__(self, protocol: H2Protocol) -> None: + self.protocol = protocol + self.writer_task: Any = None + self.handlers: dict[int, Any] = {} + self.closing = False + self.shutdown_requested = False + self.close_error_code = 0 + + def wake_writer(self) -> None: + writer = self.writer_task + if writer is not None and not getattr(writer, "done", False): + if writer.acceptSignal(_HTTP2_WRITER_SIGNAL) != 0: + raise RuntimeError("HTTP/2 writer signal failed") + + def request_shutdown(self) -> None: + self.shutdown_requested = True + self.wake_writer() class _NoThreadLock: @@ -189,6 +211,9 @@ def serve( host: str = "127.0.0.1", port: int = 8000, config: ServerConfig | None = None, + *, + protocol: str = "http1", + http2_config: HTTP2Config | None = None, ) -> ServerHandle: """Bind a TCP listener and schedule SmallOS listener/control tasks. @@ -198,7 +223,9 @@ def serve( scheduler thread instead. """ self._validate_runtime(runtime, require_start=False) - return self._bind_and_schedule(runtime, host, port, config) + return self._bind_and_schedule( + runtime, host, port, config, protocol, http2_config + ) @overload def listen( @@ -207,6 +234,8 @@ def listen( port: int = 8000, config: ServerConfig | None = None, *, + protocol: str = "http1", + http2_config: HTTP2Config | None = None, runtime: None = None, start: Literal[True] | None = None, ) -> ServerHandle: ... @@ -218,6 +247,8 @@ def listen( port: int = 8000, config: ServerConfig | None = None, *, + protocol: str = "http1", + http2_config: HTTP2Config | None = None, runtime: _RuntimeLike, start: Literal[False] | None = None, ) -> ServerHandle: ... @@ -229,6 +260,8 @@ def listen( port: int = 8000, config: ServerConfig | None = None, *, + protocol: str = "http1", + http2_config: HTTP2Config | None = None, runtime: _StartableRuntime, start: bool, ) -> ServerHandle: ... @@ -239,6 +272,8 @@ def listen( port: int = 8000, config: ServerConfig | None = None, *, + protocol: str = "http1", + http2_config: HTTP2Config | None = None, runtime: _RuntimeLike | None = None, start: bool | None = None, ) -> ServerHandle: @@ -257,7 +292,9 @@ def listen( if runtime is None: runtime = _default_runtime_factory() self._validate_runtime(runtime, require_start=should_start) - handle = self._bind_and_schedule(runtime, host, port, config) + handle = self._bind_and_schedule( + runtime, host, port, config, protocol, http2_config + ) if not should_start: return handle primary_error: BaseException | None = None @@ -300,6 +337,8 @@ def _bind_and_schedule( host: str, port: int, config: ServerConfig | None, + protocol: str, + http2_config: HTTP2Config | None, ) -> ServerHandle: """Shared validated bind-and-schedule core for ``serve`` and ``listen``.""" from SmallPackage import SmallTask @@ -310,6 +349,14 @@ def _bind_and_schedule( raise ValueError("port must be an integer between 0 and 65535") if config is not None and not isinstance(config, ServerConfig): raise TypeError("config must be a ServerConfig or None") + if protocol not in {"http1", "http2"}: + raise ValueError("protocol must be 'http1' or 'http2'") + if http2_config is not None and not isinstance(http2_config, HTTP2Config): + raise TypeError("http2_config must be an HTTP2Config or None") + if protocol == "http1" and http2_config is not None: + raise ValueError("http2_config requires protocol='http2'") + if protocol == "http2": + require_http2() marker = self._reserve_invocation() def release_marker() -> None: @@ -365,7 +412,14 @@ def release(completed: ServerHandle) -> None: try: handle = ServerHandle( - runtime, transport, listener, wakeup, config, on_finalized=release + runtime, + transport, + listener, + wakeup, + config, + on_finalized=release, + protocol=protocol, + protocol_config=http2_config or HTTP2Config(), ) except BaseException as primary_error: transaction = _CleanupTransaction() @@ -466,9 +520,14 @@ async def _accept_loop(self, task: Any, handle: ServerHandle) -> None: connection_task: Any = None try: + routine = ( + self._http2_connection_loop + if handle._protocol == "http2" + else self._connection_loop + ) connection_task = SmallTask( handle._config.connection_priority, - self._connection_loop, + routine, args=(handle, client), name="smallserver-connection", ) @@ -552,3 +611,124 @@ async def _send_response( headers["Connection"] = "close" payload = Response(response.status, response.body, headers).to_http1() await handle._transport.send_all(task, client, payload) + + async def _http2_connection_loop( + self, task: Any, handle: ServerHandle, client: TransportHandle + ) -> None: + from SmallPackage import SmallTask + + protocol = H2Protocol(handle._protocol_config) + state = _H2ConnectionState(protocol) + primary_error: BaseException | None = None + handle._graceful_connections.add(id(client)) + handle._graceful_closers[id(client)] = state.request_shutdown + try: + await handle._transport.send_all(task, client, protocol.initiate()) + writer = SmallTask( + handle._config.connection_priority, + self._http2_writer_loop, + args=(handle, client, state), + name="smallserver-http2-writer", + ) + state.writer_task = writer + handle._owned_tasks.append(writer) + handle._runtime.fork(writer) + while not handle.closed and not protocol.remote_closed: + chunk = await handle._transport.recv( + task, client, handle._config.receive_chunk_bytes + ) + if not chunk: + break + try: + ready = protocol.receive_data(chunk) + except Exception as protocol_error: + primary_error = protocol_error + state.close_error_code = 1 + break + for stream_id in protocol.take_cancelled_streams(): + handler = state.handlers.pop(stream_id, None) + if handler is not None: + handle._cancel_or_retain_task(handler) + for item in ready: + handler = SmallTask( + handle._config.connection_priority, + self._http2_handler, + args=(handle, state, item.stream_id, item.request), + name="smallserver-http2-stream-{}".format(item.stream_id), + ) + state.handlers[item.stream_id] = handler + handle._owned_tasks.append(handler) + handle._runtime.fork(handler) + state.wake_writer() + except Exception as exc: + primary_error = exc + except BaseException as exc: + primary_error = exc + raise + finally: + state.closing = True + for handler in tuple(state.handlers.values()): + handle._cancel_or_retain_task(handler) + state.handlers.clear() + if state.writer_task is not None: + state.wake_writer() + handle._cancel_or_retain_task(state.writer_task) + if state.writer_task in handle._owned_tasks: + handle._owned_tasks.remove(state.writer_task) + try: + goaway = protocol.close(state.close_error_code) + if goaway: + await handle._transport.send_all(task, client, goaway) + except BaseException: + pass + handle._connection_finished(task, client, primary_error) + + async def _http2_handler( + self, + task: Any, + handle: ServerHandle, + state: _H2ConnectionState, + stream_id: int, + request: Request, + ) -> None: + try: + try: + response = await self.dispatch(request) + except Exception: + response = Response.text("internal server error", status=500) + state.protocol.queue_response(stream_id, response) + state.wake_writer() + finally: + state.handlers.pop(stream_id, None) + if task in handle._owned_tasks: + handle._owned_tasks.remove(task) + + async def _http2_writer_loop( + self, + task: Any, + handle: ServerHandle, + client: TransportHandle, + state: _H2ConnectionState, + ) -> None: + try: + while not state.closing: + await task.wait_signal(_HTTP2_WRITER_SIGNAL) + if state.shutdown_requested: + state.closing = True + payload = state.protocol.close() + if payload: + await handle._transport.send_all(task, client, payload) + if not handle._transport.close_safely(client): + error = client.close_error or RuntimeError( + "kernel connection close failed" + ) + handle._connection_close_failed(error, task) + return + while not state.closing: + payload = state.protocol.flush() + if not payload: + break + await handle._transport.send_all(task, client, payload) + finally: + if task in handle._owned_tasks: + handle._owned_tasks.remove(task) diff --git a/smallserver/http2.py b/smallserver/http2.py new file mode 100644 index 0000000..49dc184 --- /dev/null +++ b/smallserver/http2.py @@ -0,0 +1,484 @@ +"""Lazy, bounded HTTP/2 protocol state built on optional hyper-h2.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from .errors import ServerConfigurationError +from .http import Headers, Request, Response + + +HTTP2_CLIENT_PREFACE = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" + + +@dataclass(frozen=True) +class HTTP2Config: + """Finite protocol and buffering limits for HTTP/2 connections.""" + + max_concurrent_streams: int = 100 + max_header_count: int = 100 + max_header_bytes: int = 16 * 1024 + max_compressed_header_bytes: int = 16 * 1024 + max_body_bytes: int = 1024 * 1024 + max_connection_buffer_bytes: int = 4 * 1024 * 1024 + max_pending_output_bytes: int = 4 * 1024 * 1024 + max_response_body_bytes: int = 2 * 1024 * 1024 + max_frame_size: int = 16 * 1024 + + def __post_init__(self) -> None: + for name, value in self.__dict__.items(): + if type(value) is not int or value <= 0: + raise ValueError("{} must be a positive integer".format(name)) + if not 16_384 <= self.max_frame_size <= 16_777_215: + raise ValueError("max_frame_size must be between 16384 and 16777215") + if self.max_body_bytes > self.max_connection_buffer_bytes: + raise ValueError( + "max_body_bytes cannot exceed max_connection_buffer_bytes" + ) + if self.max_response_body_bytes > self.max_pending_output_bytes: + raise ValueError( + "max_response_body_bytes cannot exceed max_pending_output_bytes" + ) + + +@dataclass(frozen=True) +class H2ReadyRequest: + """A complete request stream ready for application dispatch.""" + + stream_id: int + request: Request + + +@dataclass +class _InboundStream: + method: str + path: str + headers: Headers + body: bytearray + + +@dataclass +class _OutboundStream: + body: bytes + offset: int = 0 + + +class _FrameBudget: + """Account compressed header blocks without implementing frame semantics.""" + + def __init__(self, config: HTTP2Config) -> None: + self._config = config + self._preface = bytearray() + self._header = bytearray() + self._remaining = 0 + self._frame_type = 0 + self._frame_stream = 0 + self._frame_flags = 0 + self._header_stream: int | None = None + self._header_bytes = 0 + + def feed(self, data: bytes) -> None: + view = memoryview(data) + offset = 0 + if len(self._preface) < len(HTTP2_CLIENT_PREFACE): + needed = len(HTTP2_CLIENT_PREFACE) - len(self._preface) + take = min(needed, len(view)) + self._preface.extend(view[:take]) + offset += take + expected = HTTP2_CLIENT_PREFACE[: len(self._preface)] + if bytes(self._preface) != expected: + raise ValueError("invalid HTTP/2 client preface") + if offset == len(view): + return + + while offset < len(view): + if self._remaining == 0: + needed = 9 - len(self._header) + take = min(needed, len(view) - offset) + self._header.extend(view[offset : offset + take]) + offset += take + if len(self._header) < 9: + return + length = int.from_bytes(self._header[:3], "big") + if length > self._config.max_frame_size: + raise ValueError("HTTP/2 frame exceeds configured maximum") + self._frame_type = self._header[3] + self._frame_flags = self._header[4] + self._frame_stream = int.from_bytes(self._header[5:9], "big") & 0x7FFFFFFF + self._header.clear() + self._remaining = length + if length == 0: + self._finish_frame() + continue + + take = min(self._remaining, len(view) - offset) + if self._frame_type in (0x1, 0x9): + self._account_header_bytes(take) + self._remaining -= take + offset += take + if self._remaining == 0: + self._finish_frame() + + def _account_header_bytes(self, count: int) -> None: + if self._frame_type == 0x1 and self._header_stream is None: + self._header_stream = self._frame_stream + self._header_bytes = 0 + self._header_bytes += count + if self._header_bytes > self._config.max_compressed_header_bytes: + raise ValueError("HTTP/2 compressed header block is too large") + + def _finish_frame(self) -> None: + if self._frame_type in (0x1, 0x9) and self._frame_flags & 0x4: + self._header_stream = None + self._header_bytes = 0 + + +def require_http2() -> None: + """Fail clearly without importing hyper-h2 on HTTP/1.1 paths.""" + try: + import h2 # type: ignore[import-not-found] + except ImportError as exc: + raise ServerConfigurationError( + "HTTP/2 requires the optional dependency; install smallserver[http2]" + ) from exc + version = getattr(h2, "__version__", "") + if not isinstance(version, str) or not version.startswith("4."): + raise ServerConfigurationError("HTTP/2 requires hyper-h2 version 4.x") + + +class H2Protocol: + """One connection's sans-I/O HTTP/2 and bounded stream state.""" + + _FORBIDDEN_HEADERS = { + "connection", + "keep-alive", + "proxy-connection", + "transfer-encoding", + "upgrade", + } + + def __init__(self, config: HTTP2Config | None = None) -> None: + require_http2() + from h2.config import H2Configuration # type: ignore[import-not-found] + from h2.connection import H2Connection # type: ignore[import-not-found] + from h2.errors import ErrorCodes # type: ignore[import-not-found] + from h2.events import ( # type: ignore[import-not-found] + ConnectionTerminated, + DataReceived, + RemoteSettingsChanged, + RequestReceived, + StreamEnded, + StreamReset, + TrailersReceived, + WindowUpdated, + ) + from h2.settings import SettingCodes # type: ignore[import-not-found] + + self.config = config or HTTP2Config() + h2_config = H2Configuration( + client_side=False, + header_encoding="utf-8", + validate_inbound_headers=True, + normalize_inbound_headers=False, + ) + self.connection = H2Connection(config=h2_config) + self.connection.local_settings[SettingCodes.MAX_CONCURRENT_STREAMS] = ( + self.config.max_concurrent_streams + ) + self.connection.local_settings[SettingCodes.MAX_HEADER_LIST_SIZE] = ( + self.config.max_header_bytes + ) + self.connection.local_settings[SettingCodes.MAX_FRAME_SIZE] = ( + self.config.max_frame_size + ) + self._events = { + "request": RequestReceived, + "data": DataReceived, + "ended": StreamEnded, + "reset": StreamReset, + "trailers": TrailersReceived, + "window": WindowUpdated, + "settings": RemoteSettingsChanged, + "terminated": ConnectionTerminated, + } + self._error_codes = ErrorCodes + self._frames = _FrameBudget(self.config) + self._inbound: dict[int, _InboundStream] = {} + self._active_streams: set[int] = set() + self._outbound: dict[int, _OutboundStream] = {} + self._commands: list[tuple[str, int, Response | None]] = [] + self._buffered_request_bytes = 0 + self._pending_output_bytes = 0 + self._cancelled_streams: list[int] = [] + self.last_processed_stream_id = 0 + self.remote_closed = False + self.local_closed = False + + @property + def active_stream_count(self) -> int: + return len(self._active_streams) + + @property + def pending_output_bytes(self) -> int: + return self._pending_output_bytes + + def initiate(self) -> bytes: + self.connection.initiate_connection() + return self.connection.data_to_send() + + def receive_data(self, data: bytes) -> tuple[H2ReadyRequest, ...]: + self._frames.feed(data) + events = self.connection.receive_data(data) + ready: list[H2ReadyRequest] = [] + for event in events: + if isinstance(event, self._events["request"]): + self._request_received(event.stream_id, event.headers) + elif isinstance(event, self._events["data"]): + self._data_received( + event.stream_id, event.data, event.flow_controlled_length + ) + elif isinstance(event, self._events["ended"]): + completed = self._stream_ended(event.stream_id) + if completed is not None: + ready.append(completed) + elif isinstance(event, self._events["reset"]): + self._cancelled_streams.append(event.stream_id) + self.drop_stream(event.stream_id) + elif isinstance(event, self._events["trailers"]): + self._reset_stream(event.stream_id, self._error_codes.PROTOCOL_ERROR) + elif isinstance(event, self._events["terminated"]): + self.remote_closed = True + elif isinstance(event, (self._events["window"], self._events["settings"])): + pass + return tuple(ready) + + def take_cancelled_streams(self) -> tuple[int, ...]: + """Return peer-reset stream ids exactly once.""" + cancelled, self._cancelled_streams = self._cancelled_streams, [] + return tuple(cancelled) + + def _request_received(self, stream_id: int, raw_headers: Any) -> None: + if len(self._active_streams) >= self.config.max_concurrent_streams: + self._reset_stream(stream_id, self._error_codes.REFUSED_STREAM) + return + try: + method, path, headers = self._decode_request_headers(raw_headers) + except (TypeError, ValueError): + self._reset_stream(stream_id, self._error_codes.PROTOCOL_ERROR) + return + self._active_streams.add(stream_id) + self._inbound[stream_id] = _InboundStream(method, path, headers, bytearray()) + + def _decode_request_headers(self, raw_headers: Any) -> tuple[str, str, Headers]: + if len(raw_headers) > self.config.max_header_count: + raise ValueError("too many HTTP/2 request headers") + decoded_size = 0 + pseudo: dict[str, str] = {} + regular: list[tuple[str, str]] = [] + cookies: list[str] = [] + seen_regular = False + for name, value in raw_headers: + if not isinstance(name, str) or not isinstance(value, str): + raise TypeError("HTTP/2 headers must decode to text") + decoded_size += len(name.encode("utf-8")) + len(value.encode("utf-8")) + 32 + if decoded_size > self.config.max_header_bytes: + raise ValueError("HTTP/2 decoded headers are too large") + if name.startswith(":"): + if seen_regular or name in pseudo: + raise ValueError("invalid HTTP/2 pseudo-header ordering") + pseudo[name] = value + continue + seen_regular = True + lowered = name.lower() + if name != lowered or lowered in self._FORBIDDEN_HEADERS: + raise ValueError("forbidden HTTP/2 request header") + if lowered == "te" and value.lower() != "trailers": + raise ValueError("invalid HTTP/2 TE header") + if lowered == "cookie": + cookies.append(value) + else: + regular.append((name, value)) + allowed = {":method", ":scheme", ":authority", ":path"} + if set(pseudo) - allowed: + raise ValueError("unknown HTTP/2 pseudo-header") + if pseudo.get(":method") == "CONNECT": + raise ValueError("HTTP/2 CONNECT is not supported") + if not all(pseudo.get(name) for name in (":method", ":scheme", ":path")): + raise ValueError("missing required HTTP/2 pseudo-header") + authority = pseudo.get(":authority") + existing_host = any(name == "host" for name, _value in regular) + if authority and existing_host: + raise ValueError("HTTP/2 authority and host must not both be supplied") + if not authority and not existing_host: + raise ValueError("HTTP/2 requests require :authority or host") + if authority: + regular.append(("host", authority)) + if cookies: + regular.append(("cookie", "; ".join(cookies))) + return pseudo[":method"], pseudo[":path"], Headers(regular) + + def _data_received(self, stream_id: int, data: bytes, flow_length: int) -> None: + self.connection.acknowledge_received_data(flow_length, stream_id) + stream = self._inbound.get(stream_id) + if stream is None: + return + next_stream_size = len(stream.body) + len(data) + next_connection_size = self._buffered_request_bytes + len(data) + if ( + next_stream_size > self.config.max_body_bytes + or next_connection_size > self.config.max_connection_buffer_bytes + ): + self._reset_stream(stream_id, self._error_codes.ENHANCE_YOUR_CALM) + return + stream.body.extend(data) + self._buffered_request_bytes = next_connection_size + + def _stream_ended(self, stream_id: int) -> H2ReadyRequest | None: + stream = self._inbound.pop(stream_id, None) + if stream is None: + return None + self._buffered_request_bytes -= len(stream.body) + self.last_processed_stream_id = max(self.last_processed_stream_id, stream_id) + request = Request( + stream.method, + stream.path, + stream.headers, + bytes(stream.body), + "HTTP/2", + ) + return H2ReadyRequest(stream_id, request) + + def queue_response(self, stream_id: int, response: Response) -> bool: + if stream_id not in self._active_streams: + return False + body_size = len(response.body) + if ( + body_size > self.config.max_response_body_bytes + or self._pending_output_bytes + body_size + > self.config.max_pending_output_bytes + ): + if not any(command[1] == stream_id for command in self._commands): + self._commands.append(("reset", stream_id, None)) + return False + if len(self._commands) >= self.config.max_concurrent_streams * 2: + self._reset_stream(stream_id, self._error_codes.ENHANCE_YOUR_CALM) + return False + self._pending_output_bytes += body_size + self._commands.append(("response", stream_id, response)) + return True + + def flush(self) -> bytes: + commands, self._commands = self._commands, [] + for operation, stream_id, response in commands: + if operation == "reset": + self._reset_stream(stream_id, self._error_codes.ENHANCE_YOUR_CALM) + continue + assert response is not None + self._start_response(stream_id, response) + + for stream_id, outbound in tuple(self._outbound.items()): + remaining = len(outbound.body) - outbound.offset + if remaining <= 0: + self._outbound.pop(stream_id, None) + self._active_streams.discard(stream_id) + continue + try: + window = self.connection.local_flow_control_window(stream_id) + except Exception: + self.drop_stream(stream_id) + continue + chunk_size = min( + remaining, + max(0, window), + self.connection.max_outbound_frame_size, + ) + if chunk_size <= 0: + continue + end_stream = chunk_size == remaining + chunk = memoryview(outbound.body)[ + outbound.offset : outbound.offset + chunk_size + ] + try: + self.connection.send_data(stream_id, chunk, end_stream=end_stream) + except Exception: + self.drop_stream(stream_id) + continue + outbound.offset += chunk_size + self._pending_output_bytes -= chunk_size + if end_stream: + self._outbound.pop(stream_id, None) + self._active_streams.discard(stream_id) + return self.connection.data_to_send() + + def _start_response(self, stream_id: int, response: Response) -> None: + headers: list[tuple[str, str]] = [(":status", str(response.status))] + forbidden = self._FORBIDDEN_HEADERS | {"te"} + for name, value in response.headers.items(): + lowered = name.lower() + if lowered in forbidden: + continue + headers.append((lowered, value)) + if response.headers.get("content-length") is None: + headers.append(("content-length", str(len(response.body)))) + header_size = sum( + len(name.encode("utf-8")) + len(value.encode("utf-8")) + 32 + for name, value in headers + ) + if header_size > self.config.max_header_bytes: + self._pending_output_bytes -= len(response.body) + self._reset_stream(stream_id, self._error_codes.INTERNAL_ERROR) + return + try: + self.connection.send_headers( + stream_id, headers, end_stream=not response.body + ) + except Exception: + self._pending_output_bytes -= len(response.body) + self.drop_stream(stream_id) + return + if response.body: + self._outbound[stream_id] = _OutboundStream(response.body) + else: + self._active_streams.discard(stream_id) + + def drop_stream(self, stream_id: int) -> None: + inbound = self._inbound.pop(stream_id, None) + if inbound is not None: + self._buffered_request_bytes -= len(inbound.body) + outbound = self._outbound.pop(stream_id, None) + if outbound is not None: + self._pending_output_bytes -= len(outbound.body) - outbound.offset + kept: list[tuple[str, int, Response | None]] = [] + for command in self._commands: + if command[1] == stream_id and command[2] is not None: + self._pending_output_bytes -= len(command[2].body) + else: + kept.append(command) + self._commands = kept + self._active_streams.discard(stream_id) + + def _reset_stream(self, stream_id: int, error_code: Any) -> None: + try: + self.connection.reset_stream(stream_id, error_code=error_code) + except Exception: + pass + self.drop_stream(stream_id) + + def close(self, error_code: int = 0) -> bytes: + if not self.local_closed: + self.local_closed = True + try: + self.connection.close_connection( + error_code=error_code, + last_stream_id=self.last_processed_stream_id, + ) + except Exception: + pass + self._inbound.clear() + self._outbound.clear() + self._commands.clear() + self._active_streams.clear() + self._buffered_request_bytes = 0 + self._pending_output_bytes = 0 + return self.connection.data_to_send() diff --git a/smallserver/server.py b/smallserver/server.py index 4a1cad7..4c313ca 100644 --- a/smallserver/server.py +++ b/smallserver/server.py @@ -129,12 +129,16 @@ def __init__( wakeup: WakeupChannel | None, config: ServerConfig, on_finalized: Callable[[ServerHandle], None] | None = None, + protocol: str = "http1", + protocol_config: Any = None, ) -> None: self._runtime = runtime self._transport = transport self._listener = listener self._wakeup = wakeup self._config = config + self._protocol = protocol + self._protocol_config = protocol_config self._address = transport.local_address(listener) self._on_finalized = on_finalized self._close_requested = False @@ -152,6 +156,8 @@ def __init__( self._closing_connections: dict[int, TransportHandle] = {} self._pending_task_cancellations: dict[int, Any] = {} self._capacity_waiting = False + self._graceful_connections: set[int] = set() + self._graceful_closers: dict[int, Callable[[], None]] = {} @property def address(self) -> tuple[str, int]: @@ -322,10 +328,17 @@ def _finish_close( self._cancelled_task_ids.add(id(task)) elif task is not current_task: try: - self._runtime.resume_task(task) + closer = self._graceful_closers.get(identity) + if closer is not None: + closer() + else: + self._runtime.resume_task(task) except BaseException: pass - if task is not current_task: + if ( + task is not current_task + and not (not owner_thread and identity in self._graceful_connections) + ): self._connections.pop(identity, None) self._close_or_retain(connection, current_task) @@ -438,6 +451,8 @@ def _connection_finished( """Release a completed connection without losing failed-close ownership.""" previous_count = self.owned_connection_count entry = self._connections.pop(id(connection), None) + self._graceful_connections.discard(id(connection)) + self._graceful_closers.pop(id(connection), None) owned_task = entry[1] if entry is not None else task if owned_task in self._owned_tasks: self._owned_tasks.remove(owned_task) diff --git a/tests/test_http2.py b/tests/test_http2.py new file mode 100644 index 0000000..4a853cd --- /dev/null +++ b/tests/test_http2.py @@ -0,0 +1,276 @@ +import builtins +import socket +import threading +import unittest +from unittest.mock import patch + +from h2.config import H2Configuration +from h2.connection import H2Connection +from h2.events import ConnectionTerminated, DataReceived, ResponseReceived, StreamEnded + +from SmallPackage import SmallOS, Unix + +from smallserver import HTTP2Config, Response, SmallServer +from smallserver.errors import ServerConfigurationError +from smallserver.http2 import H2Protocol + + +class HTTP2ProtocolTests(unittest.TestCase): + def _pair(self, config=None): + client = H2Connection( + config=H2Configuration(client_side=True, header_encoding="utf-8") + ) + server = H2Protocol(config) + client.initiate_connection() + server_bytes = server.initiate() + server.receive_data(client.data_to_send()) + client.receive_data(server_bytes + server.flush()) + return client, server + + def test_dependency_is_lazy_and_missing_extra_is_actionable(self): + original = builtins.__import__ + + def reject_h2(name, *args, **kwargs): + if name == "h2" or name.startswith("h2."): + raise ImportError("missing") + return original(name, *args, **kwargs) + + with patch("builtins.__import__", side_effect=reject_h2): + with self.assertRaisesRegex(ServerConfigurationError, "smallserver\\[http2\\]"): + H2Protocol() + + def test_prior_knowledge_request_uses_shared_values_and_response(self): + client, server = self._pair() + client.send_headers( + 1, + [ + (":method", "POST"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/echo"), + ("content-type", "text/plain"), + ], + ) + client.send_data(1, b"hello", end_stream=True) + ready = server.receive_data(client.data_to_send()) + self.assertEqual(len(ready), 1) + self.assertEqual(ready[0].request.version, "HTTP/2") + self.assertEqual(ready[0].request.body, b"hello") + self.assertEqual(ready[0].request.headers["host"], "localhost") + + self.assertTrue(server.queue_response(1, Response.text("world"))) + events = client.receive_data(server.flush()) + self.assertTrue(any(isinstance(event, ResponseReceived) for event in events)) + self.assertEqual( + b"".join(event.data for event in events if isinstance(event, DataReceived)), + b"world", + ) + self.assertTrue(any(isinstance(event, StreamEnded) for event in events)) + + def test_multiplexed_streams_can_finish_out_of_order(self): + client, server = self._pair() + for stream_id, path in ((1, "/slow"), (3, "/fast")): + client.send_headers( + stream_id, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", path), + ], + end_stream=True, + ) + ready = server.receive_data(client.data_to_send()) + self.assertEqual([item.stream_id for item in ready], [1, 3]) + server.queue_response(3, Response.text("fast")) + first = client.receive_data(server.flush()) + self.assertTrue(any(isinstance(event, StreamEnded) and event.stream_id == 3 for event in first)) + server.queue_response(1, Response.text("slow")) + second = client.receive_data(server.flush()) + self.assertTrue(any(isinstance(event, StreamEnded) and event.stream_id == 1 for event in second)) + + def test_request_and_response_limits_reset_streams_without_unbounded_buffers(self): + config = HTTP2Config( + max_body_bytes=4, + max_connection_buffer_bytes=8, + max_response_body_bytes=4, + max_pending_output_bytes=8, + ) + client, server = self._pair(config) + client.send_headers( + 1, + [ + (":method", "POST"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/"), + ], + ) + client.send_data(1, b"12345", end_stream=True) + self.assertEqual(server.receive_data(client.data_to_send()), ()) + client.receive_data(server.flush()) + self.assertEqual(server.active_stream_count, 0) + self.assertEqual(server.pending_output_bytes, 0) + + def test_malformed_preface_is_a_connection_error(self): + server = H2Protocol() + server.initiate() + with self.assertRaisesRegex(ValueError, "client preface"): + server.receive_data(b"NOT HTTP/2") + + def test_peer_reset_is_reported_once_for_handler_cancellation(self): + client, server = self._pair() + client.send_headers( + 1, + [ + (":method", "POST"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/work"), + ], + ) + server.receive_data(client.data_to_send()) + client.reset_stream(1) + server.receive_data(client.data_to_send()) + self.assertEqual(server.take_cancelled_streams(), (1,)) + self.assertEqual(server.take_cancelled_streams(), ()) + + +class HTTP2ServerIntegrationTests(unittest.TestCase): + _pair = HTTP2ProtocolTests._pair + + def test_prior_knowledge_multiplexing_and_graceful_goaway(self): + runtime = SmallOS().setKernel(Unix()) + app = SmallServer() + + @app.get("/one") + async def one(request): + return Response.text("one") + + @app.get("/two") + async def two(request): + return Response.text("two") + + try: + server = app.serve( + runtime, host="127.0.0.1", port=0, protocol="http2" + ) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + + bodies = {1: bytearray(), 3: bytearray()} + ended = set() + terminated = [] + errors = [] + + def client_work(): + try: + client = H2Connection( + config=H2Configuration( + client_side=True, header_encoding="utf-8" + ) + ) + client.initiate_connection() + with socket.create_connection( + ("127.0.0.1", server.port), timeout=3 + ) as connection: + connection.sendall(client.data_to_send()) + for stream_id, path in ((1, "/one"), (3, "/two")): + client.send_headers( + stream_id, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", path), + ], + end_stream=True, + ) + connection.sendall(client.data_to_send()) + while len(ended) < 2: + data = connection.recv(65535) + if not data: + raise RuntimeError("HTTP/2 connection ended early") + for event in client.receive_data(data): + if isinstance(event, DataReceived): + bodies[event.stream_id].extend(event.data) + client.acknowledge_received_data( + event.flow_controlled_length, event.stream_id + ) + elif isinstance(event, StreamEnded): + ended.add(event.stream_id) + pending = client.data_to_send() + if pending: + connection.sendall(pending) + server.close() + while True: + data = connection.recv(65535) + if not data: + break + terminated.extend( + event + for event in client.receive_data(data) + if isinstance(event, ConnectionTerminated) + ) + except BaseException as exc: + errors.append(exc) + try: + server.close() + except BaseException: + pass + + worker = threading.Thread(target=client_work, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=3) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertEqual(bytes(bodies[1]), b"one") + self.assertEqual(bytes(bodies[3]), b"two") + self.assertTrue(terminated) + self.assertTrue(server.finished) + self.assertIsNone(server.failure) + + def test_large_response_respects_flow_control(self): + body = b"x" * 100_000 + config = HTTP2Config( + max_response_body_bytes=len(body), + max_pending_output_bytes=len(body), + ) + client, server = self._pair(config) + client.send_headers( + 1, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/"), + ], + end_stream=True, + ) + server.receive_data(client.data_to_send()) + server.queue_response(1, Response(body=body)) + received = bytearray() + ended = False + for _ in range(20): + events = client.receive_data(server.flush()) + for event in events: + if isinstance(event, DataReceived): + received.extend(event.data) + client.acknowledge_received_data( + event.flow_controlled_length, event.stream_id + ) + elif isinstance(event, StreamEnded): + ended = True + updates = client.data_to_send() + if updates: + server.receive_data(updates) + if ended: + break + self.assertTrue(ended) + self.assertEqual(bytes(received), body) + self.assertEqual(server.pending_output_bytes, 0) + + +if __name__ == "__main__": + unittest.main() From ada2a477669b9cfa06199b7d7e11c6496d0174b6 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 01:44:56 -0500 Subject: [PATCH 02/14] docs: document HTTP/2 prior-knowledge usage --- README.md | 45 ++++++++++++++++++++++++-- examples/http2_prior_knowledge.py | 27 ++++++++++++++++ guide/http2.md | 53 +++++++++++++++++++++++++++++++ 3 files changed, 122 insertions(+), 3 deletions(-) create mode 100644 examples/http2_prior_knowledge.py create mode 100644 guide/http2.md diff --git a/README.md b/README.md index 3c7598b..3dd7769 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,9 @@ # SmallServer SmallServer is a SmallOS-native web framework in early development. It provides -a bounded HTTP/1.1 server, static async routing for GET, POST, PUT, PATCH, and -DELETE, and explicit escape hatches for blocking and asyncio-native libraries. +bounded HTTP/1.1 and optional cleartext HTTP/2 servers, static async routing for +GET, POST, PUT, PATCH, and DELETE, and explicit escape hatches for blocking and +asyncio-native libraries. ## Current scope @@ -18,7 +19,8 @@ The current package provides an HTTP/1.1 baseline over a SmallOS runtime. It can - parse one `Content-Length` HTTP/1.1 request per connection and close after its response. -Keep-alive/pipelining, TLS, path parameters, and HTTP/2 are not implemented yet. +HTTP/1.1 keep-alive/pipelining, TLS, and path parameters are not implemented +yet. HTTP/2 currently supports cleartext prior knowledge only. ## Install for development @@ -28,6 +30,12 @@ python3 -m pip install -e . python3 -m unittest discover -s tests -v ``` +Install the bounded optional hyper-h2 4.x integration when serving HTTP/2: + +```bash +python3 -m pip install -e '.[http2]' +``` + SmallOS is installed from the canonical `master` branch in `requirements.txt`. It owns scheduling, socket readiness, and foreign execution adapters. @@ -65,6 +73,37 @@ async def health(request): app.listen(host="127.0.0.1", port=8000) ``` +## Cleartext HTTP/2 + +HTTP/2 uses the same routes, `Request`, `Response`, and SmallOS runtime. Select +it explicitly on a listener; protocol auto-detection and h2c upgrade are not +performed: + +```python +from smallserver import HTTP2Config, Response, SmallServer + +app = SmallServer() + +@app.get("/health") +async def health(request): + return Response.json({"status": "ok", "protocol": request.version}) + +app.listen( + host="127.0.0.1", + port=8000, + protocol="http2", + http2_config=HTTP2Config(max_concurrent_streams=32), +) +``` + +Run `python3 examples/http2_prior_knowledge.py`, then use an HTTP/2-capable +client such as `curl --http2-prior-knowledge http://127.0.0.1:8000/health`. +See [the HTTP/2 guide](guide/http2.md) for limits and lifecycle behavior. + +TLS/ALPN is explicitly deferred: SmallOS does not yet expose a server-side TLS +kernel capability. SmallServer does not bypass that boundary with direct +`ssl` or `socket` access. + Managed `listen()` blocks and catches Ctrl-C after closing its listener, wakeup channel, connections, and server tasks. It returns the closed `ServerHandle`, whose cached `address` and `port` remain available for diagnostics. Each diff --git a/examples/http2_prior_knowledge.py b/examples/http2_prior_knowledge.py new file mode 100644 index 0000000..a029a39 --- /dev/null +++ b/examples/http2_prior_knowledge.py @@ -0,0 +1,27 @@ +"""Run SmallServer's cleartext prior-knowledge HTTP/2 demo.""" + +from smallserver import HTTP2Config, Response, SmallServer + + +app = SmallServer() + + +@app.get("/health") +async def health(request): + return Response.json({"status": "ok", "protocol": request.version}) + + +@app.post("/echo") +async def echo(request): + return Response(body=request.body, headers={"Content-Type": "application/octet-stream"}) + + +if __name__ == "__main__": + print("HTTP/2 prior-knowledge server: http://127.0.0.1:8000") + print("Try: curl --http2-prior-knowledge http://127.0.0.1:8000/health") + app.listen( + host="127.0.0.1", + port=8000, + protocol="http2", + http2_config=HTTP2Config(max_concurrent_streams=32), + ) diff --git a/guide/http2.md b/guide/http2.md new file mode 100644 index 0000000..88747f8 --- /dev/null +++ b/guide/http2.md @@ -0,0 +1,53 @@ +# Cleartext HTTP/2 + +SmallServer can serve HTTP/2 with cleartext prior knowledge. The feature uses +hyper-h2 as a lazy, optional sans-I/O protocol engine while SmallOS continues +to own task scheduling and all network readiness. + +## Install and run + +```bash +python3 -m pip install -r requirements.txt +python3 -m pip install -e '.[http2]' +python3 examples/http2_prior_knowledge.py +``` + +In another terminal: + +```bash +curl --http2-prior-knowledge http://127.0.0.1:8000/health +curl --http2-prior-knowledge --data-binary hello http://127.0.0.1:8000/echo +``` + +Select HTTP/2 with `protocol="http2"` on either `listen()` or `serve()`. +HTTP/1.1 remains the default and never imports hyper-h2. A missing or +incompatible optional dependency is rejected before SmallServer binds a port. + +## Concurrency and limits + +Each TCP connection has one protocol state and one writer task. Complete +request streams are dispatched in separate SmallOS tasks, so one stream can +wait on a bounded execution adapter while unrelated streams complete. The +single writer preserves frame ordering and observes peer flow-control windows. + +`HTTP2Config` bounds concurrent streams, decoded and compressed header sizes, +per-stream and per-connection request buffering, response buffering, and frame +size. Requests and responses use the same immutable `Request`, `Headers`, and +`Response` values as HTTP/1.1. The request version is `"HTTP/2"`. + +Peer stream resets cancel the associated handler task without stopping other +streams. Protocol/resource violations reset the affected stream when possible. +Connection shutdown emits GOAWAY and then releases the connection through the +SmallOS kernel transport. + +## Current protocol boundary + +Only cleartext prior knowledge is supported. SmallServer does not implement an +HTTP/1.1 `Upgrade: h2c` transition and does not infer the protocol from bytes. +Configure one listener for one protocol. + +TLS with ALPN `h2` is deferred because SmallOS does not currently expose a +server-side TLS kernel capability. SmallServer intentionally does not import +or call platform `ssl` or `socket` APIs to work around that missing boundary. +When the kernel gains that capability, TLS/ALPN negotiation can be added +without changing route handlers or response values. From c438147b561008d1c693131fb3c485acb0ed8906 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 01:46:48 -0500 Subject: [PATCH 03/14] fix: retain HTTP/2 graceful cleanup fallback --- smallserver/server.py | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/smallserver/server.py b/smallserver/server.py index 4c313ca..7df4dc5 100644 --- a/smallserver/server.py +++ b/smallserver/server.py @@ -317,6 +317,7 @@ def _finish_close( self._cancelled_task_ids.add(identity) for identity, (connection, task) in list(self._connections.items()): + graceful_requested = False if owner_thread: if ( task is not current_task @@ -327,17 +328,24 @@ def _finish_close( if self._cancel_or_retain_task(task): self._cancelled_task_ids.add(id(task)) elif task is not current_task: - try: - closer = self._graceful_closers.get(identity) - if closer is not None: + closer = self._graceful_closers.get(identity) + if closer is not None: + try: closer() - else: + graceful_requested = True + except BaseException: + try: + self._runtime.resume_task(task) + except BaseException: + pass + else: + try: self._runtime.resume_task(task) - except BaseException: - pass + except BaseException: + pass if ( task is not current_task - and not (not owner_thread and identity in self._graceful_connections) + and not (not owner_thread and graceful_requested) ): self._connections.pop(identity, None) self._close_or_retain(connection, current_task) From 033e968b840a38f35a6d0c8db747d655efb60e77 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 02:11:22 -0500 Subject: [PATCH 04/14] fix: harden HTTP/2 stream and lifecycle bounds --- smallserver/app.py | 183 +++++++++++++++-- smallserver/http2.py | 281 +++++++++++++++++--------- tests/test_http2.py | 472 +++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 805 insertions(+), 131 deletions(-) diff --git a/smallserver/app.py b/smallserver/app.py index d413545..c039694 100644 --- a/smallserver/app.py +++ b/smallserver/app.py @@ -30,16 +30,21 @@ Handler = Callable[[Request], Awaitable[Response]] _METHODS = frozenset({"GET", "POST", "PUT", "PATCH", "DELETE"}) _HTTP2_WRITER_SIGNAL = 30 +_HTTP2_SHUTDOWN_SIGNAL = 29 class _H2ConnectionState: def __init__(self, protocol: H2Protocol) -> None: self.protocol = protocol self.writer_task: Any = None + self.shutdown_task: Any = None + self.watchdog_task: Any = None self.handlers: dict[int, Any] = {} self.closing = False self.shutdown_requested = False self.close_error_code = 0 + self.activity_epoch = 0 + self.failure: BaseException | None = None def wake_writer(self) -> None: writer = self.writer_task @@ -50,6 +55,13 @@ def wake_writer(self) -> None: def request_shutdown(self) -> None: self.shutdown_requested = True self.wake_writer() + shutdown_task = self.shutdown_task + if shutdown_task is not None and not getattr(shutdown_task, "done", False): + if shutdown_task.acceptSignal(_HTTP2_SHUTDOWN_SIGNAL) != 0: + raise RuntimeError("HTTP/2 shutdown signal failed") + + def mark_activity(self) -> None: + self.activity_epoch += 1 class _NoThreadLock: @@ -617,12 +629,12 @@ async def _http2_connection_loop( ) -> None: from SmallPackage import SmallTask - protocol = H2Protocol(handle._protocol_config) - state = _H2ConnectionState(protocol) + protocol: H2Protocol | None = None + state: _H2ConnectionState | None = None primary_error: BaseException | None = None - handle._graceful_connections.add(id(client)) - handle._graceful_closers[id(client)] = state.request_shutdown try: + protocol = H2Protocol(handle._protocol_config) + state = _H2ConnectionState(protocol) await handle._transport.send_all(task, client, protocol.initiate()) writer = SmallTask( handle._config.connection_priority, @@ -631,14 +643,32 @@ async def _http2_connection_loop( name="smallserver-http2-writer", ) state.writer_task = writer - handle._owned_tasks.append(writer) - handle._runtime.fork(writer) + shutdown_task = SmallTask( + handle._config.connection_priority + 1, + self._http2_shutdown_enforcer, + args=(handle, client, state), + name="smallserver-http2-shutdown-enforcer", + ) + state.shutdown_task = shutdown_task + watchdog = SmallTask( + handle._config.connection_priority + 1, + self._http2_watchdog, + args=(handle, client, state), + name="smallserver-http2-watchdog", + ) + state.watchdog_task = watchdog + child_tasks = [writer, shutdown_task, watchdog] + handle._owned_tasks.extend(child_tasks) + handle._runtime.fork(child_tasks) + handle._graceful_connections.add(id(client)) + handle._graceful_closers[id(client)] = state.request_shutdown while not handle.closed and not protocol.remote_closed: chunk = await handle._transport.recv( task, client, handle._config.receive_chunk_bytes ) if not chunk: break + state.mark_activity() try: ready = protocol.receive_data(chunk) except Exception as protocol_error: @@ -666,21 +696,33 @@ async def _http2_connection_loop( primary_error = exc raise finally: - state.closing = True - for handler in tuple(state.handlers.values()): - handle._cancel_or_retain_task(handler) - state.handlers.clear() - if state.writer_task is not None: - state.wake_writer() - handle._cancel_or_retain_task(state.writer_task) - if state.writer_task in handle._owned_tasks: - handle._owned_tasks.remove(state.writer_task) - try: - goaway = protocol.close(state.close_error_code) - if goaway: - await handle._transport.send_all(task, client, goaway) - except BaseException: - pass + if primary_error is None and state is not None: + primary_error = state.failure + if state is not None: + state.closing = True + for handler in tuple(state.handlers.values()): + handle._cancel_or_retain_task(handler) + state.handlers.clear() + for child in ( + state.writer_task, + state.shutdown_task, + state.watchdog_task, + ): + if child is not None and child is not task: + handle._cancel_or_retain_task(child) + if child in handle._owned_tasks: + handle._owned_tasks.remove(child) + if protocol is not None and ( + primary_error is None or isinstance(primary_error, Exception) + ): + try: + goaway = protocol.close( + state.close_error_code if state is not None else 1 + ) + if goaway and not client.closed: + await handle._transport.send_all(task, client, goaway) + except BaseException: + pass handle._connection_finished(task, client, primary_error) async def _http2_handler( @@ -691,14 +733,18 @@ async def _http2_handler( stream_id: int, request: Request, ) -> None: + response_queued = False try: try: response = await self.dispatch(request) except Exception: response = Response.text("internal server error", status=500) state.protocol.queue_response(stream_id, response) + response_queued = True state.wake_writer() finally: + if not response_queued: + state.protocol.drop_stream(stream_id) state.handlers.pop(stream_id, None) if task in handle._owned_tasks: handle._owned_tasks.remove(task) @@ -729,6 +775,101 @@ async def _http2_writer_loop( if not payload: break await handle._transport.send_all(task, client, payload) + except Exception as error: + state.failure = error + state.closing = True + handle._listener_failed(error, task) + if not handle._transport.close_safely(client): + close_error = client.close_error or RuntimeError( + "kernel connection close failed" + ) + handle._connection_close_failed(close_error, task, error) + raise + finally: + if task in handle._owned_tasks: + handle._owned_tasks.remove(task) + + async def _http2_shutdown_enforcer( + self, + task: Any, + handle: ServerHandle, + client: TransportHandle, + state: _H2ConnectionState, + ) -> None: + try: + await task.wait_signal(_HTTP2_SHUTDOWN_SIGNAL) + await task.yield_now() + if client.closed: + return + state.closing = True + if not handle._transport.close_safely(client): + error = client.close_error or RuntimeError( + "kernel connection close failed" + ) + handle._connection_close_failed(error, task, state.failure) finally: if task in handle._owned_tasks: handle._owned_tasks.remove(task) + + async def _http2_watchdog( + self, + task: Any, + handle: ServerHandle, + client: TransportHandle, + state: _H2ConnectionState, + ) -> None: + config = state.protocol.config + try: + handshake_elapsed = 0.0 + while not state.protocol.preface_received and not state.closing: + interval = min(1.0, config.handshake_timeout - handshake_elapsed) + await task.sleep(interval) + handshake_elapsed += interval + if handshake_elapsed >= config.handshake_timeout: + self._http2_force_close( + task, + handle, + client, + state, + TimeoutError("HTTP/2 client preface timed out"), + ) + return + + observed_epoch = state.activity_epoch + idle_elapsed = 0.0 + while not state.closing: + interval = min(1.0, config.idle_timeout - idle_elapsed) + await task.sleep(interval) + if observed_epoch != state.activity_epoch: + observed_epoch = state.activity_epoch + idle_elapsed = 0.0 + continue + idle_elapsed += interval + if idle_elapsed >= config.idle_timeout: + self._http2_force_close( + task, + handle, + client, + state, + TimeoutError("HTTP/2 connection was idle too long"), + ) + return + finally: + if task in handle._owned_tasks: + handle._owned_tasks.remove(task) + + @staticmethod + def _http2_force_close( + task: Any, + handle: ServerHandle, + client: TransportHandle, + state: _H2ConnectionState, + error: BaseException, + ) -> None: + state.failure = error + state.closing = True + if not handle._transport.close_safely(client): + close_error = client.close_error or RuntimeError( + "kernel connection close failed" + ) + handle._connection_close_failed(close_error, task, error) diff --git a/smallserver/http2.py b/smallserver/http2.py index 49dc184..ba5467b 100644 --- a/smallserver/http2.py +++ b/smallserver/http2.py @@ -25,11 +25,22 @@ class HTTP2Config: max_pending_output_bytes: int = 4 * 1024 * 1024 max_response_body_bytes: int = 2 * 1024 * 1024 max_frame_size: int = 16 * 1024 + handshake_timeout: float = 10.0 + idle_timeout: float = 60.0 def __post_init__(self) -> None: - for name, value in self.__dict__.items(): + integer_fields = { + name: value + for name, value in self.__dict__.items() + if name not in {"handshake_timeout", "idle_timeout"} + } + for name, value in integer_fields.items(): if type(value) is not int or value <= 0: raise ValueError("{} must be a positive integer".format(name)) + for name in ("handshake_timeout", "idle_timeout"): + value = getattr(self, name) + if not isinstance(value, (int, float)) or isinstance(value, bool) or value <= 0: + raise ValueError("{} must be a positive number".format(name)) if not 16_384 <= self.max_frame_size <= 16_777_215: raise ValueError("max_frame_size must be between 16384 and 16777215") if self.max_body_bytes > self.max_connection_buffer_bytes: @@ -56,6 +67,8 @@ class _InboundStream: path: str headers: Headers body: bytearray + expected_content_length: int | None + dispatched: bool = False @dataclass @@ -65,86 +78,91 @@ class _OutboundStream: class _FrameBudget: - """Account compressed header blocks without implementing frame semantics.""" + """Split complete frames and enforce wire-level allocation bounds.""" def __init__(self, config: HTTP2Config) -> None: self._config = config - self._preface = bytearray() - self._header = bytearray() - self._remaining = 0 - self._frame_type = 0 - self._frame_stream = 0 - self._frame_flags = 0 + self._buffer = bytearray() + self._preface_received = False self._header_stream: int | None = None self._header_bytes = 0 - def feed(self, data: bytes) -> None: - view = memoryview(data) - offset = 0 - if len(self._preface) < len(HTTP2_CLIENT_PREFACE): - needed = len(HTTP2_CLIENT_PREFACE) - len(self._preface) - take = min(needed, len(view)) - self._preface.extend(view[:take]) - offset += take - expected = HTTP2_CLIENT_PREFACE[: len(self._preface)] - if bytes(self._preface) != expected: + def feed(self, data: bytes) -> tuple[bytes, ...]: + self._buffer.extend(data) + chunks: list[bytes] = [] + if not self._preface_received: + prefix_length = min(len(self._buffer), len(HTTP2_CLIENT_PREFACE)) + if bytes(self._buffer[:prefix_length]) != HTTP2_CLIENT_PREFACE[:prefix_length]: raise ValueError("invalid HTTP/2 client preface") - if offset == len(view): - return - - while offset < len(view): - if self._remaining == 0: - needed = 9 - len(self._header) - take = min(needed, len(view) - offset) - self._header.extend(view[offset : offset + take]) - offset += take - if len(self._header) < 9: - return - length = int.from_bytes(self._header[:3], "big") - if length > self._config.max_frame_size: - raise ValueError("HTTP/2 frame exceeds configured maximum") - self._frame_type = self._header[3] - self._frame_flags = self._header[4] - self._frame_stream = int.from_bytes(self._header[5:9], "big") & 0x7FFFFFFF - self._header.clear() - self._remaining = length - if length == 0: - self._finish_frame() - continue - - take = min(self._remaining, len(view) - offset) - if self._frame_type in (0x1, 0x9): - self._account_header_bytes(take) - self._remaining -= take - offset += take - if self._remaining == 0: - self._finish_frame() - - def _account_header_bytes(self, count: int) -> None: - if self._frame_type == 0x1 and self._header_stream is None: - self._header_stream = self._frame_stream - self._header_bytes = 0 - self._header_bytes += count - if self._header_bytes > self._config.max_compressed_header_bytes: - raise ValueError("HTTP/2 compressed header block is too large") - - def _finish_frame(self) -> None: - if self._frame_type in (0x1, 0x9) and self._frame_flags & 0x4: - self._header_stream = None - self._header_bytes = 0 + if len(self._buffer) < len(HTTP2_CLIENT_PREFACE): + return () + chunks.append(bytes(self._buffer[: len(HTTP2_CLIENT_PREFACE)])) + del self._buffer[: len(HTTP2_CLIENT_PREFACE)] + self._preface_received = True + + while len(self._buffer) >= 9: + length = int.from_bytes(self._buffer[:3], "big") + if length > self._config.max_frame_size: + raise ValueError("HTTP/2 frame exceeds configured maximum") + frame_length = 9 + length + if len(self._buffer) < frame_length: + break + frame_type = self._buffer[3] + flags = self._buffer[4] + stream_id = int.from_bytes(self._buffer[5:9], "big") & 0x7FFFFFFF + if frame_type == 0x1: + if self._header_stream is not None: + raise ValueError("interleaved HTTP/2 header blocks are invalid") + self._header_stream = stream_id + self._header_bytes = length + elif frame_type == 0x9: + if self._header_stream != stream_id: + raise ValueError("invalid HTTP/2 continuation stream") + self._header_bytes += length + if self._header_bytes > self._config.max_compressed_header_bytes: + raise ValueError("HTTP/2 compressed header block is too large") + if frame_type in (0x1, 0x9) and flags & 0x4: + self._header_stream = None + self._header_bytes = 0 + chunks.append(bytes(self._buffer[:frame_length])) + del self._buffer[:frame_length] + return tuple(chunks) + + @property + def preface_received(self) -> bool: + return self._preface_received def require_http2() -> None: """Fail clearly without importing hyper-h2 on HTTP/1.1 paths.""" try: import h2 # type: ignore[import-not-found] - except ImportError as exc: + from h2.config import H2Configuration # noqa: F401 + from h2.connection import H2Connection # noqa: F401 + from h2.errors import ErrorCodes # noqa: F401 + from h2.events import ( # noqa: F401 + ConnectionTerminated, + DataReceived, + RemoteSettingsChanged, + RequestReceived, + StreamEnded, + StreamReset, + TrailersReceived, + WindowUpdated, + ) + from h2.settings import SettingCodes # noqa: F401 + except (ImportError, AttributeError) as exc: raise ServerConfigurationError( - "HTTP/2 requires the optional dependency; install smallserver[http2]" + "HTTP/2 requires a complete hyper-h2 4.x installation; " + "install smallserver[http2]" ) from exc version = getattr(h2, "__version__", "") if not isinstance(version, str) or not version.startswith("4."): raise ServerConfigurationError("HTTP/2 requires hyper-h2 version 4.x") + if not callable(getattr(H2Connection, "_begin_new_stream", None)): + raise ServerConfigurationError( + "installed hyper-h2 4.x lacks required stream validation support" + ) class H2Protocol: @@ -182,7 +200,22 @@ def __init__(self, config: HTTP2Config | None = None) -> None: validate_inbound_headers=True, normalize_inbound_headers=False, ) - self.connection = H2Connection(config=h2_config) + class _SmallServerH2Connection(H2Connection): + def _begin_new_stream(self, stream_id: Any, allowed_ids: Any) -> Any: + stream = super()._begin_new_stream(stream_id, allowed_ids) + initializer = getattr(stream, "_initialize_content_length", None) + if not callable(initializer): + raise ServerConfigurationError( + "installed hyper-h2 4.x lacks required stream " + "validation support" + ) + # hyper-h2 treats content-length mismatch as connection-fatal. + # SmallServer owns this check so malformed request metadata can + # remain a stream-scoped error as required by RFC 9113. + stream._initialize_content_length = lambda headers: None + return stream + + self.connection = _SmallServerH2Connection(config=h2_config) self.connection.local_settings[SettingCodes.MAX_CONCURRENT_STREAMS] = ( self.config.max_concurrent_streams ) @@ -223,34 +256,44 @@ def active_stream_count(self) -> int: def pending_output_bytes(self) -> int: return self._pending_output_bytes + @property + def buffered_request_bytes(self) -> int: + return self._buffered_request_bytes + + @property + def preface_received(self) -> bool: + return self._frames.preface_received + def initiate(self) -> bytes: self.connection.initiate_connection() return self.connection.data_to_send() def receive_data(self, data: bytes) -> tuple[H2ReadyRequest, ...]: - self._frames.feed(data) - events = self.connection.receive_data(data) ready: list[H2ReadyRequest] = [] - for event in events: - if isinstance(event, self._events["request"]): - self._request_received(event.stream_id, event.headers) - elif isinstance(event, self._events["data"]): - self._data_received( - event.stream_id, event.data, event.flow_controlled_length - ) - elif isinstance(event, self._events["ended"]): - completed = self._stream_ended(event.stream_id) - if completed is not None: - ready.append(completed) - elif isinstance(event, self._events["reset"]): - self._cancelled_streams.append(event.stream_id) - self.drop_stream(event.stream_id) - elif isinstance(event, self._events["trailers"]): - self._reset_stream(event.stream_id, self._error_codes.PROTOCOL_ERROR) - elif isinstance(event, self._events["terminated"]): - self.remote_closed = True - elif isinstance(event, (self._events["window"], self._events["settings"])): - pass + for wire_chunk in self._frames.feed(data): + events = self.connection.receive_data(wire_chunk) + for event in events: + if isinstance(event, self._events["request"]): + self._request_received(event.stream_id, event.headers) + elif isinstance(event, self._events["data"]): + self._data_received( + event.stream_id, event.data, event.flow_controlled_length + ) + elif isinstance(event, self._events["ended"]): + completed = self._stream_ended(event.stream_id) + if completed is not None: + ready.append(completed) + elif isinstance(event, self._events["reset"]): + self._cancelled_streams.append(event.stream_id) + self.drop_stream(event.stream_id) + elif isinstance(event, self._events["trailers"]): + self._reset_stream(event.stream_id, self._error_codes.PROTOCOL_ERROR) + elif isinstance(event, self._events["terminated"]): + self.remote_closed = True + elif isinstance( + event, (self._events["window"], self._events["settings"]) + ): + pass return tuple(ready) def take_cancelled_streams(self) -> tuple[int, ...]: @@ -263,14 +306,20 @@ def _request_received(self, stream_id: int, raw_headers: Any) -> None: self._reset_stream(stream_id, self._error_codes.REFUSED_STREAM) return try: - method, path, headers = self._decode_request_headers(raw_headers) + method, path, headers, content_length = self._decode_request_headers( + raw_headers + ) except (TypeError, ValueError): self._reset_stream(stream_id, self._error_codes.PROTOCOL_ERROR) return self._active_streams.add(stream_id) - self._inbound[stream_id] = _InboundStream(method, path, headers, bytearray()) + self._inbound[stream_id] = _InboundStream( + method, path, headers, bytearray(), content_length + ) - def _decode_request_headers(self, raw_headers: Any) -> tuple[str, str, Headers]: + def _decode_request_headers( + self, raw_headers: Any + ) -> tuple[str, str, Headers, int | None]: if len(raw_headers) > self.config.max_header_count: raise ValueError("too many HTTP/2 request headers") decoded_size = 0 @@ -306,6 +355,25 @@ def _decode_request_headers(self, raw_headers: Any) -> tuple[str, str, Headers]: raise ValueError("HTTP/2 CONNECT is not supported") if not all(pseudo.get(name) for name in (":method", ":scheme", ":path")): raise ValueError("missing required HTTP/2 pseudo-header") + method = pseudo[":method"] + path = pseudo[":path"] + if not method or any( + not ( + character.isascii() + and ( + character.isalnum() + or character in "!#$%&'*+-.^_`|~" + ) + ) + for character in method + ): + raise ValueError("invalid HTTP/2 method") + if ( + not path.startswith("/") + or "#" in path + or any(not 0x21 <= ord(character) <= 0x7E for character in path) + ): + raise ValueError("invalid HTTP/2 origin-form path") authority = pseudo.get(":authority") existing_host = any(name == "host" for name, _value in regular) if authority and existing_host: @@ -316,7 +384,16 @@ def _decode_request_headers(self, raw_headers: Any) -> tuple[str, str, Headers]: regular.append(("host", authority)) if cookies: regular.append(("cookie", "; ".join(cookies))) - return pseudo[":method"], pseudo[":path"], Headers(regular) + headers = Headers(regular) + content_length: int | None = None + raw_length = headers.get("content-length") + if raw_length is not None: + if not raw_length.isascii() or not raw_length.isdecimal(): + raise ValueError("invalid HTTP/2 content-length") + content_length = int(raw_length) + if content_length > self.config.max_body_bytes: + raise ValueError("HTTP/2 content-length exceeds configured maximum") + return method, path, headers, content_length def _data_received(self, stream_id: int, data: bytes, flow_length: int) -> None: self.connection.acknowledge_received_data(flow_length, stream_id) @@ -327,6 +404,10 @@ def _data_received(self, stream_id: int, data: bytes, flow_length: int) -> None: next_connection_size = self._buffered_request_bytes + len(data) if ( next_stream_size > self.config.max_body_bytes + or ( + stream.expected_content_length is not None + and next_stream_size > stream.expected_content_length + ) or next_connection_size > self.config.max_connection_buffer_bytes ): self._reset_stream(stream_id, self._error_codes.ENHANCE_YOUR_CALM) @@ -335,10 +416,16 @@ def _data_received(self, stream_id: int, data: bytes, flow_length: int) -> None: self._buffered_request_bytes = next_connection_size def _stream_ended(self, stream_id: int) -> H2ReadyRequest | None: - stream = self._inbound.pop(stream_id, None) + stream = self._inbound.get(stream_id) if stream is None: return None - self._buffered_request_bytes -= len(stream.body) + if ( + stream.expected_content_length is not None + and len(stream.body) != stream.expected_content_length + ): + self._reset_stream(stream_id, self._error_codes.PROTOCOL_ERROR) + return None + stream.dispatched = True self.last_processed_stream_id = max(self.last_processed_stream_id, stream_id) request = Request( stream.method, @@ -352,6 +439,7 @@ def _stream_ended(self, stream_id: int) -> H2ReadyRequest | None: def queue_response(self, stream_id: int, response: Response) -> bool: if stream_id not in self._active_streams: return False + self._release_inbound(stream_id) body_size = len(response.body) if ( body_size > self.config.max_response_body_bytes @@ -443,9 +531,7 @@ def _start_response(self, stream_id: int, response: Response) -> None: self._active_streams.discard(stream_id) def drop_stream(self, stream_id: int) -> None: - inbound = self._inbound.pop(stream_id, None) - if inbound is not None: - self._buffered_request_bytes -= len(inbound.body) + self._release_inbound(stream_id) outbound = self._outbound.pop(stream_id, None) if outbound is not None: self._pending_output_bytes -= len(outbound.body) - outbound.offset @@ -458,6 +544,11 @@ def drop_stream(self, stream_id: int) -> None: self._commands = kept self._active_streams.discard(stream_id) + def _release_inbound(self, stream_id: int) -> None: + inbound = self._inbound.pop(stream_id, None) + if inbound is not None: + self._buffered_request_bytes -= len(inbound.body) + def _reset_stream(self, stream_id: int, error_code: Any) -> None: try: self.connection.reset_stream(stream_id, error_code=error_code) diff --git a/tests/test_http2.py b/tests/test_http2.py index 4a853cd..d932fc6 100644 --- a/tests/test_http2.py +++ b/tests/test_http2.py @@ -4,29 +4,30 @@ import unittest from unittest.mock import patch -from h2.config import H2Configuration -from h2.connection import H2Connection -from h2.events import ConnectionTerminated, DataReceived, ResponseReceived, StreamEnded +try: + from h2.config import H2Configuration + from h2.connection import H2Connection + from h2.events import ( + ConnectionTerminated, + DataReceived, + ResponseReceived, + StreamEnded, + StreamReset, + ) +except ImportError: + H2_AVAILABLE = False +else: + H2_AVAILABLE = True from SmallPackage import SmallOS, Unix from smallserver import HTTP2Config, Response, SmallServer from smallserver.errors import ServerConfigurationError from smallserver.http2 import H2Protocol +from tests.kernel_fakes import FakeKernel -class HTTP2ProtocolTests(unittest.TestCase): - def _pair(self, config=None): - client = H2Connection( - config=H2Configuration(client_side=True, header_encoding="utf-8") - ) - server = H2Protocol(config) - client.initiate_connection() - server_bytes = server.initiate() - server.receive_data(client.data_to_send()) - client.receive_data(server_bytes + server.flush()) - return client, server - +class HTTP2OptionalDependencyTests(unittest.TestCase): def test_dependency_is_lazy_and_missing_extra_is_actionable(self): original = builtins.__import__ @@ -39,6 +40,67 @@ def reject_h2(name, *args, **kwargs): with self.assertRaisesRegex(ServerConfigurationError, "smallserver\\[http2\\]"): H2Protocol() + def test_incomplete_extra_is_rejected_during_preflight(self): + original = builtins.__import__ + + def reject_events(name, *args, **kwargs): + if name == "h2.events": + raise ImportError("broken events module") + return original(name, *args, **kwargs) + + with patch("builtins.__import__", side_effect=reject_events): + with self.assertRaisesRegex(ServerConfigurationError, "complete hyper-h2"): + H2Protocol() + + def test_timeout_configuration_is_finite_and_positive(self): + for values in ( + {"handshake_timeout": 0}, + {"idle_timeout": -1}, + {"idle_timeout": True}, + ): + with self.subTest(values=values): + with self.assertRaises(ValueError): + HTTP2Config(**values) + + def test_dependency_preflight_happens_before_address_resolution(self): + class Runtime: + def __init__(self): + self.kernel = FakeKernel() + + def fork(self, tasks): + return None + + def resume_task(self, task): + return None + + def cancel_task(self, task): + return None + + runtime = Runtime() + with patch( + "smallserver.app.require_http2", + side_effect=ServerConfigurationError("broken HTTP/2 dependency"), + ): + with self.assertRaisesRegex(ServerConfigurationError, "broken"): + SmallServer().serve(runtime, protocol="http2") + self.assertFalse( + any(call[0] == "resolve_passive_address" for call in runtime.kernel.calls) + ) + + +@unittest.skipUnless(H2_AVAILABLE, "install the smallserver[test] HTTP/2 extra") +class HTTP2ProtocolTests(unittest.TestCase): + def _pair(self, config=None): + client = H2Connection( + config=H2Configuration(client_side=True, header_encoding="utf-8") + ) + server = H2Protocol(config) + client.initiate_connection() + server_bytes = server.initiate() + server.receive_data(client.data_to_send()) + client.receive_data(server_bytes + server.flush()) + return client, server + def test_prior_knowledge_request_uses_shared_values_and_response(self): client, server = self._pair() client.send_headers( @@ -135,7 +197,135 @@ def test_peer_reset_is_reported_once_for_handler_cancellation(self): self.assertEqual(server.take_cancelled_streams(), (1,)) self.assertEqual(server.take_cancelled_streams(), ()) + def test_completed_slow_handler_body_remains_in_connection_budget(self): + config = HTTP2Config( + max_body_bytes=4, + max_connection_buffer_bytes=6, + ) + client, server = self._pair(config) + for stream_id, body in ((1, b"1234"), (3, b"5678")): + client.send_headers( + stream_id, + [ + (":method", "POST"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/slow"), + ("content-length", "4"), + ], + ) + client.send_data(stream_id, body, end_stream=True) + ready = server.receive_data(client.data_to_send()) + if stream_id == 1: + self.assertEqual([item.stream_id for item in ready], [1]) + self.assertEqual(server.buffered_request_bytes, 4) + else: + self.assertEqual(ready, ()) + events = client.receive_data(server.flush()) + self.assertTrue( + any(isinstance(event, StreamReset) and event.stream_id == 3 for event in events) + ) + self.assertEqual(server.buffered_request_bytes, 4) + server.queue_response(1, Response.text("done")) + self.assertEqual(server.buffered_request_bytes, 0) + def test_bad_stream_metadata_resets_only_that_stream(self): + client, server = self._pair() + client.send_headers( + 1, + [ + (":method", "POST"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/bad"), + ("content-length", "2"), + ], + ) + client.send_data(1, b"x", end_stream=True) + client.send_headers( + 3, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/good"), + ], + end_stream=True, + ) + ready = server.receive_data(client.data_to_send()) + self.assertEqual([item.stream_id for item in ready], [3]) + events = client.receive_data(server.flush()) + self.assertTrue( + any(isinstance(event, StreamReset) and event.stream_id == 1 for event in events) + ) + + def test_invalid_method_and_origin_form_are_stream_errors(self): + client, server = self._pair() + client.config.validate_outbound_headers = False + for stream_id, method, path in ( + (1, "BAD METHOD", "/bad"), + (3, "GET", "/bad#fragment"), + ): + client.send_headers( + stream_id, + [ + (":method", method), + (":scheme", "http"), + (":authority", "localhost"), + (":path", path), + ], + end_stream=True, + ) + client.send_headers( + 5, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/good"), + ], + end_stream=True, + ) + ready = server.receive_data(client.data_to_send()) + self.assertEqual([item.stream_id for item in ready], [5]) + events = client.receive_data(server.flush()) + self.assertEqual( + {event.stream_id for event in events if isinstance(event, StreamReset)}, + {1, 3}, + ) + + def test_invalid_content_length_is_a_stream_error(self): + client, server = self._pair() + client.send_headers( + 1, + [ + (":method", "POST"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/bad"), + ("content-length", "-1"), + ], + end_stream=True, + ) + client.send_headers( + 3, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/good"), + ], + end_stream=True, + ) + ready = server.receive_data(client.data_to_send()) + self.assertEqual([item.stream_id for item in ready], [3]) + events = client.receive_data(server.flush()) + self.assertTrue( + any(isinstance(event, StreamReset) and event.stream_id == 1 for event in events) + ) + + +@unittest.skipUnless(H2_AVAILABLE, "install the smallserver[test] HTTP/2 extra") class HTTP2ServerIntegrationTests(unittest.TestCase): _pair = HTTP2ProtocolTests._pair @@ -271,6 +461,258 @@ def test_large_response_respects_flow_control(self): self.assertEqual(bytes(received), body) self.assertEqual(server.pending_output_bytes, 0) + def test_writer_send_failure_is_fatal_and_releases_capacity(self): + runtime = SmallOS().setKernel(Unix()) + app = SmallServer() + + @app.get("/fail") + async def fail(request): + return Response.text("response") + + try: + server = app.serve( + runtime, host="127.0.0.1", port=0, protocol="http2" + ) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + original_transport = server._transport + + class FailingWriterTransport: + def __getattr__(self, name): + return getattr(original_transport, name) + + async def send_all(self, task, stream, data): + if getattr(task, "name", "") == "smallserver-http2-writer": + raise RuntimeError("injected HTTP/2 writer failure") + await original_transport.send_all(task, stream, data) + + server._transport = FailingWriterTransport() + errors = [] + + def client_work(): + try: + client = H2Connection(config=H2Configuration(client_side=True)) + client.initiate_connection() + with socket.create_connection( + ("127.0.0.1", server.port), timeout=3 + ) as connection: + connection.sendall(client.data_to_send()) + client.send_headers( + 1, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/fail"), + ], + end_stream=True, + ) + connection.sendall(client.data_to_send()) + while connection.recv(65535): + pass + except BaseException as exc: + errors.append(exc) + try: + server.close() + except BaseException: + pass + + worker = threading.Thread(target=client_work, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=3) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertIsInstance(server.failure, RuntimeError) + self.assertIn("writer failure", str(server.failure)) + self.assertEqual(server.owned_connection_count, 0) + self.assertTrue(server.finished) + + def test_shutdown_force_closes_a_blocked_writer(self): + runtime = SmallOS().setKernel(Unix()) + app = SmallServer() + + @app.get("/blocked") + async def blocked(request): + return Response.text("response") + + try: + server = app.serve( + runtime, host="127.0.0.1", port=0, protocol="http2" + ) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + original_transport = server._transport + writer_blocked = threading.Event() + + class BlockingWriterTransport: + def __getattr__(self, name): + return getattr(original_transport, name) + + async def send_all(self, task, stream, data): + if getattr(task, "name", "") == "smallserver-http2-writer": + writer_blocked.set() + await task.wait_signal(28) + return + await original_transport.send_all(task, stream, data) + + server._transport = BlockingWriterTransport() + errors = [] + + def client_work(): + try: + client = H2Connection(config=H2Configuration(client_side=True)) + client.initiate_connection() + with socket.create_connection( + ("127.0.0.1", server.port), timeout=3 + ) as connection: + connection.sendall(client.data_to_send()) + client.send_headers( + 1, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/blocked"), + ], + end_stream=True, + ) + connection.sendall(client.data_to_send()) + if not writer_blocked.wait(2): + raise TimeoutError("writer did not enter its blocked wait") + server.close() + while connection.recv(65535): + pass + except BaseException as exc: + errors.append(exc) + try: + server.close() + except BaseException: + pass + + worker = threading.Thread(target=client_work, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=3) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertTrue(server.finished) + self.assertEqual(server.owned_connection_count, 0) + + def test_protocol_construction_failure_releases_accepted_connection(self): + runtime = SmallOS().setKernel(Unix()) + app = SmallServer() + try: + with patch( + "smallserver.app.H2Protocol", + side_effect=RuntimeError("injected constructor failure"), + ): + server = app.serve( + runtime, host="127.0.0.1", port=0, protocol="http2" + ) + + def client_work(): + with socket.create_connection( + ("127.0.0.1", server.port), timeout=3 + ) as connection: + while connection.recv(1024): + pass + server.close() + + worker = threading.Thread(target=client_work, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=3) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + self.assertFalse(worker.is_alive()) + self.assertTrue(server.finished) + self.assertEqual(server.owned_connection_count, 0) + + def test_handshake_timeout_closes_silent_client_and_releases_capacity(self): + runtime = SmallOS().setKernel(Unix()) + app = SmallServer() + try: + server = app.serve( + runtime, + host="127.0.0.1", + port=0, + protocol="http2", + http2_config=HTTP2Config(handshake_timeout=0.01), + ) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + errors = [] + + def client_work(): + try: + with socket.create_connection( + ("127.0.0.1", server.port), timeout=3 + ) as connection: + while connection.recv(1024): + pass + server.close() + except BaseException as exc: + errors.append(exc) + try: + server.close() + except BaseException: + pass + + worker = threading.Thread(target=client_work, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=3) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertEqual(server.owned_connection_count, 0) + self.assertTrue(server.finished) + + def test_idle_timeout_closes_prefaced_client_and_releases_capacity(self): + runtime = SmallOS().setKernel(Unix()) + app = SmallServer() + try: + server = app.serve( + runtime, + host="127.0.0.1", + port=0, + protocol="http2", + http2_config=HTTP2Config( + handshake_timeout=1, + idle_timeout=0.01, + ), + ) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + errors = [] + + def client_work(): + try: + client = H2Connection(config=H2Configuration(client_side=True)) + client.initiate_connection() + with socket.create_connection( + ("127.0.0.1", server.port), timeout=3 + ) as connection: + connection.sendall(client.data_to_send()) + while connection.recv(1024): + pass + server.close() + except BaseException as exc: + errors.append(exc) + try: + server.close() + except BaseException: + pass + + worker = threading.Thread(target=client_work, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=3) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertEqual(server.owned_connection_count, 0) + self.assertTrue(server.finished) + if __name__ == "__main__": unittest.main() From e50aac4258fb024ae7888f70e3353d0284329eae Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 02:11:29 -0500 Subject: [PATCH 05/14] docs: describe HTTP/2 test and timeout contracts --- README.md | 8 ++++++-- guide/http2.md | 14 ++++++++++---- pyproject.toml | 1 + 3 files changed, 17 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 3dd7769..8913505 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,7 @@ yet. HTTP/2 currently supports cleartext prior knowledge only. ```bash python3 -m pip install -r requirements.txt -python3 -m pip install -e . +python3 -m pip install -e '.[test]' python3 -m unittest discover -s tests -v ``` @@ -92,7 +92,11 @@ app.listen( host="127.0.0.1", port=8000, protocol="http2", - http2_config=HTTP2Config(max_concurrent_streams=32), + http2_config=HTTP2Config( + max_concurrent_streams=32, + handshake_timeout=10, + idle_timeout=60, + ), ) ``` diff --git a/guide/http2.md b/guide/http2.md index 88747f8..18fca33 100644 --- a/guide/http2.md +++ b/guide/http2.md @@ -8,7 +8,7 @@ to own task scheduling and all network readiness. ```bash python3 -m pip install -r requirements.txt -python3 -m pip install -e '.[http2]' +python3 -m pip install -e '.[test]' python3 examples/http2_prior_knowledge.py ``` @@ -32,13 +32,19 @@ single writer preserves frame ordering and observes peer flow-control windows. `HTTP2Config` bounds concurrent streams, decoded and compressed header sizes, per-stream and per-connection request buffering, response buffering, and frame -size. Requests and responses use the same immutable `Request`, `Headers`, and -`Response` values as HTTP/1.1. The request version is `"HTTP/2"`. +size. Completed request bodies remain charged to the connection budget while +their handler is running. `handshake_timeout` bounds receipt of the client +preface and `idle_timeout` bounds inactive established connections; both use +SmallOS scheduler timers. Requests and responses use the same immutable +`Request`, `Headers`, and `Response` values as HTTP/1.1. The request version is +`"HTTP/2"`. Peer stream resets cancel the associated handler task without stopping other streams. Protocol/resource violations reset the affected stream when possible. Connection shutdown emits GOAWAY and then releases the connection through the -SmallOS kernel transport. +SmallOS kernel transport. If a writer is already blocked on kernel +writability, a lower-priority scheduler task force-closes the stream after the +writer's graceful scheduling opportunity so shutdown remains bounded. ## Current protocol boundary diff --git a/pyproject.toml b/pyproject.toml index df79c26..4f007c4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,7 @@ dependencies = [] [project.optional-dependencies] dev = ["build>=1.2"] http2 = ["h2>=4,<5"] +test = ["h2>=4,<5"] [tool.setuptools.packages.find] include = ["smallserver*"] From 35cd8c5be40f1173712c8a0c04b61368bce0882f Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 02:16:38 -0500 Subject: [PATCH 06/14] fix: require finite HTTP/2 timeouts --- smallserver/http2.py | 8 +++++++- tests/test_http2.py | 3 +++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/smallserver/http2.py b/smallserver/http2.py index ba5467b..0109a25 100644 --- a/smallserver/http2.py +++ b/smallserver/http2.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import dataclass +import math from typing import Any from .errors import ServerConfigurationError @@ -39,7 +40,12 @@ def __post_init__(self) -> None: raise ValueError("{} must be a positive integer".format(name)) for name in ("handshake_timeout", "idle_timeout"): value = getattr(self, name) - if not isinstance(value, (int, float)) or isinstance(value, bool) or value <= 0: + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or not math.isfinite(value) + or value <= 0 + ): raise ValueError("{} must be a positive number".format(name)) if not 16_384 <= self.max_frame_size <= 16_777_215: raise ValueError("max_frame_size must be between 16384 and 16777215") diff --git a/tests/test_http2.py b/tests/test_http2.py index d932fc6..d0e629b 100644 --- a/tests/test_http2.py +++ b/tests/test_http2.py @@ -57,6 +57,9 @@ def test_timeout_configuration_is_finite_and_positive(self): {"handshake_timeout": 0}, {"idle_timeout": -1}, {"idle_timeout": True}, + {"handshake_timeout": float("nan")}, + {"handshake_timeout": float("inf")}, + {"idle_timeout": float("-inf")}, ): with self.subTest(values=values): with self.assertRaises(ValueError): From ce36d941ad108d9b754b330a4399d85315d90b06 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 02:29:17 -0500 Subject: [PATCH 07/14] fix: bound HTTP/2 fairness and connection failures --- smallserver/app.py | 80 ++++++++++----------- smallserver/http2.py | 163 +++++++++++++++++++++++++++++++----------- smallserver/server.py | 15 +++- tests/test_http2.py | 162 ++++++++++++++++++++++++++++++++++------- 4 files changed, 309 insertions(+), 111 deletions(-) diff --git a/smallserver/app.py b/smallserver/app.py index c039694..5ff9ef6 100644 --- a/smallserver/app.py +++ b/smallserver/app.py @@ -662,6 +662,7 @@ async def _http2_connection_loop( handle._runtime.fork(child_tasks) handle._graceful_connections.add(id(client)) handle._graceful_closers[id(client)] = state.request_shutdown + reader_batches = 0 while not handle.closed and not protocol.remote_closed: chunk = await handle._transport.recv( task, client, handle._config.receive_chunk_bytes @@ -669,27 +670,41 @@ async def _http2_connection_loop( if not chunk: break state.mark_activity() - try: - ready = protocol.receive_data(chunk) - except Exception as protocol_error: - primary_error = protocol_error - state.close_error_code = 1 + next_data = chunk + while True: + try: + ready = protocol.receive_data(next_data) + except Exception as protocol_error: + primary_error = protocol_error + state.close_error_code = 1 + break + for stream_id in protocol.take_cancelled_streams(): + handler = state.handlers.pop(stream_id, None) + if handler is not None: + handle._cancel_or_retain_task(handler) + for item in ready: + handler = SmallTask( + handle._config.connection_priority, + self._http2_handler, + args=(handle, state, item.stream_id, item.request), + name="smallserver-http2-stream-{}".format(item.stream_id), + ) + state.handlers[item.stream_id] = handler + handle._owned_tasks.append(handler) + handle._runtime.fork(handler) + state.wake_writer() + reader_batches += 1 + if protocol.has_pending_input: + await task.yield_now() + reader_batches = 0 + next_data = b"" + continue + if reader_batches >= protocol.config.reader_frame_batch_size: + await task.yield_now() + reader_batches = 0 + break + if primary_error is not None: break - for stream_id in protocol.take_cancelled_streams(): - handler = state.handlers.pop(stream_id, None) - if handler is not None: - handle._cancel_or_retain_task(handler) - for item in ready: - handler = SmallTask( - handle._config.connection_priority, - self._http2_handler, - args=(handle, state, item.stream_id, item.request), - name="smallserver-http2-stream-{}".format(item.stream_id), - ) - state.handlers[item.stream_id] = handler - handle._owned_tasks.append(handler) - handle._runtime.fork(handler) - state.wake_writer() except Exception as exc: primary_error = exc except BaseException as exc: @@ -764,11 +779,7 @@ async def _http2_writer_loop( payload = state.protocol.close() if payload: await handle._transport.send_all(task, client, payload) - if not handle._transport.close_safely(client): - error = client.close_error or RuntimeError( - "kernel connection close failed" - ) - handle._connection_close_failed(error, task) + handle._force_connection_close(client, task) return while not state.closing: payload = state.protocol.flush() @@ -778,12 +789,7 @@ async def _http2_writer_loop( except Exception as error: state.failure = error state.closing = True - handle._listener_failed(error, task) - if not handle._transport.close_safely(client): - close_error = client.close_error or RuntimeError( - "kernel connection close failed" - ) - handle._connection_close_failed(close_error, task, error) + handle._force_connection_close(client, task, error) raise finally: if task in handle._owned_tasks: @@ -802,11 +808,7 @@ async def _http2_shutdown_enforcer( if client.closed: return state.closing = True - if not handle._transport.close_safely(client): - error = client.close_error or RuntimeError( - "kernel connection close failed" - ) - handle._connection_close_failed(error, task, state.failure) + handle._force_connection_close(client, task, state.failure) finally: if task in handle._owned_tasks: handle._owned_tasks.remove(task) @@ -868,8 +870,4 @@ def _http2_force_close( ) -> None: state.failure = error state.closing = True - if not handle._transport.close_safely(client): - close_error = client.close_error or RuntimeError( - "kernel connection close failed" - ) - handle._connection_close_failed(close_error, task, error) + handle._force_connection_close(client, task, error) diff --git a/smallserver/http2.py b/smallserver/http2.py index 0109a25..3a41fcc 100644 --- a/smallserver/http2.py +++ b/smallserver/http2.py @@ -25,7 +25,9 @@ class HTTP2Config: max_connection_buffer_bytes: int = 4 * 1024 * 1024 max_pending_output_bytes: int = 4 * 1024 * 1024 max_response_body_bytes: int = 2 * 1024 * 1024 + max_control_output_bytes: int = 64 * 1024 max_frame_size: int = 16 * 1024 + reader_frame_batch_size: int = 32 handshake_timeout: float = 10.0 idle_timeout: float = 60.0 @@ -57,6 +59,14 @@ def __post_init__(self) -> None: raise ValueError( "max_response_body_bytes cannot exceed max_pending_output_bytes" ) + if self.max_control_output_bytes > self.max_pending_output_bytes: + raise ValueError( + "max_control_output_bytes cannot exceed max_pending_output_bytes" + ) + if self.max_control_output_bytes < 9: + raise ValueError( + "max_control_output_bytes must allow one HTTP/2 control frame" + ) @dataclass(frozen=True) @@ -72,7 +82,7 @@ class _InboundStream: method: str path: str headers: Headers - body: bytearray + body: bytearray | bytes expected_content_length: int | None dispatched: bool = False @@ -88,52 +98,92 @@ class _FrameBudget: def __init__(self, config: HTTP2Config) -> None: self._config = config - self._buffer = bytearray() + self._preface = bytearray() + self._header = bytearray() + self._payload = bytearray() + self._frame_length = 0 + self._frame_type = 0 + self._frame_flags = 0 + self._frame_stream = 0 self._preface_received = False self._header_stream: int | None = None self._header_bytes = 0 + self._ready: list[bytes] = [] - def feed(self, data: bytes) -> tuple[bytes, ...]: - self._buffer.extend(data) - chunks: list[bytes] = [] + def feed(self, data: bytes) -> None: + view = memoryview(data) + offset = 0 if not self._preface_received: - prefix_length = min(len(self._buffer), len(HTTP2_CLIENT_PREFACE)) - if bytes(self._buffer[:prefix_length]) != HTTP2_CLIENT_PREFACE[:prefix_length]: + needed = len(HTTP2_CLIENT_PREFACE) - len(self._preface) + take = min(needed, len(view)) + self._preface.extend(view[:take]) + offset += take + if bytes(self._preface) != HTTP2_CLIENT_PREFACE[: len(self._preface)]: raise ValueError("invalid HTTP/2 client preface") - if len(self._buffer) < len(HTTP2_CLIENT_PREFACE): - return () - chunks.append(bytes(self._buffer[: len(HTTP2_CLIENT_PREFACE)])) - del self._buffer[: len(HTTP2_CLIENT_PREFACE)] + if len(self._preface) < len(HTTP2_CLIENT_PREFACE): + return + self._ready.append(bytes(self._preface)) + self._preface.clear() self._preface_received = True - while len(self._buffer) >= 9: - length = int.from_bytes(self._buffer[:3], "big") - if length > self._config.max_frame_size: - raise ValueError("HTTP/2 frame exceeds configured maximum") - frame_length = 9 + length - if len(self._buffer) < frame_length: - break - frame_type = self._buffer[3] - flags = self._buffer[4] - stream_id = int.from_bytes(self._buffer[5:9], "big") & 0x7FFFFFFF - if frame_type == 0x1: - if self._header_stream is not None: - raise ValueError("interleaved HTTP/2 header blocks are invalid") - self._header_stream = stream_id - self._header_bytes = length - elif frame_type == 0x9: - if self._header_stream != stream_id: - raise ValueError("invalid HTTP/2 continuation stream") - self._header_bytes += length - if self._header_bytes > self._config.max_compressed_header_bytes: - raise ValueError("HTTP/2 compressed header block is too large") - if frame_type in (0x1, 0x9) and flags & 0x4: - self._header_stream = None - self._header_bytes = 0 - chunks.append(bytes(self._buffer[:frame_length])) - del self._buffer[:frame_length] + while offset < len(view): + if len(self._header) < 9: + take = min(9 - len(self._header), len(view) - offset) + self._header.extend(view[offset : offset + take]) + offset += take + if len(self._header) < 9: + return + self._start_frame() + if self._frame_length == 0: + self._finish_frame() + continue + take = min( + self._frame_length - len(self._payload), + len(view) - offset, + ) + self._payload.extend(view[offset : offset + take]) + offset += take + if len(self._payload) == self._frame_length: + self._finish_frame() + + def _start_frame(self) -> None: + length = int.from_bytes(self._header[:3], "big") + if length > self._config.max_frame_size: + raise ValueError("HTTP/2 frame exceeds configured maximum") + self._frame_length = length + self._frame_type = self._header[3] + self._frame_flags = self._header[4] + self._frame_stream = int.from_bytes(self._header[5:9], "big") & 0x7FFFFFFF + if self._frame_type == 0x1: + if self._header_stream is not None: + raise ValueError("interleaved HTTP/2 header blocks are invalid") + self._header_stream = self._frame_stream + self._header_bytes = length + elif self._frame_type == 0x9: + if self._header_stream != self._frame_stream: + raise ValueError("invalid HTTP/2 continuation stream") + self._header_bytes += length + if self._header_bytes > self._config.max_compressed_header_bytes: + raise ValueError("HTTP/2 compressed header block is too large") + + def _finish_frame(self) -> None: + self._ready.append(bytes(self._header + self._payload)) + if self._frame_type in (0x1, 0x9) and self._frame_flags & 0x4: + self._header_stream = None + self._header_bytes = 0 + self._header.clear() + self._payload.clear() + self._frame_length = 0 + + def take(self, limit: int) -> tuple[bytes, ...]: + chunks = self._ready[:limit] + del self._ready[:limit] return tuple(chunks) + @property + def has_ready_frames(self) -> bool: + return bool(self._ready) + @property def preface_received(self) -> bool: return self._preface_received @@ -249,6 +299,7 @@ def _begin_new_stream(self, stream_id: Any, allowed_ids: Any) -> Any: self._commands: list[tuple[str, int, Response | None]] = [] self._buffered_request_bytes = 0 self._pending_output_bytes = 0 + self._control_output = bytearray() self._cancelled_streams: list[int] = [] self.last_processed_stream_id = 0 self.remote_closed = False @@ -260,7 +311,7 @@ def active_stream_count(self) -> int: @property def pending_output_bytes(self) -> int: - return self._pending_output_bytes + return self._pending_output_bytes + len(self._control_output) @property def buffered_request_bytes(self) -> int: @@ -270,13 +321,18 @@ def buffered_request_bytes(self) -> int: def preface_received(self) -> bool: return self._frames.preface_received + @property + def has_pending_input(self) -> bool: + return self._frames.has_ready_frames + def initiate(self) -> bytes: self.connection.initiate_connection() return self.connection.data_to_send() def receive_data(self, data: bytes) -> tuple[H2ReadyRequest, ...]: ready: list[H2ReadyRequest] = [] - for wire_chunk in self._frames.feed(data): + self._frames.feed(data) + for wire_chunk in self._frames.take(self.config.reader_frame_batch_size): events = self.connection.receive_data(wire_chunk) for event in events: if isinstance(event, self._events["request"]): @@ -300,8 +356,20 @@ def receive_data(self, data: bytes) -> tuple[H2ReadyRequest, ...]: event, (self._events["window"], self._events["settings"]) ): pass + self._capture_control_output() return tuple(ready) + def _capture_control_output(self) -> None: + produced = self.connection.data_to_send() + next_control_size = len(self._control_output) + len(produced) + if ( + next_control_size > self.config.max_control_output_bytes + or next_control_size + self._pending_output_bytes + > self.config.max_pending_output_bytes + ): + raise ValueError("HTTP/2 control output exceeds configured maximum") + self._control_output.extend(produced) + def take_cancelled_streams(self) -> tuple[int, ...]: """Return peer-reset stream ids exactly once.""" cancelled, self._cancelled_streams = self._cancelled_streams, [] @@ -418,6 +486,9 @@ def _data_received(self, stream_id: int, data: bytes, flow_length: int) -> None: ): self._reset_stream(stream_id, self._error_codes.ENHANCE_YOUR_CALM) return + if not isinstance(stream.body, bytearray): + self._reset_stream(stream_id, self._error_codes.STREAM_CLOSED) + return stream.body.extend(data) self._buffered_request_bytes = next_connection_size @@ -433,11 +504,13 @@ def _stream_ended(self, stream_id: int) -> H2ReadyRequest | None: return None stream.dispatched = True self.last_processed_stream_id = max(self.last_processed_stream_id, stream_id) + body = bytes(stream.body) + stream.body = body request = Request( stream.method, stream.path, stream.headers, - bytes(stream.body), + body, "HTTP/2", ) return H2ReadyRequest(stream_id, request) @@ -449,7 +522,7 @@ def queue_response(self, stream_id: int, response: Response) -> bool: body_size = len(response.body) if ( body_size > self.config.max_response_body_bytes - or self._pending_output_bytes + body_size + or len(self._control_output) + self._pending_output_bytes + body_size > self.config.max_pending_output_bytes ): if not any(command[1] == stream_id for command in self._commands): @@ -463,6 +536,8 @@ def queue_response(self, stream_id: int, response: Response) -> bool: return True def flush(self) -> bytes: + control = bytes(self._control_output) + self._control_output.clear() commands, self._commands = self._commands, [] for operation, stream_id, response in commands: if operation == "reset": @@ -503,7 +578,7 @@ def flush(self) -> bytes: if end_stream: self._outbound.pop(stream_id, None) self._active_streams.discard(stream_id) - return self.connection.data_to_send() + return control + self.connection.data_to_send() def _start_response(self, stream_id: int, response: Response) -> None: headers: list[tuple[str, str]] = [(":status", str(response.status))] @@ -578,4 +653,6 @@ def close(self, error_code: int = 0) -> bytes: self._active_streams.clear() self._buffered_request_bytes = 0 self._pending_output_bytes = 0 - return self.connection.data_to_send() + control = bytes(self._control_output) + self._control_output.clear() + return control + self.connection.data_to_send() diff --git a/smallserver/server.py b/smallserver/server.py index 7df4dc5..f2434aa 100644 --- a/smallserver/server.py +++ b/smallserver/server.py @@ -395,12 +395,25 @@ def _close_or_retain( self._closing_connections.pop(identity, None) self._cleanup_errors.pop("connection:{}".format(identity), None) return True - self._closing_connections[identity] = connection + if identity not in self._connections: + self._closing_connections[identity] = connection error = connection.close_error or RuntimeError("kernel connection close failed") self._cleanup_errors["connection:{}".format(identity)] = error self._connection_close_failed(error, task, primary_error) return False + def _force_connection_close( + self, + connection: TransportHandle, + task: Any = None, + primary_error: BaseException | None = None, + ) -> bool: + """Stop graceful handling and close through retryable ownership.""" + identity = id(connection) + self._graceful_connections.discard(identity) + self._graceful_closers.pop(identity, None) + return self._close_or_retain(connection, task, primary_error) + def _connection_close_failed( self, error: BaseException, diff --git a/tests/test_http2.py b/tests/test_http2.py index d0e629b..e8b32b3 100644 --- a/tests/test_http2.py +++ b/tests/test_http2.py @@ -22,9 +22,11 @@ from SmallPackage import SmallOS, Unix from smallserver import HTTP2Config, Response, SmallServer +from smallserver._transport import KernelTransport, TransportHandle from smallserver.errors import ServerConfigurationError -from smallserver.http2 import H2Protocol -from tests.kernel_fakes import FakeKernel +from smallserver.http2 import H2Protocol, _FrameBudget +from smallserver.server import ServerConfig, ServerHandle +from tests.kernel_fakes import FakeKernel, OpaqueHandle class HTTP2OptionalDependencyTests(unittest.TestCase): @@ -90,6 +92,42 @@ def cancel_task(self, task): any(call[0] == "resolve_passive_address" for call in runtime.kernel.calls) ) + def test_h2_force_close_failure_retains_one_owner_until_retry(self): + class Runtime: + def resume_task(self, task): + return None + + def cancel_task(self, task): + return None + + kernel = FakeKernel() + transport = KernelTransport(kernel) + listener = transport.open_listener("127.0.0.1", 0, 2) + wakeup = transport.create_wakeup_channel() + handle = ServerHandle(Runtime(), transport, listener, wakeup, ServerConfig()) + raw_client = OpaqueHandle("h2-client") + client = TransportHandle(raw_client) + reader_task = object() + writer_error = RuntimeError("writer failed") + handle._connections[id(client)] = (client, reader_task) + handle._graceful_connections.add(id(client)) + handle._graceful_closers[id(client)] = lambda: None + kernel.close_failures[id(raw_client)] = 2 + + self.assertFalse( + handle._force_connection_close(client, object(), writer_error) + ) + self.assertEqual(handle.owned_connection_count, 1) + self.assertEqual(len(handle.cleanup_errors), 1) + self.assertIs(handle.failure, writer_error) + + handle._finish_close() + self.assertFalse(handle.finished) + self.assertEqual(handle.owned_connection_count, 1) + handle._finish_close() + self.assertTrue(handle.finished) + self.assertEqual(handle.owned_connection_count, 0) + @unittest.skipUnless(H2_AVAILABLE, "install the smallserver[test] HTTP/2 extra") class HTTP2ProtocolTests(unittest.TestCase): @@ -159,7 +197,8 @@ def test_request_and_response_limits_reset_streams_without_unbounded_buffers(sel max_body_bytes=4, max_connection_buffer_bytes=8, max_response_body_bytes=4, - max_pending_output_bytes=8, + max_pending_output_bytes=32, + max_control_output_bytes=16, ) client, server = self._pair(config) client.send_headers( @@ -232,6 +271,23 @@ def test_completed_slow_handler_body_remains_in_connection_budget(self): server.queue_response(1, Response.text("done")) self.assertEqual(server.buffered_request_bytes, 0) + def test_completed_body_has_one_retained_payload_object(self): + client, server = self._pair() + client.send_headers( + 1, + [ + (":method", "POST"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/slow"), + ], + ) + client.send_data(1, b"retained", end_stream=True) + ready = server.receive_data(client.data_to_send()) + retained = server._inbound[1].body + self.assertIsInstance(retained, bytes) + self.assertIs(retained, ready[0].request.body) + def test_bad_stream_metadata_resets_only_that_stream(self): client, server = self._pair() client.send_headers( @@ -327,6 +383,40 @@ def test_invalid_content_length_is_a_stream_error(self): any(isinstance(event, StreamReset) and event.stream_id == 1 for event in events) ) + def test_control_output_is_bounded_and_frames_are_processed_in_batches(self): + config = HTTP2Config( + reader_frame_batch_size=2, + max_control_output_bytes=64, + ) + client, server = self._pair(config) + for value in range(6): + client.ping(value.to_bytes(8, "big")) + server.receive_data(client.data_to_send()) + self.assertTrue(server.has_pending_input) + batches = 1 + while server.has_pending_input: + client.receive_data(server.flush()) + server.receive_data(b"") + batches += 1 + client.receive_data(server.flush()) + self.assertGreaterEqual(batches, 3) + self.assertEqual(server.pending_output_bytes, 0) + + limited_client, limited_server = self._pair( + HTTP2Config(max_control_output_bytes=16) + ) + limited_client.ping(b"12345678") + with self.assertRaisesRegex(ValueError, "control output"): + limited_server.receive_data(limited_client.data_to_send()) + + def test_compressed_header_budget_rejects_declared_size_before_payload(self): + budget = _FrameBudget(HTTP2Config(max_compressed_header_bytes=4)) + budget.feed(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n") + header = b"\x00\x00\x05" + b"\x01\x04" + b"\x00\x00\x00\x01" + with self.assertRaisesRegex(ValueError, "compressed header"): + budget.feed(header) + self.assertEqual(len(budget._payload), 0) + @unittest.skipUnless(H2_AVAILABLE, "install the smallserver[test] HTTP/2 extra") class HTTP2ServerIntegrationTests(unittest.TestCase): @@ -464,7 +554,7 @@ def test_large_response_respects_flow_control(self): self.assertEqual(bytes(received), body) self.assertEqual(server.pending_output_bytes, 0) - def test_writer_send_failure_is_fatal_and_releases_capacity(self): + def test_writer_send_failure_closes_only_client_and_listener_stays_healthy(self): runtime = SmallOS().setKernel(Unix()) app = SmallServer() @@ -479,13 +569,19 @@ async def fail(request): except PermissionError: self.skipTest("the current sandbox does not permit loopback TCP binds") original_transport = server._transport + failure_injected = False class FailingWriterTransport: def __getattr__(self, name): return getattr(original_transport, name) async def send_all(self, task, stream, data): - if getattr(task, "name", "") == "smallserver-http2-writer": + nonlocal failure_injected + if ( + getattr(task, "name", "") == "smallserver-http2-writer" + and not failure_injected + ): + failure_injected = True raise RuntimeError("injected HTTP/2 writer failure") await original_transport.send_all(task, stream, data) @@ -494,25 +590,39 @@ async def send_all(self, task, stream, data): def client_work(): try: - client = H2Connection(config=H2Configuration(client_side=True)) - client.initiate_connection() - with socket.create_connection( - ("127.0.0.1", server.port), timeout=3 - ) as connection: - connection.sendall(client.data_to_send()) - client.send_headers( - 1, - [ - (":method", "GET"), - (":scheme", "http"), - (":authority", "localhost"), - (":path", "/fail"), - ], - end_stream=True, - ) - connection.sendall(client.data_to_send()) - while connection.recv(65535): - pass + responses = [] + for _attempt in range(2): + client = H2Connection(config=H2Configuration(client_side=True)) + client.initiate_connection() + body = bytearray() + with socket.create_connection( + ("127.0.0.1", server.port), timeout=3 + ) as connection: + connection.sendall(client.data_to_send()) + client.send_headers( + 1, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/fail"), + ], + end_stream=True, + ) + connection.sendall(client.data_to_send()) + ended = False + while not ended: + data = connection.recv(65535) + if not data: + break + for event in client.receive_data(data): + if isinstance(event, DataReceived): + body.extend(event.data) + elif isinstance(event, StreamEnded): + ended = True + responses.append(bytes(body)) + self.assertEqual(responses, [b"", b"response"]) + server.close() except BaseException as exc: errors.append(exc) try: @@ -526,8 +636,8 @@ def client_work(): worker.join(timeout=3) self.assertFalse(worker.is_alive()) self.assertEqual(errors, []) - self.assertIsInstance(server.failure, RuntimeError) - self.assertIn("writer failure", str(server.failure)) + self.assertTrue(failure_injected) + self.assertIsNone(server.failure) self.assertEqual(server.owned_connection_count, 0) self.assertTrue(server.finished) From 90dc551d4d5ab0c24666541b30cf2420e1c71342 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 02:39:13 -0500 Subject: [PATCH 08/14] fix: close HTTP/2 batch and output budget gaps --- smallserver/app.py | 2 + smallserver/http2.py | 64 +++++++++++++--- tests/test_http2.py | 175 +++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 226 insertions(+), 15 deletions(-) diff --git a/smallserver/app.py b/smallserver/app.py index 5ff9ef6..3fa627f 100644 --- a/smallserver/app.py +++ b/smallserver/app.py @@ -683,6 +683,8 @@ async def _http2_connection_loop( if handler is not None: handle._cancel_or_retain_task(handler) for item in ready: + if not protocol.is_stream_active(item.stream_id): + continue handler = SmallTask( handle._config.connection_priority, self._http2_handler, diff --git a/smallserver/http2.py b/smallserver/http2.py index 3a41fcc..87ce226 100644 --- a/smallserver/http2.py +++ b/smallserver/http2.py @@ -63,9 +63,9 @@ def __post_init__(self) -> None: raise ValueError( "max_control_output_bytes cannot exceed max_pending_output_bytes" ) - if self.max_control_output_bytes < 9: + if self.max_control_output_bytes < 51: raise ValueError( - "max_control_output_bytes must allow one HTTP/2 control frame" + "max_control_output_bytes must allow initial HTTP/2 settings" ) @@ -301,6 +301,7 @@ def _begin_new_stream(self, stream_id: Any, allowed_ids: Any) -> Any: self._pending_output_bytes = 0 self._control_output = bytearray() self._cancelled_streams: list[int] = [] + self._ready_requests: list[H2ReadyRequest] = [] self.last_processed_stream_id = 0 self.remote_closed = False self.local_closed = False @@ -327,10 +328,13 @@ def has_pending_input(self) -> bool: def initiate(self) -> bytes: self.connection.initiate_connection() - return self.connection.data_to_send() + output = self.connection.data_to_send() + if len(output) > self.config.max_control_output_bytes: + raise ValueError("HTTP/2 control output exceeds configured maximum") + self._validate_wire_output(len(output), 0) + return output def receive_data(self, data: bytes) -> tuple[H2ReadyRequest, ...]: - ready: list[H2ReadyRequest] = [] self._frames.feed(data) for wire_chunk in self._frames.take(self.config.reader_frame_batch_size): events = self.connection.receive_data(wire_chunk) @@ -344,7 +348,7 @@ def receive_data(self, data: bytes) -> tuple[H2ReadyRequest, ...]: elif isinstance(event, self._events["ended"]): completed = self._stream_ended(event.stream_id) if completed is not None: - ready.append(completed) + self._ready_requests.append(completed) elif isinstance(event, self._events["reset"]): self._cancelled_streams.append(event.stream_id) self.drop_stream(event.stream_id) @@ -357,7 +361,20 @@ def receive_data(self, data: bytes) -> tuple[H2ReadyRequest, ...]: ): pass self._capture_control_output() - return tuple(ready) + if self._frames.has_ready_frames: + return () + cancelled = set(self._cancelled_streams) + ready = tuple( + item + for item in self._ready_requests + if item.stream_id not in cancelled + and item.stream_id in self._active_streams + ) + self._ready_requests.clear() + return ready + + def is_stream_active(self, stream_id: int) -> bool: + return stream_id in self._active_streams def _capture_control_output(self) -> None: produced = self.connection.data_to_send() @@ -370,6 +387,13 @@ def _capture_control_output(self) -> None: raise ValueError("HTTP/2 control output exceeds configured maximum") self._control_output.extend(produced) + def _validate_wire_output(self, new_bytes: int, already_buffered: int) -> None: + if ( + new_bytes + already_buffered + self._pending_output_bytes + > self.config.max_pending_output_bytes + ): + raise ValueError("HTTP/2 generated output exceeds configured maximum") + def take_cancelled_streams(self) -> tuple[int, ...]: """Return peer-reset stream ids exactly once.""" cancelled, self._cancelled_streams = self._cancelled_streams, [] @@ -536,15 +560,18 @@ def queue_response(self, stream_id: int, response: Response) -> bool: return True def flush(self) -> bytes: - control = bytes(self._control_output) - self._control_output.clear() commands, self._commands = self._commands, [] for operation, stream_id, response in commands: if operation == "reset": self._reset_stream(stream_id, self._error_codes.ENHANCE_YOUR_CALM) + self._capture_control_output() continue assert response is not None self._start_response(stream_id, response) + self._capture_control_output() + + output = bytearray(self._control_output) + self._control_output.clear() for stream_id, outbound in tuple(self._outbound.items()): remaining = len(outbound.body) - outbound.offset @@ -575,10 +602,16 @@ def flush(self) -> bytes: continue outbound.offset += chunk_size self._pending_output_bytes -= chunk_size + generated = self.connection.data_to_send() + self._validate_wire_output(len(generated), len(output)) + output.extend(generated) if end_stream: self._outbound.pop(stream_id, None) self._active_streams.discard(stream_id) - return control + self.connection.data_to_send() + generated = self.connection.data_to_send() + self._validate_wire_output(len(generated), len(output)) + output.extend(generated) + return bytes(output) def _start_response(self, stream_id: int, response: Response) -> None: headers: list[tuple[str, str]] = [(":status", str(response.status))] @@ -613,6 +646,9 @@ def _start_response(self, stream_id: int, response: Response) -> None: def drop_stream(self, stream_id: int) -> None: self._release_inbound(stream_id) + self._ready_requests = [ + item for item in self._ready_requests if item.stream_id != stream_id + ] outbound = self._outbound.pop(stream_id, None) if outbound is not None: self._pending_output_bytes -= len(outbound.body) - outbound.offset @@ -650,9 +686,17 @@ def close(self, error_code: int = 0) -> bytes: self._inbound.clear() self._outbound.clear() self._commands.clear() + self._ready_requests.clear() self._active_streams.clear() self._buffered_request_bytes = 0 self._pending_output_bytes = 0 control = bytes(self._control_output) self._control_output.clear() - return control + self.connection.data_to_send() + generated = self.connection.data_to_send() + try: + self._validate_wire_output(len(generated), len(control)) + if len(control) + len(generated) > self.config.max_control_output_bytes: + raise ValueError("HTTP/2 control output exceeds configured maximum") + except ValueError: + return b"" + return control + generated diff --git a/tests/test_http2.py b/tests/test_http2.py index e8b32b3..5543226 100644 --- a/tests/test_http2.py +++ b/tests/test_http2.py @@ -197,8 +197,8 @@ def test_request_and_response_limits_reset_streams_without_unbounded_buffers(sel max_body_bytes=4, max_connection_buffer_bytes=8, max_response_body_bytes=4, - max_pending_output_bytes=32, - max_control_output_bytes=16, + max_pending_output_bytes=128, + max_control_output_bytes=64, ) client, server = self._pair(config) client.send_headers( @@ -239,6 +239,38 @@ def test_peer_reset_is_reported_once_for_handler_cancellation(self): self.assertEqual(server.take_cancelled_streams(), (1,)) self.assertEqual(server.take_cancelled_streams(), ()) + def test_same_batch_end_then_reset_drops_ready_request_but_keeps_sibling(self): + client, server = self._pair(HTTP2Config(reader_frame_batch_size=1)) + client.send_headers( + 1, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/cancelled"), + ], + end_stream=True, + ) + client.reset_stream(1) + client.send_headers( + 3, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/healthy"), + ], + end_stream=True, + ) + ready = server.receive_data(client.data_to_send()) + self.assertEqual(ready, ()) + while server.has_pending_input: + ready = server.receive_data(b"") + self.assertEqual([item.stream_id for item in ready], [3]) + self.assertEqual(server.take_cancelled_streams(), (1,)) + self.assertFalse(server.is_stream_active(1)) + self.assertTrue(server.is_stream_active(3)) + def test_completed_slow_handler_body_remains_in_connection_budget(self): config = HTTP2Config( max_body_bytes=4, @@ -403,9 +435,10 @@ def test_control_output_is_bounded_and_frames_are_processed_in_batches(self): self.assertEqual(server.pending_output_bytes, 0) limited_client, limited_server = self._pair( - HTTP2Config(max_control_output_bytes=16) + HTTP2Config(max_control_output_bytes=52) ) - limited_client.ping(b"12345678") + for value in range(4): + limited_client.ping(value.to_bytes(8, "big")) with self.assertRaisesRegex(ValueError, "control output"): limited_server.receive_data(limited_client.data_to_send()) @@ -417,6 +450,57 @@ def test_compressed_header_budget_rejects_declared_size_before_payload(self): budget.feed(header) self.assertEqual(len(budget._payload), 0) + def test_response_headers_and_command_resets_obey_output_budget(self): + header_client, header_server = self._pair( + HTTP2Config( + max_pending_output_bytes=64, + max_control_output_bytes=64, + max_response_body_bytes=1, + ) + ) + header_client.send_headers( + 1, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/"), + ], + end_stream=True, + ) + header_server.receive_data(header_client.data_to_send()) + header_server.flush() + header_server.queue_response( + 1, + Response(headers={"x-large": "abcdefghijklmnopqrstuvwxyz" * 8}), + ) + with self.assertRaisesRegex(ValueError, "control output"): + header_server.flush() + + reset_client, reset_server = self._pair( + HTTP2Config( + max_pending_output_bytes=52, + max_control_output_bytes=52, + max_response_body_bytes=1, + ) + ) + reset_client.send_headers( + 1, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/"), + ], + end_stream=True, + ) + for value in range(3): + reset_client.ping(value.to_bytes(8, "big")) + reset_server.receive_data(reset_client.data_to_send()) + reset_server.queue_response(1, Response(body=b"xx")) + with self.assertRaisesRegex(ValueError, "control output"): + reset_server.flush() + @unittest.skipUnless(H2_AVAILABLE, "install the smallserver[test] HTTP/2 extra") class HTTP2ServerIntegrationTests(unittest.TestCase): @@ -514,11 +598,92 @@ def client_work(): self.assertTrue(server.finished) self.assertIsNone(server.failure) + def test_same_batch_reset_never_spawns_cancelled_handler(self): + runtime = SmallOS().setKernel(Unix()) + app = SmallServer() + called = [] + + @app.get("/cancelled") + async def cancelled(request): + called.append("cancelled") + return Response.text("wrong") + + @app.get("/healthy") + async def healthy(request): + called.append("healthy") + return Response.text("ok") + + try: + server = app.serve( + runtime, host="127.0.0.1", port=0, protocol="http2" + ) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + errors = [] + healthy_body = bytearray() + + def client_work(): + try: + client = H2Connection(config=H2Configuration(client_side=True)) + client.initiate_connection() + with socket.create_connection( + ("127.0.0.1", server.port), timeout=3 + ) as connection: + connection.sendall(client.data_to_send()) + client.send_headers( + 1, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/cancelled"), + ], + end_stream=True, + ) + client.reset_stream(1) + client.send_headers( + 3, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/healthy"), + ], + end_stream=True, + ) + connection.sendall(client.data_to_send()) + ended = False + while not ended: + for event in client.receive_data(connection.recv(65535)): + if isinstance(event, DataReceived) and event.stream_id == 3: + healthy_body.extend(event.data) + elif isinstance(event, StreamEnded) and event.stream_id == 3: + ended = True + server.close() + while connection.recv(65535): + pass + except BaseException as exc: + errors.append(exc) + try: + server.close() + except BaseException: + pass + + worker = threading.Thread(target=client_work, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=3) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertEqual(called, ["healthy"]) + self.assertEqual(bytes(healthy_body), b"ok") + self.assertEqual(server.owned_connection_count, 0) + def test_large_response_respects_flow_control(self): body = b"x" * 100_000 config = HTTP2Config( max_response_body_bytes=len(body), - max_pending_output_bytes=len(body), + max_pending_output_bytes=len(body) + 64 * 1024, ) client, server = self._pair(config) client.send_headers( From 0882caeb48fa02492dc48ed8d0c147a7e8031790 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 02:46:01 -0500 Subject: [PATCH 09/14] docs: clarify HTTP/2 install and request semantics --- guide/development.md | 14 +++++++++++++- guide/http2.md | 15 ++++++++------- guide/requests-and-responses.md | 4 ++-- 3 files changed, 23 insertions(+), 10 deletions(-) diff --git a/guide/development.md b/guide/development.md index 68689c3..f9b0fd8 100644 --- a/guide/development.md +++ b/guide/development.md @@ -7,7 +7,7 @@ SmallServer in editable mode: ```console python3 -m pip install -r requirements.txt -python3 -m pip install -e . +python3 -m pip install -e '.[test]' ``` For reproducible validation, put the canonical SmallOS checkout at the front of @@ -27,6 +27,18 @@ ownership, kernel transport behavior, and real loopback serving when the local environment permits binds. Documentation tests verify the tracked guide set, relative Markdown links, and Python code-block syntax. +In a separate clean environment, verify the lazy optional-dependency boundary +without installing the test or HTTP/2 extras: + +```console +python3 -m pip install -r requirements.txt +python3 -m pip install -e . +python3 -m unittest tests.test_http2 -v +``` + +The dependency-contract tests run and HTTP/2 interoperability cases skip +cleanly; importing and testing HTTP/1.1 must not require hyper-h2. + Run the examples when their platform requirements are available: ```console diff --git a/guide/http2.md b/guide/http2.md index fd95c89..f426355 100644 --- a/guide/http2.md +++ b/guide/http2.md @@ -8,7 +8,7 @@ to own task scheduling and all network readiness. ```bash python3 -m pip install -r requirements.txt -python3 -m pip install -e '.[test]' +python3 -m pip install -e '.[http2]' python3 examples/http2_prior_knowledge.py ``` @@ -35,12 +35,13 @@ per-stream and per-connection request buffering, response buffering, and frame size. `max_control_output_bytes` bounds generated SETTINGS/PING acknowledgments, and `reader_frame_batch_size` forces a cooperative yield during continuously readable frame floods. Compressed header-block limits are enforced from the -frame header before payload buffering. Completed request bodies remain charged to the connection budget while -their handler is running. `handshake_timeout` bounds receipt of the client -preface and `idle_timeout` bounds inactive established connections; both use -SmallOS scheduler timers. Requests and responses use the same immutable -`Request`, `Headers`, and `Response` values as HTTP/1.1. The request version is -`"HTTP/2"`. +frame header before payload buffering. Completed request bodies remain charged +to the connection budget while their handler is running. `handshake_timeout` +bounds receipt of the client preface. `idle_timeout` is the maximum interval +without inbound connection bytes or frames; outbound-only response progress +does not reset it. Both timeouts use SmallOS scheduler timers. Requests and +responses use the same immutable `Request`, `Headers`, and `Response` values as +HTTP/1.1. The request version is `"HTTP/2"`. Peer stream resets cancel the associated handler task without stopping other streams. Protocol/resource violations reset the affected stream when possible. diff --git a/guide/requests-and-responses.md b/guide/requests-and-responses.md index 4144a70..49a3ab3 100644 --- a/guide/requests-and-responses.md +++ b/guide/requests-and-responses.md @@ -11,9 +11,9 @@ A handler receives: - `path`: the request target, beginning with `/`; - `headers`: a case-insensitive `Headers` mapping; - `body`: complete request bytes; -- `version`: `HTTP/1.1` for the current network server. +- `version`: `HTTP/1.1` or `HTTP/2`, selected by the listener protocol. -The base parser accepts one origin-form HTTP/1.1 request framed by zero or one +The HTTP/1.1 parser accepts one origin-form request framed by zero or one `Content-Length` header. It rejects transfer encoding, multiple content lengths, missing `Host`, invalid targets, oversized input, and pipelined bytes. It does not decode JSON, forms, query parameters, or text for you. From 4b3540f7da3bcc9d47f7349d649f6377d2482e9d Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 03:09:34 -0500 Subject: [PATCH 10/14] Test regex observer lifecycle ownership --- tests/test_server.py | 79 ++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 77 insertions(+), 2 deletions(-) diff --git a/tests/test_server.py b/tests/test_server.py index ad225b2..5bb3771 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -4,9 +4,14 @@ import warnings from unittest.mock import patch -from smallserver import ServerStartupError, SmallServer +from smallserver import RouteErrorEvent, ServerStartupError, SmallServer from smallserver.errors import _CleanupTransaction -from smallserver.server import HTTPParseError, HTTPRequestParser, ServerConfig +from smallserver.server import ( + HTTPParseError, + HTTPRequestParser, + RouteObserverChannel, + ServerConfig, +) from tests.kernel_fakes import FakeKernel @@ -70,6 +75,76 @@ def resume_task(self, task) -> None: self.assertEqual([handle.name for handle in runtime.kernel.closed], ["listener"]) self.assertEqual(runtime.kernel.wakeup.close_calls, 1) + def test_observer_task_is_owned_by_startup_rollback(self) -> None: + class Runtime: + def __init__(self) -> None: + self.kernel = FakeKernel() + self.tasks = [] + self.cancelled = [] + + def fork(self, tasks) -> None: + self.tasks = list(tasks) + raise RuntimeError("no task capacity") + + def cancel_task(self, task) -> None: + self.cancelled.append(task) + task.cancel() + + def resume_task(self, task) -> None: + pass + + runtime = Runtime() + app = SmallServer(route_error_observer=lambda event: None) + with self.assertRaisesRegex(RuntimeError, "capacity"): + app.serve(runtime) + + self.assertEqual(runtime.cancelled, runtime.tasks) + self.assertEqual( + [task.name for task in runtime.tasks], + [ + "smallserver-listener", + "smallserver-close-watcher", + "smallserver-route-observer", + ], + ) + self.assertEqual([handle.name for handle in runtime.kernel.closed], ["listener"]) + self.assertEqual(runtime.kernel.wakeup.close_calls, 1) + + def test_route_observer_channel_is_bounded_and_stop_wakes_task(self) -> None: + class ObserverTask: + done = False + signals = [] + + @staticmethod + def getID() -> int: + return 9 + + def acceptSignal(self, signal) -> int: + self.signals.append(signal) + return 0 + + class SourceTask: + signals = [] + + def sendSignal(self, task_id, signal) -> int: + self.signals.append((task_id, signal)) + return 0 + + observer_task = ObserverTask() + source_task = SourceTask() + channel = RouteObserverChannel(lambda event: None, max_events=1) + channel.bind(observer_task) + event = RouteErrorEvent("regex-route-1", "route_match_timeout") + + self.assertTrue(channel.enqueue(event, source_task)) + self.assertFalse(channel.enqueue(event, source_task)) + channel.stop() + + self.assertEqual(source_task.signals, [(9, 31)]) + self.assertEqual(observer_task.signals, [31]) + self.assertEqual(channel.dropped, 2) + self.assertEqual(list(channel.events), []) + def test_serve_closes_kernel_resources_when_task_construction_fails(self) -> None: from SmallPackage import SmallTask as RealSmallTask From 9a7e144aa30f89e7d39d19f0aaee47f82649fb28 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 03:20:12 -0500 Subject: [PATCH 11/14] Restore integrated regex protocol coverage --- README.md | 5 +- guide/api-reference.md | 4 +- guide/development.md | 6 +- guide/errors-observability.md | 12 +- guide/index.md | 12 +- tests/test_http2.py | 172 +++++++++++++++++++++++ tests/test_server_runtime.py | 253 +++++++++++++++++++++++++++++++++- 7 files changed, 442 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 411e655..202f25b 100644 --- a/README.md +++ b/README.md @@ -63,8 +63,9 @@ async def get_user(request): ``` Pattern, path, capture, and matching-time limits are configurable with -`RegexRouteConfig`. A match timeout becomes a sanitized 500 response. An -optional `route_error_observer` receives only an immutable `RouteErrorEvent` +`RegexRouteConfig`. On HTTP/1.1 and HTTP/2 listeners, a match timeout becomes a +sanitized 500 response; direct `dispatch()` raises `RouteMatchTimeout`. An +optional network-listener `route_error_observer` receives only an immutable `RouteErrorEvent` with an opaque route ID and category; it never receives the request target, headers, body, traceback, or exception graph. diff --git a/guide/api-reference.md b/guide/api-reference.md index 4dce332..095d8d1 100644 --- a/guide/api-reference.md +++ b/guide/api-reference.md @@ -54,14 +54,14 @@ Read-only properties: `address`, `port`, `closed`, `failure`, `finished`, `cleanup_errors`, `owned_connection_count`, `dropped_route_error_events`, and `route_observer_failures`. +Operations: `close()`, `async close_from_task(task)`, and `finalize()`. + ### `RegexRouteConfig(...)` Finite optional-regex limits. `RouteErrorEvent`, `RouteMatchTimeout`, `RoutePathTooLarge`, and `RegexRoutesUnavailable` describe its bounded error surface. Runtime regex matching requires `smallserver[regex-routes]`. -Operations: `close()`, `async close_from_task(task)`, and `finalize()`. - ## Adapters ### `AdapterRegistry(**adapters)` diff --git a/guide/development.md b/guide/development.md index f9b0fd8..bb8d551 100644 --- a/guide/development.md +++ b/guide/development.md @@ -34,10 +34,12 @@ without installing the test or HTTP/2 extras: python3 -m pip install -r requirements.txt python3 -m pip install -e . python3 -m unittest tests.test_http2 -v +python3 -m unittest tests.test_regex_routing -v ``` -The dependency-contract tests run and HTTP/2 interoperability cases skip -cleanly; importing and testing HTTP/1.1 must not require hyper-h2. +The dependency-contract tests run and HTTP/2 interoperability and regex-engine +cases skip cleanly; importing and testing HTTP/1.1 must require neither +hyper-h2 nor regex. Run the examples when their platform requirements are available: diff --git a/guide/errors-observability.md b/guide/errors-observability.md index ac72a4a..7069de2 100644 --- a/guide/errors-observability.md +++ b/guide/errors-observability.md @@ -13,11 +13,13 @@ The network server converts ordinary handler exceptions into a generic 500. `app.dispatch()` only catches `HTTPError`, so direct dispatch in tests preserves programming errors. -A regex match timeout also becomes a generic 500. If configured, -`route_error_observer` receives exactly one immutable, traceback-free -`RouteErrorEvent` containing only an opaque route ID and category. Delivery is -bounded and scheduler-local; dropped events and observer callback failures are -reported by the corresponding `ServerHandle` counters. +On HTTP/1.1 and HTTP/2 listeners, a regex match timeout becomes a generic 500. +A direct `await app.dispatch(request)` instead raises `RouteMatchTimeout`. If a +network listener has an observer configured, `route_error_observer` receives +exactly one immutable, traceback-free `RouteErrorEvent` containing only an +opaque route ID and category. Delivery is bounded and scheduler-local; dropped +events and observer callback failures are reported by the corresponding +`ServerHandle` counters. ## Configuration errors diff --git a/guide/index.md b/guide/index.md index 1b8e8e0..7fc674a 100644 --- a/guide/index.md +++ b/guide/index.md @@ -1,12 +1,13 @@ # SmallServer guide -This guide documents the API available on the lifecycle base. Start with the -managed server path, then open the focused page for the part you are changing. +This guide documents the integrated lifecycle, regex-routing, and HTTP/2 +feature set. Start with the managed server path, then open the focused page for +the part you are changing. ## Learn SmallServer 1. [Getting started](getting-started.md) — install, create an app, and run it. -2. [Routing](routing.md) — exact paths, methods, 404, and 405 behavior. +2. [Routing](routing.md) — exact and optional regex paths, captures, methods, 404, and 405 behavior. 3. [Requests and responses](requests-and-responses.md) — immutable HTTP values. 4. [Runtime and lifecycle](runtime-lifecycle.md) — managed and caller-owned modes. 5. [Configuration](configuration.md) — finite parser and connection limits. @@ -24,5 +25,6 @@ managed server path, then open the focused page for the part you are changing. - [Protocol roadmap](protocol-roadmap.md) - [Development](development.md) -This branch supports bounded HTTP/1.1 and optional cleartext prior-knowledge -HTTP/2. TLS/ALPN and h2c upgrade remain outside the current protocol boundary. +This branch supports bounded HTTP/1.1, optional timeout-bounded regex routing, +and optional cleartext prior-knowledge HTTP/2. TLS/ALPN and h2c upgrade remain +outside the current protocol boundary. diff --git a/tests/test_http2.py b/tests/test_http2.py index 305a05b..8907d77 100644 --- a/tests/test_http2.py +++ b/tests/test_http2.py @@ -26,6 +26,7 @@ from smallserver import ( HTTP2Config, Headers, + RegexRouteConfig, Request, Response, RouteErrorEvent, @@ -602,6 +603,177 @@ def observe(event): class HTTP2ServerIntegrationTests(unittest.TestCase): _pair = HTTP2ProtocolTests._pair + @unittest.skipUnless( + importlib.util.find_spec("regex") is not None, + "install the smallserver[test] regex extra", + ) + def test_regex_timeout_is_observed_once_without_harming_other_streams(self): + runtime = SmallOS().setKernel(Unix()) + observed = [] + observer_finished = threading.Event() + + def observe(event): + observed.append(event) + observer_finished.set() + raise RuntimeError("intentional observer failure") + + app = SmallServer( + RegexRouteConfig(match_timeout=0.001, total_match_timeout=0.005), + route_error_observer=observe, + ) + + @app.post_regex(r"/(a+)+$") + async def expensive(request): + return Response.text("must not run") + + @app.get("/healthy") + async def healthy(request): + return Response.text("healthy") + + try: + server = app.serve( + runtime, host="127.0.0.1", port=0, protocol="http2" + ) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + + hostile_path = "/" + "a" * 5000 + "!" + authorization_secret = "Bearer h2-private-authorization" + body_secret = b"h2-private-body" + statuses = {} + bodies = {1: bytearray(), 3: bytearray(), 5: bytearray()} + errors = [] + + def client_work(): + try: + client = H2Connection( + config=H2Configuration( + client_side=True, header_encoding="utf-8" + ) + ) + client.initiate_connection() + with socket.create_connection( + ("127.0.0.1", server.port), timeout=3 + ) as connection: + connection.sendall(client.data_to_send()) + client.send_headers( + 1, + [ + (":method", "POST"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", hostile_path), + ("authorization", authorization_secret), + ("content-length", str(len(body_secret))), + ], + ) + client.send_data(1, body_secret, end_stream=True) + client.send_headers( + 3, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/healthy"), + ], + end_stream=True, + ) + connection.sendall(client.data_to_send()) + ended = set() + while not {1, 3}.issubset(ended): + data = connection.recv(65535) + if not data: + raise RuntimeError("HTTP/2 connection ended before sibling response") + for event in client.receive_data(data): + if isinstance(event, ResponseReceived): + statuses[event.stream_id] = dict(event.headers)[":status"] + elif isinstance(event, DataReceived): + bodies[event.stream_id].extend(event.data) + client.acknowledge_received_data( + event.flow_controlled_length, event.stream_id + ) + elif isinstance(event, StreamEnded): + ended.add(event.stream_id) + pending = client.data_to_send() + if pending: + connection.sendall(pending) + + client.send_headers( + 5, + [ + (":method", "GET"), + (":scheme", "http"), + (":authority", "localhost"), + (":path", "/healthy"), + ], + end_stream=True, + ) + connection.sendall(client.data_to_send()) + while 5 not in ended: + data = connection.recv(65535) + if not data: + raise RuntimeError("HTTP/2 connection ended before later response") + for event in client.receive_data(data): + if isinstance(event, ResponseReceived): + statuses[event.stream_id] = dict(event.headers)[":status"] + elif isinstance(event, DataReceived): + bodies[event.stream_id].extend(event.data) + client.acknowledge_received_data( + event.flow_controlled_length, event.stream_id + ) + elif isinstance(event, StreamEnded): + ended.add(event.stream_id) + pending = client.data_to_send() + if pending: + connection.sendall(pending) + + if not observer_finished.wait(2): + raise TimeoutError("route observer did not run") + server.close() + while connection.recv(65535): + pass + except BaseException as exc: + errors.append(exc) + try: + server.close() + except BaseException: + pass + + worker = threading.Thread(target=client_work, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=4) + + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertEqual(statuses, {1: "500", 3: "200", 5: "200"}) + self.assertEqual(bytes(bodies[3]), b"healthy") + self.assertEqual(bytes(bodies[5]), b"healthy") + self.assertNotIn(hostile_path.encode("ascii"), bytes(bodies[1])) + self.assertEqual( + observed, + [RouteErrorEvent("regex-route-1", "route_match_timeout")], + ) + self.assertEqual( + vars(observed[0]), + {"route_id": "regex-route-1", "category": "route_match_timeout"}, + ) + for secret in (hostile_path, authorization_secret, body_secret.decode("ascii")): + self.assertNotIn(secret, repr(observed[0])) + self.assertFalse(hasattr(observed[0], "__traceback__")) + self.assertEqual(server.route_observer_failures, 1) + self.assertEqual(server.dropped_route_error_events, 0) + self.assertTrue(server.finished) + self.assertIsNone(server.failure) + self.assertEqual(server.owned_connection_count, 0) + self.assertEqual(runtime.ioReadWaiters, {}) + self.assertEqual(runtime.ioWriteWaiters, {}) + channel = server._route_observer_channel + self.assertIsNotNone(channel) + assert channel is not None + self.assertIsNone(channel.task) + self.assertEqual(list(channel.events), []) + def test_prior_knowledge_multiplexing_and_graceful_goaway(self): runtime = SmallOS().setKernel(Unix()) app = SmallServer() diff --git a/tests/test_server_runtime.py b/tests/test_server_runtime.py index 07d8e59..ef01f0b 100644 --- a/tests/test_server_runtime.py +++ b/tests/test_server_runtime.py @@ -1,3 +1,6 @@ +import importlib.util +from dataclasses import FrozenInstanceError +import inspect import socket import threading import time @@ -6,16 +9,25 @@ from SmallPackage import SmallOS, Unix from SmallPackage.adapters.threads import ThreadAdapter -from smallserver import AdapterRegistry, Response, SmallServer -from smallserver.server import ServerHandle +from smallserver import ( + AdapterRegistry, + RegexRouteConfig, + Request, + Response, + RouteErrorEvent, + RouteMatchTimeout, + SmallServer, +) +from smallserver.server import ServerHandle, run_route_observer + + +HAS_REGEX = importlib.util.find_spec("regex") is not None class SmallOSServerIntegrationTests(unittest.TestCase): - def _request(self, port: int, path: str) -> bytes: + def _exchange(self, port: int, payload: bytes) -> bytes: with socket.create_connection(("127.0.0.1", port), timeout=3) as connection: - connection.sendall( - "GET {} HTTP/1.1\r\nHost: localhost\r\n\r\n".format(path).encode("ascii") - ) + connection.sendall(payload) chunks = [] while True: chunk = connection.recv(4096) @@ -23,6 +35,12 @@ def _request(self, port: int, path: str) -> bytes: return b"".join(chunks) chunks.append(chunk) + def _request(self, port: int, path: str) -> bytes: + return self._exchange( + port, + "GET {} HTTP/1.1\r\nHost: localhost\r\n\r\n".format(path).encode("ascii"), + ) + def test_loopback_server_accepts_fragmented_request_and_shuts_down(self) -> None: runtime = SmallOS().setKernel(Unix()) app = SmallServer() @@ -186,6 +204,188 @@ def clients() -> None: self.assertEqual(runtime.ioReadWaiters, {}) self.assertEqual(runtime.ioWriteWaiters, {}) + @unittest.skipUnless(HAS_REGEX, "regex-routes extra is not installed") + def test_loopback_regex_route_uses_path_without_query(self) -> None: + runtime = SmallOS().setKernel(Unix()) + app = SmallServer() + + @app.get_regex(r"/files/(?P[^/]+)") + async def file(request): + return Response.text(request.path_params["name"] + "?" + request.query_string) + + try: + server = app.serve(runtime, host="127.0.0.1", port=0) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + + received = [] + errors = [] + + def client() -> None: + try: + received.append(self._request(server.port, "/files/a%2Fb?download=1")) + except BaseException as exc: + errors.append(exc) + finally: + server.close() + + worker = threading.Thread(target=client, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=2) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertIn(b"\r\n\r\na%2Fb?download=1", b"".join(received)) + + @unittest.skipUnless(HAS_REGEX, "regex-routes extra is not installed") + def test_regex_timeout_is_observed_once_and_does_not_stop_server(self) -> None: + runtime = SmallOS().setKernel(Unix()) + observed: list[RouteErrorEvent] = [] + observer_graph = [] + observer_finished = threading.Event() + observer_threads = [] + + def observe(event: RouteErrorEvent) -> None: + observed.append(event) + observer_threads.append(threading.current_thread()) + caller_locals = [] + frame = inspect.currentframe() + while frame is not None: + caller_locals.append(dict(frame.f_locals)) + if frame.f_code is run_route_observer.__code__: + break + frame = frame.f_back + observer_graph.extend(_reachable_container_values(caller_locals)) + observer_finished.set() + raise RuntimeError("intentional observer failure") + + app = SmallServer( + RegexRouteConfig(match_timeout=0.001, total_match_timeout=0.005), + route_error_observer=observe, + ) + + pattern_secret = "sensitive-pattern-marker" + + @app.post_regex(r"/(a+)+$(?#sensitive-pattern-marker)") + async def expensive(request): + return Response() + + @app.get("/health") + async def health(request): + return Response.text("healthy") + + try: + server = app.serve(runtime, host="127.0.0.1", port=0) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + + hostile_path = "/" + "a" * 5000 + "!" + authorization_secret = "Bearer sensitive-authorization-marker" + body_secret = b"sensitive-body-marker" + runtime_thread = threading.Thread( + target=runtime.start, + name="smallos-runtime-test", + daemon=True, + ) + runtime_thread.start() + request = ( + "POST {} HTTP/1.1\r\n" + "Host: localhost\r\n" + "Authorization: {}\r\n" + "Content-Length: {}\r\n\r\n" + ).format(hostile_path, authorization_secret, len(body_secret)).encode("ascii") + received = [self._exchange(server.port, request + body_secret)] + received.append(self._request(server.port, "/health")) + self.assertTrue(observer_finished.wait(2), "route observer did not run") + server.close() + runtime_thread.join(timeout=3) + self.assertFalse(runtime_thread.is_alive()) + self.assertEqual(len(observed), 1) + event = observed[0] + self.assertEqual(event.route_id, "regex-route-1") + self.assertEqual(event.category, "route_match_timeout") + with self.assertRaises(FrozenInstanceError): + event.route_id = "changed" # type: ignore[misc] + self.assertFalse(hasattr(event, "__traceback__")) + self.assertFalse(hasattr(event, "__cause__")) + self.assertFalse(hasattr(event, "__context__")) + + reachable = _reachable_objects(event) + reachable_strings = {value for value in reachable if isinstance(value, str)} + self.assertEqual( + reachable_strings, + {"route_id", "category", "regex-route-1", "route_match_timeout"}, + ) + self.assertFalse(any(isinstance(value, Request) for value in reachable)) + for secret in ( + hostile_path, + authorization_secret, + body_secret.decode("ascii"), + pattern_secret, + ): + self.assertNotIn(secret, reachable_strings) + + caller_strings = {value for value in observer_graph if isinstance(value, str)} + self.assertFalse(any(isinstance(value, Request) for value in observer_graph)) + self.assertFalse( + any(isinstance(value, RouteMatchTimeout) for value in observer_graph) + ) + for secret in ( + hostile_path, + authorization_secret, + body_secret.decode("ascii"), + pattern_secret, + ): + self.assertNotIn(secret, caller_strings) + self.assertNotIn(body_secret, observer_graph) + self.assertEqual(server.route_observer_failures, 1) + self.assertEqual(server.dropped_route_error_events, 0) + self.assertEqual(observer_threads, [runtime_thread]) + channel = server._route_observer_channel + self.assertIsNotNone(channel) + assert channel is not None + self.assertFalse(channel.accepting) + self.assertEqual(list(channel.events), []) + self.assertIsNone(channel.task) + self.assertTrue(received[0].startswith(b"HTTP/1.1 500 Internal Server Error\r\n")) + self.assertNotIn(hostile_path.encode("ascii"), received[0]) + self.assertTrue(received[1].startswith(b"HTTP/1.1 200 OK\r\n")) + self.assertTrue(received[1].endswith(b"healthy")) + + @unittest.skipUnless(HAS_REGEX, "regex-routes extra is not installed") + def test_regex_path_limit_returns_414_before_matching(self) -> None: + runtime = SmallOS().setKernel(Unix()) + app = SmallServer(RegexRouteConfig(max_path_bytes=8)) + + @app.get_regex(r"/.*") + async def route(request): + return Response.text("must not run") + + try: + server = app.serve(runtime, host="127.0.0.1", port=0) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + + received = [] + errors = [] + + def client() -> None: + try: + received.append(self._request(server.port, "/12345678")) + except BaseException as exc: + errors.append(exc) + finally: + server.close() + + worker = threading.Thread(target=client, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=2) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertTrue(received[0].startswith(b"HTTP/1.1 414 URI Too Long\r\n")) + self.assertNotIn(b"must not run", received[0]) + def test_managed_listen_serves_loopback_and_returns_closed_handle(self) -> None: app = SmallServer() @@ -229,3 +429,44 @@ def run_server() -> None: self.assertTrue(handle.closed) self.assertEqual(handle.port, returned[0].port) self.assertIn(b"HTTP/1.1 200 OK", response) + + +def _reachable_objects(root): + pending = [root] + seen = set() + result = [] + while pending: + value = pending.pop() + identity = id(value) + if identity in seen: + continue + seen.add(identity) + result.append(value) + if isinstance(value, dict): + pending.extend(value.keys()) + pending.extend(value.values()) + elif isinstance(value, (tuple, list, set, frozenset)): + pending.extend(value) + elif hasattr(value, "__dict__"): + pending.append(vars(value)) + return result + + +def _reachable_container_values(root): + """Walk frame-local containers without traversing scheduler object graphs.""" + pending = [root] + seen = set() + result = [] + while pending: + value = pending.pop() + identity = id(value) + if identity in seen: + continue + seen.add(identity) + result.append(value) + if isinstance(value, dict): + pending.extend(value.keys()) + pending.extend(value.values()) + elif isinstance(value, (tuple, list, set, frozenset)): + pending.extend(value) + return result From 83e42f0fa015e9d2a39cb8857af8d32bf0f9c7cb Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 03:22:37 -0500 Subject: [PATCH 12/14] docs: qualify bounded route event delivery --- guide/errors-observability.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/guide/errors-observability.md b/guide/errors-observability.md index 7069de2..4e8ae56 100644 --- a/guide/errors-observability.md +++ b/guide/errors-observability.md @@ -15,11 +15,11 @@ programming errors. On HTTP/1.1 and HTTP/2 listeners, a regex match timeout becomes a generic 500. A direct `await app.dispatch(request)` instead raises `RouteMatchTimeout`. If a -network listener has an observer configured, `route_error_observer` receives -exactly one immutable, traceback-free `RouteErrorEvent` containing only an -opaque route ID and category. Delivery is bounded and scheduler-local; dropped -events and observer callback failures are reported by the corresponding -`ServerHandle` counters. +network listener has an observer configured, SmallServer attempts to enqueue at +most one immutable, traceback-free `RouteErrorEvent` containing only an opaque +route ID and category. Delivery is bounded and scheduler-local, so saturation, +signal failure, or shutdown may drop the event. Dropped events and observer +callback failures are reported by the corresponding `ServerHandle` counters. ## Configuration errors From 085aafba6a66d8d6076ca3f370f7c141c1b61f65 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sun, 23 Aug 2026 16:10:31 -0500 Subject: [PATCH 13/14] docs: integrate managed runtime configuration --- README.md | 11 ++++++----- guide/api-reference.md | 5 +++++ guide/configuration.md | 26 ++++++++++++++++++++++---- guide/runtime-lifecycle.md | 10 ++++++---- 4 files changed, 39 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 1604011..d009368 100644 --- a/README.md +++ b/README.md @@ -96,11 +96,12 @@ app.listen(host="127.0.0.1", port=8000, config=config) This bridge is only for SmallServer-owned runtimes. If you supply `runtime=`, configure it directly with `SmallOS(config=...)`; SmallServer rejects `managed_runtime` rather than mutating caller-owned scheduler state. -`task_capacity` must reserve at least `max_connections + 2` task slots for the -listener and shutdown-control tasks, and both server task priorities must be -below `priority_levels`. Configuring a regex route-error observer adds one -dedicated SmallOS task, so that mode requires at least `max_connections + 3` -slots. +For HTTP/1.1, `task_capacity` must reserve at least `max_connections + 2` task +slots for the listener and shutdown-control tasks, and both server task +priorities must be below `priority_levels`. Configuring a regex route-error +observer adds one dedicated SmallOS task, so that mode requires at least +`max_connections + 3` slots. HTTP/2 needs additional headroom for its bounded +connection-control and stream-handler tasks. ## Current boundaries diff --git a/guide/api-reference.md b/guide/api-reference.md index 095d8d1..92f6e37 100644 --- a/guide/api-reference.md +++ b/guide/api-reference.md @@ -42,6 +42,11 @@ provide common construction and serialization paths. Frozen finite-limit configuration. See [Configuration](configuration.md). +### `ManagedRuntimeConfig(...)` + +Frozen SmallOS settings used only when `listen()` creates the runtime. A +caller-supplied runtime retains its own configuration. + ### `HTTP2Config(...)` Optional cleartext HTTP/2 stream, buffer, frame-batch, and timeout limits. See diff --git a/guide/configuration.md b/guide/configuration.md index 0dfe9f1..a3a1e88 100644 --- a/guide/configuration.md +++ b/guide/configuration.md @@ -4,7 +4,7 @@ Pass a `ServerConfig` to `listen()` or `serve()` to tune finite listener, parser, and scheduling limits. ```python -from smallserver import ServerConfig, SmallServer +from smallserver import ManagedRuntimeConfig, ServerConfig, SmallServer app = SmallServer() config = ServerConfig( @@ -18,6 +18,7 @@ config = ServerConfig( accept_batch_size=16, max_request_target_bytes=8 * 1024, max_route_error_events=16, + managed_runtime=ManagedRuntimeConfig(task_capacity=256), ) ``` @@ -33,10 +34,12 @@ config = ServerConfig( | `accept_batch_size` | 16 | Accepts before the listener explicitly yields. | | `max_request_target_bytes` | 8 KiB | Maximum HTTP/1.1 origin-form request target. | | `max_route_error_events` | 16 | Bounded sanitized regex-timeout observer queue. | +| `managed_runtime` | `None` | Optional SmallOS settings used only when `listen()` creates the runtime. | -Every field must be a positive integer; booleans are rejected. The public port -must be an integer from 0 through 65535. `port=0` delegates port selection to -the kernel. +Every numeric `ServerConfig` field must be a positive integer; booleans are +rejected. `managed_runtime` must be `None` or a `ManagedRuntimeConfig`. The +public port must be an integer from 0 through 65535. `port=0` delegates port +selection to the kernel. At connection capacity, the listener waits on a scheduler signal instead of accepting and discarding more streams. Connections whose close failed still @@ -47,6 +50,21 @@ Limits are per `ServerHandle`. They bound HTTP input and framework-owned connections, but they do not limit memory allocated by your handlers, response bodies, adapter queues, or downstream libraries; configure those separately. +## Managed runtime configuration + +`ManagedRuntimeConfig` controls the SmallOS instance created by blocking +`app.listen()` when no runtime is supplied. It exposes `task_capacity`, +`priority_levels`, `io_buffer_length`, `eternal_watchers`, and immutable +per-client `client_defaults`. Caller-owned runtimes must be configured directly; +SmallServer rejects `ServerConfig(managed_runtime=...)` when `runtime=` is +provided. + +The managed task capacity must cover `max_connections + 2` for HTTP/1.1's +listener and shutdown-control tasks. Configuring `route_error_observer` adds +one dedicated task, raising that floor to `max_connections + 3`. HTTP/2 also +creates bounded connection-control and stream-handler tasks, so configure +additional capacity from the selected `HTTP2Config` concurrency limits. + ## HTTP/2 configuration Pass `protocol="http2"` and an optional `HTTP2Config` for cleartext diff --git a/guide/runtime-lifecycle.md b/guide/runtime-lifecycle.md index d67611e..348d5b9 100644 --- a/guide/runtime-lifecycle.md +++ b/guide/runtime-lifecycle.md @@ -23,10 +23,12 @@ async def index(request): app.listen(host="127.0.0.1", port=8000) ``` -With no `runtime`, `listen()` lazily creates `SmallOS().setKernel(Unix())`, -starts it, blocks until shutdown, and finalizes server-owned resources. In this -managed mode, Ctrl-C is consumed after successful cleanup and the closed -`ServerHandle` is returned. +With no `runtime`, `listen()` lazily creates a configured `SmallOS` with the +Unix kernel, starts it, blocks until shutdown, and finalizes server-owned +resources. `ServerConfig.managed_runtime` accepts a `ManagedRuntimeConfig` for +scheduler capacity, priority, I/O-buffer, watcher, and client-default settings. +In this managed mode, Ctrl-C is consumed after successful cleanup and the +closed `ServerHandle` is returned. ## Caller-owned runtime From 5ba6cb50b7fd95585a5004c54795d048c6fac115 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sun, 23 Aug 2026 16:25:39 -0500 Subject: [PATCH 14/14] Fix coalesced WebSocket close test race --- tests/test_websocket.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/tests/test_websocket.py b/tests/test_websocket.py index 9755560..3edd8c0 100644 --- a/tests/test_websocket.py +++ b/tests/test_websocket.py @@ -1126,8 +1126,14 @@ def client_work() -> None: response = b"" while b"\r\n\r\n" not in response: response += stream.recv(4096) + _, _, websocket_data = response.partition(b"\r\n\r\n") client = api.Connection(api.ConnectionType.CLIENT) - events = _receive_events(stream, client, api.CloseConnection) + events = _receive_events( + stream, + client, + api.CloseConnection, + initial_data=websocket_data, + ) close_events.extend(events) close_event = next( event @@ -1155,11 +1161,13 @@ def client_work() -> None: self.assertTrue(server.finished) -def _receive_events(stream, connection, event_type): +def _receive_events(stream, connection, event_type, *, initial_data=b""): deadline = time.monotonic() + 3 received = [] + pending = initial_data while time.monotonic() < deadline: - data = stream.recv(4096) + data = pending or stream.recv(4096) + pending = b"" if not data: return received connection.receive_data(data)