fix(transport): Enable TCP keepalive for long partition requests

Refresh idle NAT and load-balancer flows while partition requests wait for
response headers. Enable SO_KEEPALIVE with 60s idle, 30s interval and five
probes on SDK-owned sync, async and split-PDF transports. Resolve supported
platform options defensively and expose validated environment overrides.
Caller-provided clients retain their existing transport configuration.

HTTPX 0.28 disables environment proxy discovery when transport is supplied.
Rebuild its environment proxy mounts with get_environment_proxies, including
NO_PROXY exclusions, and apply socket options to proxy transports as well.
Move split-PDF limits, verify and cert into every transport so these settings
remain effective. Log resolved probe settings alongside pool and TLS settings.

Keep the changelog entry under 0.46.3: no corresponding remote tag or GitHub
release exists. Generation was decommissioned in cb867f0; .genignore is absent
from the current checkout, and the SDK entry point is maintained by hand.

Validation: make lint (pylint 10.00/10, mypy clean); make test-unit (353 passed,
1 xfailed). A Docker MASQUERADE router with a 20s conntrack timeout and a
45s server delay returned 200 at 45.0s with keepalive idle/interval 5s;
disabling keepalive produced ReadTimeout at 90.0s.

Author: CyMule

CHANGELOG.md

@@ -1,6 +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.
 * 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_split_collection_urls.py

@@ -62,7 +62,7 @@ def respond(request):
     monkeypatch.setattr(
         httpx,
         "AsyncClient",
-        lambda **kwargs: original_async_client(transport=transport, **kwargs),
+        lambda **kwargs: original_async_client(**dict(kwargs, transport=transport, mounts={})),
     )
     return sdk, sent, sync_client, async_client
 

_test_unstructured_client/unit/test_split_pdf_hook.py

@@ -498,15 +498,15 @@ async def test_unit_run_tasks_pool_limits_configurable_via_env(
     monkeypatch.setenv("UNSTRUCTURED_CLIENT_KEEPALIVE_EXPIRY", "30.0")
 
     captured: dict[str, httpx.Limits] = {}
-    real_async_client = httpx.AsyncClient
+    real_async_transport = httpx.AsyncHTTPTransport
 
-    def _capturing_async_client(*args, **kwargs):
+    def _capturing_async_transport(*args, **kwargs):
         captured["limits"] = kwargs.get("limits")
-        return real_async_client(*args, **kwargs)
+        return real_async_transport(*args, **kwargs)
 
     with patch(
-        "unstructured_client._hooks.custom.split_pdf_hook.httpx.AsyncClient",
-        side_effect=_capturing_async_client,
+        "unstructured_client._hooks.custom.http_transport.httpx.AsyncHTTPTransport",
+        side_effect=_capturing_async_transport,
     ):
         await run_tasks(
             [partial(_request_mock, fails=False, content="ok")],
@@ -622,24 +622,22 @@ def test_unit_resolve_tls_config_client_cert_and_key_split(
 
 
 @pytest.mark.asyncio
-async def test_unit_run_tasks_forwards_tls_config_to_httpx_async_client(
+async def test_unit_run_tasks_forwards_tls_config_to_httpx_async_transport(
     monkeypatch: pytest.MonkeyPatch, tmp_path
 ):
-    """run_tasks() actually wires verify+cert into httpx.AsyncClient(). End-to-
-    end check that _resolve_tls_config is called and its result reaches the
-    client construction. AsyncClient is fully mocked so we don't have to feed
-    httpx a real cert chain."""
+    """Split requests apply TLS and pool settings to the transport."""
     ca_bundle = tmp_path / "ca.pem"
     ca_bundle.write_text("-----BEGIN CERTIFICATE-----\nfake\n-----END CERTIFICATE-----\n")
     monkeypatch.setenv("SSL_CERT_FILE", str(ca_bundle))
     monkeypatch.setenv("UNSTRUCTURED_CLIENT_TLS_CLIENT_CERT", "/etc/ssl/client.pem")
 
+    monkeypatch.setenv("NO_PROXY", "*")
+    monkeypatch.setenv("UNSTRUCTURED_CLIENT_MAX_CONNECTIONS", "17")
     captured: dict = {}
 
     class _MockAsyncClient:
         def __init__(self, *args, **kwargs):
-            captured["verify"] = kwargs.get("verify")
-            captured["cert"] = kwargs.get("cert")
+            captured["transport"] = kwargs.get("transport")
 
         async def __aenter__(self):
             return self
@@ -648,16 +646,22 @@ async def __aexit__(self, *exc):
             return False
 
     with patch(
-        "unstructured_client._hooks.custom.split_pdf_hook.httpx.AsyncClient",
+        "unstructured_client._hooks.custom.http_transport.httpx.AsyncClient",
         new=_MockAsyncClient,
-    ):
+    ), patch(
+        "unstructured_client._hooks.custom.http_transport.httpx.AsyncHTTPTransport"
+    ) as transport:
         await run_tasks(
             [partial(_request_mock, fails=False, content="ok")],
             allow_failed=True,
         )
 
-    assert captured["verify"] == str(ca_bundle)
-    assert captured["cert"] == "/etc/ssl/client.pem"
+    kwargs = transport.call_args.kwargs
+    assert captured["transport"] is transport.return_value
+    assert kwargs["verify"] == str(ca_bundle)
+    assert kwargs["cert"] == "/etc/ssl/client.pem"
+    assert kwargs["limits"].max_connections == 17
+    assert kwargs["socket_options"]
 
 
 @patch("unstructured_client._hooks.custom.form_utils.Path")

_test_unstructured_client/unit/test_tcp_keepalive.py

@@ -0,0 +1,179 @@
+"""TCP keepalive configuration and transport integration."""
+
+import asyncio
+import socket
+import sys
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from threading import Thread
+
+import httpcore
+import httpx
+import pytest
+
+from unstructured_client import UnstructuredClient
+from unstructured_client._hooks.custom import http_transport as transport
+
+
+@pytest.fixture(autouse=True)
+def clean_env(monkeypatch):
+    for suffix in ("", "_IDLE", "_INTERVAL", "_COUNT"):
+        monkeypatch.delenv("UNSTRUCTURED_CLIENT_TCP_KEEPALIVE" + suffix, raising=False)
+    monkeypatch.setenv("NO_PROXY", "*")
+
+
+@pytest.mark.parametrize("platform", ["linux", "macos", "macos_old", "windows"])
+def test_unit_platform_options(monkeypatch, platform):
+    for name in ("TCP_KEEPIDLE", "TCP_KEEPALIVE", "TCP_KEEPINTVL", "TCP_KEEPCNT"):
+        monkeypatch.delattr(socket, name, raising=False)
+    expected = [(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)]
+    if platform != "windows":
+        monkeypatch.setattr(socket, "TCP_KEEPIDLE" if platform == "linux" else "TCP_KEEPALIVE", 100, raising=False)
+        expected.append((socket.IPPROTO_TCP, 100, 60))
+        if platform != "macos_old":
+            monkeypatch.setattr(socket, "TCP_KEEPINTVL", 101, raising=False)
+            monkeypatch.setattr(socket, "TCP_KEEPCNT", 102, raising=False)
+            expected += [(socket.IPPROTO_TCP, 101, 30), (socket.IPPROTO_TCP, 102, 5)]
+    assert transport.resolve_tcp_keepalive().socket_options() == expected
+
+
+@pytest.mark.parametrize("value", ["0", "false", "False", " FALSE "])
+def test_unit_disabled(monkeypatch, value):
+    monkeypatch.setenv("UNSTRUCTURED_CLIENT_TCP_KEEPALIVE", value)
+    assert transport.resolve_tcp_keepalive().socket_options() == []
+
+
+@pytest.mark.parametrize("suffix,default", [("IDLE", 60), ("INTERVAL", 30), ("COUNT", 5)])
+@pytest.mark.parametrize("value", ["0", "-1", "bad", "1.5", ""])
+def test_unit_invalid_value(monkeypatch, caplog, suffix, default, value):
+    monkeypatch.setenv("UNSTRUCTURED_CLIENT_TCP_KEEPALIVE_" + suffix, value)
+    assert getattr(transport.resolve_tcp_keepalive(), suffix.lower()) == default
+    assert "using default" in caplog.text
+
+
+def test_unit_custom_values(monkeypatch):
+    for suffix, value in (("IDLE", "5"), ("INTERVAL", "6"), ("COUNT", "7")):
+        monkeypatch.setenv("UNSTRUCTURED_CLIENT_TCP_KEEPALIVE_" + suffix, value)
+    assert transport.resolve_tcp_keepalive() == transport.TCPKeepAlive(True, 5, 6, 7)
+
+
+def test_unit_sdk_owned_clients():
+    sdk = UnstructuredClient()
+    options = transport.resolve_tcp_keepalive().socket_options()
+    assert sdk.sdk_configuration.client._transport.base_transport._pool._socket_options == options
+    assert sdk.sdk_configuration.async_client._transport._pool._socket_options == options
+    sdk.sdk_configuration.client.close()
+    asyncio.run(sdk.sdk_configuration.async_client.aclose())
+
+
+def test_unit_supplied_clients_untouched():
+    sync = httpx.Client()
+    asynchronous = httpx.AsyncClient()
+    sync_transport = sync._transport
+    async_transport = asynchronous._transport
+    sdk = UnstructuredClient(client=sync, async_client=asynchronous)
+    assert sdk.sdk_configuration.client is sync
+    assert sdk.sdk_configuration.async_client is asynchronous
+    assert sync._transport.base_transport is sync_transport
+    assert asynchronous._transport is async_transport
+    assert sync_transport._pool._socket_options is None
+    assert asynchronous._transport._pool._socket_options is None
+    sync.close()
+    asyncio.run(asynchronous.aclose())
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_unit_environment_proxy_routes(monkeypatch, asynchronous):
+    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()
+    assert client._transport_for_url(httpx.URL("https://excluded.example")) is client._transport
+    if asynchronous:
+        asyncio.run(client.aclose())
+    else:
+        client.close()
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_unit_real_socket_keepalive(asynchronous):
+    if not hasattr(socket, "SO_KEEPALIVE"):
+        pytest.skip("SO_KEEPALIVE unavailable")
+
+    class Handler(BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
+        def do_GET(self):
+            self.send_response(200)
+            self.send_header("Content-Length", "2")
+            self.end_headers()
+            self.wfile.write(b"ok")
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    url = f"http://127.0.0.1:{server.server_port}"
+
+    def check(response):
+        stream = response.extensions["network_stream"]
+        sock = stream.get_extra_info("socket")
+        enabled = sock.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE)
+        # Darwin reports the option bit mask rather than a normalized boolean.
+        if sys.platform == "darwin":
+            assert enabled != 0
+        else:
+            assert enabled == 1
+
+    async def request():
+        async with transport.create_async_http_client() as client:
+            async with client.stream("GET", url) as response:
+                check(response)
+                await response.aread()
+
+    try:
+        if asynchronous:
+            asyncio.run(request())
+        else:
+            with transport.create_http_client() as client:
+                with client.stream("GET", url) as response:
+                    check(response)
+                    response.read()
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join()
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_unit_https_proxy_receives_connect(monkeypatch, asynchronous):
+    destinations = []
+
+    class ProxyHandler(BaseHTTPRequestHandler):
+        def do_CONNECT(self):
+            destinations.append(self.path)
+            self.send_error(502)
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), ProxyHandler)
+    thread = Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{server.server_port}")
+    monkeypatch.setenv("NO_PROXY", "")
+
+    async def request():
+        async with transport.create_async_http_client() as client:
+            await client.get("https://proxy-target.invalid/")
+
+    try:
+        with pytest.raises(httpx.ProxyError, match="502"):
+            if asynchronous:
+                asyncio.run(request())
+            else:
+                with transport.create_http_client() as client:
+                    client.get("https://proxy-target.invalid/")
+        assert destinations == ["proxy-target.invalid:443"]
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join()

src/unstructured_client/_hooks/custom/http_transport.py

@@ -0,0 +1,100 @@
+"""HTTP clients with TCP keepalive and HTTPX environment proxy routing."""
+
+import logging
+import os
+import socket
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+from httpx._utils import get_environment_proxies
+
+from .common import UNSTRUCTURED_CLIENT_LOGGER_NAME
+
+logger = logging.getLogger(UNSTRUCTURED_CLIENT_LOGGER_NAME)
+
+
+def _positive_int(name: str, default: int) -> int:
+    value = os.getenv(name, str(default))
+    try:
+        parsed = int(value)
+        if parsed > 0:
+            return parsed
+    except ValueError:
+        pass
+    logger.warning("Invalid %s=%r; using default %d", name, value, default)
+    return default
+
+
+@dataclass(frozen=True)
+class TCPKeepAlive:
+    """Resolved TCP probe settings, in seconds and number of probes."""
+
+    enabled: bool
+    idle: int
+    interval: int
+    count: int
+
+    def socket_options(self) -> list[tuple[int, int, int]]:
+        """Use only the TCP options available on the current platform."""
+        if not self.enabled:
+            return []
+        options = [(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)]
+        idle_option = getattr(socket, "TCP_KEEPIDLE", None)
+        if idle_option is None:
+            idle_option = getattr(socket, "TCP_KEEPALIVE", None)
+        for option, value in (
+            (idle_option, self.idle),
+            (getattr(socket, "TCP_KEEPINTVL", None), self.interval),
+            (getattr(socket, "TCP_KEEPCNT", None), self.count),
+        ):
+            if option is not None:
+                options.append((socket.IPPROTO_TCP, option, value))
+        return options
+
+
+def resolve_tcp_keepalive() -> TCPKeepAlive:
+    """Read keepalive settings, falling back on invalid positive integers."""
+    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),
+    )
+
+
+def _client_transport_options(
+    transport_type: Any,
+    keepalive: TCPKeepAlive | None,
+    transport_options: dict[str, Any],
+) -> dict[str, Any]:
+    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.
+    mounts = {
+        pattern: transport_type(proxy=proxy, **options) if proxy else None
+        for pattern, proxy in get_environment_proxies().items()
+    }
+    return {"transport": transport_type(**options), "mounts": mounts}
+
+
+def create_http_client() -> httpx.Client:
+    """Construct the SDK's synchronous client with HTTPX defaults."""
+    return httpx.Client(**_client_transport_options(httpx.HTTPTransport, None, {}))
+
+
+def create_async_http_client(
+    *,
+    keepalive: TCPKeepAlive | None = None,
+    transport_options: dict[str, Any] | None = None,
+    **client_options: Any,
+) -> httpx.AsyncClient:
+    """Construct an async client, applying TLS and pool settings to every route."""
+    return httpx.AsyncClient(
+        **client_options,
+        **_client_transport_options(
+            httpx.AsyncHTTPTransport, keepalive, transport_options or {}
+        ),
+    )