fix(python-sdk): stop buffering streamed request bodies on the retrying transports

.changeset/python-unbuffered-retry-streamed-bodies.md

@@ -0,0 +1,14 @@
+---
+"@e2b/python-sdk": patch
+---
+
+Stop mirroring streamed request bodies in memory while they are being sent, on a
+pyqwest whose retry middleware exposes `RetryMode` (older releases keep the
+buffered replay they ship, so this is inert until the `pyqwest` dependency
+allows one). The shared retrying transports declare pyqwest's unbuffered retry
+policy, so a streamed `files.write` of a file-like object or `volume.write_file`
+is no longer copied into a replay buffer as it goes to the wire, and peak memory
+stays flat instead of scaling with file size. Connect retries are unchanged: a
+connect error means the body was never read, so the same body is handed to the
+next attempt, and `bytes` bodies (unary RPC payloads, in-memory writes) were
+already replayable for free.

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

@@ -4,12 +4,13 @@
 import re
 from dataclasses import dataclass
 from types import TracebackType
-from typing import NamedTuple, Optional, Protocol, Tuple, Union
+from typing import Any, NamedTuple, Optional, Protocol, Tuple, Union
 from urllib.parse import quote
 
 import httpx
 from httpx import AsyncBaseTransport, BaseTransport, Timeout
 from pyqwest import Proxy
+from pyqwest.middleware import retry as retry_middleware
 
 from e2b.api.client.client import AuthenticatedClient
 from e2b.api.client.types import Response
@@ -86,6 +87,42 @@ async def on_response(response: Response) -> None:
 pool_max_idle_per_host = int(os.getenv("E2B_MAX_KEEPALIVE_CONNECTIONS") or "20")
 
 
+# The retry policy the shared transports declare per request (see
+# `ConnectionRetryTransport` in client_sync/client_async).
+#
+# pyqwest's retry middleware keeps a request replayable, and for a body that
+# isn't already `bytes` it does that by growing a copy of it in memory as it is
+# sent. A streamed upload therefore went to the wire incrementally *and* was
+# mirrored whole in RAM, so peak memory scaled with file size for `files.write`
+# of a file-like object and for `volume.write_file`, even though nothing in the
+# SDK buffers (SDK-332). Template context uploads were never affected: they
+# build their own transport with no retry middleware in it.
+#
+# `RetryMode.UNBUFFERED` drops the copy: a streamed body is replayed only while
+# nothing has been read from it, which is all a connect-only policy ever needs,
+# since a connect error means the request was never written. `bytes` bodies —
+# every unary RPC payload and every in-memory write — are replayable as they
+# are and keep their retries either way.
+#
+# Releases without `RetryMode` keep the buffered behavior they ship — `True` is
+# what the middleware understood before the mode existed — because going
+# unbuffered there would trade the copy for the connect retries rather than get
+# them both: reqwest used to read ahead into its body channel while connecting,
+# so a connect error surfaced with the body already started (measured: the one
+# chunk of a unary RPC body, up to a whole small-chunked upload) and neither
+# buffered nor rewindable.
+#
+# Typed `Any` rather than the union it is: the pinned release declares
+# `should_retry_request` as `-> bool`, so annotating the override with a union
+# that admits `RetryMode` fails the Liskov check until the pin moves.
+def _resolve_retry_request_policy(middleware: Any) -> Any:
+    retry_mode = getattr(middleware, "RetryMode", None)
+    return True if retry_mode is None else retry_mode.UNBUFFERED
+
+
+_retry_request_policy: Any = _resolve_retry_request_policy(retry_middleware)
+
+
 class ProxyConfig(NamedTuple):
     """The ``proxy`` connection option in the shape pyqwest transports take.
 

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

@@ -1,5 +1,5 @@
 import threading
-from typing import Dict, Optional, Tuple, Union
+from typing import Any, Dict, Optional, Tuple, Union
 
 import httpx
 
@@ -10,6 +10,7 @@
 from e2b.api import (
     AsyncApiClient,
     ProxyConfig,
+    _retry_request_policy,
     connection_retries,
     make_async_logging_event_hooks,
     pool_idle_timeout,
@@ -32,7 +33,15 @@ class ConnectionRetryTransport(RetryTransport):
     call or unary RPC like ``SendInput``). This matches the connect-only
     ``retries`` of the httpx transports this replaced; the retry middleware's
     default policy would otherwise also retry I/O errors and 429/5xx responses
-    for idempotent methods."""
+    for idempotent methods.
+
+    On a pyqwest whose retry middleware exposes ``RetryMode``, streamed bodies
+    are retried without being buffered, so an upload is not mirrored in memory
+    as it is sent; earlier releases keep the buffered replay they ship (see
+    ``_retry_request_policy``)."""
+
+    def should_retry_request(self, request: Request) -> Any:
+        return _retry_request_policy
 
     def should_retry_response(
         self, request: Request, response: Union[Response, Exception]

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

@@ -1,4 +1,4 @@
-from typing import Dict, Optional, Tuple, Union
+from typing import Any, Dict, Optional, Tuple, Union
 
 import httpx
 import threading
@@ -10,6 +10,7 @@
 from e2b.api import (
     ApiClient,
     ProxyConfig,
+    _retry_request_policy,
     connection_retries,
     make_logging_event_hooks,
     pool_idle_timeout,
@@ -32,7 +33,15 @@ class ConnectionRetryTransport(SyncRetryTransport):
     call or unary RPC like ``SendInput``). This matches the connect-only
     ``retries`` of the httpx transports this replaced; the retry middleware's
     default policy would otherwise also retry I/O errors and 429/5xx responses
-    for idempotent methods."""
+    for idempotent methods.
+
+    On a pyqwest whose retry middleware exposes ``RetryMode``, streamed bodies
+    are retried without being buffered, so an upload is not mirrored in memory
+    as it is sent; earlier releases keep the buffered replay they ship (see
+    ``_retry_request_policy``)."""
+
+    def should_retry_request(self, request: SyncRequest) -> Any:
+        return _retry_request_policy
 
     def should_retry_response(
         self, request: SyncRequest, response: Union[SyncResponse, Exception]