diff --git a/benchmarks/fetch/response-creation.mjs b/benchmarks/fetch/response-creation.mjs new file mode 100644 index 00000000000..a9ff6bff76b --- /dev/null +++ b/benchmarks/fetch/response-creation.mjs @@ -0,0 +1,8 @@ +import { bench, run } from 'mitata' +import { Response } from '../../lib/web/fetch/response.js' + +bench('new Response()', () => new Response()) +bench('new Response(body)', () => new Response('hello')) +bench('new Response(null, { status: 201 })', () => new Response(null, { status: 201 })) + +await run() diff --git a/lib/dispatcher/client-h1.js b/lib/dispatcher/client-h1.js index 9f6f17c1579..bab9835d12c 100644 --- a/lib/dispatcher/client-h1.js +++ b/lib/dispatcher/client-h1.js @@ -817,13 +817,24 @@ class Parser { // have been queued since then. util.destroy(socket, new InformationalError('reset')) return constants.ERROR.PAUSED - } else if (client[kPipelining] == null || client[kPipelining] === 1) { - // We must wait a full event loop cycle to reuse this socket to make sure - // that non-spec compliant servers are not closing the connection even if they - // said they won't. - setImmediate(client[kResume]) } else { - client[kResume]() + // Arm idle-socket validation as soon as the socket becomes idle so + // the event-loop yield (GHSA-35p6-xmwp-9g52) runs during body + // consumption instead of after the next dispatch. Do not mark the + // client busy here — BalancedPool uses busy() to pick an upstream, + // and a background timer must not change weighted-round-robin. + if (socket[kSocketUsed]) { + armIdleSocketValidation(client, socket) + } + + if (client[kPipelining] == null || client[kPipelining] === 1) { + // We must wait a full event loop cycle to reuse this socket to make sure + // that non-spec compliant servers are not closing the connection even if they + // said they won't. + setImmediate(client[kResume]) + } else { + client[kResume]() + } } return 0 @@ -1059,8 +1070,11 @@ function clearIdleSocketValidation (socket) { socket[kIdleSocketValidation] = 0 } -function scheduleIdleSocketValidation (client, socket) { - socket[kIdleSocketValidation] = 1 +function armIdleSocketValidation (client, socket) { + if (socket[kIdleSocketValidationTimeout] || socket[kIdleSocketValidation] === 2) { + return + } + socket[kIdleSocketValidationTimeout] = setTimeout(() => { socket[kIdleSocketValidationTimeout] = null socket[kIdleSocketValidation] = 2 @@ -1072,6 +1086,12 @@ function scheduleIdleSocketValidation (client, socket) { socket[kIdleSocketValidationTimeout].unref?.() } +function scheduleIdleSocketValidation (client, socket) { + // Block further writes on this socket until the poll yield completes. + socket[kIdleSocketValidation] = 1 + armIdleSocketValidation(client, socket) +} + /** * @param {import('./client.js')} client */ diff --git a/lib/web/fetch/body.js b/lib/web/fetch/body.js index a81ddfba437..f022acc5ef6 100644 --- a/lib/web/fetch/body.js +++ b/lib/web/fetch/body.js @@ -446,6 +446,17 @@ function mixinBody (prototype, getInternalState) { Object.assign(prototype.prototype, bodyMixinMethods(prototype, getInternalState)) } +/** + * True when the body was extracted from a string or bytes and can be + * sent without incrementally reading its Web ReadableStream. + * + * @param {unknown} source + * @returns {source is string | Uint8Array} + */ +function isBufferedBodySource (source) { + return typeof source === 'string' || isUint8Array(source) +} + /** * @see https://fetch.spec.whatwg.org/#concept-body-consume-body * @param {any} object internal state @@ -543,5 +554,6 @@ module.exports = { cloneBody, mixinBody, streamRegistry, - bodyUnusable + bodyUnusable, + isBufferedBodySource } diff --git a/lib/web/fetch/index.js b/lib/web/fetch/index.js index 935bc9d4c90..8551f7d8b02 100644 --- a/lib/web/fetch/index.js +++ b/lib/web/fetch/index.js @@ -48,7 +48,7 @@ const { isTraversableNavigable } = require('./util') const assert = require('node:assert') -const { safelyExtractBody, extractBody } = require('./body') +const { safelyExtractBody, extractBody, isBufferedBodySource } = require('./body') const { redirectStatusSet, nullBodyStatus, @@ -1435,7 +1435,14 @@ async function httpNetworkOrCacheFetch ( // when request’s body’s source is null as only a single body is needed in // that case. E.g., when request’s body’s source is null, redirects and // authentication will end up failing the fetch. - if (request.body?.source != null) { + // + // When source is a string / bytes we send those bytes directly, so teeing + // the companion Web Stream is wasted work (and extractBody already kept + // the source for this reason). + if (isBufferedBodySource(request.body?.source)) { + httpRequest = cloneRequest({ ...request, body: null }) + httpRequest.body = request.body + } else if (request.body?.source != null) { httpRequest = cloneRequest(request) } else { httpRequest = cloneRequest({ ...request, body: null }) @@ -1935,6 +1942,23 @@ async function httpNetworkFetch ( // end-of-body and fetchParams’s task destination. if (request.body == null && fetchParams.processRequestEndOfBody) { queueMicrotask(() => fetchParams.processRequestEndOfBody()) + } else if (isBufferedBodySource(request.body?.source)) { + // The body was extracted from a string or BufferSource. Pass those + // bytes to the dispatcher instead of incrementally reading the Web + // Stream that extractBody() built only to satisfy the Body mixin. + requestBody = request.body.source + if (fetchParams.processRequestEndOfBody || fetchParams.processRequestBodyChunkLength) { + const length = typeof requestBody === 'string' + ? Buffer.byteLength(requestBody) + : requestBody.byteLength + queueMicrotask(() => { + if (isCancelled(fetchParams)) { + return + } + fetchParams.processRequestBodyChunkLength?.(length) + fetchParams.processRequestEndOfBody?.() + }) + } } else if (request.body != null) { // 2. Otherwise, if body is non-null: diff --git a/lib/web/fetch/response.js b/lib/web/fetch/response.js index f555ea94b15..2e407ba7ade 100644 --- a/lib/web/fetch/response.js +++ b/lib/web/fetch/response.js @@ -24,6 +24,24 @@ const { isomorphicEncode, serializeJavascriptValueToJSONString } = require('../i const textEncoder = new TextEncoder('utf-8') +/** + * Fetch's ResponseInit dictionary defaults `status` to 200 and + * `statusText` to "". Those defaults must not count as + * user-specified members — otherwise every `new Response()` / + * `new Response(body)` pays WebIDL conversion and reason-phrase + * validation for values `makeResponse()` already has. + * + * @param {object | null | undefined} init + * @returns {boolean} + */ +function responseInitHasUserMembers (init) { + return init != null && ( + init.status !== undefined || + init.statusText !== undefined || + init.headers !== undefined + ) +} + // https://fetch.spec.whatwg.org/#response-class class Response { /** @type {Headers} */ @@ -118,7 +136,13 @@ class Response { body = webidl.converters.BodyInit(body, 'Response', 'body') } - init = webidl.converters.ResponseInit(init) + // Capture this before WebIDL conversion fills dictionary defaults. + const initHasKey = responseInitHasUserMembers(init) + // Non-object init (e.g. `new Response(null, 0)`) must still go through + // the converter so it throws. Empty `{}` / omitted init can skip it. + if (initHasKey || (init != null && typeof init !== 'object')) { + init = webidl.converters.ResponseInit(init) + } // 1. Set this’s response to a new response. this.#state = makeResponse({}) @@ -140,7 +164,10 @@ class Response { } // 5. Perform initialize a response given this, init, and bodyWithType. - initializeResponse(this, init, bodyWithType) + // makeResponse() already has status 200 / statusText "". + if (initHasKey || bodyWithType) { + initializeResponse(this, initHasKey ? init : {}, bodyWithType) + } } // Returns response’s type, e.g., "cors". diff --git a/test/fetch/buffered-body.js b/test/fetch/buffered-body.js new file mode 100644 index 00000000000..9f71c61bf1b --- /dev/null +++ b/test/fetch/buffered-body.js @@ -0,0 +1,83 @@ +'use strict' + +const { test } = require('node:test') +const { createServer } = require('node:http') +const { once } = require('node:events') +const { Request } = require('../..') +const { fetching } = require('../../lib/web/fetch/index.js') +const { getRequestState } = require('../../lib/web/fetch/request.js') +const { closeServerAsPromise } = require('../utils/node-http') + +async function fetchWithBodyHooks (url, body, hooks) { + const requestObject = new Request(url, { method: 'POST', body }) + const request = getRequestState(requestObject) + + await new Promise((resolve, reject) => { + fetching({ + request, + requestObject, + processRequestBodyChunkLength: hooks.processRequestBodyChunkLength, + processRequestEndOfBody: hooks.processRequestEndOfBody, + processResponse (response) { + if (response.type === 'error') { + reject(response.error) + return + } + resolve(response) + } + }) + }) + + // The buffered-body path reports length on a microtask. + await new Promise((resolve) => queueMicrotask(resolve)) +} + +test('buffered string body reports processRequest callbacks', async (t) => { + const server = createServer((req, res) => { + req.resume() + req.on('end', () => res.end('ok')) + }).listen(0) + t.after(closeServerAsPromise(server)) + await once(server, 'listening') + + const body = 'hello' + let chunkLength = 0 + let ended = false + + await fetchWithBodyHooks(`http://127.0.0.1:${server.address().port}`, body, { + processRequestBodyChunkLength (n) { + chunkLength += n + }, + processRequestEndOfBody () { + ended = true + } + }) + + t.assert.strictEqual(chunkLength, Buffer.byteLength(body)) + t.assert.strictEqual(ended, true) +}) + +test('buffered Uint8Array body reports processRequest callbacks', async (t) => { + const server = createServer((req, res) => { + req.resume() + req.on('end', () => res.end('ok')) + }).listen(0) + t.after(closeServerAsPromise(server)) + await once(server, 'listening') + + const body = new TextEncoder().encode('hello') + let chunkLength = 0 + let ended = false + + await fetchWithBodyHooks(`http://127.0.0.1:${server.address().port}`, body, { + processRequestBodyChunkLength (n) { + chunkLength += n + }, + processRequestEndOfBody () { + ended = true + } + }) + + t.assert.strictEqual(chunkLength, body.byteLength) + t.assert.strictEqual(ended, true) +}) diff --git a/test/fetch/client-fetch.js b/test/fetch/client-fetch.js index 81a8bb4028a..526e487ffb1 100644 --- a/test/fetch/client-fetch.js +++ b/test/fetch/client-fetch.js @@ -711,3 +711,45 @@ test('Receiving non-Latin1 headers', async (t) => { t.assert.deepStrictEqual(cdHeaders, ContentDisposition) t.assert.deepStrictEqual(lengths, [30, 34, 94, 104, 90]) }) + +test('POST string body is delivered without reading the request stream', async (t) => { + const server = createServer({ joinDuplicateHeaders: true }, (req, res) => { + const chunks = [] + req.on('data', chunk => chunks.push(chunk)) + req.on('end', () => { + t.assert.strictEqual(Buffer.concat(chunks).toString(), '{"hello":"world"}') + t.assert.strictEqual(req.headers['content-type'], 'text/plain;charset=UTF-8') + res.end('ok') + }) + }).listen(0) + + t.after(closeServerAsPromise(server)) + await once(server, 'listening') + + const response = await fetch(`http://127.0.0.1:${server.address().port}`, { + method: 'POST', + body: '{"hello":"world"}' + }) + t.assert.strictEqual(await response.text(), 'ok') +}) + +test('POST Uint8Array body is delivered without reading the request stream', async (t) => { + const payload = new TextEncoder().encode('{"hello":"bytes"}') + const server = createServer({ joinDuplicateHeaders: true }, (req, res) => { + const chunks = [] + req.on('data', chunk => chunks.push(chunk)) + req.on('end', () => { + t.assert.deepStrictEqual(Buffer.concat(chunks), Buffer.from(payload)) + res.end('ok') + }) + }).listen(0) + + t.after(closeServerAsPromise(server)) + await once(server, 'listening') + + const response = await fetch(`http://127.0.0.1:${server.address().port}`, { + method: 'POST', + body: payload + }) + t.assert.strictEqual(await response.text(), 'ok') +}) diff --git a/test/fetch/response.js b/test/fetch/response.js index 5c7ce840d87..f99cb500e14 100644 --- a/test/fetch/response.js +++ b/test/fetch/response.js @@ -116,6 +116,53 @@ test('response clone', (t) => { t.assert.strictEqual(response2.body, null) }) +// Dictionary defaults (status: 200, statusText: "") must not force +// initialize-a-response work on every `new Response()` / `new Response(body)`. +test('omitted ResponseInit uses makeResponse defaults', (t) => { + const res = new Response() + t.assert.strictEqual(res.status, 200) + t.assert.strictEqual(res.statusText, '') + t.assert.strictEqual(res.ok, true) + t.assert.deepStrictEqual([...res.headers], []) +}) + +test('empty ResponseInit object is treated as omitted', (t) => { + const res = new Response(null, {}) + t.assert.strictEqual(res.status, 200) + t.assert.strictEqual(res.statusText, '') + t.assert.deepStrictEqual([...res.headers], []) +}) + +test('explicit ResponseInit members are still applied', (t) => { + const res = new Response(null, { status: 201, statusText: 'Created', headers: { 'x-a': '1' } }) + t.assert.strictEqual(res.status, 201) + t.assert.strictEqual(res.statusText, 'Created') + t.assert.deepStrictEqual([...res.headers], [['x-a', '1']]) +}) + +test('string body consume marks bodyUsed', async (t) => { + const res = new Response('hello') + t.assert.strictEqual(res.bodyUsed, false) + t.assert.strictEqual(res.headers.get('content-type'), 'text/plain;charset=UTF-8') + t.assert.strictEqual(await res.text(), 'hello') + t.assert.strictEqual(res.bodyUsed, true) + await t.assert.rejects(res.text(), TypeError) +}) + +test('cloned string body can be consumed independently', async (t) => { + const res = new Response('hello') + const clone = res.clone() + t.assert.strictEqual(await Promise.all([res.text(), clone.text()]).then((v) => v.join(',')), 'hello,hello') +}) + +test('Uint8Array body consume', async (t) => { + const res = new Response(new TextEncoder().encode('{"a":1}'), { + headers: { 'content-type': 'application/json' } + }) + t.assert.deepStrictEqual(await res.json(), { a: 1 }) + t.assert.strictEqual(res.bodyUsed, true) +}) + test('Symbol.toStringTag', (t) => { const resp = new Response() diff --git a/test/node-test/keep-alive-reuse.js b/test/node-test/keep-alive-reuse.js index 5a92f6003c0..a4a059461b7 100644 --- a/test/node-test/keep-alive-reuse.js +++ b/test/node-test/keep-alive-reuse.js @@ -4,7 +4,7 @@ const { test } = require('node:test') const assert = require('node:assert') const { createServer } = require('node:http') const { once } = require('node:events') -const { Pool } = require('../..') +const { Agent, Pool, fetch } = require('../..') // Regression for #5600 / #5606: // Reusing an idle keep-alive socket must not stall behind the poll phase. @@ -66,3 +66,48 @@ test('reusing an idle keep-alive socket must not stall', { timeout: 1000 }, asyn assert.strictEqual(connections, 1, 'keep-alive socket must be reused') } }) + +test('fetch reusing an idle keep-alive socket must not stall', { timeout: 1000 }, async (t) => { + let connections = 0 + + const server = createServer((req, res) => { + res.writeHead(200, { 'content-length': 2 }) + res.end('ok') + }) + + server.on('connection', () => { + connections++ + }) + + server.listen(0) + await once(server, 'listening') + + const agent = new Agent({ + connections: 1 + }) + + t.after(async () => { + await agent.close() + server.close() + }) + + const url = `http://127.0.0.1:${server.address().port}` + + { + const res = await fetch(`${url}/0`, { dispatcher: agent }) + assert.strictEqual(await res.text(), 'ok') + } + assert.strictEqual(connections, 1) + + for (let i = 1; i <= REUSES; i++) { + const requested = once(server, 'request') + const resPromise = fetch(`${url}/${i}`, { dispatcher: agent }) + resPromise.catch(() => {}) + + await requested + + const res = await resPromise + assert.strictEqual(await res.text(), 'ok') + assert.strictEqual(connections, 1, 'keep-alive socket must be reused') + } +})