Stream SDK file uploads and downloads instead of buffering in memory

.changeset/cuddly-pots-stream.md

@@ -0,0 +1,12 @@
+---
+"e2b": patch
+"@e2b/python-sdk": patch
+---
+
+Stream uploads instead of buffering streaming input entirely in memory:
+
+- `Volume.writeFile()` / `Volume.write_file()`: `ReadableStream` data (JS, outside the browser) and file-like objects (Python) are now streamed to the API in chunks.
+- `Sandbox.files.write()` / `write_files()` with `useOctetStream`/`use_octet_stream`: `ReadableStream` data (JS, outside the browser) and file-like objects (Python) are streamed to the sandbox, including when `gzip` is enabled (compression now happens chunk by chunk).
+- Python `Sandbox.files.read(format="stream")`: the response body is now streamed from the sandbox instead of being downloaded into memory before iteration (sync and async).
+- JS `Sandbox.files.read({ format: 'stream' })`: the request timeout now bounds only the initial handshake instead of killing the stream while it's being consumed; pass `signal` to cancel an in-flight stream.
+- JS `Sandbox.files.read()` with `blob` or `stream` format now returns an empty `Blob`/`ReadableStream` for empty files instead of `""`.

packages/js-sdk/src/connectionConfig.ts

@@ -7,6 +7,10 @@ const supportedDomains = ['e2b.app', 'e2b.dev', 'e2b.pro', 'e2b-staging.dev']
 
 export const REQUEST_TIMEOUT_MS = 60_000 // 60 seconds
 export const DEFAULT_SANDBOX_TIMEOUT_MS = 300_000 // 300 seconds
+// Default timeout for streaming file transfers (uploads/downloads). A streamed
+// body can take far longer than a regular request, so it must not inherit the
+// short `REQUEST_TIMEOUT_MS`.
+export const FILE_TIMEOUT_MS = 3_600_000 // 1 hour
 export const KEEPALIVE_PING_INTERVAL_SEC = 50 // 50 seconds
 
 export const KEEPALIVE_PING_HEADER = 'Keepalive-Ping-Interval'
@@ -191,6 +195,80 @@ export function setupRequestController(
   return { controller, clearStartTimeout, cleanup }
 }
 
+// GC safety net for streamed reads: if the consumer drops a streamed response
+// body without reading it to completion or cancelling it, the registered
+// cleanup releases the underlying connection when the stream is garbage
+// collected. This mirrors the Python SDK's `weakref.finalize` on
+// `FileStreamReader`. The held value is the cleanup function, which must not
+// reference the stream itself or it would never be collected.
+const streamReadFinalizers = new FinalizationRegistry<() => void>((cleanup) =>
+  cleanup()
+)
+
+/**
+ * Wrap a streaming response body so its pooled connection is released when the
+ * stream is fully read, cancelled, errors, or (as a GC safety net) abandoned.
+ *
+ * The request timeout configured via {@link setupRequestController} bounds only
+ * the initial handshake; this clears that timeout so consuming the body is not
+ * killed by it. Call once the handshake has succeeded (after error handling).
+ *
+ * @internal
+ */
+export function wrapStreamWithConnectionCleanup(
+  body: ReadableStream<Uint8Array> | null,
+  {
+    clearStartTimeout,
+    cleanup,
+  }: { clearStartTimeout: () => void; cleanup: () => void }
+): ReadableStream<Uint8Array> {
+  clearStartTimeout()
+
+  if (!body) {
+    cleanup()
+    return new Blob([]).stream()
+  }
+
+  const reader = body.getReader()
+  const unregisterToken = {}
+  // Detach the GC finalizer and release the connection. Idempotent via
+  // `cleanup`, so it's safe to call from multiple stream callbacks.
+  const release = () => {
+    streamReadFinalizers.unregister(unregisterToken)
+    cleanup()
+  }
+
+  const stream = new ReadableStream<Uint8Array>({
+    async pull(streamController) {
+      try {
+        const { done, value } = await reader.read()
+        if (done) {
+          streamController.close()
+          release()
+        } else {
+          streamController.enqueue(value)
+        }
+      } catch (err) {
+        release()
+        streamController.error(err)
+      }
+    },
+    async cancel(reason) {
+      try {
+        await reader.cancel(reason)
+      } finally {
+        release()
+      }
+    },
+  })
+
+  // Release the connection if the consumer abandons the stream without
+  // reading it to completion or cancelling it.
+  streamReadFinalizers.register(stream, cleanup, unregisterToken)
+
+  return stream
+}
+
 /**
  * Configuration for connecting to the API.
  */

packages/js-sdk/src/sandbox/filesystem/index.ts

@@ -9,10 +9,12 @@ import {
   ConnectionConfig,
   ConnectionOpts,
   defaultUsername,
+  FILE_TIMEOUT_MS,
   KEEPALIVE_PING_HEADER,
   KEEPALIVE_PING_INTERVAL_SEC,
   setupRequestController,
   Username,
+  wrapStreamWithConnectionCleanup,
 } from '../../connectionConfig'
 
 import {
@@ -271,6 +273,8 @@ export interface FilesystemWriteOpts extends FilesystemRequestOpts {
   gzip?: boolean
   /**
    * When true, the upload uses `application/octet-stream` instead of `multipart/form-data`.
+   * Outside the browser, `ReadableStream` data is then streamed to the sandbox
+   * instead of being buffered in memory.
    *
    * Defaults to `false`. Requires envd 0.5.7 or later — when not supported by
    * the sandbox's envd version, the upload falls back to `multipart/form-data`.
@@ -412,6 +416,10 @@ export class Filesystem {
    *
    * You can pass `text`, `bytes`, `blob`, or `stream` to `opts.format` to change the return type.
    *
+   * The request timeout bounds only the initial handshake—the returned
+   * stream is not killed by it while being consumed. Use `opts.signal` to
+   * cancel an in-flight stream.
+   *
    * @param path path to the file.
    * @param opts connection options.
    * @param [opts.format] format of the file content—`stream`.
@@ -443,6 +451,58 @@ export class Filesystem {
       headers['Accept-Encoding'] = 'gzip'
     }
 
+    if (format === 'stream') {
+      // The request timeout bounds only the initial handshake; once the
+      // response arrives, the stream lives until it's consumed, cancelled,
+      // or the user signal aborts.
+      const { controller, clearStartTimeout, cleanup } = setupRequestController(
+        opts?.requestTimeoutMs ?? this.connectionConfig.requestTimeoutMs,
+        opts?.signal
+      )
+
+      try {
+        const res = await this.envdApi.api
+          .GET('/files', {
+            params: {
+              query: {
+                path,
+                username: user,
+              },
+            },
+            parseAs: 'stream',
+            signal: controller.signal,
+            headers,
+          })
+          .catch(async (err) => {
+            // Map a dropped connection during the handshake (e.g. killed
+            // sandbox) to a typed error via the health check, matching the
+            // non-stream read path below.
+            throw await handleEnvdApiFetchError(err, this.checkHealth)
+          })
+
+        const err = await handleFilesystemEnvdApiError(res)
+        if (err) {
+          // Cancel the unconsumed error body so the pooled connection is
+          // released before we propagate, matching the Python stream path's
+          // `r.close()`. `cleanup()`'s abort would also release it, but
+          // cancelling is explicit and independent of runtime abort semantics.
+          if (res.response.body && !res.response.bodyUsed) {
+            await res.response.body.cancel().catch(() => {})
+          }
+          cleanup()
+          throw err
+        }
+
+        return wrapStreamWithConnectionCleanup(
+          res.data as ReadableStream<Uint8Array> | null,
+          { clearStartTimeout, cleanup }
+        )
+      } catch (err) {
+        cleanup()
+        throw err
+      }
+    }
+
     const res = await this.envdApi.api
       .GET('/files', {
         params: {
@@ -467,13 +527,17 @@ export class Filesystem {
       throw err
     }
 
-    if (format === 'bytes') {
-      return new Uint8Array(res.data as ArrayBuffer)
+    // When the file is empty, the response body is skipped and `res.data` is
+    // `undefined`. Return the proper empty value for the requested format.
+    if (res.response.headers.get('content-length') === '0') {
+      if (format === 'bytes') {
+        return new Uint8Array(0)
+      }
+      return format === 'blob' ? new Blob([]) : ''
     }
 
-    // When the file is empty, res.data is parsed as `{}`. This is a workaround to return an empty string.
-    if (res.response.headers.get('content-length') === '0') {
-      return ''
+    if (format === 'bytes') {
+      return new Uint8Array(res.data as ArrayBuffer)
     }
 
     return res.data
@@ -593,6 +657,15 @@ export class Filesystem {
         writeFiles.map(async (file) => {
           const filePath = path ?? (file as WriteEntry).path
           const body = await toUploadBody(file.data, useGzip)
+          const isStream = body instanceof ReadableStream
+          // A streamed upload can take far longer than the 60s request default,
+          // so fall back to the file-transfer timeout (matching volume writes)
+          // unless the caller set one explicitly. The signal is a total
+          // deadline—unlike downloads there's no post-handshake point to clear
+          // it, since the response only arrives once the body has been sent.
+          const uploadTimeoutMs =
+            writeOpts?.requestTimeoutMs ??
+            (isStream ? FILE_TIMEOUT_MS : undefined)
 
           const res = await this.envdApi.api
             .POST('/files', {
@@ -605,10 +678,14 @@ export class Filesystem {
               bodySerializer: () => body,
               headers,
               signal: this.connectionConfig.getSignal(
-                writeOpts?.requestTimeoutMs,
+                uploadTimeoutMs,
                 writeOpts?.signal
               ),
               body: {},
+              // Streaming request bodies require half-duplex mode.
+              ...(isStream && {
+                duplex: 'half' as const,
+              }),
             })
             .catch(async (err) => {
               throw await handleEnvdApiFetchError(err, this.checkHealth)

packages/js-sdk/src/utils.ts

@@ -145,7 +145,11 @@ export function shellQuote(s: string): string {
 
 /**
  * Prepare data for upload as a BodyInit, optionally gzip-compressed.
- * When gzip is enabled, compresses the data and returns a Blob.
+ *
+ * Outside the browser, streams (and gzip-compressed data) are returned as
+ * `ReadableStream` so they can be uploaded without buffering in memory.
+ * Browsers don't support streaming request bodies, so data is buffered into
+ * a Blob there.
  */
 export async function toUploadBody(
   data: string | ArrayBuffer | Blob | ReadableStream,
@@ -159,7 +163,11 @@ export async function toUploadBody(
           ? data.stream()
           : new Blob([data]).stream()
     const compressed = stream.pipeThrough(new CompressionStream('gzip'))
-    return new Response(compressed).blob()
+    return runtime === 'browser' ? new Response(compressed).blob() : compressed
+  }
+
+  if (data instanceof ReadableStream && runtime !== 'browser') {
+    return data
   }
 
   return toBlob(data)

packages/js-sdk/src/volume/client.ts

@@ -3,12 +3,11 @@ import createClient from 'openapi-fetch'
 import type { components, paths } from './schema.gen'
 import { defaultHeaders, getEnvVar } from '../api/metadata'
 import { createApiFetch } from '../api/http2'
-import { buildRequestSignal } from '../connectionConfig'
+import { buildRequestSignal, FILE_TIMEOUT_MS } from '../connectionConfig'
 import { createApiLogger, Logger } from '../logs'
 import type { Volume } from './index'
 
 const REQUEST_TIMEOUT_MS = 60_000 // 60 seconds
-const FILE_TIMEOUT_MS = 3_600_000 // 1 hour
 
 export interface VolumeApiOpts {
   /**