diff --git a/.changeset/calm-envd-http2-sessions.md b/.changeset/calm-envd-http2-sessions.md new file mode 100644 index 0000000000..42c3f53ccd --- /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 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 312da43dab..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,9 +20,9 @@ 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( 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 = {} +): 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', diff --git a/packages/js-sdk/src/envd/uploadBudget.ts b/packages/js-sdk/src/envd/uploadBudget.ts new file mode 100644 index 0000000000..73196bae42 --- /dev/null +++ b/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( + 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..abf65f7768 100644 --- a/packages/js-sdk/src/sandbox/filesystem/index.ts +++ b/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 + } { + 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) { diff --git a/packages/js-sdk/src/undici.ts b/packages/js-sdk/src/undici.ts index 9dacebd5ea..d45def1ce5 100644 --- a/packages/js-sdk/src/undici.ts +++ b/packages/js-sdk/src/undici.ts @@ -9,17 +9,46 @@ type UndiciRequestInit = RequestInit & { duplex?: 'half' } +type UndiciDispatcher = { close(): Promise } + +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( @@ -81,48 +110,120 @@ 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). 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 - 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, + } + + 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) => { + const request = toUndiciRequestInput(input, init) + const used = dispatcher + + try { + return await undiciFetch(request.input, { + ...request.init, + dispatcher: used, }) - const fetchWithDispatcher = undiciFetch as unknown as ( - input: RequestInfo | URL, - init?: UndiciRequestInit - ) => Promise + } 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 +} + +/** + * 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() + } - return ((input, init) => { + const { createDispatcher, undiciFetch } = factory + + return (async (input, init) => { const request = toUndiciRequestInput(input, init) + const dispatcher = createDispatcher() - return fetchWithDispatcher(request.input, { - ...request.init, - dispatcher, - }) + 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 } 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/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/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 () => { diff --git a/packages/js-sdk/tests/undici.test.ts b/packages/js-sdk/tests/undici.test.ts index e63d607c7d..1be770073d 100644 --- a/packages/js-sdk/tests/undici.test.ts +++ b/packages/js-sdk/tests/undici.test.ts @@ -2,7 +2,9 @@ import { expect, test, vi } from 'vitest' import { buildDispatchedFetch, + buildSingleUseDispatchedFetch, createRuntimeFetch, + isPoisonedHTTP2SessionError, loadUndici, type UndiciModule, } from '../src/undici' @@ -209,3 +211,152 @@ 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('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() +}) + +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() +})