fix(transport): Bound keepalive settings and guard HTTPX internals

Author: CyMule

CHANGELOG.md

@@ -1,7 +1,7 @@
 ## 0.46.3
 
 ### Fixes
-* Long partitions behind a NAT or load balancer idle timeout could never return, leaving the client waiting for its full read timeout. SDK-owned HTTP clients now enable TCP keepalive, including split-PDF and environment-proxy connections. Configure probes with `UNSTRUCTURED_CLIENT_TCP_KEEPALIVE_IDLE` (60 seconds), `UNSTRUCTURED_CLIENT_TCP_KEEPALIVE_INTERVAL` (30 seconds), and `UNSTRUCTURED_CLIENT_TCP_KEEPALIVE_COUNT` (5); set `UNSTRUCTURED_CLIENT_TCP_KEEPALIVE=0` or `false` to disable. Caller-supplied clients are unchanged.
+* Long partitions behind a NAT or load balancer idle timeout could never return, leaving the client waiting for its full read timeout. SDK-owned HTTP clients now enable TCP keepalive, including split-PDF requests. Connections made through an HTTP(S)/SOCKS proxy are not covered. Configure probes with `UNSTRUCTURED_CLIENT_TCP_KEEPALIVE_IDLE` (60 seconds), `UNSTRUCTURED_CLIENT_TCP_KEEPALIVE_INTERVAL` (30 seconds), and `UNSTRUCTURED_CLIENT_TCP_KEEPALIVE_COUNT` (5); set `UNSTRUCTURED_CLIENT_TCP_KEEPALIVE=0` or `false` to disable. Caller-supplied clients are unchanged.
 * Keep the deployment path in split-PDF requests. With a `server_url` that includes a path, such as `https://host/deployment`, splitting a PDF sent its collection request to `https://host/general/docs` instead of `https://host/deployment/general/docs`, which returned 404 before any chunks were processed. The collection URL is now derived from the partition request by replacing only its trailing `/general/v0/general` route, for sync and async calls and for a `server_url` passed to the operation.
 
 ## 0.46.2

_test_unstructured_client/unit/test_tcp_keepalive.py

@@ -5,6 +5,7 @@
 import sys
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from threading import Thread
+from unittest.mock import patch
 
 import httpcore
 import httpx
@@ -16,6 +17,8 @@
 
 @pytest.fixture(autouse=True)
 def clean_env(monkeypatch):
+    if hasattr(transport, "_log_keepalive_unavailable"):
+        transport._log_keepalive_unavailable.cache_clear()
     for suffix in ("", "_IDLE", "_INTERVAL", "_COUNT"):
         monkeypatch.delenv("UNSTRUCTURED_CLIENT_TCP_KEEPALIVE" + suffix, raising=False)
     monkeypatch.setenv("NO_PROXY", "*")
@@ -50,6 +53,43 @@ def test_unit_invalid_value(monkeypatch, caplog, suffix, default, value):
     assert "using default" in caplog.text
 
 
+@pytest.mark.parametrize("suffix,maximum,default", [
+    ("IDLE", 32767, 60), ("INTERVAL", 32767, 30), ("COUNT", 127, 5),
+])
+@pytest.mark.parametrize("above_maximum", [False, True])
+def test_unit_value_bounds(monkeypatch, caplog, suffix, maximum, default, above_maximum):
+    monkeypatch.setenv("UNSTRUCTURED_CLIENT_TCP_KEEPALIVE_" + suffix,
+                       str(maximum + int(above_maximum)))
+    expected = default if above_maximum else maximum
+    assert getattr(transport.resolve_tcp_keepalive(), suffix.lower()) == expected
+    assert ("using default" in caplog.text) == above_maximum
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_unit_missing_proxy_helper_client_options(monkeypatch, asynchronous):
+    monkeypatch.setattr(transport, "get_environment_proxies", None)
+    if asynchronous:
+        options = {"verify": False, "cert": None, "limits": httpx.Limits(max_connections=3)}
+        with patch.object(httpx, "AsyncClient") as constructor:
+            transport.create_async_http_client(timeout=42, transport_options=options)
+        constructor.assert_called_once_with(timeout=42, **options)
+    else:
+        with patch.object(httpx, "Client") as constructor:
+            transport.create_http_client()
+        constructor.assert_called_once_with()
+
+
+def test_unit_missing_proxy_helper_logs_once(monkeypatch, caplog):
+    monkeypatch.setattr(transport, "get_environment_proxies", None)
+    caplog.set_level("DEBUG", logger=transport.logger.name)
+    with transport.create_http_client():
+        pass
+    asyncio.run(transport.create_async_http_client().aclose())
+    messages = [record for record in caplog.records if "TCP keepalive is unavailable" in record.message]
+    assert len(messages) == 1
+    assert messages[0].levelname == "DEBUG"
+
+
 def test_unit_custom_values(monkeypatch):
     for suffix, value in (("IDLE", "5"), ("INTERVAL", "6"), ("COUNT", "7")):
         monkeypatch.setenv("UNSTRUCTURED_CLIENT_TCP_KEEPALIVE_" + suffix, value)
@@ -81,15 +121,20 @@ def test_unit_supplied_clients_untouched():
     asyncio.run(asynchronous.aclose())
 
 
+@pytest.mark.parametrize("missing_proxy_helper", [False, True])
 @pytest.mark.parametrize("asynchronous", [False, True])
-def test_unit_environment_proxy_routes(monkeypatch, asynchronous):
+def test_unit_environment_proxy_routes(monkeypatch, asynchronous, missing_proxy_helper):
+    if missing_proxy_helper:
+        monkeypatch.setattr(transport, "get_environment_proxies", None)
     monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:8123")
     monkeypatch.setenv("NO_PROXY", "excluded.example")
     client = transport.create_async_http_client() if asynchronous else transport.create_http_client()
     proxied = client._transport_for_url(httpx.URL("https://api.example"))
     assert isinstance(proxied._pool, httpcore.AsyncHTTPProxy if asynchronous else httpcore.HTTPProxy)
     assert proxied._pool._proxy_url.host == b"127.0.0.1"
-    assert proxied._pool._socket_options == transport.resolve_tcp_keepalive().socket_options()
+    if missing_proxy_helper:
+        assert client._transport._pool._socket_options is None
+        assert proxied._pool._socket_options is None
     assert client._transport_for_url(httpx.URL("https://excluded.example")) is client._transport
     if asynchronous:
         asyncio.run(client.aclose())

src/unstructured_client/_hooks/custom/http_transport.py

@@ -1,24 +1,28 @@
-"""HTTP clients with TCP keepalive and HTTPX environment proxy routing."""
+"""HTTP clients with TCP keepalive for direct connections and environment proxy routing."""
 
 import logging
 import os
 import socket
 from dataclasses import dataclass
+from functools import lru_cache
 from typing import Any
 
 import httpx
-from httpx._utils import get_environment_proxies
+try:
+    from httpx._utils import get_environment_proxies
+except ImportError:
+    get_environment_proxies = None  # type: ignore[assignment]
 
 from .common import UNSTRUCTURED_CLIENT_LOGGER_NAME
 
 logger = logging.getLogger(UNSTRUCTURED_CLIENT_LOGGER_NAME)
 
 
-def _positive_int(name: str, default: int) -> int:
+def _positive_int(name: str, default: int, maximum: int) -> int:
     value = os.getenv(name, str(default))
     try:
         parsed = int(value)
-        if parsed > 0:
+        if 1 <= parsed <= maximum:
             return parsed
     except ValueError:
         pass
@@ -54,25 +58,34 @@ def socket_options(self) -> list[tuple[int, int, int]]:
 
 
 def resolve_tcp_keepalive() -> TCPKeepAlive:
-    """Read keepalive settings, falling back on invalid positive integers."""
+    """Read keepalive settings, falling back on values outside the supported bounds."""
     prefix = "UNSTRUCTURED_CLIENT_TCP_KEEPALIVE"
     return TCPKeepAlive(
         enabled=os.getenv(prefix, "1").strip().lower() not in ("0", "false"),
-        idle=_positive_int(prefix + "_IDLE", 60),
-        interval=_positive_int(prefix + "_INTERVAL", 30),
-        count=_positive_int(prefix + "_COUNT", 5),
+        idle=_positive_int(prefix + "_IDLE", 60, 32767),
+        interval=_positive_int(prefix + "_INTERVAL", 30, 32767),
+        count=_positive_int(prefix + "_COUNT", 5, 127),
     )
 
 
+@lru_cache(maxsize=1)
+def _log_keepalive_unavailable() -> None:
+    logger.debug("TCP keepalive is unavailable with this httpx version")
+
+
 def _client_transport_options(
     transport_type: Any,
     keepalive: TCPKeepAlive | None,
     transport_options: dict[str, Any],
 ) -> dict[str, Any]:
+    if get_environment_proxies is None:
+        _log_keepalive_unavailable()
+        return transport_options
     settings = keepalive if keepalive is not None else resolve_tcp_keepalive()
     options = dict(transport_options, socket_options=settings.socket_options())
     # HTTPX disables automatic environment proxy discovery with a custom transport.
     # Reuse its URL patterns, including None mounts for NO_PROXY exclusions.
+    # HTTP and SOCKS proxy connections do not apply these socket options.
     mounts = {
         pattern: transport_type(proxy=proxy, **options) if proxy else None
         for pattern, proxy in get_environment_proxies().items()