From a46b14379fc59ba499e739f99f40aad292d80cf6 Mon Sep 17 00:00:00 2001 From: Dinh Le Date: Sat, 26 Sep 2026 09:28:30 +0700 Subject: [PATCH] fix(node): drain cancelled request bodies so connections don't stall Cancelling a `toWebReadableStream` request body (e.g. rejecting an oversized upload after the first chunk) left the rest of the upload unread. Over HTTP/1 the next request on the keep-alive connection was never served and failed with ECONNRESET at the keep-alive timeout. Over HTTP/2, destroying the `Http2ServerRequest` never touched its `Http2Stream`, which stayed paused at the flow-control window, so the stream never closed and `sendStandardResponse` never settled. Server requests (HTTP/1 and HTTP/2) are now read to the end and discarded on cancel, the way Node drains a body a handler never reads. Draining pulls the existing async iterator, so no listeners are removed. --- packages/node/src/utils.test.ts | 84 +++++++++++++++++++++++++++++++++ packages/node/src/utils.ts | 27 ++++++++--- 2 files changed, 104 insertions(+), 7 deletions(-) diff --git a/packages/node/src/utils.test.ts b/packages/node/src/utils.test.ts index 850023f..a836241 100644 --- a/packages/node/src/utils.test.ts +++ b/packages/node/src/utils.test.ts @@ -7,6 +7,7 @@ import net, { connect } from 'node:net' import { tmpdir } from 'node:os' import path from 'node:path' import { Readable } from 'node:stream' +import { text } from 'node:stream/consumers' import { canWriteToNodeResponse, getNodeResponseError, toWebReadableStream } from './utils' describe('canWriteToNodeResponse', () => { @@ -529,4 +530,87 @@ describe('toWebReadableStream', () => { expect(crashes).toEqual([]) expect(handled).toBe(25) }, 30_000) + + /** Beyond socket buffers and the HTTP/2 flow-control window, so an unread upload stalls. */ + const UPLOAD = Buffer.alloc(1024 * 1024, 0x61) + + /** Reads one chunk of a request body, then cancels it like a handler rejecting the upload. */ + async function readOneChunkThenCancel(req: Readable): Promise { + const reader = toWebReadableStream(req).getReader() + await reader.read() + await reader.cancel() + } + + it('keeps an HTTP/1 keep-alive connection usable after cancelling a request body', async ({ onTestFinished }) => { + const otherListener = vi.fn() + let readableListeners: unknown[] = [] + + const server = createServer(async (req, res) => { + if (req.method === 'POST') { + req.on('readable', otherListener) // e.g. added by a framework; draining must leave it alone + await readOneChunkThenCancel(req) + readableListeners = req.listeners('readable') + res.statusCode = 413 + res.end('too large') + } + else { + res.end('ok') + } + }) + onTestFinished(() => new Promise((r) => { + server.closeAllConnections() + server.close(r) + })) + + await new Promise(resolve => server.listen(0, resolve)) + const { port } = server.address() as AddressInfo + + const socket = net.connect(port, '127.0.0.1') + socket.write(`POST / HTTP/1.1\r\nHost: x\r\nContent-Length: ${UPLOAD.byteLength}\r\n\r\n`) + socket.write(UPLOAD) + // Queued behind the rejected body, so it's only answered once that body is off the wire + socket.write('GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n') + + const received = await text(socket) + + expect(received).toMatch(/^HTTP\/1\.1 413 [\s\S]*too large/) + expect(received).toMatch(/HTTP\/1\.1 200 [\s\S]*ok$/) + expect(readableListeners).toContain(otherListener) + }) + + it('lets an HTTP/2 stream close after cancelling a request body', async ({ onTestFinished }) => { + const server = createHttp2Server() + + const responseClosed = new Promise((resolve) => { + server.on('request', async (req, res) => { + await readOneChunkThenCancel(req) + res.once('close', () => resolve()) // what `sendStandardResponse` settles on + res.end('ok') + }) + }) + + await new Promise(resolve => server.listen(0, resolve)) + const { port } = server.address() as AddressInfo + + const client = http2Connect(`http://127.0.0.1:${port}`) + onTestFinished(() => new Promise((r) => { + client.destroy() + server.close(r) + })) + + const request = client.request({ ':method': 'POST', ':path': '/' }) + request.end(UPLOAD) + request.setEncoding('utf8') + + let received = '' + request.on('data', (chunk) => { + received += chunk + }) + + await new Promise(resolve => request.once('close', resolve)) + await responseClosed + + expect(received).toBe('ok') + expect(request.rstCode).toBe(http2.constants.NGHTTP2_NO_ERROR) + }) }) diff --git a/packages/node/src/utils.ts b/packages/node/src/utils.ts index e0f9b50..66def83 100644 --- a/packages/node/src/utils.ts +++ b/packages/node/src/utils.ts @@ -2,6 +2,7 @@ import type { Readable } from 'node:stream' import type Stream from 'node:stream' import type { NodeHttpResponse } from './types' import { IncomingMessage } from 'node:http' +import { Http2ServerRequest } from 'node:http2' /** * A cancel-safe alternative to `Readable.toWeb`. @@ -17,10 +18,11 @@ import { IncomingMessage } from 'node:http' * Fixed upstream in Node 26.10 (nodejs/node#62773); switch back to * `Readable.toWeb` once every supported Node release has the fix. * - * 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. + * Cancel destroys the source, except server requests, which are read to the + * end and discarded, the way Node drains a body the handler never reads: + * destroying an http1 request kills the socket its response shares, and an + * unread body stalls on backpressure, blocking the next keep-alive request + * (http1) or the response's `close` (http2). */ export function toWebReadableStream(stream: Readable): ReadableStream> { const iterator = stream[Symbol.asyncIterator]() @@ -31,7 +33,7 @@ export function toWebReadableStream(stream: Readable): ReadableStream {}) + } + else { stream.destroy(reason instanceof Error ? reason : undefined) } }, }) } +async function _drainIterator(iterator: AsyncIterator): Promise { + while (!(await iterator.next()).done) { + // discard + } +} + /** * Check both the response itself and its underlying stream (http2) are still writable. */