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..2ca30f6 100644 --- a/packages/node/src/utils.ts +++ b/packages/node/src/utils.ts @@ -1,5 +1,54 @@ +import type { Readable } from 'node:stream' import type Stream from 'node:stream' import type { NodeHttpResponse } from './types' +import { IncomingMessage } from 'node:http' + +/** + * A cancel-safe alternative to `Readable.toWeb`. + * + * 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> { + 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) + } + }, + }) +} /** * Check both the response itself and its underlying stream (http2) are still writable.