feat(python-sdk): bound volume streams with a 60s idle read timeout on a dedicated transport

reqwest's read timer keeps ticking while a request body is sent and while
waiting for the response head, so a shared read_timeout would kill slow
uploads; streamed reads get their own transport with read_timeout=60s
(JS SDK parity), uploads and unary calls stay unbounded-per-read.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: mishushakov

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

@@ -16,10 +16,12 @@
 from e2b.api.metadata import default_headers
 from e2b.exceptions import AuthenticationException
 from e2b.volume.client.client import AuthenticatedClient as AsyncVolumeApiClient
-from e2b.volume.connection_config import FILE_TIMEOUT, VolumeConnectionConfig
+from e2b.volume.connection_config import READ_TIMEOUT, VolumeConnectionConfig
 
 
-def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiClient:
+def get_api_client(
+    config: VolumeConnectionConfig, *, for_streaming: bool = False, **kwargs
+) -> AsyncVolumeApiClient:
     if config.access_token is None:
         raise AuthenticationException(
             "Volume token is required for volume content operations. "
@@ -46,33 +48,41 @@ def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiCl
         httpx_args={
             # The proxy lives in the cached transport; passing `proxy` here too
             # would mount a fresh, never-closed proxy transport per client.
-            "transport": get_transport(config),
+            "transport": get_transport(config, for_streaming=for_streaming),
             "event_hooks": make_async_logging_event_hooks(config.logger),
         },
         **kwargs,
     )
 
 
 _transport_lock = threading.Lock()
-# One transport (= one connection pool) per proxy; None is the direct pool.
-# pyqwest's I/O runs on its own Rust runtime, so unlike the httpx transport
-# this replaced, the transport is not bound to an event loop and the cache
-# is process-global rather than per-loop.
-_transports: Dict[Optional[ProxyConfig], AsyncApiPyqwestTransport] = {}
+# One transport (= one connection pool) per (proxy, streaming) pair; None is
+# the direct pool. pyqwest's I/O runs on its own Rust runtime, so unlike the
+# httpx transport this replaced, the transport is not bound to an event loop
+# and the cache is process-global rather than per-loop.
+_transports: Dict[tuple[Optional[ProxyConfig], bool], AsyncApiPyqwestTransport] = {}
 
 
-def get_transport(config: VolumeConnectionConfig) -> AsyncApiPyqwestTransport:
-    """The shared pyqwest-backed httpx transport for volume content API calls.
+def get_transport(
+    config: VolumeConnectionConfig, *, for_streaming: bool = False
+) -> AsyncApiPyqwestTransport:
+    """The shared pyqwest-backed httpx transports for volume content API calls.
 
-    ``read_timeout`` is the idle bound on every read: it resets after each
-    successful read, so it caps how long a streamed download may stall
-    without limiting total transfer time. It is fixed per transport — the
-    adapter's per-request timeouts are whole-request deadlines, and the sync
-    adapter does not bound body reads at all.
+    The streaming transport carries ``read_timeout``, the idle bound on every
+    read: it resets after each successful read, so it caps how long a
+    streamed download may stall without limiting total transfer time. It is
+    fixed per transport — the adapter's per-request timeouts are
+    whole-request deadlines, and the sync adapter does not bound body reads
+    at all. Only streamed downloads use it: reqwest's read timer keeps
+    running while a request body is sent and while waiting for the response
+    head, so on the regular transport it would cut off uploads and slow
+    unary responses longer than the idle bound (those stay bounded by their
+    whole-request deadlines instead).
     """
     proxy = proxy_to_config(config.proxy)
+    key = (proxy, for_streaming)
     with _transport_lock:
-        transport = _transports.get(proxy)
+        transport = _transports.get(key)
         if transport is None:
             transport = AsyncApiPyqwestTransport(
                 ConnectionRetryTransport(
@@ -81,10 +91,10 @@ def get_transport(config: VolumeConnectionConfig) -> AsyncApiPyqwestTransport:
                         proxy=proxy.to_pyqwest() if proxy is not None else None,
                         pool_idle_timeout=pool_idle_timeout,
                         pool_max_idle_per_host=pool_max_idle_per_host,
-                        read_timeout=FILE_TIMEOUT,
+                        read_timeout=READ_TIMEOUT if for_streaming else None,
                     ),
                     max_retries=connection_retries,
                 )
             )
-            _transports[proxy] = transport
+            _transports[key] = transport
         return transport

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

@@ -16,10 +16,12 @@
 from e2b.api.metadata import default_headers
 from e2b.exceptions import AuthenticationException
 from e2b.volume.client.client import AuthenticatedClient as VolumeApiClient
-from e2b.volume.connection_config import FILE_TIMEOUT, VolumeConnectionConfig
+from e2b.volume.connection_config import READ_TIMEOUT, VolumeConnectionConfig
 
 
-def get_api_client(config: VolumeConnectionConfig, **kwargs) -> VolumeApiClient:
+def get_api_client(
+    config: VolumeConnectionConfig, *, for_streaming: bool = False, **kwargs
+) -> VolumeApiClient:
     if config.access_token is None:
         raise AuthenticationException(
             "Volume token is required for volume content operations. "
@@ -46,32 +48,40 @@ def get_api_client(config: VolumeConnectionConfig, **kwargs) -> VolumeApiClient:
         httpx_args={
             # The proxy lives in the cached transport; passing `proxy` here too
             # would mount a fresh, never-closed proxy transport per client.
-            "transport": get_transport(config),
+            "transport": get_transport(config, for_streaming=for_streaming),
             "event_hooks": make_logging_event_hooks(config.logger),
         },
         **kwargs,
     )
 
 
 _transport_lock = threading.Lock()
-# One transport (= one connection pool) per proxy; None is the direct pool.
-# pyqwest transports are thread-safe, so unlike the httpx transport this
-# replaced, the cache is process-global rather than per-thread.
-_transports: Dict[Optional[ProxyConfig], ApiPyqwestTransport] = {}
+# One transport (= one connection pool) per (proxy, streaming) pair; None is
+# the direct pool. pyqwest transports are thread-safe, so unlike the httpx
+# transport this replaced, the cache is process-global rather than per-thread.
+_transports: Dict[tuple[Optional[ProxyConfig], bool], ApiPyqwestTransport] = {}
 
 
-def get_transport(config: VolumeConnectionConfig) -> ApiPyqwestTransport:
-    """The shared pyqwest-backed httpx transport for volume content API calls.
+def get_transport(
+    config: VolumeConnectionConfig, *, for_streaming: bool = False
+) -> ApiPyqwestTransport:
+    """The shared pyqwest-backed httpx transports for volume content API calls.
 
-    ``read_timeout`` is the idle bound on every read: it resets after each
-    successful read, so it caps how long a streamed download may stall
-    without limiting total transfer time. It is fixed per transport — the
-    adapter's per-request timeouts are whole-request deadlines, and the sync
-    adapter does not bound body reads at all.
+    The streaming transport carries ``read_timeout``, the idle bound on every
+    read: it resets after each successful read, so it caps how long a
+    streamed download may stall without limiting total transfer time. It is
+    fixed per transport — the adapter's per-request timeouts are
+    whole-request deadlines, and the sync adapter does not bound body reads
+    at all. Only streamed downloads use it: reqwest's read timer keeps
+    running while a request body is sent and while waiting for the response
+    head, so on the regular transport it would cut off uploads and slow
+    unary responses longer than the idle bound (those stay bounded by their
+    whole-request deadlines instead).
     """
     proxy = proxy_to_config(config.proxy)
+    key = (proxy, for_streaming)
     with _transport_lock:
-        transport = _transports.get(proxy)
+        transport = _transports.get(key)
         if transport is None:
             transport = ApiPyqwestTransport(
                 ConnectionRetryTransport(
@@ -80,10 +90,10 @@ def get_transport(config: VolumeConnectionConfig) -> ApiPyqwestTransport:
                         proxy=proxy.to_pyqwest() if proxy is not None else None,
                         pool_idle_timeout=pool_idle_timeout,
                         pool_max_idle_per_host=pool_max_idle_per_host,
-                        read_timeout=FILE_TIMEOUT,
+                        read_timeout=READ_TIMEOUT if for_streaming else None,
                     ),
                     max_retries=connection_retries,
                 )
             )
-            _transports[proxy] = transport
+            _transports[key] = transport
         return transport

packages/python-sdk/e2b/volume/connection_config.py

@@ -15,6 +15,12 @@
 # bounds each chunk by the request timeout and leaves the total to the server.)
 FILE_TIMEOUT: float = 3600.0  # 1 hour
 
+# Idle bound for every read on the volume content transports: the transfer is
+# aborted when no bytes at all arrive for this long. It resets on each chunk,
+# so it never limits total transfer time — only a fully stalled connection.
+# Matches the JS SDK's default stream idle timeout (REQUEST_TIMEOUT_MS).
+READ_TIMEOUT: float = 60.0  # 60 seconds
+
 
 class VolumeApiParams(TypedDict, total=False):
     """

packages/python-sdk/e2b/volume/volume_async.py

@@ -467,8 +467,8 @@ async def read_file(
         :param path: Path to the file
         :param format: Format of the file content—`text` by default
         :param stream_idle_timeout: Deprecated and ignored. A stalled streamed
-            read is bounded by a transport-wide idle read timeout instead,
-            which resets on every chunk.
+            read is bounded by a transport-wide idle read timeout instead
+            (60 seconds), which resets on every chunk.
         :param opts: Connection options
 
         :return: File content as string, bytes, or async iterator of bytes
@@ -492,12 +492,16 @@ async def read_file(
             stream_timeout = VolumeConnectionConfig._get_request_timeout(
                 None, opts.get("request_timeout")
             )
+            # The streaming transport carries the idle read timeout; the
+            # regular one must not (it would cut off slow uploads and
+            # responses), so streamed reads get their own client.
+            stream_client = get_volume_api_client(config, for_streaming=True)
 
             async def stream_file() -> AsyncIterator[bytes]:
                 # pyqwest raises the builtin TimeoutError; keep the httpx
                 # exception the streamed-read contract established.
                 try:
-                    async with api_client.get_async_httpx_client().stream(
+                    async with stream_client.get_async_httpx_client().stream(
                         method="GET",
                         url=f"/volumecontent/{self._volume_id}/file",
                         params=params,

packages/python-sdk/e2b/volume/volume_sync.py

@@ -464,8 +464,8 @@ def read_file(
         :param path: Path to the file
         :param format: Format of the file content—`text` by default
         :param stream_idle_timeout: Deprecated and ignored. A stalled streamed
-            read is bounded by a transport-wide idle read timeout instead,
-            which resets on every chunk.
+            read is bounded by a transport-wide idle read timeout instead
+            (60 seconds), which resets on every chunk.
         :param opts: Connection options
 
         :return: File content as string, bytes, or iterator of bytes
@@ -489,12 +489,16 @@ def read_file(
             stream_timeout = VolumeConnectionConfig._get_request_timeout(
                 None, opts.get("request_timeout")
             )
+            # The streaming transport carries the idle read timeout; the
+            # regular one must not (it would cut off slow uploads and
+            # responses), so streamed reads get their own client.
+            stream_client = get_volume_api_client(config, for_streaming=True)
 
             def stream_file() -> Iterator[bytes]:
                 # pyqwest raises the builtin TimeoutError; keep the httpx
                 # exception the streamed-read contract established.
                 try:
-                    with api_client.get_httpx_client().stream(
+                    with stream_client.get_httpx_client().stream(
                         method="GET",
                         url=f"/volumecontent/{self._volume_id}/file",
                         params=params,

packages/python-sdk/tests/test_volume_client.py

@@ -135,9 +135,12 @@ async def get_transports():
 CHUNK = b"x" * 1024
 
 
-def _start_volume_file_server(chunk_delays: List[float]) -> str:
+def _start_volume_file_server(
+    chunk_delays: List[float], ttfb_delay: float = 0.0
+) -> str:
     """One-shot HTTP server streaming a chunked volume-file body, sleeping
-    ``chunk_delays[i]`` before sending chunk ``i``. Returns its base URL."""
+    ``ttfb_delay`` before the response head and ``chunk_delays[i]`` before
+    sending chunk ``i``. Returns its base URL."""
     sock = socket.socket()
     sock.bind(("127.0.0.1", 0))
     sock.listen(1)
@@ -148,6 +151,7 @@ def serve():
             conn, _ = sock.accept()
             while b"\r\n\r\n" not in conn.recv(65536):
                 pass
+            time.sleep(ttfb_delay)
             conn.sendall(
                 b"HTTP/1.1 200 OK\r\n"
                 b"Content-Type: application/octet-stream\r\n"
@@ -171,8 +175,8 @@ def serve():
 def short_read_timeout(monkeypatch):
     """Rebuild the volume transports with a short idle read timeout."""
     reset_volume_transports()
-    monkeypatch.setattr(client_sync, "FILE_TIMEOUT", 0.3)
-    monkeypatch.setattr(client_async, "FILE_TIMEOUT", 0.3)
+    monkeypatch.setattr(client_sync, "READ_TIMEOUT", 0.3)
+    monkeypatch.setattr(client_async, "READ_TIMEOUT", 0.3)
     yield 0.3
     reset_volume_transports()
 
@@ -231,3 +235,48 @@ async def run():
         return received
 
     assert asyncio.run(run()) == [CHUNK]
+
+
+def test_stream_transport_is_separate_from_regular_transport():
+    # reqwest's read timer keeps running while a request body is sent and
+    # while waiting for the response head, so the idle read timeout lives on
+    # a dedicated streaming transport — putting it on the shared one would
+    # cut off uploads and slow unary responses longer than the idle bound.
+    reset_volume_transports()
+    config = VolumeConnectionConfig(token="vol-token")
+
+    try:
+        regular = get_sync_transport(config)
+        streaming = get_sync_transport(config, for_streaming=True)
+        assert regular is not streaming
+        assert get_sync_transport(config) is regular
+        assert get_sync_transport(config, for_streaming=True) is streaming
+
+        async_regular = get_async_transport(config)
+        async_streaming = get_async_transport(config, for_streaming=True)
+        assert async_regular is not async_streaming
+    finally:
+        reset_volume_transports()
+
+
+def test_sync_non_stream_read_survives_response_slower_than_idle_timeout(
+    short_read_timeout,
+):
+    # Non-streamed requests go through the regular transport, which has no
+    # idle read timeout: a server that takes longer than the streaming idle
+    # bound to start responding must not be cut off.
+    api_url = _start_volume_file_server([0.0], ttfb_delay=short_read_timeout * 3)
+    volume = Volume(volume_id="v1", name="test", token="vol-token")
+
+    assert volume.read_file("file.bin", format="bytes", api_url=api_url) == CHUNK
+
+
+def test_sync_stream_response_head_is_bounded_by_idle_timeout(short_read_timeout):
+    # For streamed reads the idle read timeout also bounds waiting for the
+    # response head (like the JS SDK's handshake timeout on stream start).
+    api_url = _start_volume_file_server([0.0], ttfb_delay=5.0)
+    volume = Volume(volume_id="v1", name="test", token="vol-token")
+
+    stream = volume.read_file("file.bin", format="stream", api_url=api_url)
+    with pytest.raises(httpx.ReadTimeout):
+        next(iter(stream))