fix(insight): make HttpExporter timeout_ms a request deadline

- http_send: use http.client with a timer that shuts the socket down when
  the deadline expires, then raises TimeoutError; a peer trickling bytes can
  no longer keep a request alive past timeout_ms
- redirects stay unfollowed (http.client never follows them); the urllib
  opener is removed
- HttpExporter docstring states the deadline semantics
- tests: trickling status line ends in ~0.5 s with timeout_ms=500;
  unsupported URL scheme is rejected

Author: Alex Wang

packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/_common.py

@@ -11,34 +11,21 @@
 from __future__ import annotations
 
 import datetime
+import http.client
 import json
 import re
-import urllib.error
-import urllib.request
+import socket
+import ssl
+import threading
 from typing import Any
+from urllib.parse import urlsplit
 
 
 _SQL_IDENTIFIER = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
 # Upper bound on how much of a non-2xx response body is read for diagnostics.
 _MAX_ERROR_BODY_BYTES = 64 * 1024
 
 
-class _NoRedirect(urllib.request.HTTPRedirectHandler):
-    """Refuse every redirect so a 3xx surfaces as a failed status.
-
-    Following a redirect would re-send a POST as a body-less GET and forward
-    configured credential headers to the new location.
-    """
-
-    def redirect_request(  # type: ignore[override]  # stdlib signature has no hints
-        self, req: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str
-    ) -> None:
-        return None
-
-
-_OPENER = urllib.request.build_opener(_NoRedirect())
-
-
 def compact_dumps(value: Any) -> str:
     """Serialize ``value`` as compact JSON (no whitespace, non-ASCII preserved).
 
@@ -86,20 +73,74 @@ def http_send(
     """Send one HTTP request and return ``(status, reason, error_text)``.
 
     A non-2xx status is returned, not raised, so callers build their own error
-    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.
+    message. Redirects are never followed: a 3xx is returned like any other
+    failure, so a POST is not re-sent as a body-less GET and configured
+    credential headers never reach another origin. ``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.
+
+    ``timeout`` (seconds) is a deadline for the whole request: connect, send,
+    and receiving the status and headers. When it expires the connection is
+    shut down and ``TimeoutError`` is raised, even against a peer that keeps
+    the socket alive by trickling bytes. ``None`` means no limit. Network
+    errors propagate.
     """
-    request = urllib.request.Request(url, data=body, method=method)
-    for key, value in headers.items():
-        request.add_header(key, value)
+    parts = urlsplit(url)
+    if parts.scheme not in ("http", "https") or not parts.hostname:
+        msg = f"Unsupported URL: {url!r} (need http:// or https:// with a host)"
+        raise ValueError(msg)
+    path = parts.path or "/"
+    if parts.query:
+        path = f"{path}?{parts.query}"
+    conn: http.client.HTTPConnection
+    if parts.scheme == "https":
+        conn = http.client.HTTPSConnection(
+            parts.hostname,
+            parts.port,
+            timeout=timeout,
+            context=ssl.create_default_context(),
+        )
+    else:
+        conn = http.client.HTTPConnection(parts.hostname, parts.port, timeout=timeout)
+
+    expired = threading.Event()
+
+    def _expire() -> None:
+        # Wake any blocked read; the request then fails and is reported as a
+        # timeout below.
+        expired.set()
+        sock = conn.sock
+        if sock is not None:
+            try:
+                sock.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+
+    timer: threading.Timer | None = None
+    if timeout is not None:
+        timer = threading.Timer(timeout, _expire)
+        timer.daemon = True
+        timer.start()
     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
+        conn.request(method, path, body=body, headers=headers)
+        response = conn.getresponse()
+        status = int(response.status)
+        reason = str(response.reason or "")
+        detail = ""
+        if not 200 <= status < 300:
+            try:
+                detail = response.read(_MAX_ERROR_BODY_BYTES).decode(
+                    "utf-8", errors="replace"
+                )
+            except Exception:  # noqa: BLE001 - the body is best-effort detail only
+                detail = ""
+        return status, reason, detail
+    except Exception as exc:
+        if expired.is_set():
+            msg = f"request to {url} exceeded {timeout}s"
+            raise TimeoutError(msg) from exc
+        raise
+    finally:
+        if timer is not None:
+            timer.cancel()
+        conn.close()

packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/exporters/http_exporter.py

@@ -33,8 +33,10 @@ 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, and response headers;
+    default 10 seconds); on expiry the connection is shut down and
+    ``TimeoutError`` is raised. ``max_record_size_bytes``
     has no default because a generic endpoint has no known limit.
     """
 

packages/aws-durable-execution-sdk-python-insight/tests/test_http_exporter.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import json
+import socket
 import threading
 import time
 from dataclasses import dataclass, field
@@ -222,3 +223,59 @@ 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[str]:
+    """Accept one request, then send the status line one byte every 200 ms.
+
+    Each byte arrives well inside any per-read socket timeout, so only a
+    whole-request deadline can end the exchange early.
+    """
+    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(5)
+    stop = threading.Event()
+
+    def serve() -> None:
+        try:
+            conn, _ = listener.accept()
+        except OSError:
+            return
+        with conn:
+            conn.settimeout(5)
+            try:
+                conn.recv(65536)  # the request; content is irrelevant
+                for byte in b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n":
+                    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()
+    try:
+        yield f"http://127.0.0.1:{listener.getsockname()[1]}"
+    finally:
+        stop.set()
+        listener.close()
+        thread.join(timeout=5)
+
+
+def test_timeout_is_a_whole_request_deadline(trickle_server: str) -> None:
+    exporter = HttpExporter(url=trickle_server, 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 < 2.0, elapsed
+
+
+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())