From eca913f3da7ce85ced83ea604728bbb879b3a39c Mon Sep 17 00:00:00 2001 From: Dinh Le Date: Wed, 26 Aug 2026 10:14:41 +0700 Subject: [PATCH 1/2] fix(node): prevent uncatchable crash when canceling a request body stream A consumer canceling a streamed request body mid-upload could crash the whole process: Node's Readable.toWeb adapter enqueues from 'data' events, so a chunk arriving after cancel hits an already-closed controller and throws an uncaught ERR_INVALID_STATE (nodejs/node#54205). Route all Readable-to-web conversions in body and event-stream through a new cancel-safe toWebReadableStream util, which pipes the adapter through a TransformStream so cancellation never reaches its controller mid-chunk, and copies each chunk out of Node's pooled Buffer memory. Pinned by flood-and-abort upload tests over http1 and http2, including tests documenting that the bare adapter still crashes. --- packages/node/src/body.test.ts | 6 + packages/node/src/body.ts | 3 +- packages/node/src/event-stream.test.ts | 4 + packages/node/src/event-stream.ts | 3 +- packages/node/src/utils.test.ts | 248 ++++++++++++++++++++++++- packages/node/src/utils.ts | 19 ++ 6 files changed, 277 insertions(+), 6 deletions(-) diff --git a/packages/node/src/body.test.ts b/packages/node/src/body.test.ts index bf578ce..27231d3 100644 --- a/packages/node/src/body.test.ts +++ b/packages/node/src/body.test.ts @@ -10,8 +10,10 @@ import { isAsyncIteratorObject } from '@standardserver/shared' import request from 'supertest' import { toNodeHttpBody, toStandardBody } from './body' import * as EventStreamModule from './event-stream' +import * as UtilsModule from './utils' const toEventStreamSpy = vi.spyOn(EventStreamModule, 'toEventStream') +const toWebReadableStreamSpy = vi.spyOn(UtilsModule, 'toWebReadableStream') const generateContentDispositionSpy = vi.spyOn(StandardServerModule, 'generateContentDisposition') const getFilenameFromContentDispositionSpy = vi.spyOn(StandardServerModule, 'getFilenameFromContentDisposition') @@ -309,6 +311,8 @@ describe('toStandardBody', () => { expect(result).toBeInstanceOf(ReadableStream) expect(streamedBytes).toEqual(new Uint8Array(body)) + expect(toWebReadableStreamSpy).toHaveBeenCalledTimes(1) + expect(result).toBe(toWebReadableStreamSpy.mock.results[0]!.value) }) }) @@ -342,6 +346,8 @@ describe('toStandardBody', () => { .send('hello') expect(standardBody).toBeInstanceOf(ReadableStream) + expect(toWebReadableStreamSpy).toHaveBeenCalledTimes(1) + expect(standardBody).toBe(toWebReadableStreamSpy.mock.results[0]!.value) const reader = (standardBody as ReadableStream).pipeThrough(new TextDecoderStream()).getReader() expect(await reader.read()).toEqual({ done: false, value: 'hello' }) }) diff --git a/packages/node/src/body.ts b/packages/node/src/body.ts index cdbbc32..532a899 100644 --- a/packages/node/src/body.ts +++ b/packages/node/src/body.ts @@ -6,6 +6,7 @@ import { Readable } from 'node:stream' import { generateContentDisposition, getFilenameFromContentDisposition, resolveStandardBodyHint } from '@standardserver/core' import { isAsyncIteratorObject, parseEmptyableJSON, stringifyJSON } from '@standardserver/shared' import { toAsyncIteratorObject, toEventStream } from './event-stream' +import { toWebReadableStream } from './utils' export interface ToStandardBodyOptions { /** @@ -71,7 +72,7 @@ export async function toStandardBody( return _streamToFile(req, fileName ?? 'blob', contentType ?? '') } - return Readable.toWeb(req as Readable) + return toWebReadableStream(req) } export interface ToNodeHttpBodyOptions { diff --git a/packages/node/src/event-stream.test.ts b/packages/node/src/event-stream.test.ts index 57a9bd1..653fdb5 100644 --- a/packages/node/src/event-stream.test.ts +++ b/packages/node/src/event-stream.test.ts @@ -2,9 +2,11 @@ import { Readable } from 'node:stream' import * as FetchAdapter from '@standardserver/fetch' import { isAsyncIteratorObject } from '@standardserver/shared' import { toAsyncIteratorObject, toEventStream } from './event-stream' +import * as UtilsModule from './utils' const toAsyncIteratorObjectFetch = vi.spyOn(FetchAdapter, 'toAsyncIteratorObject') const toEventStreamFetch = vi.spyOn(FetchAdapter, 'toEventStream') +const toWebReadableStreamSpy = vi.spyOn(UtilsModule, 'toWebReadableStream') beforeEach(() => { vi.clearAllMocks() @@ -28,7 +30,9 @@ it('toAsyncIteratorObject', async () => { expect(await generator.next()).toEqual({ done: false, value: 3 }) expect(await generator.next()).toEqual({ done: true, value: undefined }) + expect(toWebReadableStreamSpy).toBeCalledTimes(1) expect(toAsyncIteratorObjectFetch).toBeCalledTimes(1) + expect(toAsyncIteratorObjectFetch).toHaveBeenCalledWith(toWebReadableStreamSpy.mock.results[0]!.value) }) it('toEventStream', async () => { diff --git a/packages/node/src/event-stream.ts b/packages/node/src/event-stream.ts index 9de9096..79b44d6 100644 --- a/packages/node/src/event-stream.ts +++ b/packages/node/src/event-stream.ts @@ -7,11 +7,12 @@ import { toAsyncIteratorObject as toAsyncIteratorObjectFetch, toEventStream as toEventStreamFetch, } from '@standardserver/fetch' +import { toWebReadableStream } from './utils' export function toAsyncIteratorObject( stream: Readable, ): AsyncIteratorClass { - return toAsyncIteratorObjectFetch(Readable.toWeb(stream)) + return toAsyncIteratorObjectFetch(toWebReadableStream(stream)) } export interface ToEventStreamOptions extends ToEventStreamOptionsFetch {} diff --git a/packages/node/src/utils.test.ts b/packages/node/src/utils.test.ts index 34ae286..7beb033 100644 --- a/packages/node/src/utils.test.ts +++ b/packages/node/src/utils.test.ts @@ -1,7 +1,13 @@ -import http from 'node:http' -import http2 from 'node:http2' -import { connect } from 'node:net' -import { canWriteToNodeResponse, getNodeResponseError } from './utils' +import type { AddressInfo } from 'node:net' +import { Buffer } from 'node:buffer' +import { appendFile, mkdtemp, rm } from 'node:fs/promises' +import http, { createServer } from 'node:http' +import http2, { createServer as createHttp2Server, connect as http2Connect } from 'node:http2' +import net, { connect } from 'node:net' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { Readable } from 'node:stream' +import { canWriteToNodeResponse, getNodeResponseError, toWebReadableStream } from './utils' describe('canWriteToNodeResponse', () => { it('on http1 response aborted by client', async ({ onTestFinished }) => { @@ -312,3 +318,237 @@ describe('getNodeResponseError', () => { await handled }) }) + +describe('toWebReadableStream', () => { + /** + * Below the 256 KiB flood chunk, so the consumer cancels on its first read + * while the request is still streaming — the condition that crashes a bare + * `Readable.toWeb`. + */ + const LIMIT = 64 * 1024 + + /** + * Runs `fn` while recording `uncaughtException`/`unhandledRejection` into the + * returned array (also passed to `fn`), so an uncatchable adapter crash lands + * there instead of failing the worker. Restores the runner's listeners after. + */ + async function recordUncaught(fn: (crashes: Error[]) => Promise): Promise { + const crashes: Error[] = [] + const record = (err: unknown): void => { + crashes.push(err as Error) + } + + const prevExceptions = process.listeners('uncaughtException') + const prevRejections = process.listeners('unhandledRejection') + process.removeAllListeners('uncaughtException') + process.removeAllListeners('unhandledRejection') + process.on('uncaughtException', record) + process.on('unhandledRejection', record) + + try { + await fn(crashes) + // Let any queued 'data' event reach a (possibly closed) controller. + await new Promise(resolve => setTimeout(resolve, 50)) + return crashes + } + finally { + process.off('uncaughtException', record) + process.off('unhandledRejection', record) + prevExceptions.forEach(listener => process.on('uncaughtException', listener)) + prevRejections.forEach(listener => process.on('unhandledRejection', listener)) + } + } + + /** + * Consumes a body like an upload handler: spools each chunk to disk (an async + * gap that lets more data queue up) and rejects past `LIMIT`, cancelling the + * stream while bytes are still arriving. + */ + async function spoolUntilRejected(body: ReadableStream, tmpDir: string): Promise { + const sink = path.join(await mkdtemp(path.join(tmpDir, 'chunk-')), 'sink') + let total = 0 + try { + for await (const chunk of body) { + total += chunk.byteLength + if (total > LIMIT) { + throw new Error('PAYLOAD_TOO_LARGE') + } + await appendFile(sink, chunk) + } + } + catch { + // Mirrors the plugin turning the oversized body into a 413. + } + } + + /** Streams an oversized HTTP/1 upload, then destroys the socket mid-flight. */ + function floodAndAbortHttp1(port: number): Promise { + return new Promise((resolve) => { + const socket = net.connect(port, '127.0.0.1', () => { + socket.write('POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 104857600\r\n\r\n') + const blob = Buffer.alloc(256 * 1024, 0x61) + const interval = setInterval(() => { + if (socket.destroyed || !socket.writable) { + clearInterval(interval) + return + } + socket.write(blob) + }, 0) + socket.on('close', () => clearInterval(interval)) + }) + socket.on('error', () => {}) + socket.on('close', () => resolve()) + setTimeout(() => socket.destroy(), 15) + }) + } + + /** Streams an oversized HTTP/2 upload, then destroys the request mid-flight. */ + function floodAndAbortHttp2(port: number): Promise { + return new Promise((resolve) => { + const client = http2Connect(`http://127.0.0.1:${port}`) + client.on('error', () => {}) + const request = client.request({ ':method': 'POST', ':path': '/', 'content-length': '104857600' }) + const blob = Buffer.alloc(256 * 1024, 0x61) + const interval = setInterval(() => { + if (request.destroyed || request.closed) { + clearInterval(interval) + return + } + request.write(blob) + }, 0) + request.on('error', () => clearInterval(interval)) + + let settled = false + const settle = (): void => { + if (settled) { + return + } + settled = true + clearInterval(interval) + client.destroy() + resolve() + } + request.on('close', settle) + setTimeout(() => { + request.destroy() + settle() + }, 15) + }) + } + + /** + * Boots a server that spools each request body via `wrap`, fires `iterations` + * aborted uploads at it, and reports how many completed plus any crashes. + * `stopOnCrash` ends the flood at the first crash. + */ + async function runUploadServer( + kind: 'http1' | 'http2', + wrap: (req: Readable) => ReadableStream, + iterations: number, + stopOnCrash = false, + ): Promise<{ handled: number, crashes: Error[] }> { + const tmpDir = await mkdtemp(path.join(tmpdir(), `toweb-${kind}-`)) + let handled = 0 + + const listener = async (req: any, res: any): Promise => { + await spoolUntilRejected(wrap(req), tmpDir) + handled++ + try { + if (!res.headersSent) { + res.statusCode = 413 + res.end('too large') + } + } + catch { + // The request stream may already be torn down; the response is best effort. + } + } + + const server = kind === 'http1' ? createServer(listener) : createHttp2Server(listener) + const flood = kind === 'http1' ? floodAndAbortHttp1 : floodAndAbortHttp2 + + const crashes = await recordUncaught(async (crashes) => { + await new Promise(resolve => server.listen(0, resolve)) + const { port } = server.address() as AddressInfo + for (let i = 0; i < iterations; i++) { + if (stopOnCrash && crashes.length) { + break + } + await flood(port) + await new Promise(resolve => setTimeout(resolve, 5)) + } + }) + + await new Promise(resolve => server.close(() => resolve())) + await rm(tmpDir, { recursive: true, force: true }) + return { handled, crashes } + } + + const bare = (req: Readable): ReadableStream => Readable.toWeb(req) as ReadableStream + const wrapped = (req: Readable): ReadableStream => toWebReadableStream(req) + + it('converts a raw buffer stream and preserves its bytes as copies', async () => { + const chunks = [Buffer.from('hello '), Buffer.from('world'), Buffer.alloc(1024, 7)] + const source = Readable.from(chunks) + + const received: Uint8Array[] = [] + for await (const chunk of toWebReadableStream(source)) { + expect(chunk).toBeInstanceOf(Uint8Array) + expect(Buffer.isBuffer(chunk)).toBe(false) // copied out of the Node Buffer + received.push(chunk) + } + + expect(Buffer.concat(received).equals(Buffer.concat(chunks))).toBe(true) + }) + + it('does not throw when a raw buffer stream is cancelled mid-read', async () => { + const source = Readable.from((async function* () { + for (let i = 0; i < 10_000; i++) { + yield Buffer.alloc(64 * 1024, 0x61) + } + })()) + + let read = 0 + const crashes = await recordUncaught(async () => { + for await (const chunk of toWebReadableStream(source)) { + read += chunk.byteLength + if (read >= 256 * 1024) { + break // cancels the web stream while the source still has data + } + } + await new Promise(resolve => setImmediate(resolve)) + }) + + expect(crashes).toEqual([]) + expect(read).toBeGreaterThanOrEqual(256 * 1024) + expect(source.destroyed).toBe(true) // cancellation still tears the source down + }) + + it('lets a bare Readable.toWeb crash an aborted HTTP/1 upload (documents the bug)', async () => { + const { crashes } = await runUploadServer('http1', bare, 25, true) + + expect(crashes.length).toBeGreaterThan(0) + expect(crashes[0]).toMatchObject({ code: 'ERR_INVALID_STATE' }) + }, 30_000) + + it('keeps an aborted HTTP/1 upload from crashing the process', async () => { + const { handled, crashes } = await runUploadServer('http1', wrapped, 25) + + expect(crashes).toEqual([]) + expect(handled).toBe(25) + }, 30_000) + + it('lets a bare Readable.toWeb crash an aborted HTTP/2 upload (documents the bug)', async () => { + const { crashes } = await runUploadServer('http2', bare, 25, true) + + expect(crashes.length).toBeGreaterThan(0) + expect(crashes[0]).toMatchObject({ code: 'ERR_INVALID_STATE' }) + }, 30_000) + + it('keeps an aborted HTTP/2 upload from crashing the process', async () => { + const { handled, crashes } = await runUploadServer('http2', wrapped, 25) + + expect(crashes).toEqual([]) + expect(handled).toBe(25) + }, 30_000) +}) diff --git a/packages/node/src/utils.ts b/packages/node/src/utils.ts index b55a717..3ceaa70 100644 --- a/packages/node/src/utils.ts +++ b/packages/node/src/utils.ts @@ -1,5 +1,24 @@ import type Stream from 'node:stream' import type { NodeHttpResponse } from './types' +import { Readable } from 'node:stream' + +/** + * A cancel-safe alternative to a bare `Readable.toWeb`. + * + * Bare `Readable.toWeb` enqueues from `'data'` events, so a chunk arriving + * after the consumer cancels hits a closed controller and crashes the process + * with an uncaught `ERR_INVALID_STATE` (nodejs/node#54205). The intermediate + * `TransformStream` shields the adapter's controller from that cancellation + * (which still destroys the source), and the per-chunk copy detaches chunks + * from Node's pooled `Buffer` memory. + */ +export function toWebReadableStream(stream: Readable): ReadableStream> { + return Readable.toWeb(stream).pipeThrough(new TransformStream({ + transform(chunk, controller) { + controller.enqueue(new Uint8Array(chunk)) + }, + })) +} /** * Check both the response itself and its underlying stream (http2) are still writable. From 65fd468662ff46980221d714596c9cae56fba933 Mon Sep 17 00:00:00 2001 From: Dinh Le Date: Wed, 26 Aug 2026 10:47:03 +0700 Subject: [PATCH 2/2] fix(node): make toWebReadableStream cancel-safe on every Node release MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Readable.toWeb + TransformStream approach only shielded the adapter's controller on some Node releases: on Node 24 a chunk arriving after cancel still hit the closed controller and crashed the process with an uncaught ERR_INVALID_STATE, even for plain Readable sources. Hand-roll the adapter instead of wrapping Readable.toWeb: chunks are only enqueued inside pull(), which never runs after cancel, so the crash is impossible by construction regardless of Node's adapter internals. Cancel destroys the source, except http1 server requests — they share their socket with the response, so they are abandoned (stalled by backpressure, reclaimed on connection teardown) to keep in-flight responses deliverable. All package and e2e suites pass on Node 22, 24, and 26. --- packages/node/src/utils.ts | 54 +++++++++++++++++++++++++++++--------- 1 file changed, 42 insertions(+), 12 deletions(-) diff --git a/packages/node/src/utils.ts b/packages/node/src/utils.ts index 3ceaa70..2ca30f6 100644 --- a/packages/node/src/utils.ts +++ b/packages/node/src/utils.ts @@ -1,23 +1,53 @@ +import type { Readable } from 'node:stream' import type Stream from 'node:stream' import type { NodeHttpResponse } from './types' -import { Readable } from 'node:stream' +import { IncomingMessage } from 'node:http' /** - * A cancel-safe alternative to a bare `Readable.toWeb`. + * A cancel-safe alternative to `Readable.toWeb`. * - * Bare `Readable.toWeb` enqueues from `'data'` events, so a chunk arriving - * after the consumer cancels hits a closed controller and crashes the process - * with an uncaught `ERR_INVALID_STATE` (nodejs/node#54205). The intermediate - * `TransformStream` shields the adapter's controller from that cancellation - * (which still destroys the source), and the per-chunk copy detaches chunks - * from Node's pooled `Buffer` memory. + * Node's adapter enqueues from `'data'` events, so a chunk arriving after the + * consumer cancels hits a closed controller and crashes the process with an + * uncaught `ERR_INVALID_STATE` (nodejs/node#54205) — on some Node releases even + * through an intermediate `TransformStream`. Pulling through the stream's async + * iterator makes that impossible: chunks are only enqueued inside `pull`, never + * after cancel. The per-chunk copy detaches chunks from Node's pooled `Buffer` + * memory. + * + * Cancel destroys the source, except http1 server requests: they share their + * socket with the response, so destroying them would kill an in-flight + * response. They are abandoned instead — stalled by backpressure and reclaimed + * on connection teardown. */ export function toWebReadableStream(stream: Readable): ReadableStream> { - return Readable.toWeb(stream).pipeThrough(new TransformStream({ - transform(chunk, controller) { - controller.enqueue(new Uint8Array(chunk)) + const iterator = stream[Symbol.asyncIterator]() + let canceled = false + + return new ReadableStream({ + async pull(controller) { + const { done, value } = await iterator.next() + + if (canceled) { + return // a chunk in flight while cancel happened; drop it + } + + if (done) { + controller.close() + } + else { + controller.enqueue(new Uint8Array(value)) + } + }, + cancel(reason) { + canceled = true + + const isHttp1ServerRequest = stream instanceof IncomingMessage && stream.method !== null + + if (!isHttp1ServerRequest) { + stream.destroy(reason instanceof Error ? reason : undefined) + } }, - })) + }) } /**