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/sandbox/filesystem/index.ts

@@ -267,6 +267,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`.
@@ -408,6 +410,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`.
@@ -439,6 +445,71 @@ 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,
+        })
+
+        const err = await handleFilesystemEnvdApiError(res)
+        if (err) {
+          throw err
+        }
+
+        clearStartTimeout()
+
+        const body = res.data as ReadableStream<Uint8Array> | null
+        if (!body) {
+          cleanup()
+          return new Blob([]).stream()
+        }
+
+        const reader = body.getReader()
+        return new ReadableStream<Uint8Array>({
+          async pull(streamController) {
+            try {
+              const { done, value } = await reader.read()
+              if (done) {
+                streamController.close()
+                cleanup()
+              } else {
+                streamController.enqueue(value)
+              }
+            } catch (err) {
+              cleanup()
+              streamController.error(err)
+            }
+          },
+          async cancel(reason) {
+            try {
+              await reader.cancel(reason)
+            } finally {
+              cleanup()
+            }
+          },
+        })
+      } catch (err) {
+        cleanup()
+        throw err
+      }
+    }
+
     const res = await this.envdApi.api
       .GET('/files', {
         params: {
@@ -463,13 +534,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
@@ -602,6 +677,10 @@ export class Filesystem {
                 writeOpts?.signal
               ),
               body: {},
+              // Streaming request bodies require half-duplex mode.
+              ...(body instanceof ReadableStream && {
+                duplex: 'half' as const,
+              }),
             })
             .catch(async (err) => {
               throw await handleEnvdApiFetchError(err, this.checkHealth)

packages/js-sdk/src/utils.ts

@@ -131,7 +131,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,
@@ -145,7 +149,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/index.ts

@@ -8,7 +8,7 @@ import {
 } from './client'
 import { ConnectionConfig, ConnectionOpts } from '../connectionConfig'
 import { NotFoundError, VolumeError } from '../errors'
-import { toBlob } from '../utils'
+import { runtime, toBlob } from '../utils'
 import { VolumeFileType } from './types'
 import type {
   VolumeAndToken,
@@ -587,7 +587,7 @@ export class Volume {
    * Writing to a file that already exists overwrites the file.
    *
    * @param path path to the file.
-   * @param data data to write to the file. Data can be a string, `ArrayBuffer`, `Blob`, or `ReadableStream`.
+   * @param data data to write to the file. Data can be a string, `ArrayBuffer`, `Blob`, or `ReadableStream`. Outside the browser, `ReadableStream` data is streamed to the API instead of being buffered in memory.
    * @param options file creation options.
    * @param opts connection options.
    *
@@ -604,7 +604,9 @@ export class Volume {
     })
     const client = new VolumeApiClient(config)
 
-    const blob = await toBlob(data)
+    // Browsers don't support streaming request bodies, so buffer there.
+    const isStream = data instanceof ReadableStream && runtime !== 'browser'
+    const body = isStream ? data : await toBlob(data)
 
     const res = await client.api.PUT('/volumecontent/{volumeID}/file', {
       params: {
@@ -619,12 +621,14 @@ export class Volume {
           force: opts?.force,
         },
       },
-      bodySerializer: () => blob,
+      bodySerializer: () => body,
       body: {} as any,
       headers: {
         'Content-Type': 'application/octet-stream',
       },
       signal: config.getSignal(),
+      // Streaming request bodies require half-duplex mode.
+      ...(isStream && { duplex: 'half' as const }),
     })
 
     if (res.response.status === 404) {

packages/js-sdk/tests/sandbox/files/read.test.ts

@@ -38,3 +38,50 @@ sandboxTest('empty file', async ({ sandbox }) => {
   const content = await sandbox.files.read(filename)
   expect(content).toBe('')
 })
+
+sandboxTest('read file as stream', async ({ sandbox }) => {
+  const filename = 'test_read_stream.txt'
+  const content = 'Streamed read content. '.repeat(10_000)
+
+  await sandbox.files.write(filename, content)
+  const stream = await sandbox.files.read(filename, { format: 'stream' })
+
+  const chunks: Uint8Array[] = []
+  for await (const chunk of stream as unknown as AsyncIterable<Uint8Array>) {
+    chunks.push(chunk)
+  }
+  const readContent = Buffer.concat(chunks).toString('utf-8')
+  assert.equal(readContent, content)
+})
+
+sandboxTest('read non-existing file as stream', async ({ sandbox }) => {
+  const filename = 'non_existing_file.txt'
+
+  await expect(
+    sandbox.files.read(filename, { format: 'stream' })
+  ).rejects.toThrowError(FileNotFoundError)
+})
+
+sandboxTest('read empty file in all formats', async ({ sandbox }) => {
+  const filename = 'empty-file-formats.txt'
+  await sandbox.commands.run(`touch ${filename}`)
+
+  const text = await sandbox.files.read(filename, { format: 'text' })
+  expect(text).toBe('')
+
+  const bytes = await sandbox.files.read(filename, { format: 'bytes' })
+  expect(bytes).toBeInstanceOf(Uint8Array)
+  expect(bytes.length).toBe(0)
+
+  const blob = await sandbox.files.read(filename, { format: 'blob' })
+  expect(blob).toBeInstanceOf(Blob)
+  expect(blob.size).toBe(0)
+
+  const stream = await sandbox.files.read(filename, { format: 'stream' })
+  expect(stream).toBeInstanceOf(ReadableStream)
+  const chunks: Uint8Array[] = []
+  for await (const chunk of stream as unknown as AsyncIterable<Uint8Array>) {
+    chunks.push(chunk)
+  }
+  expect(Buffer.concat(chunks).length).toBe(0)
+})

packages/js-sdk/tests/sandbox/files/write.test.ts

@@ -321,3 +321,46 @@ sandboxTest('writeFiles overwrites existing files', async ({ sandbox }) => {
     await sandbox.files.remove(filename)
   }
 })
+
+sandboxTest(
+  'write ReadableStream with octet stream upload',
+  async ({ sandbox }) => {
+    const filename = 'test_write_octet_stream.bin'
+    const content = 'Streamed octet-stream upload. '.repeat(10_000)
+    const stream = new Blob([content]).stream()
+
+    const info = await sandbox.files.write(filename, stream, {
+      useOctetStream: true,
+    })
+    assert.equal(info.path, `/home/user/${filename}`)
+
+    const readContent = await sandbox.files.read(filename)
+    assert.equal(readContent, content)
+
+    if (isDebug) {
+      await sandbox.files.remove(filename)
+    }
+  }
+)
+
+sandboxTest(
+  'write ReadableStream with octet stream upload and gzip',
+  async ({ sandbox }) => {
+    const filename = 'test_write_octet_stream_gzip.bin'
+    const content = 'Streamed gzipped octet-stream upload. '.repeat(10_000)
+    const stream = new Blob([content]).stream()
+
+    const info = await sandbox.files.write(filename, stream, {
+      useOctetStream: true,
+      gzip: true,
+    })
+    assert.equal(info.path, `/home/user/${filename}`)
+
+    const readContent = await sandbox.files.read(filename)
+    assert.equal(readContent, content)
+
+    if (isDebug) {
+      await sandbox.files.remove(filename)
+    }
+  }
+)

packages/js-sdk/tests/volume/file.test.ts

@@ -58,6 +58,20 @@ describe('Volume File Operations', () => {
       }
     )
 
+    volumeTest(
+      'should write and read a file from a ReadableStream',
+      async ({ volume }) => {
+        const path = '/test-stream.txt'
+        const content = 'Test stream content'
+        const stream = new Blob([content]).stream()
+
+        await volume.writeFile(path, stream)
+        const readContent = await volume.readFile(path, { format: 'text' })
+
+        expect(readContent).toBe(content)
+      }
+    )
+
     volumeTest('should write and read an empty file', async ({ volume }) => {
       const path = '/empty.txt'
       const content = ''
@@ -68,6 +82,30 @@ describe('Volume File Operations', () => {
       expect(readContent).toBe(content)
     })
 
+    volumeTest(
+      'should read an empty file in all formats',
+      async ({ volume }) => {
+        const path = '/empty-formats.txt'
+        await volume.writeFile(path, '')
+
+        const bytes = await volume.readFile(path, { format: 'bytes' })
+        expect(bytes).toBeInstanceOf(Uint8Array)
+        expect(bytes.length).toBe(0)
+
+        const blob = await volume.readFile(path, { format: 'blob' })
+        expect(blob).toBeInstanceOf(Blob)
+        expect(blob.size).toBe(0)
+
+        const stream = await volume.readFile(path, { format: 'stream' })
+        expect(stream).toBeInstanceOf(ReadableStream)
+        const chunks: Uint8Array[] = []
+        for await (const chunk of stream as unknown as AsyncIterable<Uint8Array>) {
+          chunks.push(chunk)
+        }
+        expect(chunks.reduce((n, c) => n + c.length, 0)).toBe(0)
+      }
+    )
+
     volumeTest(
       'should overwrite an existing file with force option',
       async ({ volume }) => {