diff --git a/packages/runtime/src/native/bridge.ts b/packages/runtime/src/native/bridge.ts index 559a8a6a8..57874567b 100644 --- a/packages/runtime/src/native/bridge.ts +++ b/packages/runtime/src/native/bridge.ts @@ -8,6 +8,7 @@ import { type Server, type ServerResponse, } from 'node:http'; +import type { Socket } from 'node:net'; import { Readable } from 'node:stream'; import { pipeline } from 'node:stream/promises'; import type { ReadableStream as NodeReadableStream } from 'node:stream/web'; @@ -80,6 +81,8 @@ interface Invocation { accepted: boolean; handlerFinished: boolean; transportFinished: boolean; + /** The request as the server read it, and the server's end of its connection. */ + served?: { readonly request: IncomingMessage; readonly socket: Socket } | undefined; } function boundedInteger(value: number, name: string): number { @@ -209,6 +212,7 @@ class NativeBridge implements NodeBridge { }); this.#clients.add(client); let upload: Readable | undefined; + let connection: Socket | undefined; let answered = false; const timer = setTimeout(() => { client.destroy( @@ -225,6 +229,22 @@ class NativeBridge implements NodeBridge { client.on('error', (error) => { if (!answered) reject(error); }); + client.once('socket', (socket) => { + connection = socket; + }); + /** + * A server may also stop reading a request before its end, as Next.js does with a Server + * Action over its size limit, even where all of it was sent. The rest stays on the connection, + * which reads nothing more: a call the Agent sent on it from its pool waited for its headers + * deadline, and the server's end, its idle connections being kept, was never closed. So both + * ends are dropped, once the answer has been read whole or the call has ended without it. + */ + const dropUnread = (): void => { + const { served } = invocation; + if (served === undefined || served.request.complete) return; + connection?.destroy(); + served.socket.destroy(); + }; client.once('close', () => { if (!answered) reject(new Error('native transport closed before response headers')); clearTimeout(timer); @@ -233,6 +253,7 @@ class NativeBridge implements NodeBridge { // A server may finish its answer before consuming an upload. A pooled socket must never // retain that producer, nor be handed to the next call with unfinished request framing. if (upload !== undefined && !upload.readableEnded) upload.destroy(); + dropUnread(); invocation.transportFinished = true; this.#release(id, invocation); }); @@ -251,6 +272,8 @@ class NativeBridge implements NodeBridge { upload.destroy(); client.destroy(); } + // Before the Agent pools the connection, which it does on the next tick. + dropUnread(); }); clearTimeout(timer); try { @@ -318,6 +341,7 @@ class NativeBridge implements NodeBridge { // would otherwise wait for this one until the drain's deadline. Ended, not destroyed, so an // answer still in the socket's buffer reaches the caller before the connection closes. const { socket } = req; + invocation.served = { request: req, socket }; let answered = false; let received = false; const endIfIdle = (): void => { diff --git a/packages/runtime/test/native/bridge.test.ts b/packages/runtime/test/native/bridge.test.ts index 412ec4b09..c79724d6c 100644 --- a/packages/runtime/test/native/bridge.test.ts +++ b/packages/runtime/test/native/bridge.test.ts @@ -3,6 +3,8 @@ import assert from 'node:assert/strict'; import { AsyncLocalStorage } from 'node:async_hooks'; import { once } from 'node:events'; import { type IncomingMessage, request as httpRequest, type ServerResponse } from 'node:http'; +import { Transform, Writable } from 'node:stream'; +import { pipeline } from 'node:stream/promises'; import { setTimeout as delay } from 'node:timers/promises'; import { gzipSync } from 'node:zlib'; @@ -84,6 +86,37 @@ async function answerUpload(req: IncomingMessage, res: ServerResponse): Promise< res.end(await textOf(req)); } +/** + * A handler that reads its body as Next.js reads a Server Action's: through a size limit. Past the + * limit, the pipeline's failure destroys the request, and Node.js reads its connection no further. + */ +async function overLimit(req: IncomingMessage, res: ServerResponse): Promise { + let size = 0; + const limit = new Transform({ + transform(chunk: Buffer, _encoding, callback) { + size += chunk.byteLength; + callback(size > 64 * 1024 ? new Error('Body exceeded the limit') : null, chunk); + }, + }); + const sink = new Writable({ + write(_chunk, _encoding, callback) { + callback(); + }, + }); + try { + await pipeline(req, limit, sink); + res.end('read'); + } catch { + res.statusCode = 413; + res.end('too large'); + } +} + +/** A body the loopback's buffers hold whole, so the bridge sends all of it. */ +function megabyteUpload(): Request { + return upload(new Blob([new Uint8Array(1024 * 1024)]).stream()); +} + describe('native Node bridge', () => { it('restores rewritten URL and public headers on real Node request/response objects', async () => { const bridge = fixture(); @@ -207,6 +240,52 @@ describe('native Node bridge', () => { assert.ok(performance.now() - closingAt < 2000); }); + it('does not pool a connection whose server left its request unread', async () => { + const bridge = fixture({ headersTimeoutMs: 3000 }); + const sockets: unknown[] = []; + const first = await bridge.invokeNodeHandler( + input( + async (req, res) => { + sockets.push(req.socket); + await overLimit(req, res); + }, + { request: megabyteUpload() }, + ), + ); + assert.equal(first.status, 413); + assert.equal(await first.text(), 'too large'); + + const second = await bridge.invokeNodeHandler( + input(async (req, res) => { + sockets.push(req.socket); + res.end('second'); + }), + ); + + assert.equal(await second.text(), 'second'); + assert.equal(sockets.length, 2); + assert.notEqual(sockets[1], sockets[0]); + }); + + it('closes the server end of a connection whose request was left unread', async () => { + const bridge = fixture(); + const served = Promise.withResolvers(); + const response = await bridge.invokeNodeHandler( + input( + async (req, res) => { + served.resolve(req.socket); + await overLimit(req, res); + }, + { request: megabyteUpload() }, + ), + ); + const socket = await served.promise; + const closed = once(socket, 'close'); + + assert.equal(await response.text(), 'too large'); + await closed; + }); + it('preserves an explicit Host instead of replacing it with the loopback host', async () => { const bridge = fixture(); const response = await bridge.invokeNodeHandler(