From 5b77c2ec8163524d8e3541d0555336ec7c57bc8c Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sat, 24 Jan 2026 10:19:42 +0100 Subject: [PATCH 1/4] http: add serve() API with Fetch Request/Response model Add a modern HTTP server API that uses the Fetch API's Request and Response objects, bypassing the legacy IncomingMessage/ServerResponse stream-based model. Features: - serve(options, handler) creates HTTP or HTTPS servers - Handler receives Request, returns Response (sync or async) - getRemoteMetadata(request) retrieves connection info - Automatic chunked transfer encoding when no Content-Length - Keep-alive connection support - Graceful shutdown via AbortSignal - Custom error handling via onError callback The implementation reuses the battle-tested HTTPParser but creates Fetch API objects directly from parser output. Requests are serialized per connection (no HTTP pipelining) - parser is paused after headers and resumed after response is written. Example usage: const { serve, getRemoteMetadata } = require('http'); const server = serve({}, async (request) => { const body = await request.json(); return Response.json({ received: body }); }); server.listen(3000); --- lib/_http_common.js | 1 + lib/http.js | 6 + lib/internal/http_serve.js | 566 +++++++++++++++++++++ test/parallel/test-http-serve-basic.js | 281 ++++++++++ test/parallel/test-http-serve-errors.js | 261 ++++++++++ test/parallel/test-http-serve-streaming.js | 249 +++++++++ 6 files changed, 1364 insertions(+) create mode 100644 lib/internal/http_serve.js create mode 100644 test/parallel/test-http-serve-basic.js create mode 100644 test/parallel/test-http-serve-errors.js create mode 100644 test/parallel/test-http-serve-streaming.js diff --git a/lib/_http_common.js b/lib/_http_common.js index d5e7bdedee39..02cac9c3572a 100644 --- a/lib/_http_common.js +++ b/lib/_http_common.js @@ -330,6 +330,7 @@ function calculateLenientFlags(httpValidation, insecureHTTPParserOption) { module.exports = { _checkInvalidHeaderChar: checkInvalidHeaderChar, _checkIsHttpToken: checkIsHttpToken, + allMethods, chunkExpression: /(?:^|\W)chunked(?:$|\W)/i, continueExpression: /(?:^|\W)100-continue(?:$|\W)/i, CRLF: '\r\n', // TODO: Deprecate this. diff --git a/lib/http.js b/lib/http.js index 934d0b14bfdc..ff744bf92248 100644 --- a/lib/http.js +++ b/lib/http.js @@ -42,6 +42,10 @@ const { Server, ServerResponse, } = require('_http_server'); +const { + serve, + getRemoteMetadata, +} = require('internal/http_serve'); const { parseProxyUrl, getGlobalAgent, @@ -201,6 +205,8 @@ module.exports = { parsers.max = max; }, setGlobalProxyFromEnv, + serve, + getRemoteMetadata, }; ObjectDefineProperty(module.exports, 'maxHeaderSize', { diff --git a/lib/internal/http_serve.js b/lib/internal/http_serve.js new file mode 100644 index 000000000000..103dfb3149b8 --- /dev/null +++ b/lib/internal/http_serve.js @@ -0,0 +1,566 @@ +'use strict'; + +const { + Error: PrimordialError, + Promise, + SafeWeakMap, + Symbol, + Uint8Array, +} = primordials; + +const net = require('net'); +const tls = require('tls'); +const { once } = require('events'); +const { ReadableStream } = require('internal/webstreams/readablestream'); +const { + Headers, + Request, + Response, +} = require('internal/deps/undici/undici'); + +const { + parsers, + freeParser, + HTTPParser, + isLenient, + prepareError, + allMethods, +} = require('_http_common'); +const { ConnectionsList } = internalBinding('http_parser'); + +const { + utcDate, +} = require('internal/http'); + +const { validateFunction, validateObject } = require('internal/validators'); +const { + codes: { + ERR_INVALID_ARG_TYPE, + }, +} = require('internal/errors'); + +const dc = require('diagnostics_channel'); +const onRequestStartChannel = dc.channel('http.server.request.start'); + +// Symbols for private properties on server object +const kHandler = Symbol('kHandler'); +const kOnError = Symbol('kOnError'); +const kSignal = Symbol('kSignal'); +const kConnections = Symbol('kConnections'); + +// Parser callback indexes +const kOnHeaders = HTTPParser.kOnHeaders | 0; +const kOnHeadersComplete = HTTPParser.kOnHeadersComplete | 0; +const kOnBody = HTTPParser.kOnBody | 0; +const kOnMessageComplete = HTTPParser.kOnMessageComplete | 0; +const kOnExecute = HTTPParser.kOnExecute | 0; +const kOnTimeout = HTTPParser.kOnTimeout | 0; +const kLenientAll = HTTPParser.kLenientAll | 0; +const kLenientNone = HTTPParser.kLenientNone | 0; + +// WeakMap for storing request metadata (O(1) lookup, no memory leak) +const requestMetadata = new SafeWeakMap(); + +// HTTP status codes +const STATUS_CODES = { + 100: 'Continue', + 101: 'Switching Protocols', + 200: 'OK', + 201: 'Created', + 202: 'Accepted', + 204: 'No Content', + 301: 'Moved Permanently', + 302: 'Found', + 304: 'Not Modified', + 400: 'Bad Request', + 401: 'Unauthorized', + 403: 'Forbidden', + 404: 'Not Found', + 405: 'Method Not Allowed', + 408: 'Request Timeout', + 500: 'Internal Server Error', + 501: 'Not Implemented', + 502: 'Bad Gateway', + 503: 'Service Unavailable', +}; + +// Async resource for parser initialization +class HTTPServerAsyncResource { + constructor(type, socket) { + this.type = type; + this.socket = socket; + } +} + +/** + * Get the Host header value from raw headers array. + * @param {string[]} headers - Raw headers array (key-value pairs) + * @returns {string|undefined} + */ +function getHostHeader(headers) { + for (let i = 0; i < headers.length; i += 2) { + if (headers[i].toLowerCase() === 'host') { + return headers[i + 1]; + } + } + return undefined; +} + +/** + * Convert raw headers array to Headers object. + * @param {string[]} rawHeaders - Raw headers array (key-value pairs) + * @returns {Headers} + */ +function convertHeaders(rawHeaders) { + const headersObj = new Headers(); + for (let i = 0; i < rawHeaders.length; i += 2) { + headersObj.append(rawHeaders[i], rawHeaders[i + 1]); + } + return headersObj; +} + +/** + * Store metadata for a Request object. + * @param {Request} request + * @param {net.Socket|tls.TLSSocket} socket + */ +function setRequestMetadata(request, socket) { + requestMetadata.set(request, { + remoteAddress: socket.remoteAddress, + remotePort: socket.remotePort, + localAddress: socket.localAddress, + localPort: socket.localPort, + encrypted: !!socket.encrypted, + }); +} + +/** + * Get metadata for a Request object created by serve(). + * @param {Request} request + * @returns {{remoteAddress: string, remotePort: number, localAddress: string, localPort: number, encrypted: boolean}} + */ +function getRemoteMetadata(request) { + const meta = requestMetadata.get(request); + if (!meta) { + throw new ERR_INVALID_ARG_TYPE('request', 'Request from serve() handler', request); + } + return meta; +} + +/** + * Create a ReadableStream that bridges parser body events. + * @param {HTTPParser} parser + * @param {net.Socket} socket + * @returns {{stream: ReadableStream, setCallbacks: Function}} + */ +function createRequestBodyStream(parser, socket) { + let controller; + let onBody; + let onComplete; + let closed = false; + + const stream = new ReadableStream({ + start(c) { + controller = c; + }, + pull() { + // Resume parser when consumer is ready for more data + if (!closed) { + parser.resume(); + } + }, + cancel() { + closed = true; + }, + }); + + function setCallbacks() { + onBody = (chunk) => { + if (closed) return; + controller.enqueue(new Uint8Array(chunk)); + // Backpressure: pause parser if the queue is full + if (controller.desiredSize <= 0) { + parser.pause(); + } + }; + + onComplete = () => { + if (closed) return; + closed = true; + controller.close(); + }; + + parser[kOnBody] = onBody; + parser[kOnMessageComplete] = onComplete; + } + + return { stream, setCallbacks }; +} + +/** + * Create a Request object from parser output. + * @param {string[]} headers - Raw headers array + * @param {number} method - HTTP method index + * @param {string} url - Request URL path + * @param {net.Socket} socket + * @param {HTTPParser} parser + * @returns {Request} + */ +function createRequest(headers, method, url, socket, parser) { + // Build full URL + const host = getHostHeader(headers) || `${socket.localAddress}:${socket.localPort}`; + const protocol = socket.encrypted ? 'https:' : 'http:'; + const fullUrl = `${protocol}//${host}${url}`; + + // Convert raw headers to Headers object + const headersObj = convertHeaders(headers); + + // Determine method name + const methodName = allMethods[method]; + + // Create body stream for methods that can have a body + let body = null; + let setCallbacks = null; + if (methodName !== 'GET' && methodName !== 'HEAD') { + const result = createRequestBodyStream(parser, socket); + body = result.stream; + setCallbacks = result.setCallbacks; + } + + // Create Request object + const request = new Request(fullUrl, { + method: methodName, + headers: headersObj, + body, + duplex: body ? 'half' : undefined, + }); + + // Store metadata for getRemoteMetadata() + setRequestMetadata(request, socket); + + return { request, setCallbacks }; +} + +/** + * Write a Response to the socket. + * @param {Response} response + * @param {net.Socket} socket + * @param {boolean} keepAlive + */ +async function writeResponse(response, socket, keepAlive) { + // Status line + const statusText = response.statusText || STATUS_CODES[response.status] || 'Unknown'; + socket.write(`HTTP/1.1 ${response.status} ${statusText}\r\n`); + + // Write headers + let hasContentLength = false; + let hasDate = false; + let hasConnection = false; + + for (const { 0: name, 1: value } of response.headers) { + const lowerName = name.toLowerCase(); + if (lowerName === 'content-length') hasContentLength = true; + if (lowerName === 'date') hasDate = true; + if (lowerName === 'connection') hasConnection = true; + socket.write(`${name}: ${value}\r\n`); + } + + // Determine if we need chunked encoding + const chunked = !hasContentLength && response.body; + + if (chunked) { + socket.write('Transfer-Encoding: chunked\r\n'); + } + + if (!hasConnection) { + socket.write(keepAlive ? 'Connection: keep-alive\r\n' : 'Connection: close\r\n'); + } + + if (!hasDate) { + socket.write(`Date: ${utcDate()}\r\n`); + } + + // End headers + socket.write('\r\n'); + + // Write body if present + if (response.body) { + const reader = response.body.getReader(); + try { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + + if (chunked) { + socket.write(`${value.length.toString(16)}\r\n`); + socket.write(value); + socket.write('\r\n'); + } else { + socket.write(value); + } + + // Handle backpressure + if (socket.writableNeedDrain) { + await once(socket, 'drain'); + } + } + if (chunked) { + socket.write('0\r\n\r\n'); + } + } finally { + reader.releaseLock(); + } + } +} + +/** + * Invoke the handler and manage the response. + * @param {net.Server|tls.Server} server + * @param {net.Socket} socket + * @param {Request} request + * @param {boolean} keepAlive + * @param {HTTPParser} parser + * @param {Function|null} setBodyCallbacks + */ +async function invokeHandler(server, socket, request, keepAlive, parser, setBodyCallbacks) { + let headersSent = false; + + try { + // Set up body stream callbacks if there's a body + if (setBodyCallbacks) { + setBodyCallbacks(); + } else { + // No body expected - set up callbacks to handle any unexpected body data + parser[kOnBody] = () => {}; + parser[kOnMessageComplete] = () => {}; + } + + // Call the handler + let response = server[kHandler](request); + if (response instanceof Promise) { + response = await response; + } + + // Validate response + if (!(response instanceof Response)) { + throw new ERR_INVALID_ARG_TYPE('handler return value', 'Response', response); + } + + // Write response + headersSent = true; + await writeResponse(response, socket, keepAlive); + + } catch (error) { + // Handle errors + if (!headersSent) { + const onError = server[kOnError]; + let errorResponse; + + if (onError) { + try { + errorResponse = await onError(error, request); + } catch { + errorResponse = new Response('Internal Server Error', { status: 500 }); + } + } else { + errorResponse = new Response('Internal Server Error', { status: 500 }); + } + + try { + await writeResponse(errorResponse, socket, false); + } catch { + // Ignore write errors when sending error response + } + } + + // Emit error on server + server.emit('error', error); + } + + // Resume parser for next request or close connection + if (keepAlive && !socket.destroyed) { + parser.resume(); + } else if (!socket.destroyed) { + socket.end(); + } +} + +/** + * Handle a new connection. + * @param {net.Server|tls.Server} server + * @param {net.Socket|tls.TLSSocket} socket + */ +function handleConnection(server, socket) { + // Allocate and initialize parser + const parser = parsers.alloc(); + + const lenient = isLenient(); + + parser.initialize( + HTTPParser.REQUEST, + new HTTPServerAsyncResource('HTTPINCOMINGMESSAGE', socket), + 0, // maxHeaderSize (0 = use default) + lenient ? kLenientAll : kLenientNone, + server[kConnections], + ); + + parser.socket = socket; + socket.parser = parser; + + // Track parser state + parser._headers = []; + parser._url = ''; + + // Handle fragmented headers + parser[kOnHeaders] = function onHeaders(headers, url) { + this._headers.push(...headers); + this._url += url; + }; + + // Main callback when headers are complete + parser[kOnHeadersComplete] = function onHeadersComplete( + versionMajor, versionMinor, headers, method, + url, statusCode, statusMessage, upgrade, shouldKeepAlive, + ) { + // Use accumulated headers if fragmented + if (headers === undefined) { + headers = this._headers; + this._headers = []; + } + if (url === undefined) { + url = this._url; + this._url = ''; + } + + // Handle upgrade requests (WebSocket, etc.) + if (upgrade) { + server.emit('upgrade', { headers, method: allMethods[method], url }, socket); + return 1; // Skip body parsing + } + + // Pause parser immediately - we control when to resume + parser.pause(); + + // Create Request and invoke handler + const { request, setCallbacks } = createRequest(headers, method, url, socket, parser); + + // Publish to diagnostics channel + if (onRequestStartChannel.hasSubscribers) { + onRequestStartChannel.publish({ + request, + socket, + server, + }); + } + + // Invoke handler asynchronously + invokeHandler(server, socket, request, shouldKeepAlive, parser, setCallbacks); + + return 0; + }; + + // Parser execution callback (for consumed sockets) + parser[kOnExecute] = function onParserExecute(ret) { + socket._unrefTimer?.(); + if (ret instanceof PrimordialError) { + prepareError(ret, parser, undefined); + socketOnError(socket, server, ret); + } + }; + + // Parser timeout callback + parser[kOnTimeout] = function onParserTimeout() { + const serverTimeout = server.emit('timeout', socket); + if (!serverTimeout) { + socket.destroy(); + } + }; + + // Consume the socket for zero-copy parsing + if (socket._handle?.isStreamBase && !socket._handle._consumed) { + parser._consumed = true; + socket._handle._consumed = true; + parser.consume(socket._handle); + } + + // Socket event handlers + socket.on('error', (err) => socketOnError(socket, server, err)); + socket.on('close', () => { + freeParser(parser, null, socket); + }); +} + +/** + * Handle socket errors. + * @param {net.Socket} socket + * @param {net.Server} server + * @param {Error} err + */ +function socketOnError(socket, server, err) { + if (!server.emit('clientError', err, socket)) { + // Default error handling + if (socket.writable && !socket._httpMessage) { + socket.write('HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n'); + } + socket.destroy(err); + } +} + +/** + * Create an HTTP server that handles requests using the Fetch API model. + * @param {object} options - Server options + * @param {object} [options.tls] - TLS options (key, cert, etc.) for HTTPS + * @param {AbortSignal} [options.signal] - Signal for graceful shutdown + * @param {Function} [options.onError] - Error handler (error, request) => Response + * @param {Function} handler - Request handler (request) => Response + * @returns {net.Server|tls.Server} + */ +function serve(options, handler) { + // Validate arguments + if (typeof options === 'function') { + handler = options; + options = {}; + } + + validateObject(options, 'options'); + validateFunction(handler, 'handler'); + + // Create base server (net or tls) + let baseServer; + if (options.tls) { + validateObject(options.tls, 'options.tls'); + baseServer = tls.createServer(options.tls); + } else { + baseServer = net.createServer({ allowHalfOpen: true }); + } + + // Attach handler and options to server + baseServer[kHandler] = handler; + baseServer[kOnError] = options.onError; + baseServer[kSignal] = options.signal; + + // Initialize connections tracking + baseServer[kConnections] = new ConnectionsList(); + + // Set up connection listener + const connectionEvent = options.tls ? 'secureConnection' : 'connection'; + baseServer.on(connectionEvent, (socket) => handleConnection(baseServer, socket)); + + // Handle abort signal for graceful shutdown + if (options.signal) { + if (options.signal.aborted) { + process.nextTick(() => baseServer.close()); + } else { + options.signal.addEventListener('abort', () => { + baseServer.close(); + }, { once: true }); + } + } + + return baseServer; +} + +module.exports = { + serve, + getRemoteMetadata, +}; diff --git a/test/parallel/test-http-serve-basic.js b/test/parallel/test-http-serve-basic.js new file mode 100644 index 000000000000..4c66e1cd9820 --- /dev/null +++ b/test/parallel/test-http-serve-basic.js @@ -0,0 +1,281 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); +const net = require('net'); + +// Test that serve is exported +assert.strictEqual(typeof http.serve, 'function'); +assert.strictEqual(typeof http.getRemoteMetadata, 'function'); + +// Test basic request/response +{ + const server = http.serve({}, common.mustCall((request) => { + assert.strictEqual(request.method, 'GET'); + assert.strictEqual(new URL(request.url).pathname, '/test'); + return new Response('Hello World'); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET /test HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /Hello World/); + server.close(); + })); + })); +} + +// Test async handler +{ + const server = http.serve({}, common.mustCall(async (request) => { + await new Promise((resolve) => setTimeout(resolve, 10)); + return new Response('Async Hello'); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /Async Hello/); + server.close(); + })); + })); +} + +// Test Response.json() +{ + const server = http.serve({}, common.mustCall((request) => { + return Response.json({ message: 'Hello JSON' }); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /application\/json/); + assert.match(data, /"message":"Hello JSON"/); + server.close(); + })); + })); +} + +// Test getRemoteMetadata +{ + const server = http.serve({}, common.mustCall((request) => { + const meta = http.getRemoteMetadata(request); + assert.strictEqual(typeof meta.remoteAddress, 'string'); + assert.strictEqual(typeof meta.remotePort, 'number'); + assert.strictEqual(typeof meta.localAddress, 'string'); + assert.strictEqual(typeof meta.localPort, 'number'); + assert.strictEqual(meta.encrypted, false); + return new Response(`Hello from ${meta.remoteAddress}`); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /Hello from/); + server.close(); + })); + })); +} + +// Test custom status code +{ + const server = http.serve({}, common.mustCall((request) => { + return new Response('Created', { status: 201, statusText: 'Created' }); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 201 Created/); + server.close(); + })); + })); +} + +// Test custom headers +{ + const server = http.serve({}, common.mustCall((request) => { + return new Response('With Headers', { + headers: { + 'X-Custom-Header': 'CustomValue', + 'Content-Type': 'text/plain', + }, + }); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /x-custom-header: CustomValue/i); + assert.match(data, /content-type: text\/plain/i); + server.close(); + })); + })); +} + +// Test reading request headers +{ + const server = http.serve({}, common.mustCall((request) => { + const customHeader = request.headers.get('X-Test-Header'); + return new Response(`Header: ${customHeader}`); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nX-Test-Header: TestValue\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /Header: TestValue/); + server.close(); + })); + })); +} + +// Test keep-alive connection (multiple requests) +{ + let requestCount = 0; + const server = http.serve({}, common.mustCall((request) => { + requestCount++; + return new Response(`Request ${requestCount}`); + }, 2)); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + // First request + client.write('GET /1 HTTP/1.1\r\nHost: localhost\r\n\r\n'); + })); + + let data = ''; + let firstResponseReceived = false; + client.setEncoding('utf8'); + client.on('data', (chunk) => { + data += chunk; + if (!firstResponseReceived && data.includes('Request 1')) { + firstResponseReceived = true; + // Send second request + client.write('GET /2 HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + } + }); + client.on('end', common.mustCall(() => { + assert.match(data, /Request 1/); + assert.match(data, /Request 2/); + assert.strictEqual(requestCount, 2); + server.close(); + })); + })); +} + +// Test POST method +{ + const server = http.serve({}, common.mustCall((request) => { + assert.strictEqual(request.method, 'POST'); + return new Response('POST received'); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('POST / HTTP/1.1\r\nHost: localhost\r\nContent-Length: 0\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /POST received/); + server.close(); + })); + })); +} + +// Test handler as first argument (convenience API) +{ + const server = http.serve(common.mustCall((request) => { + return new Response('Convenience API'); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /Convenience API/); + server.close(); + })); + })); +} + +// Test signal for graceful shutdown +{ + const ac = new AbortController(); + const server = http.serve({ signal: ac.signal }, (request) => { + return new Response('OK'); + }); + + server.listen(0, common.mustCall(() => { + // Abort should close the server + ac.abort(); + })); + + server.on('close', common.mustCall()); +} diff --git a/test/parallel/test-http-serve-errors.js b/test/parallel/test-http-serve-errors.js new file mode 100644 index 000000000000..936642e9d0d4 --- /dev/null +++ b/test/parallel/test-http-serve-errors.js @@ -0,0 +1,261 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); +const net = require('net'); + +// Test handler throws error (default 500 response) +{ + const server = http.serve({}, common.mustCall((request) => { + throw new Error('Handler error'); + })); + + server.on('error', common.mustCall((err) => { + assert.strictEqual(err.message, 'Handler error'); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 500/); + assert.match(data, /Internal Server Error/); + server.close(); + })); + })); +} + +// Test async handler throws error +{ + const server = http.serve({}, common.mustCall(async (request) => { + await new Promise((resolve) => setTimeout(resolve, 10)); + throw new Error('Async handler error'); + })); + + server.on('error', common.mustCall((err) => { + assert.strictEqual(err.message, 'Async handler error'); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 500/); + server.close(); + })); + })); +} + +// Test custom onError handler +{ + const server = http.serve({ + onError: common.mustCall((error, request) => { + assert.strictEqual(error.message, 'Custom error'); + return new Response('Custom Error Page', { status: 503 }); + }), + }, common.mustCall((request) => { + throw new Error('Custom error'); + })); + + server.on('error', common.mustCall()); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 503/); + assert.match(data, /Custom Error Page/); + server.close(); + })); + })); +} + +// Test handler returns non-Response +{ + const server = http.serve({}, common.mustCall((request) => { + return 'not a Response'; // Invalid return value + })); + + server.on('error', common.mustCall((err) => { + assert.strictEqual(err.code, 'ERR_INVALID_ARG_TYPE'); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 500/); + server.close(); + })); + })); +} + +// Test handler returns null +{ + const server = http.serve({}, common.mustCall((request) => { + return null; + })); + + server.on('error', common.mustCall()); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 500/); + server.close(); + })); + })); +} + +// Test getRemoteMetadata with non-serve request +{ + assert.throws(() => { + const request = new Request('http://example.com/'); + http.getRemoteMetadata(request); + }, { + code: 'ERR_INVALID_ARG_TYPE', + }); +} + +// Test invalid arguments to serve() +{ + assert.throws(() => { + http.serve({}, 'not a function'); + }, { + code: 'ERR_INVALID_ARG_TYPE', + }); +} + +// Test onError handler throws +{ + const server = http.serve({ + onError: common.mustCall((error, request) => { + throw new Error('onError also throws'); + }), + }, common.mustCall((request) => { + throw new Error('Handler error'); + })); + + server.on('error', common.mustCall()); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + // Should fallback to default 500 response + assert.match(data, /HTTP\/1\.1 500/); + server.close(); + })); + })); +} + +// Test clientError event +{ + const server = http.serve({}, (request) => { + return new Response('OK'); + }); + + server.on('clientError', common.mustCall((err, socket) => { + assert.ok(err); + socket.end('HTTP/1.1 400 Bad Request\r\n\r\n'); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + // Send malformed HTTP request + client.write('INVALID HTTP\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /400/); + server.close(); + })); + })); +} + +// Test signal already aborted +{ + const ac = new AbortController(); + ac.abort(); + + const server = http.serve({ signal: ac.signal }, (request) => { + return new Response('OK'); + }); + + // Set up close handler before listen since server closes immediately + server.on('close', common.mustCall()); + + server.listen(0); +} + +// Test async onError handler +{ + const server = http.serve({ + onError: common.mustCall(async (error, request) => { + await new Promise((resolve) => setTimeout(resolve, 10)); + return new Response('Async Error Handler', { status: 502 }); + }), + }, common.mustCall((request) => { + throw new Error('Handler error'); + })); + + server.on('error', common.mustCall()); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 502/); + assert.match(data, /Async Error Handler/); + server.close(); + })); + })); +} diff --git a/test/parallel/test-http-serve-streaming.js b/test/parallel/test-http-serve-streaming.js new file mode 100644 index 000000000000..670da8795ece --- /dev/null +++ b/test/parallel/test-http-serve-streaming.js @@ -0,0 +1,249 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); +const net = require('net'); + +// Test reading request body +{ + const server = http.serve({}, common.mustCall(async (request) => { + const body = await request.text(); + return new Response(`Received: ${body}`); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const body = 'Hello Server'; + const client = net.createConnection(port, common.mustCall(() => { + client.write(`POST / HTTP/1.1\r\nHost: localhost\r\nContent-Length: ${body.length}\r\nConnection: close\r\n\r\n${body}`); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /Received: Hello Server/); + server.close(); + })); + })); +} + +// Test reading JSON request body +{ + const server = http.serve({}, common.mustCall(async (request) => { + const body = await request.json(); + return Response.json({ received: body }); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const body = JSON.stringify({ message: 'Hello' }); + const client = net.createConnection(port, common.mustCall(() => { + client.write(`POST / HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: ${body.length}\r\nConnection: close\r\n\r\n${body}`); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /"received":\{"message":"Hello"\}/); + server.close(); + })); + })); +} + +// Test streaming response body +{ + const server = http.serve({}, common.mustCall((request) => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('chunk1')); + controller.enqueue(new TextEncoder().encode('chunk2')); + controller.enqueue(new TextEncoder().encode('chunk3')); + controller.close(); + }, + }); + return new Response(stream); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /Transfer-Encoding: chunked/i); + assert.match(data, /chunk1/); + assert.match(data, /chunk2/); + assert.match(data, /chunk3/); + server.close(); + })); + })); +} + +// Test async streaming response +{ + const server = http.serve({}, common.mustCall((request) => { + const stream = new ReadableStream({ + async start(controller) { + await new Promise((resolve) => setTimeout(resolve, 10)); + controller.enqueue(new TextEncoder().encode('async1')); + await new Promise((resolve) => setTimeout(resolve, 10)); + controller.enqueue(new TextEncoder().encode('async2')); + controller.close(); + }, + }); + return new Response(stream); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /async1/); + assert.match(data, /async2/); + server.close(); + })); + })); +} + +// Test response with explicit Content-Length (no chunked encoding) +{ + const responseBody = 'Fixed length body'; + const server = http.serve({}, common.mustCall((request) => { + return new Response(responseBody, { + headers: { + 'Content-Length': responseBody.length.toString(), + }, + }); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /Content-Length: 17/i); + assert.ok(!data.toLowerCase().includes('transfer-encoding')); + assert.match(data, /Fixed length body/); + server.close(); + })); + })); +} + +// Test empty response body +{ + const server = http.serve({}, common.mustCall((request) => { + return new Response(null, { status: 204 }); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 204/); + server.close(); + })); + })); +} + +// Test reading request body as arrayBuffer +{ + const server = http.serve({}, common.mustCall(async (request) => { + const buffer = await request.arrayBuffer(); + const text = new TextDecoder().decode(buffer); + return new Response(`ArrayBuffer: ${text}`); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const body = 'binary data'; + const client = net.createConnection(port, common.mustCall(() => { + client.write(`POST / HTTP/1.1\r\nHost: localhost\r\nContent-Length: ${body.length}\r\nConnection: close\r\n\r\n${body}`); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /ArrayBuffer: binary data/); + server.close(); + })); + })); +} + +// Test chunked request body +{ + const server = http.serve({}, common.mustCall(async (request) => { + const body = await request.text(); + return new Response(`Chunked: ${body}`); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('POST / HTTP/1.1\r\nHost: localhost\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n'); + client.write('5\r\nHello\r\n'); + client.write('6\r\n World\r\n'); + client.write('0\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /Chunked: Hello World/); + server.close(); + })); + })); +} + +// Test large request body +{ + const largeBody = 'x'.repeat(65536); // 64KB + const server = http.serve({}, common.mustCall(async (request) => { + const body = await request.text(); + return new Response(`Length: ${body.length}`); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write(`POST / HTTP/1.1\r\nHost: localhost\r\nContent-Length: ${largeBody.length}\r\nConnection: close\r\n\r\n`); + client.write(largeBody); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /Length: 65536/); + server.close(); + })); + })); +} From 3af3279fd31885432d28ff8ee2a22b30ee5b7d61 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Mon, 26 Jan 2026 09:59:44 +0100 Subject: [PATCH 2/4] http: optimize serve() performance - Enable TCP_NODELAY (noDelay: true) to prevent Nagle's algorithm from batching small writes, which was causing 40ms delays - Batch header writes into a single socket.write() call - Use cork()/uncork() to batch chunked body writes These optimizations improve hello-world throughput from ~24 req/sec to ~13,600 req/sec (567x improvement). The remaining 2.3x gap vs createServer() is due to Response body streaming overhead. --- lib/internal/http_serve.js | 26 ++++++++++++++++---------- 1 file changed, 16 insertions(+), 10 deletions(-) diff --git a/lib/internal/http_serve.js b/lib/internal/http_serve.js index 103dfb3149b8..cfb8461a4de9 100644 --- a/lib/internal/http_serve.js +++ b/lib/internal/http_serve.js @@ -248,11 +248,11 @@ function createRequest(headers, method, url, socket, parser) { * @param {boolean} keepAlive */ async function writeResponse(response, socket, keepAlive) { - // Status line + // Build headers as a single string to minimize syscalls const statusText = response.statusText || STATUS_CODES[response.status] || 'Unknown'; - socket.write(`HTTP/1.1 ${response.status} ${statusText}\r\n`); + let head = `HTTP/1.1 ${response.status} ${statusText}\r\n`; - // Write headers + // Collect headers let hasContentLength = false; let hasDate = false; let hasConnection = false; @@ -262,26 +262,29 @@ async function writeResponse(response, socket, keepAlive) { if (lowerName === 'content-length') hasContentLength = true; if (lowerName === 'date') hasDate = true; if (lowerName === 'connection') hasConnection = true; - socket.write(`${name}: ${value}\r\n`); + head += `${name}: ${value}\r\n`; } // Determine if we need chunked encoding const chunked = !hasContentLength && response.body; if (chunked) { - socket.write('Transfer-Encoding: chunked\r\n'); + head += 'Transfer-Encoding: chunked\r\n'; } if (!hasConnection) { - socket.write(keepAlive ? 'Connection: keep-alive\r\n' : 'Connection: close\r\n'); + head += keepAlive ? 'Connection: keep-alive\r\n' : 'Connection: close\r\n'; } if (!hasDate) { - socket.write(`Date: ${utcDate()}\r\n`); + head += `Date: ${utcDate()}\r\n`; } // End headers - socket.write('\r\n'); + head += '\r\n'; + + // Write headers in a single call + socket.write(head); // Write body if present if (response.body) { @@ -292,9 +295,12 @@ async function writeResponse(response, socket, keepAlive) { if (done) break; if (chunked) { + // Cork to batch chunk writes into single TCP packet + socket.cork(); socket.write(`${value.length.toString(16)}\r\n`); socket.write(value); socket.write('\r\n'); + socket.uncork(); } else { socket.write(value); } @@ -529,9 +535,9 @@ function serve(options, handler) { let baseServer; if (options.tls) { validateObject(options.tls, 'options.tls'); - baseServer = tls.createServer(options.tls); + baseServer = tls.createServer({ noDelay: true, ...options.tls }); } else { - baseServer = net.createServer({ allowHalfOpen: true }); + baseServer = net.createServer({ allowHalfOpen: true, noDelay: true }); } // Attach handler and options to server From 06956edb0d0e9eb64240d259d9b8944e662585bb Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Mon, 20 Jul 2026 01:18:27 +0200 Subject: [PATCH 3/4] deps: expose undici serverKit for http.serve() Add a single frozen serverKit namespace to the bundle exports, exposing the existing fetch internals (kConstruct, HeadersList, headers guard and list accessors, request/response state accessors) that the http.serve() implementation needs to construct Request/Response subclasses without going through the public constructors. No undici logic is changed; the request and response modules only gain exports for accessors they already define. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RGGHNWBpZyJEEN1Pb2msxY --- deps/undici/undici.js | 26 +++++++++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/deps/undici/undici.js b/deps/undici/undici.js index 02f7f6da92ff..d9aa28840363 100644 --- a/deps/undici/undici.js +++ b/deps/undici/undici.js @@ -12561,7 +12561,9 @@ var require_response = __commonJS({ Response, cloneResponse, fromInnerResponse, - getResponseState + getResponseState, + setResponseState, + setResponseHeaders }; } }); @@ -13347,6 +13349,9 @@ var require_request2 = __commonJS({ cloneRequest, getRequestDispatcher, getRequestState, + setRequestState, + setRequestHeaders, + setRequestSignal, removeRequestAbortListener }; } @@ -18572,6 +18577,25 @@ module.exports.FormData = require_formdata().FormData; module.exports.Headers = require_headers().Headers; module.exports.Response = require_response().Response; module.exports.Request = require_request2().Request; +// Exposed only through Node.js' internal Undici bundle for HTTP server use. +// This is the single boundary through which Node.js core may reach fetch +// internals; keep it in sync with lib/internal/http_serve_classes.js. +module.exports.serverKit = Object.freeze({ + kConstruct: require_symbols().kConstruct, + HeadersList: require_headers().HeadersList, + fillHeaders: require_headers().fill, + getHeadersList: require_headers().getHeadersList, + setHeadersList: require_headers().setHeadersList, + getHeadersGuard: require_headers().getHeadersGuard, + setHeadersGuard: require_headers().setHeadersGuard, + getRequestState: require_request2().getRequestState, + setRequestState: require_request2().setRequestState, + setRequestHeaders: require_request2().setRequestHeaders, + setRequestSignal: require_request2().setRequestSignal, + getResponseState: require_response().getResponseState, + setResponseState: require_response().setResponseState, + setResponseHeaders: require_response().setResponseHeaders +}); var { CloseEvent, ErrorEvent, MessageEvent, createFastMessageEvent } = require_events(); module.exports.WebSocket = require_websocket().WebSocket; module.exports.CloseEvent = CloseEvent; From 125f08f3ff63307b4ed7f60eef556699fdd92014 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Mon, 20 Jul 2026 01:18:41 +0200 Subject: [PATCH 4/4] http: add NodeRequest and NodeResponse classes to serve() Introduce NodeRequest and NodeResponse, real subclasses of the fetch Request/Response built through undici's kConstruct path, and use them in serve(). Incoming requests skip the public Request constructor entirely: the HeadersList is filled directly from llhttp-validated parser output, the URL is parsed lazily, and the AbortSignal is only materialized on access. NodeResponse adds a fast construction path for null, string and Uint8Array bodies with no WebIDL conversion and no ReadableStream until one is requested, and writeResponse writes body sources directly instead of entering the Web Streams reader loop. Also: - request.signal now aborts and the request body stream errors when the client disconnects before the request completes, following http.Server's half-open connection semantics - socket metadata is exposed as getters on NodeRequest and cached per connection; getRemoteMetadata is unchanged - multiple Set-Cookie response headers are written as separate header lines instead of one comma-joined line - Response.error() is rejected instead of serializing as status 0 - fetch-compatible Response objects from a different undici instance are serialized through their public API - serve() is lazily loaded from lib/http.js; loading the undici bundle while http.js is still initializing let undici capture the module without maxHeaderSize, breaking fetch's default dispatcher benchmark/http/serve.js gains a serve-fast variant using NodeResponse: with wrk at c=500 it roughly doubles throughput against plain Response handlers, and plain Response handlers gain 8-16% over the previous implementation. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01RGGHNWBpZyJEEN1Pb2msxY --- benchmark/http/serve.js | 40 ++ lib/http.js | 32 +- lib/internal/http_serve.js | 673 +++++++++++++------ lib/internal/http_serve_classes.js | 461 +++++++++++++ test/parallel/test-http-serve-basic.js | 25 + test/parallel/test-http-serve-classes.js | 319 +++++++++ test/parallel/test-http-serve-parser-pool.js | 38 ++ test/parallel/test-http-serve-unread-body.js | 36 + 8 files changed, 1430 insertions(+), 194 deletions(-) create mode 100644 benchmark/http/serve.js create mode 100644 lib/internal/http_serve_classes.js create mode 100644 test/parallel/test-http-serve-classes.js create mode 100644 test/parallel/test-http-serve-parser-pool.js create mode 100644 test/parallel/test-http-serve-unread-body.js diff --git a/benchmark/http/serve.js b/benchmark/http/serve.js new file mode 100644 index 000000000000..8c38af97a460 --- /dev/null +++ b/benchmark/http/serve.js @@ -0,0 +1,40 @@ +'use strict'; + +const common = require('../common.js'); +const http = require('http'); + +const bench = common.createBenchmark(main, { + server: ['createServer', 'serve', 'serve-fast'], + type: ['string', 'buffer'], + len: [4, 1024, 102400], + c: [50, 500], + duration: 5, +}); + +function main({ server: serverType, type, len, c, duration }) { + const body = type === 'string' ? 'C'.repeat(len) : Buffer.alloc(len, 67); + const headers = { + 'Content-Length': `${len}`, + 'Content-Type': 'application/octet-stream', + }; + + let server; + if (serverType === 'serve') { + server = http.serve(() => new Response(body, { headers })); + } else if (serverType === 'serve-fast') { + server = http.serve(() => new http.NodeResponse(body, { headers })); + } else { + server = http.createServer((request, response) => { + response.writeHead(200, headers); + response.end(body); + }); + } + + server.listen(0, () => { + bench.http({ + connections: c, + duration, + port: server.address().port, + }, () => server.close()); + }); +} diff --git a/lib/http.js b/lib/http.js index ff744bf92248..d4b966fe94ab 100644 --- a/lib/http.js +++ b/lib/http.js @@ -42,10 +42,16 @@ const { Server, ServerResponse, } = require('_http_server'); -const { - serve, - getRemoteMetadata, -} = require('internal/http_serve'); +// Loaded lazily: pulling in the serve() implementation loads the undici +// bundle, and doing that while this module is still initializing would let +// undici capture a half-initialized http module (e.g. no maxHeaderSize). +function serve(options, handler) { + return require('internal/http_serve').serve(options, handler); +} + +function getRemoteMetadata(request) { + return require('internal/http_serve').getRemoteMetadata(request); +} const { parseProxyUrl, getGlobalAgent, @@ -209,6 +215,24 @@ module.exports = { getRemoteMetadata, }; +ObjectDefineProperty(module.exports, 'NodeRequest', { + __proto__: null, + configurable: true, + enumerable: true, + get() { + return require('internal/http_serve_classes').NodeRequest; + }, +}); + +ObjectDefineProperty(module.exports, 'NodeResponse', { + __proto__: null, + configurable: true, + enumerable: true, + get() { + return require('internal/http_serve_classes').NodeResponse; + }, +}); + ObjectDefineProperty(module.exports, 'maxHeaderSize', { __proto__: null, configurable: true, diff --git a/lib/internal/http_serve.js b/lib/internal/http_serve.js index cfb8461a4de9..7bcb7cfdbfd5 100644 --- a/lib/internal/http_serve.js +++ b/lib/internal/http_serve.js @@ -3,7 +3,7 @@ const { Error: PrimordialError, Promise, - SafeWeakMap, + StringPrototypeToLowerCase, Symbol, Uint8Array, } = primordials; @@ -11,22 +11,29 @@ const { const net = require('net'); const tls = require('tls'); const { once } = require('events'); +const { setImmediate } = require('timers'); const { ReadableStream } = require('internal/webstreams/readablestream'); +const { Response } = require('internal/deps/undici/undici'); const { - Headers, - Request, - Response, -} = require('internal/deps/undici/undici'); + NodeResponse, + createServeRequest, + getServeMetadata, + abortServeRequest, + getResponseState, + HeadersList, + isLazyResponseBody, + hasMaterializedStream, +} = require('internal/http_serve_classes'); const { - parsers, - freeParser, HTTPParser, isLenient, prepareError, allMethods, } = require('_http_common'); +const { STATUS_CODES } = require('_http_server'); const { ConnectionsList } = internalBinding('http_parser'); +const FreeList = require('internal/freelist'); const { utcDate, @@ -36,6 +43,7 @@ const { validateFunction, validateObject } = require('internal/validators'); const { codes: { ERR_INVALID_ARG_TYPE, + ERR_INVALID_STATE, }, } = require('internal/errors'); @@ -48,6 +56,10 @@ const kOnError = Symbol('kOnError'); const kSignal = Symbol('kSignal'); const kConnections = Symbol('kConnections'); +// Symbols for private properties on sockets +const kConnMeta = Symbol('kConnMeta'); +const kInFlight = Symbol('kInFlight'); + // Parser callback indexes const kOnHeaders = HTTPParser.kOnHeaders | 0; const kOnHeadersComplete = HTTPParser.kOnHeadersComplete | 0; @@ -58,31 +70,9 @@ const kOnTimeout = HTTPParser.kOnTimeout | 0; const kLenientAll = HTTPParser.kLenientAll | 0; const kLenientNone = HTTPParser.kLenientNone | 0; -// WeakMap for storing request metadata (O(1) lookup, no memory leak) -const requestMetadata = new SafeWeakMap(); - -// HTTP status codes -const STATUS_CODES = { - 100: 'Continue', - 101: 'Switching Protocols', - 200: 'OK', - 201: 'Created', - 202: 'Accepted', - 204: 'No Content', - 301: 'Moved Permanently', - 302: 'Found', - 304: 'Not Modified', - 400: 'Bad Request', - 401: 'Unauthorized', - 403: 'Forbidden', - 404: 'Not Found', - 405: 'Method Not Allowed', - 408: 'Request Timeout', - 500: 'Internal Server Error', - 501: 'Not Implemented', - 502: 'Bad Gateway', - 503: 'Service Unavailable', -}; +// serve() installs different callbacks from the regular HTTP client and server, +// so its parsers must not share their callback pool. +const serveParsers = new FreeList('serveParsers', 1000, () => new HTTPParser()); // Async resource for parser initialization class HTTPServerAsyncResource { @@ -93,45 +83,51 @@ class HTTPServerAsyncResource { } /** - * Get the Host header value from raw headers array. - * @param {string[]} headers - Raw headers array (key-value pairs) - * @returns {string|undefined} - */ -function getHostHeader(headers) { - for (let i = 0; i < headers.length; i += 2) { - if (headers[i].toLowerCase() === 'host') { - return headers[i + 1]; - } - } - return undefined; -} - -/** - * Convert raw headers array to Headers object. + * Build the request HeadersList directly from the parser output in a single + * pass, detecting Host, Content-Length and Transfer-Encoding along the way. + * Names and values were already validated by llhttp, so the WebIDL-level + * re-validation done by the public Headers API is skipped on purpose. * @param {string[]} rawHeaders - Raw headers array (key-value pairs) - * @returns {Headers} + * @returns {{headersList: HeadersList, host: string|undefined, + * hasBody: boolean, contentLength: number|null}} */ -function convertHeaders(rawHeaders) { - const headersObj = new Headers(); +function scanHeaders(rawHeaders) { + const headersList = new HeadersList(); + let host; + let hasBody = false; + let contentLength = null; for (let i = 0; i < rawHeaders.length; i += 2) { - headersObj.append(rawHeaders[i], rawHeaders[i + 1]); + const name = StringPrototypeToLowerCase(rawHeaders[i]); + const value = rawHeaders[i + 1]; + if (name === 'host') { + host ??= value; + } else if (name === 'content-length') { + hasBody = value !== '0'; + contentLength = hasBody ? +value : 0; + } else if (name === 'transfer-encoding') { + hasBody = true; + contentLength = null; + } + headersList.append(name, value, true); } - return headersObj; + return { headersList, host, hasBody, contentLength }; } /** - * Store metadata for a Request object. - * @param {Request} request + * Get (and lazily cache) the address metadata for a connection. The object is + * shared by every request on a keep-alive connection and stays valid after + * the socket is torn down. * @param {net.Socket|tls.TLSSocket} socket + * @returns {{remoteAddress: string, remotePort: number, localAddress: string, localPort: number, encrypted: boolean}} */ -function setRequestMetadata(request, socket) { - requestMetadata.set(request, { +function getConnectionMetadata(socket) { + return socket[kConnMeta] ??= { remoteAddress: socket.remoteAddress, remotePort: socket.remotePort, localAddress: socket.localAddress, localPort: socket.localPort, encrypted: !!socket.encrypted, - }); + }; } /** @@ -140,11 +136,11 @@ function setRequestMetadata(request, socket) { * @returns {{remoteAddress: string, remotePort: number, localAddress: string, localPort: number, encrypted: boolean}} */ function getRemoteMetadata(request) { - const meta = requestMetadata.get(request); - if (!meta) { + const metadata = getServeMetadata(request); + if (metadata === undefined) { throw new ERR_INVALID_ARG_TYPE('request', 'Request from serve() handler', request); } - return meta; + return metadata; } /** @@ -158,19 +154,21 @@ function createRequestBodyStream(parser, socket) { let onBody; let onComplete; let closed = false; + let complete = false; const stream = new ReadableStream({ start(c) { controller = c; }, pull() { - // Resume parser when consumer is ready for more data - if (!closed) { + // Resume parser when consumer is ready for more data. + if (!closed && socket.parser === parser) { parser.resume(); } }, cancel() { closed = true; + if (socket.parser === parser) parser.resume(); }, }); @@ -185,6 +183,7 @@ function createRequestBodyStream(parser, socket) { }; onComplete = () => { + complete = true; if (closed) return; closed = true; controller.close(); @@ -194,51 +193,68 @@ function createRequestBodyStream(parser, socket) { parser[kOnMessageComplete] = onComplete; } - return { stream, setCallbacks }; + function discard(onDiscarded) { + if (complete) return false; + closed = true; + parser[kOnBody] = noop; + parser[kOnMessageComplete] = () => { + complete = true; + controller.close(); + onDiscarded(); + }; + return true; + } + + function abort(reason) { + if (closed) return; + closed = true; + controller.error(reason); + } + + function isComplete() { + return complete; + } + + return { stream, setCallbacks, discard, abort, isComplete }; } /** * Create a Request object from parser output. - * @param {string[]} headers - Raw headers array + * @param {string[]} rawHeaders - Raw headers array * @param {number} method - HTTP method index * @param {string} url - Request URL path * @param {net.Socket} socket * @param {HTTPParser} parser * @returns {Request} */ -function createRequest(headers, method, url, socket, parser) { - // Build full URL - const host = getHostHeader(headers) || `${socket.localAddress}:${socket.localPort}`; +function createRequest(rawHeaders, method, url, socket, parser) { + // Build the headers list and the full URL. + const scanned = scanHeaders(rawHeaders); + const host = scanned.host || `${socket.localAddress}:${socket.localPort}`; const protocol = socket.encrypted ? 'https:' : 'http:'; const fullUrl = `${protocol}//${host}${url}`; - // Convert raw headers to Headers object - const headersObj = convertHeaders(headers); - // Determine method name const methodName = allMethods[method]; // Create body stream for methods that can have a body let body = null; - let setCallbacks = null; - if (methodName !== 'GET' && methodName !== 'HEAD') { - const result = createRequestBodyStream(parser, socket); - body = result.stream; - setCallbacks = result.setCallbacks; - } - - // Create Request object - const request = new Request(fullUrl, { - method: methodName, - headers: headersObj, - body, - duplex: body ? 'half' : undefined, - }); + let bodyControl = null; + if (methodName !== 'GET' && methodName !== 'HEAD' && scanned.hasBody) { + bodyControl = createRequestBodyStream(parser, socket); + body = { + stream: bodyControl.stream, + source: null, + length: scanned.contentLength, + }; + } - // Store metadata for getRemoteMetadata() - setRequestMetadata(request, socket); + const request = createServeRequest( + methodName, fullUrl, scanned.headersList, body, + getConnectionMetadata(socket), + ); - return { request, setCallbacks }; + return { request, bodyControl }; } /** @@ -246,28 +262,77 @@ function createRequest(headers, method, url, socket, parser) { * @param {Response} response * @param {net.Socket} socket * @param {boolean} keepAlive + * @returns {Promise|undefined} */ -async function writeResponse(response, socket, keepAlive) { - // Build headers as a single string to minimize syscalls - const statusText = response.statusText || STATUS_CODES[response.status] || 'Unknown'; - let head = `HTTP/1.1 ${response.status} ${statusText}\r\n`; +function writeResponse(response, socket, keepAlive, headRequest = false) { + if (!(response instanceof Response)) { + return writeForeignResponse(response, socket, keepAlive, headRequest); + } + + const state = getResponseState(response); + if (state.status === 0) { + // Response.error() and filtered responses cannot be serialized. + throw new ERR_INVALID_STATE('Network error responses cannot be sent'); + } + const body = state.body ?? null; + const statusText = state.statusText || STATUS_CODES[state.status] || 'Unknown'; + let head = `HTTP/1.1 ${state.status} ${statusText}\r\n`; - // Collect headers let hasContentLength = false; + let hasTransferEncoding = false; let hasDate = false; let hasConnection = false; - for (const { 0: name, 1: value } of response.headers) { - const lowerName = name.toLowerCase(); - if (lowerName === 'content-length') hasContentLength = true; - if (lowerName === 'date') hasDate = true; - if (lowerName === 'connection') hasConnection = true; - head += `${name}: ${value}\r\n`; + const headersList = state.headersList; + for (const { 0: name, 1: entry } of headersList.headersMap) { + if (name === 'set-cookie') continue; // Written individually below. + if (name === 'content-length') hasContentLength = true; + else if (name === 'transfer-encoding') hasTransferEncoding = true; + else if (name === 'date') hasDate = true; + else if (name === 'connection') hasConnection = true; + head += `${name}: ${entry.value}\r\n`; + } + // Multiple Set-Cookie headers must each go on their own line; the + // headersMap entry holds them joined with ', ', which clients misparse. + const cookies = headersList.cookies; + if (cookies !== null) { + for (let i = 0; i < cookies.length; i++) { + head += `set-cookie: ${cookies[i]}\r\n`; + } + } + + // A body whose source bytes are available can be written directly, + // without touching (or, for lazy bodies, even creating) its stream. + let source; + if (body !== null) { + if (isLazyResponseBody(body) && !hasMaterializedStream(body)) { + if (body.used) { + throw new ERR_INVALID_STATE('Response body is unusable'); + } + source = body.source; + } else { + if (response.bodyUsed || body.stream.locked) { + throw new ERR_INVALID_STATE('Response body is unusable'); + } + const candidate = body.source; + if (candidate instanceof Uint8Array) { + source = candidate; + } else if (typeof candidate === 'string' && candidate.length <= 8192) { + // The stream already holds the encoded bytes; writing the string + // re-encodes it, which only beats the stream loop for small bodies. + source = candidate; + } + } } - // Determine if we need chunked encoding - const chunked = !hasContentLength && response.body; + if (!hasContentLength && !hasTransferEncoding && + body !== null && body.length !== null) { + head += `Content-Length: ${body.length}\r\n`; + hasContentLength = true; + } + const chunked = !headRequest && body !== null && + !hasContentLength && !hasTransferEncoding; if (chunked) { head += 'Transfer-Encoding: chunked\r\n'; } @@ -279,43 +344,235 @@ async function writeResponse(response, socket, keepAlive) { if (!hasDate) { head += `Date: ${utcDate()}\r\n`; } + head += '\r\n'; + + if (headRequest || body === null || body.length === 0) { + const headWritten = socket.write(head); + markResponseBodyUsed(body); + if (!headWritten) return once(socket, 'drain'); + return; + } + + // Write the source bytes directly, skipping the Web Streams reader loop. + if (source !== undefined) { + socket.cork(); + socket.write(head); + const bodyWritten = socket.write(source); + socket.uncork(); + markResponseBodyUsed(body); + if (!bodyWritten) return once(socket, 'drain'); + return; + } + + const headWritten = socket.write(head); + if (!headWritten) { + return writeStreamingBodyAfterDrain(body.stream, socket, chunked); + } + return writeStreamingBody(body.stream, socket, chunked); +} + +function markResponseBodyUsed(body) { + if (body === null) return; + if (isLazyResponseBody(body) && !hasMaterializedStream(body)) { + body.used = true; + } else { + body.stream.cancel().catch(noop); + } +} + +/** + * Serialize a fetch-compatible Response-like object that is not an instance + * of the bundled undici Response (e.g. one produced by a copy of undici + * shipped inside a framework's node_modules) using only its public API. + * @param {object} response + * @param {net.Socket} socket + * @param {boolean} keepAlive + * @param {boolean} headRequest + * @returns {Promise|undefined} + */ +function writeForeignResponse(response, socket, keepAlive, headRequest) { + const status = response.status; + if (status === 0) { + throw new ERR_INVALID_STATE('Network error responses cannot be sent'); + } + const statusText = response.statusText || STATUS_CODES[status] || 'Unknown'; + let head = `HTTP/1.1 ${status} ${statusText}\r\n`; - // End headers + let hasContentLength = false; + let hasTransferEncoding = false; + let hasDate = false; + let hasConnection = false; + + // Public Headers iteration is sorted, lowercased and yields set-cookie + // entries individually. + for (const { 0: name, 1: value } of response.headers) { + if (name === 'content-length') hasContentLength = true; + else if (name === 'transfer-encoding') hasTransferEncoding = true; + else if (name === 'date') hasDate = true; + else if (name === 'connection') hasConnection = true; + head += `${name}: ${value}\r\n`; + } + + const body = response.body ?? null; + if (body !== null && (response.bodyUsed || body.locked)) { + throw new ERR_INVALID_STATE('Response body is unusable'); + } + + const noBody = headRequest || body === null || + status === 204 || status === 304; + const chunked = !noBody && !hasContentLength && !hasTransferEncoding; + if (chunked) { + head += 'Transfer-Encoding: chunked\r\n'; + } + if (!hasConnection) { + head += keepAlive ? 'Connection: keep-alive\r\n' : 'Connection: close\r\n'; + } + if (!hasDate) { + head += `Date: ${utcDate()}\r\n`; + } head += '\r\n'; - // Write headers in a single call - socket.write(head); + if (noBody) { + const headWritten = socket.write(head); + if (body !== null) body.cancel().catch(noop); + if (!headWritten) return once(socket, 'drain'); + return; + } + + const headWritten = socket.write(head); + if (!headWritten) { + return writeStreamingBodyAfterDrain(body, socket, chunked); + } + return writeStreamingBody(body, socket, chunked); +} + +async function writeStreamingBodyAfterDrain(stream, socket, chunked) { + await once(socket, 'drain'); + return writeStreamingBody(stream, socket, chunked); +} + +async function writeStreamingBody(stream, socket, chunked) { + const reader = stream.getReader(); + try { + while (true) { + const { value, done } = await reader.read(); + if (done) break; - // Write body if present - if (response.body) { - const reader = response.body.getReader(); - try { - while (true) { - const { value, done } = await reader.read(); - if (done) break; - - if (chunked) { - // Cork to batch chunk writes into single TCP packet - socket.cork(); - socket.write(`${value.length.toString(16)}\r\n`); - socket.write(value); - socket.write('\r\n'); - socket.uncork(); - } else { - socket.write(value); - } - - // Handle backpressure - if (socket.writableNeedDrain) { - await once(socket, 'drain'); - } - } if (chunked) { - socket.write('0\r\n\r\n'); + socket.cork(); + socket.write(`${value.length.toString(16)}\r\n`); + socket.write(value); + socket.write('\r\n'); + socket.uncork(); + } else { + socket.write(value); + } + + if (socket.writableNeedDrain) { + await once(socket, 'drain'); } - } finally { - reader.releaseLock(); } + if (chunked) { + socket.write('0\r\n\r\n'); + } + } finally { + reader.releaseLock(); + } +} + +function noop() {} + +function completeRequest(socket, keepAlive, parser, resume = true) { + // A half-closed connection cannot receive further requests, so finish it + // even if the client asked for keep-alive. + if (keepAlive && !socket.destroyed && !socket.readableEnded && + socket.parser === parser) { + if (resume) parser.resume(); + } else if (!socket.destroyed) { + socket.end(); + } +} + +function completeResponse(socket, keepAlive, parser, bodyControl, resume) { + // The response is fully written: client disconnects are no longer aborts. + socket[kInFlight] = undefined; + if (bodyControl && bodyControl.discard(() => { + completeRequest(socket, keepAlive, parser, false); + })) { + if (resume && socket.parser === parser) parser.resume(); + return; + } + completeRequest(socket, keepAlive, parser, resume); +} + +function validateResponse(response) { + if (response instanceof Response) return; + // A fetch-compatible Response from another undici instance (e.g. bundled + // inside a framework) is accepted and serialized through its public API. + if (response !== null && typeof response === 'object' && + typeof response.status === 'number' && + typeof response.headers?.getSetCookie === 'function') { + return; + } + throw new ERR_INVALID_ARG_TYPE('handler return value', 'Response', response); +} + +async function handleHandlerError( + server, socket, request, keepAlive, parser, bodyControl, error, headersSent, +) { + if (!headersSent && !socket.destroyed) { + const onError = server[kOnError]; + let errorResponse; + + if (onError) { + try { + errorResponse = await onError(error, request); + } catch { + errorResponse = new NodeResponse('Internal Server Error', { status: 500 }); + } + } else { + errorResponse = new NodeResponse('Internal Server Error', { status: 500 }); + } + + try { + await writeResponse(errorResponse, socket, false, request.method === 'HEAD'); + } catch { + // Ignore write errors when sending error response. + } + } + + server.emit('error', error); + completeResponse(socket, keepAlive, parser, bodyControl, true); +} + +async function finishAsyncResponse( + server, socket, request, keepAlive, parser, bodyControl, response, +) { + let headersSent = false; + try { + response = await response; + validateResponse(response); + headersSent = true; + await writeResponse(response, socket, keepAlive, request.method === 'HEAD'); + completeResponse(socket, keepAlive, parser, bodyControl, true); + } catch (error) { + await handleHandlerError( + server, socket, request, keepAlive, parser, bodyControl, + error, headersSent, + ); + } +} + +async function finishStreamingResponse( + server, socket, request, keepAlive, parser, bodyControl, writing, +) { + try { + await writing; + completeResponse(socket, keepAlive, parser, bodyControl, true); + } catch (error) { + await handleHandlerError( + server, socket, request, keepAlive, parser, bodyControl, error, true, + ); } } @@ -326,68 +583,69 @@ async function writeResponse(response, socket, keepAlive) { * @param {Request} request * @param {boolean} keepAlive * @param {HTTPParser} parser - * @param {Function|null} setBodyCallbacks + * @param {object|null} bodyControl + * @returns {boolean} Whether response completion is asynchronous */ -async function invokeHandler(server, socket, request, keepAlive, parser, setBodyCallbacks) { - let headersSent = false; +function invokeHandler(server, socket, request, keepAlive, parser, bodyControl) { + if (bodyControl) { + bodyControl.setCallbacks(); + } else { + parser[kOnBody] = noop; + parser[kOnMessageComplete] = noop; + } try { - // Set up body stream callbacks if there's a body - if (setBodyCallbacks) { - setBodyCallbacks(); - } else { - // No body expected - set up callbacks to handle any unexpected body data - parser[kOnBody] = () => {}; - parser[kOnMessageComplete] = () => {}; - } - - // Call the handler - let response = server[kHandler](request); + const response = server[kHandler](request); if (response instanceof Promise) { - response = await response; + finishAsyncResponse( + server, socket, request, keepAlive, parser, bodyControl, response, + ); + return true; } - // Validate response - if (!(response instanceof Response)) { - throw new ERR_INVALID_ARG_TYPE('handler return value', 'Response', response); + validateResponse(response); + const writing = writeResponse( + response, socket, keepAlive, request.method === 'HEAD', + ); + if (writing instanceof Promise) { + finishStreamingResponse( + server, socket, request, keepAlive, parser, bodyControl, writing, + ); + return true; } - // Write response - headersSent = true; - await writeResponse(response, socket, keepAlive); - + completeResponse(socket, keepAlive, parser, bodyControl, false); + return false; } catch (error) { - // Handle errors - if (!headersSent) { - const onError = server[kOnError]; - let errorResponse; - - if (onError) { - try { - errorResponse = await onError(error, request); - } catch { - errorResponse = new Response('Internal Server Error', { status: 500 }); - } - } else { - errorResponse = new Response('Internal Server Error', { status: 500 }); - } - - try { - await writeResponse(errorResponse, socket, false); - } catch { - // Ignore write errors when sending error response - } - } - - // Emit error on server - server.emit('error', error); + handleHandlerError( + server, socket, request, keepAlive, parser, bodyControl, error, false, + ); + return true; } +} - // Resume parser for next request or close connection - if (keepAlive && !socket.destroyed) { - parser.resume(); - } else if (!socket.destroyed) { - socket.end(); +function closeParser(parser) { + parser.close(); +} + +function freeServeParser(parser, socket) { + if (socket.parser !== parser) return; + + if (parser._consumed) parser.unconsume(); + parser.remove(); + parser[kOnHeaders] = null; + parser[kOnHeadersComplete] = null; + parser[kOnBody] = null; + parser[kOnMessageComplete] = null; + parser[kOnExecute] = null; + parser[kOnTimeout] = null; + parser.socket = null; + socket.parser = null; + + if (serveParsers.free(parser)) { + parser.free(); + } else { + setImmediate(closeParser, parser); } } @@ -397,8 +655,8 @@ async function invokeHandler(server, socket, request, keepAlive, parser, setBody * @param {net.Socket|tls.TLSSocket} socket */ function handleConnection(server, socket) { - // Allocate and initialize parser - const parser = parsers.alloc(); + // Allocate and initialize a parser from the serve()-specific pool. + const parser = serveParsers.alloc(); const lenient = isLenient(); @@ -411,6 +669,7 @@ function handleConnection(server, socket) { ); parser.socket = socket; + parser._consumed = false; socket.parser = parser; // Track parser state @@ -444,11 +703,12 @@ function handleConnection(server, socket) { return 1; // Skip body parsing } - // Pause parser immediately - we control when to resume - parser.pause(); + // Create Request and invoke handler. + const { request, bodyControl } = createRequest(headers, method, url, socket, parser); - // Create Request and invoke handler - const { request, setCallbacks } = createRequest(headers, method, url, socket, parser); + // Track the in-flight request so a client disconnect before the response + // completes aborts request.signal and errors the body stream. + socket[kInFlight] = { request, bodyControl }; // Publish to diagnostics channel if (onRequestStartChannel.hasSubscribers) { @@ -459,8 +719,18 @@ function handleConnection(server, socket) { }); } - // Invoke handler asynchronously - invokeHandler(server, socket, request, shouldKeepAlive, parser, setCallbacks); + // Request bodies need the parser paused before the handler starts so a + // ReadableStream pull can resume it. Bodyless synchronous responses avoid + // the pause/resume cycle entirely. + if (bodyControl) parser.pause(); + const pending = invokeHandler( + server, socket, request, shouldKeepAlive, parser, bodyControl, + ); + if (pending) { + if (!bodyControl) parser.pause(); + } else if (bodyControl) { + parser.resume(); + } return 0; }; @@ -491,8 +761,31 @@ function handleConnection(server, socket) { // Socket event handlers socket.on('error', (err) => socketOnError(socket, server, err)); + socket.on('end', () => { + // The connection allows half-open so clients may send FIN and still + // receive their response, matching http.Server. + const inFlight = socket[kInFlight]; + if (inFlight !== undefined) { + // The client stopped sending: a request body that is still incomplete + // can never complete, so the request is aborted. A complete request + // just keeps waiting for its response. + if (inFlight.bodyControl !== null && !inFlight.bodyControl.isComplete()) { + socket[kInFlight] = undefined; + abortServeRequest(inFlight.request); + inFlight.bodyControl.abort(inFlight.request.signal.reason); + } + } else if (!socket.destroyed) { + socket.end(); + } + }); socket.on('close', () => { - freeParser(parser, null, socket); + freeServeParser(parser, socket); + const inFlight = socket[kInFlight]; + if (inFlight !== undefined) { + socket[kInFlight] = undefined; + abortServeRequest(inFlight.request); + inFlight.bodyControl?.abort(inFlight.request.signal.reason); + } }); } diff --git a/lib/internal/http_serve_classes.js b/lib/internal/http_serve_classes.js new file mode 100644 index 000000000000..b2e9be59c563 --- /dev/null +++ b/lib/internal/http_serve_classes.js @@ -0,0 +1,461 @@ +'use strict'; + +const { + JSONStringify, + NumberIsInteger, + ObjectAssign, + ObjectDefineProperty, + ObjectFreeze, + ObjectPrototypeHasOwnProperty, + PromisePrototypeThen, + RangeError, + StringPrototypeCharCodeAt, + Symbol, + TypeError, +} = primordials; + +const { + Headers, + Request, + Response, + serverKit, +} = require('internal/deps/undici/undici'); + +const { + kConstruct, + HeadersList, + fillHeaders, + setHeadersList, + setHeadersGuard, + setRequestState, + setRequestHeaders, + setRequestSignal, + getResponseState, + setResponseState, + setResponseHeaders, +} = serverKit; + +const { URL } = require('internal/url'); +const { ReadableStream } = require('internal/webstreams/readablestream'); +const { AbortController } = require('internal/abort_controller'); +const { Buffer } = require('buffer'); +const { isUint8Array } = require('internal/util/types'); + +// Sentinel that routes the subclass constructors to the uninitialized +// (kConstruct) base path. Never exported, so user code cannot create +// uninitialized instances. +const kInternalConstruct = Symbol('kInternalConstruct'); + +function defineOwnValue(object, key, value) { + ObjectDefineProperty(object, key, { + __proto__: null, + value, + writable: true, + enumerable: true, + configurable: true, + }); + return value; +} + +function noop() {} + +/** + * Inner request state for requests produced by serve(). Shape-compatible with + * undici's makeRequest() record: constant fields live on the prototype and + * the expensive ones (url, urlList) materialize on first access, so creating + * one costs four stores. Consumers that copy the record with an own-property + * spread (cloneRequest) are handled by NodeRequest.prototype.clone(). + */ +class ServerRequestState { + constructor(method, fullUrl, headersList, body) { + this.method = method; + this.fullUrl = fullUrl; + this.headersList = headersList; + this.body = body; + } + + get url() { + return defineOwnValue(this, 'url', new URL(this.fullUrl)); + } + + set url(value) { + defineOwnValue(this, 'url', value); + } + + get urlList() { + return defineOwnValue(this, 'urlList', [this.url]); + } + + set urlList(value) { + defineOwnValue(this, 'urlList', value); + } +} + +// Defaults mirror undici's makeRequest(), except mode: server requests were +// previously built through the public Request constructor, which sets 'cors'. +ObjectAssign(ServerRequestState.prototype, { + localURLsOnly: false, + unsafeRequest: false, + client: null, + reservedClient: null, + replacesClientId: '', + window: 'client', + keepalive: false, + serviceWorkers: 'all', + initiator: '', + destination: '', + priority: null, + origin: 'client', + policyContainer: 'client', + referrer: 'client', + referrerPolicy: '', + mode: 'cors', + useCORSPreflightFlag: false, + credentials: 'same-origin', + useCredentials: false, + cache: 'default', + redirect: 'follow', + integrity: '', + cryptoGraphicsNonceMetadata: '', + parserMetadata: '', + reloadNavigation: false, + historyNavigation: false, + userActivation: false, + taintedOrigin: false, + redirectCount: 0, + responseTainting: 'basic', + preventNoCacheCacheControlHeaderModification: false, + done: false, + timingAllowFailed: false, + traversableForUserPrompts: 'client', +}); + +let createServeRequest; +let getServeMetadata; +let abortServeRequest; + +/** + * A Request subclass whose serve() construction path skips the public + * constructor entirely: no WebIDL conversion, no URL parse, no header + * re-validation or copies, no AbortSignal until requested. Publicly + * constructed instances (`new NodeRequest(url, init)`) behave exactly like + * `new Request(url, init)`. + */ +class NodeRequest extends Request { + #state = undefined; + #meta = undefined; + #signalController = undefined; + #aborted = false; + + constructor(input, init = undefined) { + if (input === kInternalConstruct) { + super(kConstruct); + return; + } + super(input, init); + } + + get url() { + return this.#state === undefined ? super.url : this.#state.fullUrl; + } + + // The signal is created lazily: most handlers never observe it, and wiring + // an EventTarget per request is measurable. It is written through to the + // base class field so undici-internal reads observe it afterwards. The one + // observable gap: `new Request(nodeRequest)` copies the private field + // directly, so a request derived before .signal was ever accessed does not + // follow client disconnects. + get signal() { + if (this.#state === undefined) { + return super.signal; + } + let controller = this.#signalController; + if (controller === undefined) { + controller = new AbortController(); + this.#signalController = controller; + setRequestSignal(this, controller.signal); + if (this.#aborted) { + controller.abort(); + } + } + return controller.signal; + } + + get remoteAddress() { + return this.#meta?.remoteAddress; + } + + get remotePort() { + return this.#meta?.remotePort; + } + + get localAddress() { + return this.#meta?.localAddress; + } + + get localPort() { + return this.#meta?.localPort; + } + + get encrypted() { + return this.#meta === undefined ? false : this.#meta.encrypted; + } + + clone() { + const state = this.#state; + if (state !== undefined) { + // cloneRequest() copies the inner state with an own-property spread. + // Materialize the lazy fields and the one prototype default that + // diverges from makeRequest()'s so they survive the copy, and the + // signal so the clone's signal can follow this one. + state.urlList; // eslint-disable-line no-unused-expressions + defineOwnValue(state, 'mode', state.mode); + this.signal; // eslint-disable-line no-unused-expressions + } + return super.clone(); + } + + static { + createServeRequest = (method, fullUrl, headersList, body, meta) => { + const request = new NodeRequest(kInternalConstruct); + const state = new ServerRequestState(method, fullUrl, headersList, body); + setRequestState(request, state); + const headers = new Headers(kConstruct); + setHeadersList(headers, headersList); + setHeadersGuard(headers, 'immutable'); + setRequestHeaders(request, headers); + request.#state = state; + request.#meta = meta; + return request; + }; + + getServeMetadata = (request) => { + let meta; + if (typeof request === 'object' && request !== null) { + try { + meta = request.#meta; + } catch { + // Not a NodeRequest. + } + } + return meta; + }; + + abortServeRequest = (request) => { + const controller = request.#signalController; + if (controller !== undefined) { + controller.abort(); + } else { + request.#aborted = true; + } + }; + } +} + +/** + * Inner response state, shape-compatible with undici's makeResponse() record. + */ +class ServerResponseState { + constructor(status, statusText, headersList, body) { + this.status = status; + this.statusText = statusText; + this.headersList = headersList; + this.body = body; + } +} + +const kEmptyUrlList = ObjectFreeze([]); + +ObjectAssign(ServerResponseState.prototype, { + aborted: false, + rangeRequested: false, + timingAllowPassed: false, + requestIncludesCredentials: false, + type: 'default', + timingInfo: null, + cacheState: '', + urlList: kEmptyUrlList, +}); + +/** + * Body record for string/Uint8Array response bodies: keeps the original + * source and its byte length, and only materializes the ReadableStream if + * something actually asks for it (res.body, mixins, clone). writeResponse() + * writes the source directly and sets `used` instead. + */ +class LazyResponseBody { + used = false; + + constructor(source, length) { + this.source = source; + this.length = length; + } + + get stream() { + return defineOwnValue(this, 'stream', + createSourceStream(this.source, this.used)); + } + + set stream(value) { + defineOwnValue(this, 'stream', value); + } +} + +function hasMaterializedStream(body) { + return ObjectPrototypeHasOwnProperty(body, 'stream'); +} + +function createSourceStream(source, used) { + const stream = new ReadableStream({ + start(controller) { + if (!used) { + controller.enqueue( + typeof source === 'string' ? Buffer.from(source, 'utf8') : source, + ); + } + controller.close(); + }, + }); + if (used) { + // The body was already written to the socket: hand out a stream that is + // both closed and disturbed, matching a fully consumed body. + PromisePrototypeThen(stream.cancel(), noop, noop); + } + return stream; +} + +// https://fetch.spec.whatwg.org/#reason-phrase +function isValidReasonPhrase(statusText) { + for (let i = 0; i < statusText.length; ++i) { + const c = StringPrototypeCharCodeAt(statusText, i); + if (!(c === 0x09 || (c >= 0x20 && c <= 0x7e) || (c >= 0x80 && c <= 0xff))) { + return false; + } + } + return true; +} + +function isFastResponseInit(init) { + return init == null || + (typeof init === 'object' && + (init.status === undefined || + (typeof init.status === 'number' && NumberIsInteger(init.status))) && + (init.statusText === undefined || typeof init.statusText === 'string')); +} + +/** + * A Response subclass with a fast construction path for the bodies handlers + * actually return: null, strings and Uint8Arrays, combined with plain-object + * init. That path performs no WebIDL conversion and allocates no + * ReadableStream. Anything else falls back to the standard Response + * constructor, so behavior never diverges. Note that unlike Response, a + * Uint8Array body is not copied; do not mutate it after passing it in. + */ +class NodeResponse extends Response { + #state = undefined; + + constructor(body = null, init = undefined) { + if (body === kInternalConstruct) { + super(kConstruct); + return; + } + if ((body === null || typeof body === 'string' || isUint8Array(body)) && + isFastResponseInit(init)) { + super(kConstruct); + NodeResponse.#init(this, body, init, + typeof body === 'string' ? + 'text/plain;charset=UTF-8' : null); + return; + } + super(body, init); + } + + static #init(response, body, init, defaultContentType) { + let status = 200; + let statusText = ''; + let headersInit; + if (init != null) { + if (init.status !== undefined) { + status = init.status; + if (status < 200 || status > 599) { + // Plain errors match what the Response constructor throws. + // eslint-disable-next-line no-restricted-syntax + throw new RangeError( + 'init["status"] must be in the range of 200 to 599, inclusive.'); + } + } + if (init.statusText !== undefined) { + statusText = init.statusText; + if (!isValidReasonPhrase(statusText)) { + // eslint-disable-next-line no-restricted-syntax + throw new TypeError('Invalid statusText'); + } + } + headersInit = init.headers; + } + const headersList = new HeadersList(); + const state = new ServerResponseState(status, statusText, headersList, null); + setResponseState(response, state); + const headers = new Headers(kConstruct); + setHeadersList(headers, headersList); + setHeadersGuard(headers, 'response'); + setResponseHeaders(response, headers); + if (headersInit != null) { + fillHeaders(headers, headersInit); + } + if (body !== null) { + if (status === 204 || status === 205 || status === 304) { + // eslint-disable-next-line no-restricted-syntax + throw new TypeError('Response with null body status cannot have body'); + } + state.body = new LazyResponseBody( + body, + typeof body === 'string' ? Buffer.byteLength(body) : body.byteLength, + ); + if (defaultContentType !== null && + !headersList.contains('content-type', true)) { + headersList.append('content-type', defaultContentType, true); + } + } + response.#state = state; + } + + get bodyUsed() { + const state = this.#state; + if (state !== undefined && state.body !== null && + !hasMaterializedStream(state.body)) { + return state.body.used; + } + return super.bodyUsed; + } + + static json(data, init = undefined) { + if (!isFastResponseInit(init)) { + return Response.json(data, init); + } + const text = JSONStringify(data); + if (typeof text !== 'string') { + // eslint-disable-next-line no-restricted-syntax + throw new TypeError('The data is not JSON serializable'); + } + const response = new NodeResponse(kInternalConstruct); + NodeResponse.#init(response, text, init, 'application/json'); + return response; + } +} + +function isLazyResponseBody(body) { + return body instanceof LazyResponseBody; +} + +module.exports = { + NodeRequest, + NodeResponse, + createServeRequest, + getServeMetadata, + abortServeRequest, + getResponseState, + HeadersList, + isLazyResponseBody, + hasMaterializedStream, +}; diff --git a/test/parallel/test-http-serve-basic.js b/test/parallel/test-http-serve-basic.js index 4c66e1cd9820..72f1e0717517 100644 --- a/test/parallel/test-http-serve-basic.js +++ b/test/parallel/test-http-serve-basic.js @@ -28,6 +28,8 @@ assert.strictEqual(typeof http.getRemoteMetadata, 'function'); client.on('data', (chunk) => data += chunk); client.on('end', common.mustCall(() => { assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /content-length: 11/i); + assert.doesNotMatch(data, /transfer-encoding/i); assert.match(data, /Hello World/); server.close(); })); @@ -243,6 +245,29 @@ assert.strictEqual(typeof http.getRemoteMetadata, 'function'); })); } +// Test HEAD responses omit body bytes while retaining the representation length. +{ + const server = http.serve({}, common.mustCall(() => { + return new Response('HEAD body'); + })); + + server.listen(0, common.mustCall(() => { + const client = net.createConnection(server.address().port, common.mustCall(() => { + client.write('HEAD / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + const { 0: head, 1: body } = data.split('\r\n\r\n'); + assert.match(head, /content-length: 9/i); + assert.strictEqual(body, ''); + server.close(); + })); + })); +} + // Test handler as first argument (convenience API) { const server = http.serve(common.mustCall((request) => { diff --git a/test/parallel/test-http-serve-classes.js b/test/parallel/test-http-serve-classes.js new file mode 100644 index 000000000000..abc049dbe74c --- /dev/null +++ b/test/parallel/test-http-serve-classes.js @@ -0,0 +1,319 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); +const net = require('net'); + +// NodeRequest and NodeResponse are exported. +assert.strictEqual(typeof http.NodeRequest, 'function'); +assert.strictEqual(typeof http.NodeResponse, 'function'); +assert.ok(http.NodeRequest.prototype instanceof Request); +assert.ok(http.NodeResponse.prototype instanceof Response); + +// Publicly constructed instances behave like Request/Response. +{ + const request = new http.NodeRequest('http://example.org/a?b=c', { + method: 'POST', + body: 'hello', + duplex: 'half', + headers: { 'x-test': '1' }, + }); + assert.ok(request instanceof Request); + assert.strictEqual(request.method, 'POST'); + assert.strictEqual(request.url, 'http://example.org/a?b=c'); + assert.strictEqual(request.headers.get('x-test'), '1'); + assert.strictEqual(request.remoteAddress, undefined); + assert.strictEqual(request.encrypted, false); + request.text().then(common.mustCall((body) => { + assert.strictEqual(body, 'hello'); + })); +} + +// NodeResponse fast path: string body. +(async () => { + const response = new http.NodeResponse('hello world', { status: 201 }); + assert.ok(response instanceof Response); + assert.strictEqual(response.status, 201); + assert.strictEqual(response.ok, true); + assert.strictEqual(response.statusText, ''); + assert.strictEqual(response.type, 'default'); + assert.strictEqual(response.url, ''); + assert.strictEqual(response.headers.get('content-type'), + 'text/plain;charset=UTF-8'); + assert.strictEqual(response.bodyUsed, false); + assert.strictEqual(await response.text(), 'hello world'); + assert.strictEqual(response.bodyUsed, true); +})().then(common.mustCall()); + +// NodeResponse fast path: Uint8Array body, no default content-type. +(async () => { + const bytes = new Uint8Array([1, 2, 3]); + const response = new http.NodeResponse(bytes); + assert.strictEqual(response.headers.get('content-type'), null); + assert.deepStrictEqual(new Uint8Array(await response.arrayBuffer()), bytes); +})().then(common.mustCall()); + +// NodeResponse fast path: headers init and explicit content-type. +{ + const response = new http.NodeResponse('x', { + headers: { 'content-type': 'text/csv', 'x-a': 'b' }, + }); + assert.strictEqual(response.headers.get('content-type'), 'text/csv'); + assert.strictEqual(response.headers.get('x-a'), 'b'); +} + +// NodeResponse.json(). +(async () => { + const response = http.NodeResponse.json({ a: 1 }); + assert.strictEqual(response.headers.get('content-type'), 'application/json'); + assert.deepStrictEqual(await response.json(), { a: 1 }); + assert.throws(() => http.NodeResponse.json(undefined), TypeError); +})().then(common.mustCall()); + +// NodeResponse.clone() before the body is consumed. +(async () => { + const response = new http.NodeResponse('dup'); + const clone = response.clone(); + assert.strictEqual(await response.text(), 'dup'); + assert.strictEqual(await clone.text(), 'dup'); +})().then(common.mustCall()); + +// NodeResponse validation matches Response. +{ + assert.throws(() => new http.NodeResponse(null, { status: 199 }), RangeError); + assert.throws(() => new http.NodeResponse(null, { status: 600 }), RangeError); + assert.throws(() => new http.NodeResponse('x', { status: 204 }), TypeError); + assert.throws(() => new http.NodeResponse('x', { statusText: 'bad\r\n' }), + TypeError); + // Null body statuses without body are fine. + assert.strictEqual(new http.NodeResponse(null, { status: 204 }).status, 204); +} + +// Exotic bodies fall back to the standard Response construction path. +(async () => { + const params = new URLSearchParams({ a: 'b' }); + const response = new http.NodeResponse(params); + assert.match(response.headers.get('content-type'), + /application\/x-www-form-urlencoded/); + assert.strictEqual(await response.text(), 'a=b'); +})().then(common.mustCall()); + +// Fetch class instances cannot be structured-cloned. +{ + assert.throws(() => structuredClone(new http.NodeResponse('x')), + { name: 'DataCloneError' }); +} + +// The handler receives a NodeRequest with metadata getters, an absolute URL, +// combined headers and an immutable Headers object. +{ + const server = http.serve({}, common.mustCall((request) => { + assert.ok(request instanceof http.NodeRequest); + assert.ok(request instanceof Request); + assert.strictEqual(request.method, 'GET'); + assert.strictEqual(request.url, 'http://localhost/test?x=1'); + // URL is parseable and normalized on demand. + assert.strictEqual(new URL(request.url).searchParams.get('x'), '1'); + // Duplicate headers are combined like the public Headers API does. + assert.strictEqual(request.headers.get('x-dup'), 'a, b'); + assert.strictEqual(request.headers.get('cookie'), 'k=1; j=2'); + // Incoming request headers are immutable, matching fetch events on + // other server runtimes. + assert.throws(() => request.headers.set('x-dup', 'c'), TypeError); + // Socket metadata is available directly on the request. + assert.strictEqual(typeof request.remoteAddress, 'string'); + assert.strictEqual(typeof request.remotePort, 'number'); + assert.strictEqual(request.encrypted, false); + const meta = http.getRemoteMetadata(request); + assert.strictEqual(meta.remoteAddress, request.remoteAddress); + return new http.NodeResponse('ok'); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET /test?x=1 HTTP/1.1\r\nHost: localhost\r\n' + + 'X-Dup: a\r\nX-Dup: b\r\n' + + 'Cookie: k=1\r\nCookie: j=2\r\n' + + 'Connection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /content-length: 2/i); + assert.match(data, /\r\n\r\nok$/); + server.close(); + })); + })); +} + +// A NodeRequest can be cloned and wrapped in a plain Request. +{ + const server = http.serve({}, common.mustCall(async (request) => { + const clone = request.clone(); + assert.strictEqual(clone.method, 'POST'); + assert.strictEqual(clone.url, request.url); + assert.strictEqual(clone.mode, request.mode); + assert.strictEqual(clone.headers.get('content-type'), 'text/plain'); + + // Wrapping transfers the original request's body to the new Request. + const wrapped = new Request(request, { headers: { 'x-b': '2' } }); + assert.strictEqual(wrapped.url, request.url); + assert.strictEqual(wrapped.headers.get('x-b'), '2'); + + const [first, second] = + await Promise.all([wrapped.text(), clone.text()]); + assert.strictEqual(first, 'ping'); + assert.strictEqual(second, 'ping'); + return new http.NodeResponse(first); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('POST / HTTP/1.1\r\nHost: localhost\r\n' + + 'Content-Type: text/plain\r\nContent-Length: 4\r\n' + + 'Connection: close\r\n\r\nping'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /\r\n\r\nping$/); + server.close(); + })); + })); +} + +// Multiple Set-Cookie response headers are written as separate lines. +{ + const server = http.serve({}, common.mustCall((request) => { + const response = new http.NodeResponse('c'); + response.headers.append('set-cookie', 'a=1'); + response.headers.append('set-cookie', 'b=2; Path=/'); + return response; + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /set-cookie: a=1\r\n/); + assert.match(data, /set-cookie: b=2; Path=\/\r\n/); + assert.doesNotMatch(data, /set-cookie: a=1, b=2/); + server.close(); + })); + })); +} + +// request.signal aborts and the body stream errors when the client +// disconnects before the request body is complete. +{ + let client; + const server = http.serve({}, common.mustCall((request) => { + assert.strictEqual(request.signal.aborted, false); + request.signal.addEventListener('abort', common.mustCall()); + client.destroy(); + return request.text().then(common.mustNotCall(), common.mustCall((err) => { + assert.strictEqual(err.name, 'AbortError'); + assert.strictEqual(request.signal.aborted, true); + server.close(); + return new http.NodeResponse(null); + })); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + client = net.createConnection(port, common.mustCall(() => { + client.write('POST / HTTP/1.1\r\nHost: localhost\r\n' + + 'Content-Length: 10\r\n\r\nabc'); + })); + })); +} + +// Response.error() cannot be serialized and turns into a 500. +{ + const server = http.serve({}, common.mustCall((request) => Response.error())); + server.on('error', common.mustCall()); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 500/); + server.close(); + })); + })); +} + +// Echoing the request body through a Response streams it back. +{ + const server = http.serve({}, common.mustCall((request) => { + return new Response(request.body); + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('POST / HTTP/1.1\r\nHost: localhost\r\n' + + 'Content-Length: 4\r\nConnection: close\r\n\r\necho'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 200/); + assert.match(data, /echo/); + server.close(); + })); + })); +} + +// A structurally fetch-compatible response from a different Response class +// (e.g. a framework bundling its own undici) is serialized via its public API. +{ + const server = http.serve({}, common.mustCall((request) => { + return { + status: 203, + statusText: '', + headers: new Headers({ 'x-foreign': 'yes' }), + body: null, + bodyUsed: false, + }; + })); + + server.listen(0, common.mustCall(() => { + const port = server.address().port; + const client = net.createConnection(port, common.mustCall(() => { + client.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /HTTP\/1\.1 203/); + assert.match(data, /x-foreign: yes/); + server.close(); + })); + })); +} diff --git a/test/parallel/test-http-serve-parser-pool.js b/test/parallel/test-http-serve-parser-pool.js new file mode 100644 index 000000000000..a0e9fa2c7cde --- /dev/null +++ b/test/parallel/test-http-serve-parser-pool.js @@ -0,0 +1,38 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); +const net = require('net'); + +const serveServer = http.serve(common.mustCall(() => new Response('serve'))); + +serveServer.listen(0, common.mustCall(() => { + const client = net.createConnection(serveServer.address().port, common.mustCall(() => { + client.end('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + client.resume(); + client.on('end', common.mustCall(() => { + serveServer.close(common.mustCall(runCreateServer)); + })); +})); + +function runCreateServer() { + const server = http.createServer(common.mustCall((request, response) => { + response.end('createServer'); + })); + + server.listen(0, common.mustCall(() => { + const client = net.createConnection(server.address().port, common.mustCall(() => { + client.end('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n'); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.match(data, /createServer/); + server.close(); + })); + })); +} diff --git a/test/parallel/test-http-serve-unread-body.js b/test/parallel/test-http-serve-unread-body.js new file mode 100644 index 000000000000..e7853a456227 --- /dev/null +++ b/test/parallel/test-http-serve-unread-body.js @@ -0,0 +1,36 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); +const net = require('net'); + +let requests = 0; +const server = http.serve(common.mustCall((request) => { + requests++; + return new Response(new URL(request.url).pathname); +}, 2)); + +server.listen(0, common.mustCall(() => { + const client = net.createConnection(server.address().port, common.mustCall(() => { + client.write( + 'POST /one HTTP/1.1\r\n' + + 'Host: localhost\r\n' + + 'Content-Length: 4\r\n\r\n' + + 'body' + + 'GET /two HTTP/1.1\r\n' + + 'Host: localhost\r\n' + + 'Connection: close\r\n\r\n', + ); + })); + + let data = ''; + client.setEncoding('utf8'); + client.on('data', (chunk) => data += chunk); + client.on('end', common.mustCall(() => { + assert.strictEqual(requests, 2); + assert.match(data, /\/one/); + assert.match(data, /\/two/); + server.close(); + })); +}));