fix(python-sdk): download files over HTTP/1.1 to avoid h2 too_many_data_frames

.changeset/python-file-read-http1.md

@@ -0,0 +1,9 @@
+---
+'@e2b/python-sdk': patch
+---
+
+- Fixed `Sandbox.files.read` / `AsyncSandbox.files.read` failing at random for
+  files over ~500 KB (and always for a few MB) with `httpx.ReadError ...
+  too_many_data_frames`. File downloads now go over HTTP/1.1 regardless of
+  `http_version`, as h2 closed HTTP/2 connections after receiving 100 empty
+  DATA frames, which the sandbox proxy sends throughout file responses.

packages/python-sdk/e2b/api/client_async/__init__.py

@@ -217,6 +217,7 @@ def get_envd_api(
     *,
     for_streaming: bool = False,
     retry_connect: bool = True,
+    http_version: Optional[HttpVersion] = None,
 ) -> httpx.AsyncClient:
     """An httpx client for a sandbox's envd HTTP API (file transfers, health
     checks) on the shared transports. The client itself is a cheap stateless
@@ -225,7 +226,10 @@ def get_envd_api(
     return httpx.AsyncClient(
         base_url=base_url,
         transport=get_transport(
-            config, for_streaming=for_streaming, retry_connect=retry_connect
+            config,
+            http_version=http_version,
+            for_streaming=for_streaming,
+            retry_connect=retry_connect,
         ),
         headers=config.sandbox_headers,
         event_hooks=make_async_logging_event_hooks(config.logger),

packages/python-sdk/e2b/api/client_sync/__init__.py

@@ -222,6 +222,7 @@ def get_envd_api(
     *,
     for_streaming: bool = False,
     retry_connect: bool = True,
+    http_version: Optional[HttpVersion] = None,
 ) -> httpx.Client:
     """An httpx client for a sandbox's envd HTTP API (file transfers, health
     checks) on the shared transports. The client itself is a cheap stateless
@@ -230,7 +231,10 @@ def get_envd_api(
     return httpx.Client(
         base_url=base_url,
         transport=get_transport(
-            config, for_streaming=for_streaming, retry_connect=retry_connect
+            config,
+            http_version=http_version,
+            for_streaming=for_streaming,
+            retry_connect=retry_connect,
         ),
         headers=config.sandbox_headers,
         event_hooks=make_logging_event_hooks(config.logger),

packages/python-sdk/e2b/connection_config.py

@@ -128,7 +128,10 @@ class ApiParams(TypedDict, total=False):
     `"2"`. `"1.1"` pins them to HTTP/1.1, which
     uses one connection per concurrent request instead of multiplexing streams
     over shared connections — for example when an intermediary on the path
-    retires or mishandles long-lived HTTP/2 connections."""
+    retires or mishandles long-lived HTTP/2 connections. In the Python SDK,
+    file downloads (`files.read`) always use HTTP/1.1, overriding this
+    option, to work around an HTTP/2 empty-DATA-frame limit in the
+    underlying transport."""
 
 
 class ApiParamsWithLogger(ApiParams, total=False):

packages/python-sdk/e2b/envd/api.py

@@ -2,6 +2,8 @@
 
 from typing import Callable, Optional
 
+from e2b.connection_config import HttpVersion
+
 from e2b.envd.rpc import (
     format_sandbox_unreachable_exception,
     format_terminated_exception,
@@ -21,6 +23,13 @@
 ENVD_API_FILES_ROUTE = "/files"
 ENVD_API_HEALTH_ROUTE = "/health"
 
+# File downloads go over HTTP/1.1 whatever the configured HTTP version: the
+# sandbox proxy pads them with an empty DATA frame every ~32 KiB, and h2 (under
+# pyqwest) closes a connection with ENHANCE_YOUR_CALM ("too_many_data_frames")
+# after 100 such frames over its lifetime — failing the download and every
+# other stream multiplexed on that connection.
+ENVD_API_FILE_DOWNLOAD_HTTP_VERSION: HttpVersion = "1.1"
+
 _DEFAULT_API_ERROR_MAP: dict[int, Callable[[str], Exception]] = {
     400: InvalidArgumentException,
     401: AuthenticationException,

packages/python-sdk/e2b/sandbox_async/filesystem/filesystem.py

@@ -16,6 +16,7 @@
     default_username,
 )
 from e2b.envd.api import (
+    ENVD_API_FILE_DOWNLOAD_HTTP_VERSION,
     ENVD_API_FILES_ROUTE,
     acheck_sandbox_health,
     ahandle_envd_api_exception,
@@ -100,10 +101,20 @@ def __init__(
             connection_config,
         )
         self._envd_api = envd_api
-        # Streamed downloads default to a sibling client whose transport
-        # carries the idle read timeout (see `get_transport`).
+        # Downloads go over their own HTTP/1.1 clients (see
+        # `ENVD_API_FILE_DOWNLOAD_HTTP_VERSION`). Streamed downloads default
+        # to the one whose transport carries the idle read timeout (see
+        # `get_transport`).
+        self._envd_api_download = get_envd_api(
+            connection_config,
+            envd_api_url,
+            http_version=ENVD_API_FILE_DOWNLOAD_HTTP_VERSION,
+        )
         self._envd_api_streaming = get_envd_api(
-            connection_config, envd_api_url, for_streaming=True
+            connection_config,
+            envd_api_url,
+            for_streaming=True,
+            http_version=ENVD_API_FILE_DOWNLOAD_HTTP_VERSION,
         )
         # The health probe run after a request failed goes without the
         # connection retries, so a sandbox that cannot be reached is not tried
@@ -235,7 +246,7 @@ async def read(
             client = (
                 self._envd_api_streaming
                 if stream_idle_timeout is None
-                else self._envd_api
+                else self._envd_api_download
             )
             request = client.build_request(
                 "GET",
@@ -268,7 +279,7 @@ async def read(
             return AsyncFileStreamReader(r, idle_timeout=stream_idle_timeout)
 
         try:
-            r = await self._envd_api.get(
+            r = await self._envd_api_download.get(
                 ENVD_API_FILES_ROUTE,
                 params=params,
                 headers=headers,

packages/python-sdk/e2b/sandbox_sync/filesystem/filesystem.py

@@ -15,6 +15,7 @@
 )
 
 from e2b.envd.api import (
+    ENVD_API_FILE_DOWNLOAD_HTTP_VERSION,
     ENVD_API_FILES_ROUTE,
     check_sandbox_health,
     handle_envd_api_exception,
@@ -96,12 +97,21 @@ def __init__(
             connection_config,
         )
         self._envd_api = envd_api
-        # Streamed downloads default to a sibling client whose transport
-        # carries the idle read timeout (see `get_transport`). Like the
-        # RPC client, the pyqwest transports underneath are thread-safe, so
-        # one client serves all threads.
+        # Downloads go over their own HTTP/1.1 clients (see
+        # `ENVD_API_FILE_DOWNLOAD_HTTP_VERSION`). Streamed downloads take the
+        # one whose transport carries the idle read timeout (see
+        # `get_transport`). Like the RPC client, the pyqwest transports
+        # underneath are thread-safe, so one client serves all threads.
+        self._envd_api_download = get_envd_api(
+            connection_config,
+            envd_api_url,
+            http_version=ENVD_API_FILE_DOWNLOAD_HTTP_VERSION,
+        )
         self._envd_api_streaming = get_envd_api(
-            connection_config, envd_api_url, for_streaming=True
+            connection_config,
+            envd_api_url,
+            for_streaming=True,
+            http_version=ENVD_API_FILE_DOWNLOAD_HTTP_VERSION,
         )
         # The health probe run after a request failed goes without the
         # connection retries, so a sandbox that cannot be reached is not tried
@@ -245,7 +255,7 @@ def read(
             return FileStreamReader(r)
 
         try:
-            r = self._envd_api.get(
+            r = self._envd_api_download.get(
                 ENVD_API_FILES_ROUTE,
                 params=params,
                 headers=headers,

packages/python-sdk/tests/test_envd_file_download_empty_data_frames.py

@@ -0,0 +1,190 @@
+"""File downloads must survive a peer that pads responses with empty DATA frames.
+
+The sandbox proxy splits envd file downloads into HTTP/2 DATA frames with an
+empty, non-final one after every ~32 KiB. h2 (under pyqwest) counts those over
+the lifetime of a connection and tears the whole connection down with
+``ENHANCE_YOUR_CALM`` (``too_many_data_frames``) after 100 of them, so a few
+MB read over HTTP/2 always failed, and smaller reads failed once a pooled
+connection had accumulated the count. ``files.read`` therefore downloads over
+HTTP/1.1. The server here answers HTTP/2 (prior knowledge, standing in for
+ALPN against the real proxy) the way the proxy does, and plain HTTP/1.1
+otherwise.
+"""
+
+import contextlib
+import socket
+import threading
+from typing import Iterator, List
+
+import h2.config
+import h2.connection
+import h2.events
+import pytest
+from packaging.version import Version
+from pyqwest import HTTPVersion
+
+import e2b.api.client_async as api_client_async
+import e2b.api.client_sync as api_client_sync
+from e2b.connection_config import ConnectionConfig
+from e2b.sandbox_async.filesystem.filesystem import Filesystem as AsyncFilesystem
+from e2b.sandbox_sync.filesystem.filesystem import Filesystem
+from transport_caches import reset_transport_caches
+
+H2_PREFACE = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
+# Comfortably past h2's limit of 100 empty DATA frames per connection, with
+# chunks large enough not to touch its separate small-frame budget and a body
+# that fits the default flow-control windows.
+CHUNKS = [bytes([ord("a") + i % 26]) * 300 for i in range(150)]
+CONTENT = b"".join(CHUNKS)
+
+
+class EmptyDataFramesServer(threading.Thread):
+    def __init__(self):
+        super().__init__(daemon=True)
+        self.listener = socket.create_server(("127.0.0.1", 0))
+        self.port = self.listener.getsockname()[1]
+        self.protocols: List[str] = []
+
+    def run(self):
+        while True:
+            try:
+                sock, _ = self.listener.accept()
+            except OSError:
+                return
+            threading.Thread(target=self.serve, args=(sock,), daemon=True).start()
+
+    def serve(self, sock: socket.socket):
+        sock.settimeout(10)
+        try:
+            # Sniff without MSG_PEEK/MSG_WAITALL, which Windows can't combine.
+            buf = b""
+            while len(buf) < len(H2_PREFACE) and H2_PREFACE.startswith(buf):
+                data = sock.recv(65535)
+                if not data:
+                    return
+                buf += data
+            if buf.startswith(H2_PREFACE):
+                self.serve_h2(sock, buf)
+            else:
+                self.serve_h1(sock, buf)
+        except OSError:
+            pass
+        finally:
+            sock.close()
+
+    def serve_h1(self, sock: socket.socket, buf: bytes):
+        while True:
+            while b"\r\n\r\n" not in buf:
+                data = sock.recv(65535)
+                if not data:
+                    return
+                buf += data
+            _, buf = buf.split(b"\r\n\r\n", 1)
+            self.protocols.append("HTTP/1.1")
+            sock.sendall(
+                b"HTTP/1.1 200 OK\r\n"
+                b"Content-Type: application/octet-stream\r\n"
+                + f"Content-Length: {len(CONTENT)}\r\n\r\n".encode()
+                + CONTENT
+            )
+
+    def serve_h2(self, sock: socket.socket, data: bytes):
+        conn = h2.connection.H2Connection(
+            config=h2.config.H2Configuration(client_side=False)
+        )
+        conn.initiate_connection()
+        sock.sendall(conn.data_to_send())
+        while True:
+            for event in conn.receive_data(data):
+                if isinstance(event, h2.events.RequestReceived):
+                    self.protocols.append("HTTP/2")
+                    conn.send_headers(
+                        event.stream_id,
+                        [
+                            (":status", "200"),
+                            ("content-type", "application/octet-stream"),
+                            ("content-length", str(len(CONTENT))),
+                        ],
+                    )
+                    for chunk in CHUNKS:
+                        conn.send_data(event.stream_id, chunk)
+                        conn.send_data(event.stream_id, b"")
+                    conn.end_stream(event.stream_id)
+                elif isinstance(event, h2.events.ConnectionTerminated):
+                    return
+            out = conn.data_to_send()
+            if out:
+                sock.sendall(out)
+            data = sock.recv(65535)
+            if not data:
+                return
+
+
+@contextlib.contextmanager
+def empty_data_frames_server() -> Iterator[EmptyDataFramesServer]:
+    server = EmptyDataFramesServer()
+    server.start()
+    try:
+        yield server
+    finally:
+        server.listener.close()
+
+
+@pytest.fixture(autouse=True)
+def negotiate_http2(monkeypatch):
+    """Speak HTTP/2 wherever production would negotiate it over TLS: the
+    server is plaintext, so force prior knowledge on every transport not
+    pinned to HTTP/1.1."""
+    reset_transport_caches()
+    for module, name in (
+        (api_client_sync, "SyncHTTPTransport"),
+        (api_client_async, "HTTPTransport"),
+    ):
+        build_transport = getattr(module, name)
+
+        def build(build_transport=build_transport, **kwargs):
+            if kwargs.get("http_version") is None:
+                kwargs["http_version"] = HTTPVersion.HTTP2
+            return build_transport(**kwargs)
+
+        monkeypatch.setattr(module, name, build)
+    yield
+    reset_transport_caches()
+
+
+ENVD_VERSION = Version("0.6.2")
+
+
+@pytest.mark.parametrize("http_version", ["2", "1.1"])
+def test_sync_read_downloads_over_http1(http_version):
+    config = ConnectionConfig(api_key="e2b_" + "0" * 40, http_version=http_version)
+    with empty_data_frames_server() as server:
+        url = f"http://127.0.0.1:{server.port}"
+        fs = Filesystem(
+            url, ENVD_VERSION, config, api_client_sync.get_envd_api(config, url)
+        )
+        for _ in range(3):
+            assert fs.read("/blob", format="bytes") == CONTENT
+            assert fs.read("/blob", format="text") == CONTENT.decode()
+            with fs.read("/blob", format="stream") as reader:
+                assert b"".join(reader) == CONTENT
+        assert set(server.protocols) == {"HTTP/1.1"}
+
+
+@pytest.mark.parametrize("http_version", ["2", "1.1"])
+async def test_async_read_downloads_over_http1(http_version):
+    config = ConnectionConfig(api_key="e2b_" + "0" * 40, http_version=http_version)
+    with empty_data_frames_server() as server:
+        url = f"http://127.0.0.1:{server.port}"
+        fs = AsyncFilesystem(
+            url, ENVD_VERSION, config, api_client_async.get_envd_api(config, url)
+        )
+        for _ in range(3):
+            assert await fs.read("/blob", format="bytes") == CONTENT
+            assert await fs.read("/blob", format="text") == CONTENT.decode()
+            for idle_timeout in (None, 5):
+                async with await fs.read(
+                    "/blob", format="stream", stream_idle_timeout=idle_timeout
+                ) as reader:
+                    assert b"".join([chunk async for chunk in reader]) == CONTENT
+        assert set(server.protocols) == {"HTTP/1.1"}