fix(js-sdk): keep envd HTTP/2 sessions within Node's session memory budget

.changeset/calm-envd-http2-sessions.md

@@ -0,0 +1,5 @@
+---
+'e2b': patch
+---
+
+Fix `TypeError: fetch failed` with `NGHTTP2_ENHANCE_YOUR_CALM` on concurrent `files.write()` calls. Node's HTTP/2 client queues each buffered request body in full against a 10 MB per-session memory budget and resets whichever stream's response arrives while it is exhausted; since undici 8 every envd REST request of a process shares one session. Buffered `files.write()` uploads are now admitted so that the bodies in flight together stay within 6 MB, an upload larger than that is sent over an HTTP/2 session of its own (so it cannot reset concurrent reads), and a dispatcher whose session Node reports as out of memory is retired and replaced instead of failing every later request.

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

@@ -4,6 +4,7 @@ import type { FetchOpts } from '../api/http2'
 import { parsePositiveIntEnv } from '../api/metadata'
 import {
   buildDispatchedFetch,
+  buildSingleUseDispatchedFetch,
   createRuntimeFetch,
   type UndiciModule,
 } from '../undici'
@@ -19,9 +20,9 @@ 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 envdSoloFetchers = new Map<string, typeof fetch>()
 const DEFAULT_ENVD_CONNECTION_LIMIT = 10
 const DEFAULT_ENVD_RPC_CONNECTION_LIMIT = 200
-
 export function createEnvdFetchForRuntime(
   currentRuntime = runtime,
   options: EnvdFetchOptions = {}
@@ -36,6 +37,23 @@ export function createEnvdFetchForRuntime(
   )
 }
 
+/**
+ * Fetch for envd requests that must not share an HTTP/2 session: each request
+ * runs on a dispatcher of its own that is closed afterwards.
+ */
+export function createEnvdSoloFetchForRuntime(
+  currentRuntime = runtime,
+  options: Omit<EnvdFetchOptions, 'connectionLimit'> = {}
+): typeof fetch {
+  return createRuntimeFetch(currentRuntime, () =>
+    buildSingleUseDispatchedFetch({
+      proxy: options.proxy,
+      httpVersion: options.httpVersion,
+      loadUndici: options.loadUndici,
+    })
+  )
+}
+
 function fetcherKey(
   proxy: string | undefined,
   httpVersion: HttpVersion
@@ -54,8 +72,6 @@ export function createEnvdFetch({
     return cached
   }
 
-  // Keep one origin connection for short envd REST calls. If ALPN falls back
-  // to h1, this favors connection pressure over per-sandbox throughput.
   const envdFetch = createEnvdFetchForRuntime(runtime, {
     proxy,
     httpVersion,
@@ -86,6 +102,26 @@ export function createEnvdRpcFetch({
   return envdRpcFetch
 }
 
+export function createEnvdSoloFetch({
+  proxy,
+  httpVersion = DEFAULT_HTTP_VERSION,
+}: FetchOpts = {}): typeof fetch {
+  const key = fetcherKey(proxy, httpVersion)
+
+  const cached = envdSoloFetchers.get(key)
+  if (cached) {
+    return cached
+  }
+
+  const envdSoloFetch = createEnvdSoloFetchForRuntime(runtime, {
+    proxy,
+    httpVersion,
+  })
+  envdSoloFetchers.set(key, envdSoloFetch)
+
+  return envdSoloFetch
+}
+
 export function getEnvdRpcConnectionLimit(): number {
   return parsePositiveIntEnv(
     'E2B_ENVD_RPC_CONNECTIONS',

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

@@ -0,0 +1,127 @@
+// Node's HTTP/2 client queues a buffered request body (Blob, FormData, bytes,
+// string) in full against the session's memory budget (`maxSessionMemory`,
+// 10 MB) and resets whichever stream's response headers arrive while that
+// budget is exhausted (ENHANCE_YOUR_CALM). Since undici 8 every envd REST
+// request of a process multiplexes over one session, so concurrent uploads
+// share one budget — with every read and small write whose response may land
+// while it is full. Buffered uploads are admitted here so the bodies in flight
+// together stay well under it; a single body above this limit is sent on a
+// session of its own instead (see `Filesystem.write`). A stream body is
+// written chunk by chunk behind `drain`, so it is never queued whole.
+export const ENVD_MAX_INFLIGHT_BODY_BYTES = 6 * 1024 * 1024
+
+const utf8Length = (text: string): number =>
+  new TextEncoder().encode(text).byteLength
+
+/**
+ * Bytes of a request body Node queues in full, or 0 for a body that is
+ * streamed or of unknown size.
+ */
+export function bodyByteLength(body: unknown): number {
+  if (body == null) {
+    return 0
+  }
+  if (typeof body === 'string') {
+    return utf8Length(body)
+  }
+  if (body instanceof ArrayBuffer || ArrayBuffer.isView(body)) {
+    return body.byteLength
+  }
+  if (body instanceof Blob) {
+    return body.size
+  }
+  if (body instanceof URLSearchParams) {
+    return utf8Length(body.toString())
+  }
+  if (body instanceof FormData) {
+    let total = 0
+    body.forEach((value, name) => {
+      total +=
+        utf8Length(name) +
+        (typeof value === 'string' ? utf8Length(value) : value.size)
+    })
+    return total
+  }
+  return 0
+}
+
+export type BodyBudget = {
+  /**
+   * Run `fn` once `size` more bytes fit in flight; the bytes are released when
+   * the returned promise settles — fetch resolves on response headers, which
+   * envd sends after reading the whole body, so Node's queue is empty by then.
+   * A request aborted while waiting rejects with its signal's reason.
+   */
+  run<T>(
+    size: number,
+    signal: AbortSignal | null | undefined,
+    fn: () => Promise<T>
+  ): Promise<T>
+}
+
+/**
+ * Admit requests in order while their bodies fit in `limit` bytes together; a
+ * body larger than the limit still goes alone. Sizeless requests pass through.
+ */
+export function createBodyBudget(limit: number): BodyBudget {
+  let inflight = 0
+  const waiters: Array<{ size: number; admit: () => void }> = []
+
+  const fits = (size: number) => inflight === 0 || inflight + size <= limit
+  const admitWaiting = () => {
+    while (waiters.length > 0 && fits(waiters[0].size)) {
+      waiters.shift()!.admit()
+    }
+  }
+
+  const acquire = async (size: number, signal?: AbortSignal | null) => {
+    if (signal?.aborted) {
+      throw signal.reason
+    }
+    if (waiters.length === 0 && fits(size)) {
+      inflight += size
+      return
+    }
+
+    await new Promise<void>((resolve, reject) => {
+      const waiter = {
+        size,
+        admit: () => {
+          signal?.removeEventListener('abort', onAbort)
+          inflight += size
+          resolve()
+        },
+      }
+      const onAbort = () => {
+        const index = waiters.indexOf(waiter)
+        if (index !== -1) {
+          waiters.splice(index, 1)
+        }
+        reject(signal!.reason)
+        // The aborted waiter may have been holding smaller ones back.
+        admitWaiting()
+      }
+      signal?.addEventListener('abort', onAbort, { once: true })
+      waiters.push(waiter)
+    })
+  }
+
+  return {
+    async run(size, signal, fn) {
+      if (size <= 0) {
+        return fn()
+      }
+      await acquire(size, signal)
+      try {
+        return await fn()
+      } finally {
+        inflight -= size
+        admitWaiting()
+      }
+    },
+  }
+}
+
+// One budget per process: envd REST dispatchers are shared across sandboxes,
+// and sandboxes behind one authority share HTTP/2 sessions.
+export const envdUploadBudget = createBodyBudget(ENVD_MAX_INFLIGHT_BODY_BYTES)

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

@@ -38,6 +38,12 @@ import { FilesystemEvent, WatchHandle } from './watchHandle'
 
 import type { Timestamp } from '@bufbuild/protobuf/wkt'
 import { compareVersions } from 'compare-versions'
+import { createEnvdSoloFetch } from '../../envd/http2'
+import {
+  bodyByteLength,
+  ENVD_MAX_INFLIGHT_BODY_BYTES,
+  envdUploadBudget,
+} from '../../envd/uploadBudget'
 import {
   ENVD_DEFAULT_USER,
   ENVD_FILE_METADATA,
@@ -396,6 +402,27 @@ export class Filesystem {
     this.checkHealth = () => checkSandboxHealth(this.envdApi)
   }
 
+  /**
+   * How a buffered upload of `size` bytes is kept from exhausting Node's HTTP/2
+   * session memory: paced through `envdUploadBudget` together with the other
+   * uploads, or — when it alone would crowd out the reads and small writes on
+   * the shared session — sent over a session of its own. HTTP/1.1 and streamed
+   * bodies need neither.
+   */
+  private uploadAdmission(size: number): {
+    budgeted: number
+    fetch?: (request: Request) => Promise<Response>
+  } {
+    if (size <= 0 || this.connectionConfig.httpVersion !== '2') {
+      return { budgeted: 0 }
+    }
+    if (size > ENVD_MAX_INFLIGHT_BODY_BYTES) {
+      const soloFetch = createEnvdSoloFetch(this.connectionConfig)
+      return { budgeted: 0, fetch: (request) => soloFetch(request) }
+    }
+    return { budgeted: size }
+  }
+
   /**
    * Read file content as a `string`.
    *
@@ -722,32 +749,39 @@ export class Filesystem {
                 writeOpts?.signal
               )
 
-          const res = await this.envdApi.api
-            .POST('/files', {
-              params: {
-                query: {
-                  path: filePath,
-                  username: user,
+          const upload = this.uploadAdmission(
+            streamed ? 0 : bodyByteLength(body)
+          )
+          const res = await envdUploadBudget.run(upload.budgeted, signal, () =>
+            this.envdApi.api
+              .POST('/files', {
+                params: {
+                  query: {
+                    path: filePath,
+                    username: user,
+                  },
                 },
-              },
-              bodySerializer: () => body,
-              headers,
-              signal,
-              body: {},
-              // Streaming request bodies require half-duplex mode.
-              ...(streamed && {
-                duplex: 'half' as const,
-              }),
-            })
-            .catch(async (err) => {
-              // fetch reports an error raised while reading a streamed body as
-              // its own `TypeError`; surface the SDK's own validation error.
-              const invalidChunk = sourceError()
-              if (invalidChunk) {
-                throw invalidChunk
-              }
-              throw await handleEnvdApiFetchError(err, this.checkHealth)
-            })
+                bodySerializer: () => body,
+                headers,
+                signal,
+                body: {},
+                fetch: upload.fetch,
+                // Streaming request bodies require half-duplex mode.
+                ...(streamed && {
+                  duplex: 'half' as const,
+                }),
+              })
+              .catch(async (err) => {
+                // fetch reports an error raised while reading a streamed body
+                // as its own `TypeError`; surface the SDK's own validation
+                // error.
+                const invalidChunk = sourceError()
+                if (invalidChunk) {
+                  throw invalidChunk
+                }
+                throw await handleEnvdApiFetchError(err, this.checkHealth)
+              })
+          )
 
           const err = await handleFilesystemEnvdApiError(res)
           if (err) {
@@ -782,25 +816,30 @@ export class Filesystem {
         )
       }
 
-      const res = await this.envdApi.api
-        .POST('/files', {
-          params: {
-            query: {
-              path,
-              username: user,
+      const signal = this.connectionConfig.getSignal(
+        writeOpts?.requestTimeoutMs,
+        writeOpts?.signal
+      )
+      const upload = this.uploadAdmission(bodyByteLength(formData))
+      const res = await envdUploadBudget.run(upload.budgeted, signal, () =>
+        this.envdApi.api
+          .POST('/files', {
+            params: {
+              query: {
+                path,
+                username: user,
+              },
             },
-          },
-          bodySerializer: () => formData,
-          headers: extraHeaders,
-          signal: this.connectionConfig.getSignal(
-            writeOpts?.requestTimeoutMs,
-            writeOpts?.signal
-          ),
-          body: {},
-        })
-        .catch(async (err) => {
-          throw await handleEnvdApiFetchError(err, this.checkHealth)
-        })
+            bodySerializer: () => formData,
+            headers: extraHeaders,
+            signal,
+            body: {},
+            fetch: upload.fetch,
+          })
+          .catch(async (err) => {
+            throw await handleEnvdApiFetchError(err, this.checkHealth)
+          })
+      )
 
       const err = await handleFilesystemEnvdApiError(res)
       if (err) {