From 886e8930b2ef5ab9cc92d4fe87b117de642d912a Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Tue, 15 Sep 2026 18:39:54 +0000 Subject: [PATCH] fix(insight): make HttpExporter timeout_ms a request deadline - http_send: build the opener per request with handlers that create deadline-aware connections; a timer shuts the live socket down when the deadline expires and TimeoutError is raised, so a peer trickling bytes can no longer keep a request alive past timeout_ms - every TCP attempt is budgeted with the remaining time and registered with the deadline, so a stalled connect is aborted on time; expiry is judged by the monotonic clock after the request and after the error body read, so a late response is never reported as success - the urllib transport is otherwise unchanged (redirects refused, header merging, IPv6 hosts, proxy settings, default TLS context) - tests: trickled status line, trickled error body, slow name resolution, stalled connect, header case merge, unsupported scheme --- .../exporters/_common.py | 231 +++++++++++++++++- .../exporters/http_exporter.py | 7 +- .../tests/test_http_exporter.py | 158 +++++++++++- 3 files changed, 384 insertions(+), 12 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py index 28ef3626..74dec07f 100644 --- a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py @@ -11,11 +11,16 @@ from __future__ import annotations import datetime +import http.client import json import re +import socket +import threading +import time import urllib.error import urllib.request from typing import Any +from urllib.parse import urlsplit _SQL_IDENTIFIER = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$") @@ -36,7 +41,179 @@ def redirect_request( # type: ignore[override] # stdlib signature has no hints return None -_OPENER = urllib.request.build_opener(_NoRedirect()) +class _Deadline: + """A whole-request time budget shared by the timer and the connection. + + ``expired()`` is judged by the monotonic clock, so a response that lands + after the budget is never accepted even if the timer callback runs late. + The timer's job is only to wake a blocked socket. + """ + + def __init__(self, seconds: float) -> None: + self.seconds = seconds + self.expires_at = time.monotonic() + seconds + self._fired = threading.Event() + self._lock = threading.Lock() + self._sockets: list[socket.socket] = [] + + def expired(self) -> bool: + return self._fired.is_set() or time.monotonic() >= self.expires_at + + def remaining(self) -> float: + # Never hand the socket layer zero or a negative value: those mean + # non-blocking / blocking, not "no time left". + return max(self.expires_at - time.monotonic(), 0.001) + + def register(self, sock: socket.socket) -> None: + """Make ``sock`` reachable by ``fire`` (an in-flight connect or the live socket).""" + with self._lock: + self._sockets.append(sock) + already_fired = self._fired.is_set() + if already_fired: + _shutdown(sock) + + def fire(self) -> None: + self._fired.set() + with self._lock: + sockets = list(self._sockets) + for sock in sockets: + _shutdown(sock) + + +def _shutdown(sock: socket.socket) -> None: + try: + sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass + + +def _connect_within( + deadline: _Deadline, + address: tuple[str, int], + timeout: float | None, + source_address: tuple[str, int] | None = None, +) -> socket.socket: + """``socket.create_connection`` with the remaining budget per attempt. + + Every candidate socket is registered with the deadline before connecting, + so the timer can abort an attempt that is still waiting for the peer. Name + resolution itself cannot be interrupted. + """ + del timeout # the deadline, not the connection's static timeout, rules here + host, port = address + if deadline.expired(): + msg = "deadline expired before connecting" + raise TimeoutError(msg) + last_error: OSError | None = None + for family, kind, proto, _, sockaddr in socket.getaddrinfo( + host, port, 0, socket.SOCK_STREAM + ): + if deadline.expired(): + msg = "deadline expired while connecting" + raise TimeoutError(msg) + sock = socket.socket(family, kind, proto) + try: + sock.settimeout(deadline.remaining()) + if source_address: + sock.bind(source_address) + deadline.register(sock) + sock.connect(sockaddr) + except OSError as exc: + sock.close() + last_error = exc + continue + return sock + if last_error is not None: + raise last_error + msg = f"getaddrinfo returned no addresses for {host!r}" + raise OSError(msg) + + +def _bind_deadline( + conn: http.client.HTTPConnection, deadline: _Deadline | None +) -> None: + # ``_create_connection`` is the connection's socket factory; swapping it + # keeps the stdlib connect (TLS wrapping, ALPN, tunnelling) intact while + # every TCP attempt is budgeted and interruptible. + if deadline is not None and hasattr(conn, "_create_connection"): + conn._create_connection = ( # type: ignore[attr-defined] # stdlib hook + lambda address, timeout=None, source_address=None: _connect_within( + deadline, address, timeout, source_address + ) + ) + + +def _after_connect( + conn: http.client.HTTPConnection, deadline: _Deadline | None +) -> None: + if deadline is None: + return + if conn.sock is not None: + deadline.register(conn.sock) # the (possibly TLS-wrapped) live socket + if deadline.expired(): + conn.close() + msg = "deadline expired while connecting" + raise TimeoutError(msg) + + +class _HTTPConnection(http.client.HTTPConnection): + def __init__(self, *args: Any, deadline: _Deadline | None, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self._deadline = deadline + _bind_deadline(self, deadline) + + def connect(self) -> None: + super().connect() + _after_connect(self, self._deadline) + + +class _HTTPSConnection(http.client.HTTPSConnection): + def __init__(self, *args: Any, deadline: _Deadline | None, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self._deadline = deadline + _bind_deadline(self, deadline) + + def connect(self) -> None: + super().connect() + _after_connect(self, self._deadline) + + +class _HTTPHandler(urllib.request.HTTPHandler): + def __init__(self, deadline: _Deadline | None) -> None: + super().__init__() + self._deadline = deadline + + def http_open(self, req: urllib.request.Request) -> http.client.HTTPResponse: + deadline = self._deadline + + def factory(*args: Any, **kwargs: Any) -> _HTTPConnection: + return _HTTPConnection(*args, deadline=deadline, **kwargs) + + return self.do_open(factory, req) # type: ignore[arg-type] # factory, not class + + +class _HTTPSHandler(urllib.request.HTTPSHandler): + def __init__(self, deadline: _Deadline | None) -> None: + super().__init__() + self._deadline = deadline + + def https_open(self, req: urllib.request.Request) -> http.client.HTTPResponse: + deadline = self._deadline + + def factory(*args: Any, **kwargs: Any) -> _HTTPSConnection: + return _HTTPSConnection(*args, deadline=deadline, **kwargs) + + # No context is configured on this handler, so the connection builds the + # stdlib default (certificate verification, ALPN http/1.1). + return self.do_open(factory, req) # type: ignore[arg-type] # factory, not class + + +def _opener(deadline: _Deadline | None) -> urllib.request.OpenerDirector: + # build_opener keeps the default handlers (proxy discovery, header merging, + # IPv6 hosts, error handling) and swaps in ours where classes overlap. + return urllib.request.build_opener( + _NoRedirect(), _HTTPHandler(deadline), _HTTPSHandler(deadline) + ) def compact_dumps(value: Any) -> str: @@ -89,17 +266,53 @@ def http_send( message. Redirects are not followed: a 3xx is returned like any other failure. ``error_text`` is the first ``_MAX_ERROR_BODY_BYTES`` of a non-2xx response body and empty on success; a success body is never read. Network - errors and timeouts propagate. + errors propagate. + + ``timeout`` (seconds) is a deadline for the whole request: connecting, + sending, and receiving the status, headers and any error body. On expiry + the live socket is shut down and ``TimeoutError`` is raised, even against + a peer that keeps the connection alive by trickling bytes, and a response + that completes after the deadline is never reported as success. Name + resolution cannot be interrupted. ``None`` means no limit. """ + scheme = urlsplit(url).scheme + if scheme not in ("http", "https"): + msg = f"Unsupported URL scheme {scheme!r} in {url!r} (need http or https)" + raise ValueError(msg) request = urllib.request.Request(url, data=body, method=method) for key, value in headers.items(): request.add_header(key, value) + + deadline = _Deadline(timeout) if timeout is not None else None + timer: threading.Timer | None = None + if deadline is not None: + timer = threading.Timer(deadline.seconds, deadline.fire) + timer.daemon = True + timer.start() + timed_out = f"request to {url} exceeded {timeout}s" try: - with _OPENER.open(request, timeout=timeout) as response: # noqa: S310 - return int(response.status), str(response.reason or ""), "" - except urllib.error.HTTPError as exc: try: - detail = exc.read(_MAX_ERROR_BODY_BYTES).decode("utf-8", errors="replace") - except Exception: # noqa: BLE001 - the body is best-effort detail only - detail = "" - return int(exc.code), str(exc.reason or ""), detail + with _opener(deadline).open(request, timeout=timeout) as response: # noqa: S310 + status = int(response.status) + reason = str(response.reason or "") + detail = "" + except urllib.error.HTTPError as exc: + status = int(exc.code) + reason = str(exc.reason or "") + try: + detail = exc.read(_MAX_ERROR_BODY_BYTES).decode( + "utf-8", errors="replace" + ) + except Exception: # noqa: BLE001 - the body is best-effort detail only + detail = "" + except Exception as exc: + if deadline is not None and deadline.expired(): + raise TimeoutError(timed_out) from exc + raise + finally: + if timer is not None: + timer.cancel() + # Whatever arrived after the deadline is not a delivery. + if deadline is not None and deadline.expired(): + raise TimeoutError(timed_out) + return status, reason, detail diff --git a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/http_exporter.py b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/http_exporter.py index fec78281..4cf26a1f 100644 --- a/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/http_exporter.py +++ b/packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/http_exporter.py @@ -33,8 +33,11 @@ class HttpMethod(StrEnum): class HttpExporter: """Sends each record as a JSON body to any HTTP endpoint. - The endpoint must answer 2xx; any other status raises. ``timeout_ms`` - bounds the whole request (default 10 seconds). ``max_record_size_bytes`` + The endpoint must answer 2xx; any other status raises. ``timeout_ms`` is a + deadline for the whole request (connect, send, response headers and any + error body; default 10 seconds); on expiry the connection is shut down and + ``TimeoutError`` is raised. Name resolution is not interruptible. + ``max_record_size_bytes`` has no default because a generic endpoint has no known limit. """ diff --git a/packages/aws-durable-execution-sdk-python-insight/tests/test_http_exporter.py b/packages/aws-durable-execution-sdk-python-insight/tests/test_http_exporter.py index 6fd64d6b..aa62fdb1 100644 --- a/packages/aws-durable-execution-sdk-python-insight/tests/test_http_exporter.py +++ b/packages/aws-durable-execution-sdk-python-insight/tests/test_http_exporter.py @@ -6,11 +6,12 @@ from __future__ import annotations import json +import socket import threading import time from dataclasses import dataclass, field from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -from typing import Any, Iterator +from typing import Any, Callable, Iterator import pytest @@ -60,6 +61,7 @@ class CapturedRequest: path: str headers: dict[str, str] body: bytes + header_counts: dict[str, int] = field(default_factory=dict) @dataclass @@ -88,6 +90,10 @@ def _handle(self) -> None: path=self.path, headers={k.lower(): v for k, v in self.headers.items()}, body=body, + header_counts={ + k.lower(): len(self.headers.get_all(k) or []) + for k in self.headers + }, ) ) if capture.delay_seconds: @@ -222,3 +228,153 @@ def test_redirects_are_not_followed(http_capture: HttpCapture, status: int) -> N assert [r.path for r in http_capture.requests] == ["/insight"] assert http_capture.requests[0].method == "POST" assert http_capture.requests[0].body + + +@pytest.fixture +def trickle_server() -> Iterator[Callable[[bytes, bytes], str]]: + """Start a server that sends ``immediate`` at once, then ``trickled`` one + byte every 200 ms. + + Each trickled byte arrives well inside any per-read socket timeout, so only + a whole-request deadline can end the exchange early. Returns the URL. + """ + listeners: list[socket.socket] = [] + threads: list[threading.Thread] = [] + stop = threading.Event() + + def start(immediate: bytes, trickled: bytes) -> str: + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind(("127.0.0.1", 0)) + listener.listen(1) + listener.settimeout(10) + listeners.append(listener) + + def serve() -> None: + try: + conn, _ = listener.accept() + except OSError: + return + with conn: + conn.settimeout(10) + try: + conn.recv(65536) # the request; content is irrelevant + if immediate: + conn.sendall(immediate) + for byte in trickled: + if stop.is_set(): + return + conn.sendall(bytes([byte])) + time.sleep(0.2) + except OSError: + return + + thread = threading.Thread(target=serve, daemon=True) + thread.start() + threads.append(thread) + return f"http://127.0.0.1:{listener.getsockname()[1]}" + + try: + yield start + finally: + stop.set() + for listener in listeners: + listener.close() + for thread in threads: + thread.join(timeout=5) + + +def test_timeout_is_a_whole_request_deadline( + trickle_server: Callable[[bytes, bytes], str], +) -> None: + url = trickle_server(b"", b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") + exporter = HttpExporter(url=url, timeout_ms=500) + started = time.monotonic() + with pytest.raises(TimeoutError, match=r"exceeded 0\.5s"): + exporter.export(_record()) + elapsed = time.monotonic() - started + # ~40 bytes at 200 ms each would take ~8 s without a deadline + assert 0.4 <= elapsed < 5.0, elapsed + + +def test_deadline_applies_while_reading_an_error_body( + trickle_server: Callable[[bytes, bytes], str], +) -> None: + # Headers arrive at once; the body trickles. The exporter must report the + # deadline, not the 500. + url = trickle_server( + b"HTTP/1.1 500 Internal Server Error\r\nContent-Length: 40\r\n\r\n", + b"x" * 40, + ) + exporter = HttpExporter(url=url, timeout_ms=500) + started = time.monotonic() + with pytest.raises(TimeoutError, match=r"exceeded 0\.5s"): + exporter.export(_record()) + assert time.monotonic() - started < 5.0 + + +def test_deadline_applies_when_name_resolution_is_slow( + http_capture: HttpCapture, monkeypatch: pytest.MonkeyPatch +) -> None: + # Name resolution outlives the deadline: no socket exists when the timer + # fires, so the connect path itself must refuse to proceed, promptly, and + # no request may reach the server. + real_getaddrinfo = socket.getaddrinfo + + def slow_getaddrinfo(*args: Any, **kwargs: Any) -> Any: + time.sleep(0.8) + return real_getaddrinfo(*args, **kwargs) + + monkeypatch.setattr(socket, "getaddrinfo", slow_getaddrinfo) + exporter = HttpExporter(url=http_capture.url, timeout_ms=300) + started = time.monotonic() + with pytest.raises(TimeoutError, match=r"exceeded 0\.3s"): + exporter.export(_record()) + assert time.monotonic() - started < 2.0 + assert http_capture.requests == [] + + +def test_deadline_interrupts_a_stalled_connect(monkeypatch: pytest.MonkeyPatch) -> None: + # A peer that never completes the handshake. The in-flight socket is + # registered with the deadline, so the timer's shutdown ends the attempt at + # the deadline instead of after the per-address socket timeout. + shut: set[int] = set() + real_shutdown = socket.socket.shutdown + + def marking_shutdown(self: socket.socket, how: int) -> None: + shut.add(id(self)) + real_shutdown(self, how) + + def stalled_connect(self: socket.socket, address: Any) -> None: + give_up = time.monotonic() + 5 + while time.monotonic() < give_up: + if id(self) in shut: + msg = "connection aborted by deadline" + raise ConnectionAbortedError(msg) + time.sleep(0.02) + msg = "test peer never answered" + raise TimeoutError(msg) + + monkeypatch.setattr(socket.socket, "shutdown", marking_shutdown) + monkeypatch.setattr(socket.socket, "connect", stalled_connect) + # 2000 ms socket timeout would be the old bound; the deadline is 300 ms. + exporter = HttpExporter(url="http://127.0.0.1:9/", timeout_ms=300) + started = time.monotonic() + with pytest.raises(TimeoutError, match=r"exceeded 0\.3s"): + exporter.export(_record()) + elapsed = time.monotonic() - started + assert 0.25 <= elapsed < 2.0, elapsed + + +def test_custom_header_case_is_merged_not_duplicated(http_capture: HttpCapture) -> None: + HttpExporter( + url=http_capture.url, headers={"content-type": "application/x-ndjson"} + ).export(_record()) + req = http_capture.requests[0] + assert req.header_counts["content-type"] == 1 + assert req.headers["content-type"] == "application/x-ndjson" + + +def test_unsupported_url_scheme_is_rejected() -> None: + with pytest.raises(ValueError, match="Unsupported URL"): + HttpExporter(url="ftp://127.0.0.1/insight").export(_record())