fix(js-sdk): send envd file uploads over HTTP/1.1, retire poisoned HTTP/2 dispatchers

.changeset/envd-uploads-http1.md

@@ -0,0 +1,5 @@
+---
+'e2b': patch
+---
+
+Fix `TypeError: fetch failed` / `NGHTTP2_ENHANCE_YOUR_CALM` on concurrent `files.write()` calls. Node's HTTP/2 client queues each request body in full against a 10 MB per-session memory budget and resets the stream whose response arrives while it is exhausted; since undici 8 every envd request of a process shares one session, so parallel or large uploads reset each other and the reads next to them. `files.write()` uploads now go over HTTP/1.1 on a pool of their own (one request per connection, no shared budget; `E2B_ENVD_UPLOAD_CONNECTIONS`, default 50), and an envd dispatcher whose HTTP/2 session Node reports as out of memory is retired and replaced instead of failing every later request. Nothing is retried: a failed write may already have been applied.

packages/js-sdk/src/connectionConfig.ts

@@ -138,7 +138,10 @@ export interface ConnectionOpts {
    * filesystem, PTY). `'2'` multiplexes streams over shared connections;
    * `'1.1'` pins them to HTTP/1.1 with one connection per concurrent
    * request — for example when an intermediary on the path retires or
-   * mishandles long-lived HTTP/2 connections. Only applies in Node.
+   * mishandles long-lived HTTP/2 connections. File uploads (`files.write()`)
+   * always use HTTP/1.1, whatever this option says, so that their bodies
+   * cannot exhaust Node's per-session HTTP/2 memory and reset other requests.
+   * Only applies in Node.
    *
    * @default E2B_HTTP_VERSION // environment variable or `'2'`
    */

packages/js-sdk/src/envd/http2.ts

@@ -19,8 +19,10 @@ type EnvdFetchOptions = {
 // keep sharing a single dispatcher while each distinct proxy URL gets its own.
 const envdFetchers = new Map<string, typeof fetch>()
 const envdRpcFetchers = new Map<string, typeof fetch>()
+const envdUploadFetchers = new Map<string, typeof fetch>()
 const DEFAULT_ENVD_CONNECTION_LIMIT = 10
 const DEFAULT_ENVD_RPC_CONNECTION_LIMIT = 200
+const DEFAULT_ENVD_UPLOAD_CONNECTION_LIMIT = 50
 
 export function createEnvdFetchForRuntime(
   currentRuntime = runtime,
@@ -92,3 +94,40 @@ export function getEnvdRpcConnectionLimit(): number {
     DEFAULT_ENVD_RPC_CONNECTION_LIMIT
   )
 }
+
+/**
+ * Fetch for `/files` uploads, always HTTP/1.1. Node's HTTP/2 client queues
+ * each buffered request body in full against the session's memory budget
+ * (`maxSessionMemory`, 10 MB) and resets whichever stream's response headers
+ * arrive while it is exhausted (ENHANCE_YOUR_CALM), so uploads multiplexed
+ * with everything else on the shared envd session fail each other and the
+ * reads next to them. HTTP/1.1 gives each upload a connection of its own; the
+ * pool is sized for many concurrent uploads per process since the request
+ * timeout keeps running while a request waits for a free connection.
+ */
+export function createEnvdUploadFetch({
+  proxy,
+}: Pick<FetchOpts, 'proxy'> = {}): typeof fetch {
+  const key = fetcherKey(proxy, '1.1')
+
+  const cached = envdUploadFetchers.get(key)
+  if (cached) {
+    return cached
+  }
+
+  const envdUploadFetch = createEnvdFetchForRuntime(runtime, {
+    connectionLimit: getEnvdUploadConnectionLimit(),
+    proxy,
+    httpVersion: '1.1',
+  })
+  envdUploadFetchers.set(key, envdUploadFetch)
+
+  return envdUploadFetch
+}
+
+export function getEnvdUploadConnectionLimit(): number {
+  return parsePositiveIntEnv(
+    'E2B_ENVD_UPLOAD_CONNECTIONS',
+    DEFAULT_ENVD_UPLOAD_CONNECTION_LIMIT
+  )
+}

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

@@ -15,6 +15,7 @@ import {
   Username,
   wrapStreamWithConnectionCleanup,
 } from '../../connectionConfig'
+import { createEnvdUploadFetch } from '../../envd/http2'
 
 import {
   checkSandboxHealth,
@@ -387,11 +388,16 @@ export class Filesystem {
   private readonly defaultWatchRecursive = false
   private readonly checkHealth: SandboxHealthCheck
 
+  /** `/files` uploads go over HTTP/1.1, see {@link createEnvdUploadFetch}. */
+  private readonly uploadFetch: (request: Request) => Promise<Response>
+
   constructor(
     transport: Transport,
     private readonly envdApi: EnvdApiClient,
     private readonly connectionConfig: ConnectionConfig
   ) {
+    const uploadFetch = createEnvdUploadFetch({ proxy: connectionConfig.proxy })
+    this.uploadFetch = (request) => uploadFetch(request)
     this.rpc = createClient(FilesystemService, transport)
     this.checkHealth = () => checkSandboxHealth(this.envdApi)
   }
@@ -734,6 +740,7 @@ export class Filesystem {
               headers,
               signal,
               body: {},
+              fetch: this.uploadFetch,
               // Streaming request bodies require half-duplex mode.
               ...(streamed && {
                 duplex: 'half' as const,
@@ -797,6 +804,7 @@ export class Filesystem {
             writeOpts?.signal
           ),
           body: {},
+          fetch: this.uploadFetch,
         })
         .catch(async (err) => {
           throw await handleEnvdApiFetchError(err, this.checkHealth)

packages/js-sdk/src/undici.ts

@@ -9,17 +9,46 @@ type UndiciRequestInit = RequestInit & {
   duplex?: 'half'
 }
 
+type UndiciDispatcher = { close(): Promise<void> }
+
+type UndiciAgentOptions = {
+  allowH2: boolean
+  connections?: number
+}
+
 export type UndiciModule = {
-  Agent: new (options: { allowH2: boolean; connections?: number }) => unknown
-  ProxyAgent: new (options: {
-    uri: string
-    allowH2: boolean
-    connections?: number
-    proxyTunnel: true
-  }) => unknown
+  Agent: new (options: UndiciAgentOptions) => UndiciDispatcher
+  ProxyAgent: new (
+    options: UndiciAgentOptions & { uri: string; proxyTunnel: true }
+  ) => UndiciDispatcher
   fetch: unknown
 }
 
+// Node's HTTP/2 client charges queued request-body bytes against a per-session
+// memory budget (`maxSessionMemory`, 10 MB) and resets the stream whose
+// response headers arrive while the budget is exhausted — with
+// ENHANCE_YOUR_CALM, or INTERNAL_ERROR where the reset is deferred. Bytes of
+// streams destroyed before they were flushed stay charged until the session
+// closes (nodejs/node#67113), so such a reset marks the session as poisoned.
+const POISONED_HTTP2_SESSION_RST =
+  /NGHTTP2_(ENHANCE_YOUR_CALM|INTERNAL_ERROR)\b/
+
+/**
+ * Whether a `fetch` rejection is Node's HTTP/2 client resetting the stream
+ * itself because the session's memory budget is exhausted. The session stays
+ * unusable for every later request, so the dispatcher holding it must go.
+ */
+export function isPoisonedHTTP2SessionError(err: unknown): boolean {
+  if (!(err instanceof Error) || !(err.cause instanceof Error)) {
+    return false
+  }
+  const cause = err.cause as Error & { code?: unknown }
+  return (
+    cause.code === 'ERR_HTTP2_STREAM_ERROR' &&
+    POISONED_HTTP2_SESSION_RST.test(cause.message)
+  )
+}
+
 const UNDICI_MIN_NODE = '22.19.0'
 
 export async function loadUndici(
@@ -84,7 +113,9 @@ export function createRuntimeFetch(
 /**
  * Build a fetch bound to a bounded undici dispatcher (HTTP/2 enabled unless
  * `httpVersion` is `'1.1'`, `connections` origin connections, optional proxy
- * tunnel). Falls back to the global fetch when undici cannot be loaded.
+ * tunnel). A dispatcher whose HTTP/2 session Node reports as out of memory is
+ * retired and replaced, so one poisoned session does not fail every later
+ * request. Falls back to the global fetch when undici cannot be loaded.
  */
 export async function buildDispatchedFetch(options: {
   connections: number
@@ -99,30 +130,44 @@ export async function buildDispatchedFetch(options: {
   }
 
   const { Agent, ProxyAgent, fetch: undiciFetch } = undici
-  const allowH2 = (options.httpVersion ?? DEFAULT_HTTP_VERSION) === '2'
-  const dispatcher = options.proxy
-    ? new ProxyAgent({
-        uri: options.proxy,
-        allowH2,
-        connections: options.connections,
-        proxyTunnel: true,
-      })
-    : new Agent({
-        allowH2,
-        connections: options.connections,
-      })
+  const agentOptions: UndiciAgentOptions = {
+    allowH2: (options.httpVersion ?? DEFAULT_HTTP_VERSION) === '2',
+    connections: options.connections,
+  }
+  const createDispatcher = () =>
+    options.proxy
+      ? new ProxyAgent({
+          uri: options.proxy,
+          ...agentOptions,
+          proxyTunnel: true,
+        })
+      : new Agent(agentOptions)
   const fetchWithDispatcher = undiciFetch as unknown as (
     input: RequestInfo | URL,
     init?: UndiciRequestInit
   ) => Promise<Response>
 
-  return ((input, init) => {
+  let dispatcher = createDispatcher()
+
+  return (async (input, init) => {
     const request = toUndiciRequestInput(input, init)
+    const used = dispatcher
 
-    return fetchWithDispatcher(request.input, {
-      ...request.init,
-      dispatcher,
-    })
+    try {
+      return await fetchWithDispatcher(request.input, {
+        ...request.init,
+        dispatcher: used,
+      })
+    } catch (err) {
+      // Only the first failure of a dispatcher replaces it; its other in-flight
+      // requests fail the same way and must not retire the replacement.
+      if (used === dispatcher && isPoisonedHTTP2SessionError(err)) {
+        dispatcher = createDispatcher()
+        // Graceful: requests still running on the old session finish first.
+        void used.close().catch(() => {})
+      }
+      throw err
+    }
   }) as typeof fetch
 }
 

packages/js-sdk/tests/envd/http2.test.ts

@@ -4,6 +4,7 @@ afterEach(() => {
   vi.restoreAllMocks()
   vi.resetModules()
   delete process.env.E2B_ENVD_RPC_CONNECTIONS
+  delete process.env.E2B_ENVD_UPLOAD_CONNECTIONS
 })
 
 test('uses undici with HTTP/2 enabled in Node', async () => {
@@ -297,3 +298,16 @@ test('uses global fetch outside Node', async () => {
     undefined
   )
 })
+
+test('getEnvdUploadConnectionLimit defaults to 50 and honours E2B_ENVD_UPLOAD_CONNECTIONS', async () => {
+  const { getEnvdUploadConnectionLimit } = await import('../../src/envd/http2')
+  expect(getEnvdUploadConnectionLimit()).toBe(50)
+
+  process.env.E2B_ENVD_UPLOAD_CONNECTIONS = '8'
+  expect(getEnvdUploadConnectionLimit()).toBe(8)
+
+  process.env.E2B_ENVD_UPLOAD_CONNECTIONS = 'bogus'
+  expect(() => getEnvdUploadConnectionLimit()).toThrow(
+    /E2B_ENVD_UPLOAD_CONNECTIONS/
+  )
+})

packages/js-sdk/tests/sandbox/httpVersion.test.ts

@@ -3,6 +3,7 @@ import { afterEach, expect, test, vi } from 'vitest'
 const mocks = vi.hoisted(() => ({
   createEnvdFetch: vi.fn(() => vi.fn()),
   createEnvdRpcFetch: vi.fn(() => vi.fn()),
+  createEnvdUploadFetch: vi.fn(() => vi.fn()),
 }))
 
 vi.mock('@connectrpc/connect-web', () => ({
@@ -12,6 +13,7 @@ vi.mock('@connectrpc/connect-web', () => ({
 vi.mock('../../src/envd/http2', () => ({
   createEnvdFetch: mocks.createEnvdFetch,
   createEnvdRpcFetch: mocks.createEnvdRpcFetch,
+  createEnvdUploadFetch: mocks.createEnvdUploadFetch,
 }))
 
 afterEach(() => {
@@ -73,3 +75,14 @@ test('E2B_HTTP_VERSION=1.1 pins envd fetchers to HTTP/1.1', async () => {
     httpVersion: '1.1',
   })
 })
+
+test('files.write uploads get the HTTP/1.1 upload fetcher for the configured proxy', async () => {
+  await createSandbox({ httpVersion: '2', proxy: 'http://127.0.0.1:8080' })
+
+  expect(mocks.createEnvdUploadFetch).toHaveBeenCalledWith({
+    proxy: 'http://127.0.0.1:8080',
+  })
+  expect(mocks.createEnvdFetch.mock.calls[0][0]).toMatchObject({
+    httpVersion: '2',
+  })
+})

packages/js-sdk/tests/sandbox/rpcHeaders.test.ts

@@ -15,6 +15,7 @@ vi.mock('@connectrpc/connect-web', () => ({
 vi.mock('../../src/envd/http2', () => ({
   createEnvdFetch: vi.fn(() => vi.fn()),
   createEnvdRpcFetch: vi.fn(() => mocks.rpcFetch),
+  createEnvdUploadFetch: vi.fn(() => vi.fn()),
 }))
 
 test('does not pass custom connection headers to envd RPC requests', async () => {