perf(fetch): skip empty ResponseInit work and send buffered bodies directly

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()

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
  */

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
 }

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:
 

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".

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)
+})

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')
+})

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()