From ece2c43c851d3d946c4a949ebeb7317fa87df378 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 16:10:26 +0000 Subject: [PATCH 1/6] fix(js-sdk): keep envd HTTP/2 sessions within Node's memory budget Node charges queued request bodies against a 10 MB per-session budget and resets the stream whose response arrives while it is exhausted (NGHTTP2_ENHANCE_YOUR_CALM). Since undici 8 all envd REST requests of a process shared one session. envd REST dispatchers now carry one request per HTTP/2 session across the connection pool, and a dispatcher whose session Node reports as out of memory is retired and replaced. Co-Authored-By: mish@e2b.dev --- .changeset/calm-envd-http2-sessions.md | 5 + packages/js-sdk/src/envd/http2.ts | 11 +- packages/js-sdk/src/undici.ts | 103 ++++++++++++---- packages/js-sdk/tests/envd/http2.test.ts | 29 +++++ packages/js-sdk/tests/undici.test.ts | 145 +++++++++++++++++++++++ 5 files changed, 265 insertions(+), 28 deletions(-) create mode 100644 .changeset/calm-envd-http2-sessions.md diff --git a/.changeset/calm-envd-http2-sessions.md b/.changeset/calm-envd-http2-sessions.md new file mode 100644 index 0000000000..5d375a2133 --- /dev/null +++ b/.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 charges queued request bodies against a 10 MB per-session budget and resets whichever stream's response arrives while it is exhausted; since undici 8 every envd REST request of a process shared one session. envd REST dispatchers now carry one request per HTTP/2 session (spread over the connection pool, as with undici 7), and a dispatcher whose session Node reports as out of memory is retired and replaced instead of failing every later request. diff --git a/packages/js-sdk/src/envd/http2.ts b/packages/js-sdk/src/envd/http2.ts index 312da43dab..60e138f890 100644 --- a/packages/js-sdk/src/envd/http2.ts +++ b/packages/js-sdk/src/envd/http2.ts @@ -10,6 +10,7 @@ import { type EnvdFetchOptions = { connectionLimit?: number + maxConcurrentStreams?: number proxy?: string httpVersion?: HttpVersion loadUndici?: () => Promise @@ -21,6 +22,12 @@ const envdFetchers = new Map() const envdRpcFetchers = new Map() const DEFAULT_ENVD_CONNECTION_LIMIT = 10 const DEFAULT_ENVD_RPC_CONNECTION_LIMIT = 200 +// Node queues a request body in full against the HTTP/2 session's memory budget +// (`maxSessionMemory`, 10 MB) and resets whichever stream's response headers +// arrive while it is exhausted. envd REST carries file uploads, so streams of +// one session must not share that budget: one request per session, spread over +// `DEFAULT_ENVD_CONNECTION_LIMIT` sessions — the same shape undici 7 gave. +export const ENVD_MAX_CONCURRENT_STREAMS = 1 export function createEnvdFetchForRuntime( currentRuntime = runtime, @@ -29,6 +36,7 @@ export function createEnvdFetchForRuntime( return createRuntimeFetch(currentRuntime, () => buildDispatchedFetch({ connections: options.connectionLimit ?? DEFAULT_ENVD_CONNECTION_LIMIT, + maxConcurrentStreams: options.maxConcurrentStreams, proxy: options.proxy, httpVersion: options.httpVersion, loadUndici: options.loadUndici, @@ -54,9 +62,8 @@ 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, { + maxConcurrentStreams: ENVD_MAX_CONCURRENT_STREAMS, proxy, httpVersion, }) diff --git a/packages/js-sdk/src/undici.ts b/packages/js-sdk/src/undici.ts index 9dacebd5ea..594d9170f9 100644 --- a/packages/js-sdk/src/undici.ts +++ b/packages/js-sdk/src/undici.ts @@ -9,17 +9,47 @@ type UndiciRequestInit = RequestInit & { duplex?: 'half' } +type UndiciDispatcher = { close(): Promise } + +type UndiciAgentOptions = { + allowH2: boolean + connections?: number + maxConcurrentStreams?: 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( @@ -83,11 +113,15 @@ 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. + * `httpVersion` is `'1.1'`, `connections` origin connections, at most + * `maxConcurrentStreams` streams per HTTP/2 session, optional proxy 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 + maxConcurrentStreams?: number proxy?: string httpVersion?: HttpVersion loadUndici?: () => Promise @@ -99,30 +133,47 @@ 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, + } + if (options.maxConcurrentStreams !== undefined) { + agentOptions.maxConcurrentStreams = options.maxConcurrentStreams + } + const createDispatcher = (): UndiciDispatcher => + 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 - 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 sessions finish first. + void used.close().catch(() => {}) + } + throw err + } }) as typeof fetch } diff --git a/packages/js-sdk/tests/envd/http2.test.ts b/packages/js-sdk/tests/envd/http2.test.ts index 022673196b..ad2f81cf95 100644 --- a/packages/js-sdk/tests/envd/http2.test.ts +++ b/packages/js-sdk/tests/envd/http2.test.ts @@ -297,3 +297,32 @@ test('uses global fetch outside Node', async () => { undefined ) }) + +test('envd REST dispatchers allow one stream per HTTP/2 session, RPC ones do not', async () => { + const builds: Array<{ connections: number; maxConcurrentStreams?: number }> = + [] + vi.doMock('../../src/undici', async (importOriginal) => { + const original = await importOriginal() + return { + ...original, + buildDispatchedFetch: vi.fn((options) => { + builds.push(options) + return Promise.resolve((() => + Promise.resolve(new Response('ok'))) as typeof fetch) + }), + } + }) + + const { createEnvdFetch, createEnvdRpcFetch, ENVD_MAX_CONCURRENT_STREAMS } = + await import('../../src/envd/http2') + + await createEnvdFetch()('https://example.com/files') + await createEnvdRpcFetch()('https://example.com/rpc') + + expect(ENVD_MAX_CONCURRENT_STREAMS).toBe(1) + expect(builds).toEqual([ + expect.objectContaining({ connections: 10, maxConcurrentStreams: 1 }), + expect.objectContaining({ connections: 200 }), + ]) + expect(builds[1].maxConcurrentStreams).toBeUndefined() +}) diff --git a/packages/js-sdk/tests/undici.test.ts b/packages/js-sdk/tests/undici.test.ts index e63d607c7d..a3ed05cc4a 100644 --- a/packages/js-sdk/tests/undici.test.ts +++ b/packages/js-sdk/tests/undici.test.ts @@ -3,6 +3,7 @@ import { expect, test, vi } from 'vitest' import { buildDispatchedFetch, createRuntimeFetch, + isPoisonedHttp2SessionError, loadUndici, type UndiciModule, } from '../src/undici' @@ -209,3 +210,147 @@ test.skipIf(runtime !== 'node')( } } ) + +function poisonedSessionError(rst = 'NGHTTP2_ENHANCE_YOUR_CALM'): TypeError { + return new TypeError('fetch failed', { + cause: Object.assign(new Error(`Stream closed with error code ${rst}`), { + code: 'ERR_HTTP2_STREAM_ERROR', + }), + }) +} + +test.each([ + ['NGHTTP2_ENHANCE_YOUR_CALM', true], + ['NGHTTP2_INTERNAL_ERROR', true], + ['NGHTTP2_CANCEL', false], + ['NGHTTP2_REFUSED_STREAM', false], +])('isPoisonedHttp2SessionError on a %s stream reset is %s', (rst, poisoned) => + expect(isPoisonedHttp2SessionError(poisonedSessionError(rst))).toBe(poisoned) +) + +test('isPoisonedHttp2SessionError ignores other transport failures', () => { + expect( + isPoisonedHttp2SessionError( + new TypeError('fetch failed', { + cause: Object.assign(new Error('read ECONNRESET'), { + code: 'ECONNRESET', + }), + }) + ) + ).toBe(false) + expect(isPoisonedHttp2SessionError(new TypeError('fetch failed'))).toBe(false) + expect(isPoisonedHttp2SessionError('fetch failed')).toBe(false) +}) + +function fakeUndiciWithDispatchers(responses: Array<() => Promise>) { + const agents: Array<{ options: unknown; close: ReturnType }> = + [] + const dispatched: unknown[] = [] + + class Agent { + close = vi.fn(() => Promise.resolve()) + constructor(options: unknown) { + agents.push({ options, close: this.close }) + } + } + + const undici = { + Agent, + ProxyAgent: class extends Agent {}, + fetch: (_input: unknown, init?: { dispatcher?: unknown }) => { + dispatched.push(init?.dispatcher) + return responses.shift()!() + }, + } as unknown as UndiciModule + + return { agents, dispatched, undici } +} + +test('passes maxConcurrentStreams to the dispatcher', async () => { + const { agents, undici } = fakeUndiciWithDispatchers([ + () => Promise.resolve(new Response('ok')), + () => Promise.resolve(new Response('ok')), + ]) + + const direct = await buildDispatchedFetch({ + connections: 10, + maxConcurrentStreams: 1, + loadUndici: () => Promise.resolve(undici), + }) + await direct('https://example.com/') + const proxied = await buildDispatchedFetch({ + connections: 10, + maxConcurrentStreams: 1, + proxy: 'http://127.0.0.1:8080', + loadUndici: () => Promise.resolve(undici), + }) + await proxied('https://example.com/') + + expect(agents.map((a) => a.options)).toEqual([ + { allowH2: true, connections: 10, maxConcurrentStreams: 1 }, + { + uri: 'http://127.0.0.1:8080', + allowH2: true, + connections: 10, + maxConcurrentStreams: 1, + proxyTunnel: true, + }, + ]) +}) + +test('retires the dispatcher once after its HTTP/2 session is poisoned', async () => { + const { agents, dispatched, undici } = fakeUndiciWithDispatchers([ + () => Promise.reject(poisonedSessionError()), + () => Promise.reject(poisonedSessionError()), + () => Promise.resolve(new Response('ok')), + ]) + + const fetcher = await buildDispatchedFetch({ + connections: 1, + loadUndici: () => Promise.resolve(undici), + }) + + // Two requests in flight on the same poisoned session: both fail, the + // dispatcher is replaced once, and the replacement is not closed in turn. + const results = await Promise.allSettled([ + fetcher('https://example.com/a'), + fetcher('https://example.com/b'), + ]) + expect(results.map((r) => r.status)).toEqual(['rejected', 'rejected']) + expect(agents).toHaveLength(2) + expect(agents[0].close).toHaveBeenCalledOnce() + expect(agents[1].close).not.toHaveBeenCalled() + + const res = await fetcher('https://example.com/c') + expect(await res.text()).toBe('ok') + expect(dispatched).toEqual( + [agents[0], agents[0], agents[1]].map(() => expect.any(Object)) + ) + expect(dispatched[0]).toBe(dispatched[1]) + expect(dispatched[2]).not.toBe(dispatched[0]) + expect(dispatched[2]).toBeInstanceOf(undici.Agent) +}) + +test('keeps the dispatcher on other fetch failures', async () => { + const { agents, undici } = fakeUndiciWithDispatchers([ + () => + Promise.reject( + new TypeError('fetch failed', { + cause: Object.assign(new Error('read ECONNRESET'), { + code: 'ECONNRESET', + }), + }) + ), + () => Promise.resolve(new Response('ok')), + ]) + + const fetcher = await buildDispatchedFetch({ + connections: 1, + loadUndici: () => Promise.resolve(undici), + }) + + await expect(fetcher('https://example.com/')).rejects.toThrow('fetch failed') + await fetcher('https://example.com/') + expect(agents).toHaveLength(1) + expect(agents[0].close).not.toHaveBeenCalled() +}) From d077e5a221ae2670f45ab8d53271d24f050b9598 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 16:17:38 +0000 Subject: [PATCH 2/6] test(js-sdk): assert envd REST stream cap without module mocking (browser runner) Co-Authored-By: mish@e2b.dev --- packages/js-sdk/tests/envd/http2.test.ts | 54 +++++++++++++++--------- 1 file changed, 33 insertions(+), 21 deletions(-) diff --git a/packages/js-sdk/tests/envd/http2.test.ts b/packages/js-sdk/tests/envd/http2.test.ts index ad2f81cf95..80ff091bde 100644 --- a/packages/js-sdk/tests/envd/http2.test.ts +++ b/packages/js-sdk/tests/envd/http2.test.ts @@ -298,31 +298,43 @@ test('uses global fetch outside Node', async () => { ) }) -test('envd REST dispatchers allow one stream per HTTP/2 session, RPC ones do not', async () => { - const builds: Array<{ connections: number; maxConcurrentStreams?: number }> = - [] - vi.doMock('../../src/undici', async (importOriginal) => { - const original = await importOriginal() - return { - ...original, - buildDispatchedFetch: vi.fn((options) => { - builds.push(options) - return Promise.resolve((() => - Promise.resolve(new Response('ok'))) as typeof fetch) - }), +test('caps envd REST dispatchers at one stream per HTTP/2 session', async () => { + const agents: Array<{ + allowH2?: boolean + connections?: number + maxConcurrentStreams?: number + }> = [] + + class Agent { + constructor(options: { + allowH2?: boolean + connections?: number + maxConcurrentStreams?: number + }) { + agents.push(options) } - }) + } - const { createEnvdFetch, createEnvdRpcFetch, ENVD_MAX_CONCURRENT_STREAMS } = - await import('../../src/envd/http2') + const undiciFetch = vi.fn(() => Promise.resolve(new Response('ok'))) - await createEnvdFetch()('https://example.com/files') - await createEnvdRpcFetch()('https://example.com/rpc') + const { createEnvdFetchForRuntime, ENVD_MAX_CONCURRENT_STREAMS } = + await import('../../src/envd/http2') expect(ENVD_MAX_CONCURRENT_STREAMS).toBe(1) - expect(builds).toEqual([ - expect.objectContaining({ connections: 10, maxConcurrentStreams: 1 }), - expect.objectContaining({ connections: 200 }), + + const rest = createEnvdFetchForRuntime('node', { + maxConcurrentStreams: ENVD_MAX_CONCURRENT_STREAMS, + loadUndici: () => Promise.resolve({ Agent, fetch: undiciFetch }), + }) + await rest('https://example.com/files') + const rpc = createEnvdFetchForRuntime('node', { + connectionLimit: 200, + loadUndici: () => Promise.resolve({ Agent, fetch: undiciFetch }), + }) + await rpc('https://example.com/rpc') + + expect(agents).toEqual([ + { allowH2: true, connections: 10, maxConcurrentStreams: 1 }, + { allowH2: true, connections: 200 }, ]) - expect(builds[1].maxConcurrentStreams).toBeUndefined() }) From 8a13a2ddbea92e8fa760a34230f95ba234ec9f03 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 16:18:25 +0000 Subject: [PATCH 3/6] refactor(js-sdk): isPoisonedHttp2SessionError -> isPoisonedHTTP2SessionError Co-Authored-By: mish@e2b.dev --- packages/js-sdk/src/undici.ts | 4 ++-- packages/js-sdk/tests/undici.test.ts | 14 +++++++------- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/packages/js-sdk/src/undici.ts b/packages/js-sdk/src/undici.ts index 594d9170f9..cc5704afa5 100644 --- a/packages/js-sdk/src/undici.ts +++ b/packages/js-sdk/src/undici.ts @@ -39,7 +39,7 @@ const POISONED_HTTP2_SESSION_RST = * 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 { +export function isPoisonedHTTP2SessionError(err: unknown): boolean { if (!(err instanceof Error) || !(err.cause instanceof Error)) { return false } @@ -167,7 +167,7 @@ export async function buildDispatchedFetch(options: { } 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)) { + if (used === dispatcher && isPoisonedHTTP2SessionError(err)) { dispatcher = createDispatcher() // Graceful: requests still running on the old sessions finish first. void used.close().catch(() => {}) diff --git a/packages/js-sdk/tests/undici.test.ts b/packages/js-sdk/tests/undici.test.ts index a3ed05cc4a..27e0887e47 100644 --- a/packages/js-sdk/tests/undici.test.ts +++ b/packages/js-sdk/tests/undici.test.ts @@ -3,7 +3,7 @@ import { expect, test, vi } from 'vitest' import { buildDispatchedFetch, createRuntimeFetch, - isPoisonedHttp2SessionError, + isPoisonedHTTP2SessionError, loadUndici, type UndiciModule, } from '../src/undici' @@ -224,13 +224,13 @@ test.each([ ['NGHTTP2_INTERNAL_ERROR', true], ['NGHTTP2_CANCEL', false], ['NGHTTP2_REFUSED_STREAM', false], -])('isPoisonedHttp2SessionError on a %s stream reset is %s', (rst, poisoned) => - expect(isPoisonedHttp2SessionError(poisonedSessionError(rst))).toBe(poisoned) +])('isPoisonedHTTP2SessionError on a %s stream reset is %s', (rst, poisoned) => + expect(isPoisonedHTTP2SessionError(poisonedSessionError(rst))).toBe(poisoned) ) -test('isPoisonedHttp2SessionError ignores other transport failures', () => { +test('isPoisonedHTTP2SessionError ignores other transport failures', () => { expect( - isPoisonedHttp2SessionError( + isPoisonedHTTP2SessionError( new TypeError('fetch failed', { cause: Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET', @@ -238,8 +238,8 @@ test('isPoisonedHttp2SessionError ignores other transport failures', () => { }) ) ).toBe(false) - expect(isPoisonedHttp2SessionError(new TypeError('fetch failed'))).toBe(false) - expect(isPoisonedHttp2SessionError('fetch failed')).toBe(false) + expect(isPoisonedHTTP2SessionError(new TypeError('fetch failed'))).toBe(false) + expect(isPoisonedHTTP2SessionError('fetch failed')).toBe(false) }) function fakeUndiciWithDispatchers(responses: Array<() => Promise>) { From 5d407ea4aaef46ac165b8382ee09d578a364e091 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 17:18:02 +0000 Subject: [PATCH 4/6] fix(js-sdk): budget in-flight envd upload bytes instead of capping HTTP/2 streams A 1-stream-per-session cap starves the origin: undici 8's Pool holds every request behind its first (protocol-probe) client until it drains, so one long-lived REST request blocked all others to that sandbox. Keep sessions multiplexed and instead admit buffered upload bodies so that together they stay within 6 MB of Node's 10 MB session memory budget. Co-Authored-By: mish@e2b.dev --- .changeset/calm-envd-http2-sessions.md | 2 +- packages/js-sdk/src/envd/http2.ts | 10 -- packages/js-sdk/src/envd/uploadBudget.ts | 126 ++++++++++++++++ .../js-sdk/src/sandbox/filesystem/index.ts | 103 +++++++------ packages/js-sdk/src/undici.ts | 14 +- packages/js-sdk/tests/envd/http2.test.ts | 41 ----- .../js-sdk/tests/envd/uploadBudget.test.ts | 140 ++++++++++++++++++ packages/js-sdk/tests/undici.test.ts | 32 ---- 8 files changed, 329 insertions(+), 139 deletions(-) create mode 100644 packages/js-sdk/src/envd/uploadBudget.ts create mode 100644 packages/js-sdk/tests/envd/uploadBudget.test.ts diff --git a/.changeset/calm-envd-http2-sessions.md b/.changeset/calm-envd-http2-sessions.md index 5d375a2133..24255dc19d 100644 --- a/.changeset/calm-envd-http2-sessions.md +++ b/.changeset/calm-envd-http2-sessions.md @@ -2,4 +2,4 @@ 'e2b': patch --- -Fix `TypeError: fetch failed` with `NGHTTP2_ENHANCE_YOUR_CALM` on concurrent `files.write()` calls. Node's HTTP/2 client charges queued request bodies against a 10 MB per-session budget and resets whichever stream's response arrives while it is exhausted; since undici 8 every envd REST request of a process shared one session. envd REST dispatchers now carry one request per HTTP/2 session (spread over the connection pool, as with undici 7), and a dispatcher whose session Node reports as out of memory is retired and replaced instead of failing every later request. +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. envd uploads are now admitted so that the bodies in flight together stay within 6 MB (a single larger upload still goes alone), and a dispatcher whose session Node reports as out of memory is retired and replaced instead of failing every later request. diff --git a/packages/js-sdk/src/envd/http2.ts b/packages/js-sdk/src/envd/http2.ts index 60e138f890..87c299b589 100644 --- a/packages/js-sdk/src/envd/http2.ts +++ b/packages/js-sdk/src/envd/http2.ts @@ -10,7 +10,6 @@ import { type EnvdFetchOptions = { connectionLimit?: number - maxConcurrentStreams?: number proxy?: string httpVersion?: HttpVersion loadUndici?: () => Promise @@ -22,13 +21,6 @@ const envdFetchers = new Map() const envdRpcFetchers = new Map() const DEFAULT_ENVD_CONNECTION_LIMIT = 10 const DEFAULT_ENVD_RPC_CONNECTION_LIMIT = 200 -// Node queues a request body in full against the HTTP/2 session's memory budget -// (`maxSessionMemory`, 10 MB) and resets whichever stream's response headers -// arrive while it is exhausted. envd REST carries file uploads, so streams of -// one session must not share that budget: one request per session, spread over -// `DEFAULT_ENVD_CONNECTION_LIMIT` sessions — the same shape undici 7 gave. -export const ENVD_MAX_CONCURRENT_STREAMS = 1 - export function createEnvdFetchForRuntime( currentRuntime = runtime, options: EnvdFetchOptions = {} @@ -36,7 +28,6 @@ export function createEnvdFetchForRuntime( return createRuntimeFetch(currentRuntime, () => buildDispatchedFetch({ connections: options.connectionLimit ?? DEFAULT_ENVD_CONNECTION_LIMIT, - maxConcurrentStreams: options.maxConcurrentStreams, proxy: options.proxy, httpVersion: options.httpVersion, loadUndici: options.loadUndici, @@ -63,7 +54,6 @@ export function createEnvdFetch({ } const envdFetch = createEnvdFetchForRuntime(runtime, { - maxConcurrentStreams: ENVD_MAX_CONCURRENT_STREAMS, proxy, httpVersion, }) diff --git a/packages/js-sdk/src/envd/uploadBudget.ts b/packages/js-sdk/src/envd/uploadBudget.ts new file mode 100644 index 0000000000..fe8aa11c4e --- /dev/null +++ b/packages/js-sdk/src/envd/uploadBudget.ts @@ -0,0 +1,126 @@ +// 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. They are admitted here so the bodies in flight together +// stay well under it — headers and buffered response data count against it +// too. A single larger upload still goes through on its own, and 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( + size: number, + signal: AbortSignal | null | undefined, + fn: () => Promise + ): Promise +} + +/** + * 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((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) diff --git a/packages/js-sdk/src/sandbox/filesystem/index.ts b/packages/js-sdk/src/sandbox/filesystem/index.ts index b0279c2749..c530ce3d7d 100644 --- a/packages/js-sdk/src/sandbox/filesystem/index.ts +++ b/packages/js-sdk/src/sandbox/filesystem/index.ts @@ -38,6 +38,7 @@ import { FilesystemEvent, WatchHandle } from './watchHandle' import type { Timestamp } from '@bufbuild/protobuf/wkt' import { compareVersions } from 'compare-versions' +import { bodyByteLength, envdUploadBudget } from '../../envd/uploadBudget' import { ENVD_DEFAULT_USER, ENVD_FILE_METADATA, @@ -722,32 +723,38 @@ export class Filesystem { writeOpts?.signal ) - const res = await 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) - }) + const res = await envdUploadBudget.run( + streamed ? 0 : bodyByteLength(body), + 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) + }) + ) const err = await handleFilesystemEnvdApiError(res) if (err) { @@ -782,25 +789,31 @@ export class Filesystem { ) } - const res = await 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) - }) + const signal = this.connectionConfig.getSignal( + writeOpts?.requestTimeoutMs, + writeOpts?.signal + ) + const res = await envdUploadBudget.run( + bodyByteLength(formData), + signal, + () => + this.envdApi.api + .POST('/files', { + params: { + query: { + path, + username: user, + }, + }, + bodySerializer: () => formData, + headers: extraHeaders, + signal, + body: {}, + }) + .catch(async (err) => { + throw await handleEnvdApiFetchError(err, this.checkHealth) + }) + ) const err = await handleFilesystemEnvdApiError(res) if (err) { diff --git a/packages/js-sdk/src/undici.ts b/packages/js-sdk/src/undici.ts index cc5704afa5..bdafdb21b4 100644 --- a/packages/js-sdk/src/undici.ts +++ b/packages/js-sdk/src/undici.ts @@ -14,7 +14,6 @@ type UndiciDispatcher = { close(): Promise } type UndiciAgentOptions = { allowH2: boolean connections?: number - maxConcurrentStreams?: number } export type UndiciModule = { @@ -113,15 +112,13 @@ export function createRuntimeFetch( /** * Build a fetch bound to a bounded undici dispatcher (HTTP/2 enabled unless - * `httpVersion` is `'1.1'`, `connections` origin connections, at most - * `maxConcurrentStreams` streams per HTTP/2 session, optional proxy 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. + * `httpVersion` is `'1.1'`, `connections` origin connections, optional proxy + * 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 - maxConcurrentStreams?: number proxy?: string httpVersion?: HttpVersion loadUndici?: () => Promise @@ -137,9 +134,6 @@ export async function buildDispatchedFetch(options: { allowH2: (options.httpVersion ?? DEFAULT_HTTP_VERSION) === '2', connections: options.connections, } - if (options.maxConcurrentStreams !== undefined) { - agentOptions.maxConcurrentStreams = options.maxConcurrentStreams - } const createDispatcher = (): UndiciDispatcher => options.proxy ? new ProxyAgent({ diff --git a/packages/js-sdk/tests/envd/http2.test.ts b/packages/js-sdk/tests/envd/http2.test.ts index 80ff091bde..022673196b 100644 --- a/packages/js-sdk/tests/envd/http2.test.ts +++ b/packages/js-sdk/tests/envd/http2.test.ts @@ -297,44 +297,3 @@ test('uses global fetch outside Node', async () => { undefined ) }) - -test('caps envd REST dispatchers at one stream per HTTP/2 session', async () => { - const agents: Array<{ - allowH2?: boolean - connections?: number - maxConcurrentStreams?: number - }> = [] - - class Agent { - constructor(options: { - allowH2?: boolean - connections?: number - maxConcurrentStreams?: number - }) { - agents.push(options) - } - } - - const undiciFetch = vi.fn(() => Promise.resolve(new Response('ok'))) - - const { createEnvdFetchForRuntime, ENVD_MAX_CONCURRENT_STREAMS } = - await import('../../src/envd/http2') - - expect(ENVD_MAX_CONCURRENT_STREAMS).toBe(1) - - const rest = createEnvdFetchForRuntime('node', { - maxConcurrentStreams: ENVD_MAX_CONCURRENT_STREAMS, - loadUndici: () => Promise.resolve({ Agent, fetch: undiciFetch }), - }) - await rest('https://example.com/files') - const rpc = createEnvdFetchForRuntime('node', { - connectionLimit: 200, - loadUndici: () => Promise.resolve({ Agent, fetch: undiciFetch }), - }) - await rpc('https://example.com/rpc') - - expect(agents).toEqual([ - { allowH2: true, connections: 10, maxConcurrentStreams: 1 }, - { allowH2: true, connections: 200 }, - ]) -}) diff --git a/packages/js-sdk/tests/envd/uploadBudget.test.ts b/packages/js-sdk/tests/envd/uploadBudget.test.ts new file mode 100644 index 0000000000..3c6431a39f --- /dev/null +++ b/packages/js-sdk/tests/envd/uploadBudget.test.ts @@ -0,0 +1,140 @@ +import { expect, test } from 'vitest' + +import { + bodyByteLength, + createBodyBudget, + ENVD_MAX_INFLIGHT_BODY_BYTES, +} from '../../src/envd/uploadBudget' + +function deferred() { + let resolve!: (value: T) => void + let reject!: (err: unknown) => void + const promise = new Promise((res, rej) => { + resolve = res + reject = rej + }) + return { promise, resolve, reject } +} + +const tick = async () => { + await Promise.resolve() + await Promise.resolve() +} + +test("the envd upload budget stays under Node's 10 MB HTTP/2 session memory", () => { + expect(ENVD_MAX_INFLIGHT_BODY_BYTES).toBe(6 * 1024 * 1024) +}) + +test('bodyByteLength counts buffered bodies and ignores streamed ones', () => { + expect(bodyByteLength(undefined)).toBe(0) + expect(bodyByteLength(null)).toBe(0) + expect(bodyByteLength('héllo')).toBe(6) + expect(bodyByteLength(new Uint8Array(5))).toBe(5) + expect(bodyByteLength(new ArrayBuffer(7))).toBe(7) + expect(bodyByteLength(new Blob(['abc']))).toBe(3) + expect(bodyByteLength(new URLSearchParams({ a: 'b' }))).toBe(3) + + const form = new FormData() + form.append('file', new Blob(['0123456789']), 'a.txt') + form.append('file', new Blob(['01234']), 'b.txt') + form.append('note', 'xyz') + expect(bodyByteLength(form)).toBe(4 + 10 + 4 + 5 + 4 + 3) + + expect(bodyByteLength(new ReadableStream({ start: (c) => c.close() }))).toBe( + 0 + ) +}) + +test('admits bodies in order while they fit together', async () => { + const budget = createBodyBudget(10) + const started: string[] = [] + const first = deferred() + const second = deferred() + + const a = budget.run(6, undefined, () => { + started.push('a') + return first.promise + }) + const b = budget.run(6, undefined, () => { + started.push('b') + return second.promise + }) + // Sizeless requests are never held back by waiting uploads. + const c = budget.run(0, undefined, async () => { + started.push('c') + return 'c' + }) + await tick() + expect(started.toSorted()).toEqual(['a', 'c']) + + first.resolve('a') + await a + await tick() + expect(started.toSorted()).toEqual(['a', 'b', 'c']) + + second.resolve('b') + await expect(Promise.all([b, c])).resolves.toEqual(['b', 'c']) +}) + +test('a body larger than the budget goes through alone and is released on failure', async () => { + const budget = createBodyBudget(4) + const started: string[] = [] + const first = deferred() + + const big = budget.run(100, undefined, () => { + started.push('big') + return first.promise + }) + const small = budget.run(1, undefined, async () => { + started.push('small') + return 'small' + }) + await tick() + expect(started).toEqual(['big']) + + first.reject(new TypeError('fetch failed')) + await expect(big).rejects.toThrow('fetch failed') + await expect(small).resolves.toBe('small') +}) + +test('a request aborted while waiting rejects and unblocks the queue', async () => { + const budget = createBodyBudget(10) + const started: string[] = [] + const holder = deferred() + const controller = new AbortController() + + const a = budget.run(8, undefined, () => { + started.push('a') + return holder.promise + }) + const aborted = budget.run(8, controller.signal, async () => { + started.push('aborted') + return 'aborted' + }) + const queued = budget.run(2, undefined, async () => { + started.push('queued') + return 'queued' + }) + await tick() + expect(started).toEqual(['a']) + + controller.abort(new Error('gone')) + await expect(aborted).rejects.toThrow('gone') + // The small upload behind the aborted one now fits next to the holder. + await expect(queued).resolves.toBe('queued') + expect(started).toEqual(['a', 'queued']) + + holder.resolve('a') + await expect(a).resolves.toBe('a') +}) + +test('an already aborted request rejects without taking budget', async () => { + const budget = createBodyBudget(10) + const controller = new AbortController() + controller.abort(new Error('gone')) + + await expect( + budget.run(1, controller.signal, async () => 'never') + ).rejects.toThrow('gone') + await expect(budget.run(10, undefined, async () => 'ok')).resolves.toBe('ok') +}) diff --git a/packages/js-sdk/tests/undici.test.ts b/packages/js-sdk/tests/undici.test.ts index 27e0887e47..9264b9dc2d 100644 --- a/packages/js-sdk/tests/undici.test.ts +++ b/packages/js-sdk/tests/undici.test.ts @@ -266,38 +266,6 @@ function fakeUndiciWithDispatchers(responses: Array<() => Promise>) { return { agents, dispatched, undici } } -test('passes maxConcurrentStreams to the dispatcher', async () => { - const { agents, undici } = fakeUndiciWithDispatchers([ - () => Promise.resolve(new Response('ok')), - () => Promise.resolve(new Response('ok')), - ]) - - const direct = await buildDispatchedFetch({ - connections: 10, - maxConcurrentStreams: 1, - loadUndici: () => Promise.resolve(undici), - }) - await direct('https://example.com/') - const proxied = await buildDispatchedFetch({ - connections: 10, - maxConcurrentStreams: 1, - proxy: 'http://127.0.0.1:8080', - loadUndici: () => Promise.resolve(undici), - }) - await proxied('https://example.com/') - - expect(agents.map((a) => a.options)).toEqual([ - { allowH2: true, connections: 10, maxConcurrentStreams: 1 }, - { - uri: 'http://127.0.0.1:8080', - allowH2: true, - connections: 10, - maxConcurrentStreams: 1, - proxyTunnel: true, - }, - ]) -}) - test('retires the dispatcher once after its HTTP/2 session is poisoned', async () => { const { agents, dispatched, undici } = fakeUndiciWithDispatchers([ () => Promise.reject(poisonedSessionError()), From 861a151ff36f446fba8457894b5cbc7c4bc4d14c Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 17:32:17 +0000 Subject: [PATCH 5/6] fix(js-sdk): send over-budget envd uploads on a session of their own A buffered upload larger than the 6 MiB budget was admitted alone but still shared the pooled HTTP/2 session, so any read whose response landed while it filled Node's session memory was reset (16 MB write + 10 reads: 30/30 reads failed, also on base). Such uploads now go through a single-use dispatcher that is closed after the response, like undici 7 gave them in 2.27.1. Co-Authored-By: mish@e2b.dev --- .changeset/calm-envd-http2-sessions.md | 2 +- packages/js-sdk/src/envd/http2.ts | 39 ++++++ packages/js-sdk/src/envd/uploadBudget.ts | 9 +- .../js-sdk/src/sandbox/filesystem/index.ts | 126 +++++++++++------- packages/js-sdk/src/undici.ts | 104 +++++++++++---- packages/js-sdk/tests/envd/http2.test.ts | 35 +++++ packages/js-sdk/tests/undici.test.ts | 38 ++++++ 7 files changed, 274 insertions(+), 79 deletions(-) diff --git a/.changeset/calm-envd-http2-sessions.md b/.changeset/calm-envd-http2-sessions.md index 24255dc19d..42c3f53ccd 100644 --- a/.changeset/calm-envd-http2-sessions.md +++ b/.changeset/calm-envd-http2-sessions.md @@ -2,4 +2,4 @@ '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. envd uploads are now admitted so that the bodies in flight together stay within 6 MB (a single larger upload still goes alone), and a dispatcher whose session Node reports as out of memory is retired and replaced instead of failing every later request. +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. diff --git a/packages/js-sdk/src/envd/http2.ts b/packages/js-sdk/src/envd/http2.ts index 87c299b589..a8659c78ab 100644 --- a/packages/js-sdk/src/envd/http2.ts +++ b/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,6 +20,7 @@ type EnvdFetchOptions = { // keep sharing a single dispatcher while each distinct proxy URL gets its own. const envdFetchers = new Map() const envdRpcFetchers = new Map() +const envdSoloFetchers = new Map() const DEFAULT_ENVD_CONNECTION_LIMIT = 10 const DEFAULT_ENVD_RPC_CONNECTION_LIMIT = 200 export function createEnvdFetchForRuntime( @@ -35,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 = {} +): typeof fetch { + return createRuntimeFetch(currentRuntime, () => + buildSingleUseDispatchedFetch({ + proxy: options.proxy, + httpVersion: options.httpVersion, + loadUndici: options.loadUndici, + }) + ) +} + function fetcherKey( proxy: string | undefined, httpVersion: HttpVersion @@ -83,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', diff --git a/packages/js-sdk/src/envd/uploadBudget.ts b/packages/js-sdk/src/envd/uploadBudget.ts index fe8aa11c4e..73196bae42 100644 --- a/packages/js-sdk/src/envd/uploadBudget.ts +++ b/packages/js-sdk/src/envd/uploadBudget.ts @@ -3,10 +3,11 @@ // 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. They are admitted here so the bodies in flight together -// stay well under it — headers and buffered response data count against it -// too. A single larger upload still goes through on its own, and a stream -// body is written chunk by chunk behind `drain`, so it is never queued whole. +// 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 => diff --git a/packages/js-sdk/src/sandbox/filesystem/index.ts b/packages/js-sdk/src/sandbox/filesystem/index.ts index c530ce3d7d..abf65f7768 100644 --- a/packages/js-sdk/src/sandbox/filesystem/index.ts +++ b/packages/js-sdk/src/sandbox/filesystem/index.ts @@ -38,7 +38,12 @@ import { FilesystemEvent, WatchHandle } from './watchHandle' import type { Timestamp } from '@bufbuild/protobuf/wkt' import { compareVersions } from 'compare-versions' -import { bodyByteLength, envdUploadBudget } from '../../envd/uploadBudget' +import { createEnvdSoloFetch } from '../../envd/http2' +import { + bodyByteLength, + ENVD_MAX_INFLIGHT_BODY_BYTES, + envdUploadBudget, +} from '../../envd/uploadBudget' import { ENVD_DEFAULT_USER, ENVD_FILE_METADATA, @@ -397,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 + } { + 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`. * @@ -723,37 +749,38 @@ export class Filesystem { writeOpts?.signal ) - const res = await envdUploadBudget.run( - streamed ? 0 : bodyByteLength(body), - signal, - () => - 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) @@ -793,26 +820,25 @@ export class Filesystem { writeOpts?.requestTimeoutMs, writeOpts?.signal ) - const res = await envdUploadBudget.run( - bodyByteLength(formData), - signal, - () => - this.envdApi.api - .POST('/files', { - params: { - query: { - path, - username: user, - }, + 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, - 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) diff --git a/packages/js-sdk/src/undici.ts b/packages/js-sdk/src/undici.ts index bdafdb21b4..d45def1ce5 100644 --- a/packages/js-sdk/src/undici.ts +++ b/packages/js-sdk/src/undici.ts @@ -110,23 +110,27 @@ export function createRuntimeFetch( }) as typeof fetch } -/** - * Build a fetch bound to a bounded undici dispatcher (HTTP/2 enabled unless - * `httpVersion` is `'1.1'`, `connections` origin connections, optional proxy - * 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 +type DispatcherOptions = { proxy?: string httpVersion?: HttpVersion loadUndici?: () => Promise -}): Promise { +} + +type UndiciFetch = ( + input: RequestInfo | URL, + init?: UndiciRequestInit +) => Promise + +async function loadDispatcherFactory( + options: DispatcherOptions & { connections: number } +): Promise< + | { createDispatcher: () => UndiciDispatcher; undiciFetch: UndiciFetch } + | undefined +> { const undici = await (options.loadUndici ?? loadUndici)() if (!undici) { - return lateBoundGlobalFetch() + return undefined } const { Agent, ProxyAgent, fetch: undiciFetch } = undici @@ -134,19 +138,37 @@ export async function buildDispatchedFetch(options: { allowH2: (options.httpVersion ?? DEFAULT_HTTP_VERSION) === '2', connections: options.connections, } - const createDispatcher = (): UndiciDispatcher => - 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 + return { + createDispatcher: () => + options.proxy + ? new ProxyAgent({ + uri: options.proxy, + ...agentOptions, + proxyTunnel: true, + }) + : new Agent(agentOptions), + undiciFetch: undiciFetch as unknown as UndiciFetch, + } +} + +/** + * Build a fetch bound to a bounded undici dispatcher (HTTP/2 enabled unless + * `httpVersion` is `'1.1'`, `connections` origin connections, optional proxy + * 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: DispatcherOptions & { connections: number } +): Promise { + const factory = await loadDispatcherFactory(options) + + if (!factory) { + return lateBoundGlobalFetch() + } + + const { createDispatcher, undiciFetch } = factory let dispatcher = createDispatcher() return (async (input, init) => { @@ -154,7 +176,7 @@ export async function buildDispatchedFetch(options: { const used = dispatcher try { - return await fetchWithDispatcher(request.input, { + return await undiciFetch(request.input, { ...request.init, dispatcher: used, }) @@ -171,6 +193,40 @@ export async function buildDispatchedFetch(options: { }) as typeof fetch } +/** + * Build a fetch that gives every request its own single-connection dispatcher + * — on HTTP/2 a session of its own, with Node's whole per-session memory + * budget — and closes it once the response has been read. For requests whose + * buffered body would crowd out everything else on a shared session. Falls + * back to the global fetch when undici cannot be loaded. + */ +export async function buildSingleUseDispatchedFetch( + options: DispatcherOptions +): Promise { + const factory = await loadDispatcherFactory({ ...options, connections: 1 }) + + if (!factory) { + return lateBoundGlobalFetch() + } + + const { createDispatcher, undiciFetch } = factory + + return (async (input, init) => { + const request = toUndiciRequestInput(input, init) + const dispatcher = createDispatcher() + + try { + return await undiciFetch(request.input, { + ...request.init, + dispatcher, + }) + } finally { + // Graceful: resolves after the response body has been consumed. + void dispatcher.close().catch(() => {}) + } + }) as typeof fetch +} + function toUndiciRequestInput( input: RequestInfo | URL, init?: RequestInit diff --git a/packages/js-sdk/tests/envd/http2.test.ts b/packages/js-sdk/tests/envd/http2.test.ts index 022673196b..532021f1c7 100644 --- a/packages/js-sdk/tests/envd/http2.test.ts +++ b/packages/js-sdk/tests/envd/http2.test.ts @@ -297,3 +297,38 @@ test('uses global fetch outside Node', async () => { undefined ) }) + +test('solo envd fetch uses a fresh single-connection dispatcher per request', async () => { + const agents: Array<{ allowH2?: boolean; connections?: number }> = [] + const closed: Array = [] + class Agent { + constructor(options: { allowH2?: boolean; connections?: number }) { + agents.push(options) + } + close() { + closed.push(this) + return Promise.resolve() + } + } + const dispatchers: unknown[] = [] + const undiciFetch = vi.fn((_input, init) => { + dispatchers.push(init?.dispatcher) + return Promise.resolve(new Response('ok')) + }) + + const { createEnvdSoloFetchForRuntime } = await import('../../src/envd/http2') + + const fetcher = createEnvdSoloFetchForRuntime('node', { + loadUndici: () => Promise.resolve({ Agent, fetch: undiciFetch }), + }) + await fetcher('https://example.com/files') + await fetcher('https://example.com/files') + + expect(agents).toEqual([ + { allowH2: true, connections: 1 }, + { allowH2: true, connections: 1 }, + ]) + expect(dispatchers[0]).toBeInstanceOf(Agent) + expect(dispatchers[0]).not.toBe(dispatchers[1]) + expect(closed).toEqual(dispatchers) +}) diff --git a/packages/js-sdk/tests/undici.test.ts b/packages/js-sdk/tests/undici.test.ts index 9264b9dc2d..1be770073d 100644 --- a/packages/js-sdk/tests/undici.test.ts +++ b/packages/js-sdk/tests/undici.test.ts @@ -2,6 +2,7 @@ import { expect, test, vi } from 'vitest' import { buildDispatchedFetch, + buildSingleUseDispatchedFetch, createRuntimeFetch, isPoisonedHTTP2SessionError, loadUndici, @@ -322,3 +323,40 @@ test('keeps the dispatcher on other fetch failures', async () => { expect(agents).toHaveLength(1) expect(agents[0].close).not.toHaveBeenCalled() }) + +test('buildSingleUseDispatchedFetch gives each request its own dispatcher and closes it', async () => { + const { agents, dispatched, undici } = fakeUndiciWithDispatchers([ + () => Promise.resolve(new Response('a')), + () => Promise.reject(new TypeError('fetch failed')), + ]) + + const fetcher = await buildSingleUseDispatchedFetch({ + loadUndici: () => Promise.resolve(undici), + }) + + const res = await fetcher('https://example.com/a') + expect(await res.text()).toBe('a') + await expect(fetcher('https://example.com/b')).rejects.toThrow('fetch failed') + + expect(agents).toHaveLength(2) + expect(agents.map((a) => a.options)).toEqual([ + { allowH2: true, connections: 1 }, + { allowH2: true, connections: 1 }, + ]) + expect(dispatched[0]).not.toBe(dispatched[1]) + expect(agents[0].close).toHaveBeenCalledOnce() + expect(agents[1].close).toHaveBeenCalledOnce() +}) + +test('buildSingleUseDispatchedFetch falls back to the global fetch without undici', async () => { + const globalFetch = vi.fn(async () => new Response('global')) + vi.stubGlobal('fetch', globalFetch) + + const fetcher = await buildSingleUseDispatchedFetch({ + loadUndici: () => Promise.resolve(undefined), + }) + const res = await fetcher('https://example.com/a') + + expect(await res.text()).toBe('global') + expect(globalFetch).toHaveBeenCalledOnce() +}) From b2a5dbb47ddae4298f10a403d1940ec7f9adaa90 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 17:39:25 +0000 Subject: [PATCH 6/6] test(js-sdk): mock createEnvdSoloFetch in envd/http2 module mocks (browser runner) Co-Authored-By: mish@e2b.dev --- packages/js-sdk/tests/sandbox/httpVersion.test.ts | 1 + packages/js-sdk/tests/sandbox/rpcHeaders.test.ts | 1 + 2 files changed, 2 insertions(+) diff --git a/packages/js-sdk/tests/sandbox/httpVersion.test.ts b/packages/js-sdk/tests/sandbox/httpVersion.test.ts index d01d48e0b7..32aab21958 100644 --- a/packages/js-sdk/tests/sandbox/httpVersion.test.ts +++ b/packages/js-sdk/tests/sandbox/httpVersion.test.ts @@ -12,6 +12,7 @@ vi.mock('@connectrpc/connect-web', () => ({ vi.mock('../../src/envd/http2', () => ({ createEnvdFetch: mocks.createEnvdFetch, createEnvdRpcFetch: mocks.createEnvdRpcFetch, + createEnvdSoloFetch: vi.fn(() => vi.fn()), })) afterEach(() => { diff --git a/packages/js-sdk/tests/sandbox/rpcHeaders.test.ts b/packages/js-sdk/tests/sandbox/rpcHeaders.test.ts index a559713d38..93113f5692 100644 --- a/packages/js-sdk/tests/sandbox/rpcHeaders.test.ts +++ b/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), + createEnvdSoloFetch: vi.fn(() => vi.fn()), })) test('does not pass custom connection headers to envd RPC requests', async () => {