From 56616961bfe5ba821de6f4aa9fee885295405bcf Mon Sep 17 00:00:00 2001 From: Oaklight <13076552+Oaklight@users.noreply.github.com> Date: Mon, 28 Sep 2026 09:47:59 +0000 Subject: [PATCH] chore(deps): update vendored zerodep modules --- src/veilrender/_vendor/httpserver.py | 295 +++++++++++++++++++++++++-- 1 file changed, 281 insertions(+), 14 deletions(-) diff --git a/src/veilrender/_vendor/httpserver.py b/src/veilrender/_vendor/httpserver.py index 55e4b2b..e37a235 100644 --- a/src/veilrender/_vendor/httpserver.py +++ b/src/veilrender/_vendor/httpserver.py @@ -1,5 +1,5 @@ # /// zerodep -# version = "0.4.0" +# version = "0.5.1" # deps = [] # tier = "subsystem" # category = "network" @@ -45,6 +45,7 @@ def echo(request): import os import re import signal +import ssl import sys from collections.abc import AsyncIterator, Callable from email.utils import formatdate @@ -122,6 +123,11 @@ def echo(request): _SENTINEL = object() +# Seconds between disconnect checks while waiting for the next chunk from +# a streaming generator. Only matters when the generator is blocked on +# slow upstream I/O; fast generators never hit the timeout. +_STREAMING_DISCONNECT_CHECK_INTERVAL: float = 30.0 + # ── Exceptions ─────────────────────────────────────────────────────────────── @@ -509,6 +515,8 @@ class StreamingResponse: background: Optional callable invoked after the stream completes (including client disconnect). Accepts both sync and async callables. Exceptions are logged and suppressed. + disconnect_check_interval: Seconds between client-disconnect + checks while the generator is blocked. Defaults to 30. """ __slots__ = ( @@ -518,6 +526,7 @@ class StreamingResponse: "content_type", "background", "_cookie_headers", + "_disconnect_check_interval", ) def __init__( @@ -527,6 +536,7 @@ def __init__( headers: dict[str, str] | None = None, content_type: str = "application/octet-stream", background: Callable[[], Any] | None = None, + disconnect_check_interval: float | None = None, ): self._generator = generator self.status_code = status_code @@ -534,6 +544,11 @@ def __init__( self.content_type = content_type self.background = background self._cookie_headers: list[str] = [] + self._disconnect_check_interval = ( + disconnect_check_interval + if disconnect_check_interval is not None + else _STREAMING_DISCONNECT_CHECK_INTERVAL + ) def set_cookie( self, @@ -578,7 +593,14 @@ def delete_cookie( ) async def _write(self, writer: asyncio.StreamWriter) -> None: - """Write status line, headers, then stream the body.""" + """Write status line, headers, then stream the body. + + Raises ``BrokenPipeError``, ``ConnectionResetError``, or + ``ConnectionAbortedError`` on client disconnect. The caller + (``_handle_connection``) is responsible for handling these. + Generator cleanup and the background callback still run via + the ``finally`` block before the exception propagates. + """ reason = _STATUS_REASONS.get(self.status_code, "Unknown") is_sse = self.content_type.startswith("text/event-stream") @@ -600,7 +622,49 @@ async def _write(self, writer: asyncio.StreamWriter) -> None: await writer.drain() try: - async for chunk in self._generator: + aiter = self._generator.__aiter__() + is_closing = getattr(writer, "is_closing", None) + while True: + # Detect client disconnect between generator yields. + # When the generator blocks on slow upstream I/O (e.g. + # an LLM API producing the next SSE chunk), plain + # ``async for`` never checks the writer state. We + # race the next-chunk coroutine against a periodic + # disconnect check so CLOSE_WAIT connections don't + # accumulate. + if is_closing is not None and is_closing(): + raise BrokenPipeError("client disconnected (detected)") + + next_coro = aiter.__anext__() + if is_closing is None: + # Writer has no is_closing (e.g. mock) — fall back + # to plain await without disconnect detection. + try: + chunk = await next_coro + except StopAsyncIteration: + break + else: + next_task = asyncio.ensure_future(next_coro) + try: + chunk = await _anext_or_disconnect( + next_task, + is_closing, + self._disconnect_check_interval, + ) + except StopAsyncIteration: + break + except ( + BrokenPipeError, + ConnectionResetError, + ConnectionAbortedError, + ): + next_task.cancel() + try: + await next_task + except (asyncio.CancelledError, Exception): + pass + raise + if isinstance(chunk, str): chunk = chunk.encode("utf-8") writer.write(f"{len(chunk):x}\r\n".encode("latin-1")) @@ -610,7 +674,7 @@ async def _write(self, writer: asyncio.StreamWriter) -> None: writer.write(b"0\r\n\r\n") await writer.drain() except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): - logger.debug("Client disconnected during streaming") + raise finally: aclose = getattr(self._generator, "aclose", None) if aclose is not None: @@ -795,6 +859,36 @@ async def _read_chunked_body( # ── Utilities ──────────────────────────────────────────────────────────────── +async def _anext_or_disconnect( + next_task: asyncio.Task[Any], + is_closing: Callable[[], bool], + interval: float, +) -> Any: + """Await *next_task* while periodically checking for client disconnect. + + Polls ``is_closing()`` every *interval* seconds. If the writer is + closing, cancels the pending ``__anext__`` task and raises + ``BrokenPipeError`` so the caller can clean up the generator. + + If the task completes normally the chunk value is returned. If it + raises ``StopAsyncIteration`` (generator exhausted) that exception + propagates to the caller unchanged. + """ + while True: + try: + return await asyncio.wait_for(asyncio.shield(next_task), timeout=interval) + except asyncio.TimeoutError: + if is_closing(): + next_task.cancel() + try: + await next_task + except (asyncio.CancelledError, StopAsyncIteration, Exception): + pass + raise BrokenPipeError("client disconnected (detected)") + # Generator is still working on the next chunk; keep waiting. + continue + + def _http_date(timestamp: float | None = None) -> str: """Format a timestamp as an HTTP-date (RFC 7231).""" return formatdate(timeval=timestamp, localtime=False, usegmt=True) @@ -875,6 +969,12 @@ def _resolve_static_file( class App: """Async HTTP server application. + Request lifecycle:: + + before_request → route handler → after_request + → on_response_started → write → on_response_completed + ↘ on_client_disconnect + Args: max_body_size: Maximum request body size in bytes. read_timeout: Timeout for reading a single request (seconds). @@ -902,6 +1002,9 @@ def __init__( self._after_request_handlers: list[Callable[..., Any]] = [] self._startup_handlers: list[Callable[[], Any]] = [] self._shutdown_handlers: list[Callable[[], Any]] = [] + self._on_response_started_handlers: list[Callable[..., Any]] = [] + self._on_response_completed_handlers: list[Callable[..., Any]] = [] + self._on_client_disconnect_handlers: list[Callable[..., Any]] = [] self._error_handlers: dict[int | type, Callable[..., Any]] = {} self._server: asyncio.Server | None = None self._shutdown_event: asyncio.Event | None = None @@ -1060,6 +1163,82 @@ async def close_pool(): self._shutdown_handlers.append(handler) return handler + # ── Request Lifecycle Signals ──────────────────────────────────────── + + def on_response_started(self, handler: Callable[..., Any]) -> Callable[..., Any]: + """Register a response-started signal. + + Fired just before the response is written to the client. Useful + for TTFB (time-to-first-byte) metrics. + + The handler receives ``(request, response)`` and its return value + is ignored. Both sync and async callables are supported. + Exceptions are logged and suppressed. Handlers run sequentially; + keep them fast to avoid delaying the response write. + + Example:: + + @app.on_response_started + async def ttfb(request, response): + request.state.response_start = time.monotonic() + """ + self._on_response_started_handlers.append(handler) + return handler + + def on_response_completed(self, handler: Callable[..., Any]) -> Callable[..., Any]: + """Register a response-completed signal. + + Fired after the entire response body has been successfully sent + to the client. Useful for total transfer time and post-response + logging. + + The handler receives ``(request, response)`` and its return value + is ignored. Both sync and async callables are supported. + Exceptions are logged and suppressed. Handlers run sequentially. + + Not fired when the client disconnects mid-response (see + :meth:`on_client_disconnect` for that case). + + Example:: + + @app.on_response_completed + async def log_transfer(request, response): + elapsed = time.monotonic() - request.state.response_start + logger.info("Sent %d in %.3fs", response.status_code, elapsed) + """ + self._on_response_completed_handlers.append(handler) + return handler + + def on_client_disconnect(self, handler: Callable[..., Any]) -> Callable[..., Any]: + """Register a client-disconnect signal. + + Fired when the client closes the connection during response + delivery (broken pipe, connection reset). Does not fire for + disconnects during request dispatch (e.g. slow handler). + Useful for cleanup, metrics, and cancelling expensive work. + + The handler receives ``(request)`` and its return value is ignored. + Both sync and async callables are supported. Exceptions are logged + and suppressed. + + Example:: + + @app.on_client_disconnect + async def on_disconnect(request): + logger.info("Client %s disconnected", request.client_addr) + """ + self._on_client_disconnect_handlers.append(handler) + return handler + + async def _fire_signal( + self, name: str, handlers: list[Callable[..., Any]], *args: Any + ) -> None: + for hook in handlers: + try: + await _invoke(hook, *args) + except Exception: + logger.warning("%s hook failed", name, exc_info=True) + # ── Request Dispatch ───────────────────────────────────────────────── def _match_route( @@ -1264,10 +1443,35 @@ async def _handle_connection( logger.debug("%s %s from %s", method, path, client_addr) try: - response = await self._dispatch(request) - await response._write(writer) - except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): - logger.debug("Connection reset by %s during response", client_addr) + try: + response = await self._dispatch(request) + except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): + logger.debug("Connection reset by %s during dispatch", client_addr) + return + + await self._fire_signal( + "on_response_started", + self._on_response_started_handlers, + request, + response, + ) + + try: + await response._write(writer) + except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): + await self._fire_signal( + "on_client_disconnect", + self._on_client_disconnect_handlers, + request, + ) + logger.debug("Connection reset by %s during response", client_addr) + else: + await self._fire_signal( + "on_response_completed", + self._on_response_completed_handlers, + request, + response, + ) except Exception: logger.exception("Error writing response to %s", client_addr) finally: @@ -1322,6 +1526,10 @@ def run( port: int = DEFAULT_PORT, *, socket: str | None = None, + ssl_context: ssl.SSLContext | None = None, + backlog: int | None = None, + reuse_address: bool | None = None, + reuse_port: bool | None = None, ) -> None: """Start the server (blocking). @@ -1333,13 +1541,44 @@ def run( on a Unix socket instead of TCP. The socket file permissions are restricted to owner-only (``0o600``) after creation. Only available on Unix-like systems. + ssl_context: An :class:`ssl.SSLContext` for TLS termination. + When provided, the server accepts HTTPS connections. + The caller is responsible for configuring the context + (loading certs, setting verify mode, etc.). + backlog: Maximum number of queued connections passed to + :func:`asyncio.start_server`. ``None`` leaves the OS + default (typically 128). + reuse_address: Sets ``SO_REUSEADDR``. ``None`` lets asyncio + decide (``True`` on non-Windows). + reuse_port: Sets ``SO_REUSEPORT`` for multi-process load + balancing. ``None`` lets asyncio decide (``False``). """ try: - asyncio.run(self._serve(host, port, socket=socket)) + asyncio.run( + self._serve( + host, + port, + socket=socket, + ssl_context=ssl_context, + backlog=backlog, + reuse_address=reuse_address, + reuse_port=reuse_port, + ) + ) except KeyboardInterrupt: pass - async def _serve(self, host: str, port: int, *, socket: str | None = None) -> None: + async def _serve( + self, + host: str, + port: int, + *, + socket: str | None = None, + ssl_context: ssl.SSLContext | None = None, + backlog: int | None = None, + reuse_address: bool | None = None, + reuse_port: bool | None = None, + ) -> None: """Internal async server loop.""" self._shutdown_event = asyncio.Event() self._loop = asyncio.get_running_loop() @@ -1350,19 +1589,31 @@ async def _serve(self, host: str, port: int, *, socket: str | None = None) -> No await self._run_startup_hooks() if socket: - server = await self._start_unix_socket(socket) + server = await self._start_unix_socket( + socket, ssl_context=ssl_context, backlog=backlog + ) else: + kwargs: dict[str, Any] = {} + if backlog is not None: + kwargs["backlog"] = backlog + if reuse_address is not None: + kwargs["reuse_address"] = reuse_address + if reuse_port is not None: + kwargs["reuse_port"] = reuse_port server = await asyncio.start_server( self._handle_connection, host, port, + ssl=ssl_context, + **kwargs, ) addrs = ( server.sockets[0].getsockname() if server.sockets else (host, port) ) self.host = addrs[0] self.port = addrs[1] - logger.info("Serving on %s:%d", self.host, self.port) + scheme = "https" if ssl_context else "http" + logger.info("Serving on %s://%s:%d", scheme, self.host, self.port) self._server = server @@ -1380,7 +1631,13 @@ async def _serve(self, host: str, port: int, *, socket: str | None = None) -> No # Run shutdown hooks after server stops accepting connections await self._run_shutdown_hooks() - async def _start_unix_socket(self, socket_path: str) -> asyncio.Server: + async def _start_unix_socket( + self, + socket_path: str, + *, + ssl_context: ssl.SSLContext | None = None, + backlog: int | None = None, + ) -> asyncio.Server: """Start listening on a Unix domain socket. Handles stale socket cleanup, permission hardening (``0o600``), @@ -1388,6 +1645,10 @@ async def _start_unix_socket(self, socket_path: str) -> asyncio.Server: Args: socket_path: Path for the Unix domain socket file. + ssl_context: Optional :class:`ssl.SSLContext` for TLS over + the Unix socket. + backlog: Maximum number of queued connections. ``None`` + leaves the OS default. Returns: The ``asyncio.Server`` instance. @@ -1420,9 +1681,14 @@ async def _start_unix_socket(self, socket_path: str) -> asyncio.Server: logger.error("Socket parent directory does not exist: %s", parent) sys.exit(1) + kwargs: dict[str, Any] = {} + if backlog is not None: + kwargs["backlog"] = backlog server = await asyncio.start_unix_server( self._handle_connection, path=path, + ssl=ssl_context, + **kwargs, ) # Restrict permissions to owner-only @@ -1430,7 +1696,8 @@ async def _start_unix_socket(self, socket_path: str) -> asyncio.Server: os.chmod(path, 0o600) self._socket_path = path - logger.info("Serving on unix:%s (mode 0600)", path) + scheme = "https+unix" if ssl_context else "unix" + logger.info("Serving on %s:%s (mode 0600)", scheme, path) return server def _cleanup_socket(self) -> None: