extractBody fixes

lib/web/fetch/body.js

@@ -11,7 +11,7 @@ const { FormData, setFormDataState } = require('./formdata')
 const { webidl } = require('../webidl')
 const assert = require('node:assert')
 const { isErrored, isDisturbed } = require('node:stream')
-const { isArrayBuffer } = require('node:util/types')
+const { isUint8Array } = require('node:util/types')
 const { serializeAMimeType } = require('./data-url')
 const { multipartFormDataParser } = require('./formdata-parser')
 const { createDeferredPromise } = require('../../util/promise')
@@ -45,6 +45,7 @@ const streamRegistry = new FinalizationRegistry((weakRef) => {
 function extractBody (object, keepalive = false) {
   // 1. Let stream be null.
   let stream = null
+  let controller = null
 
   // 2. If object is a ReadableStream object, then set stream to object.
   if (webidl.is.ReadableStream(object)) {
@@ -57,16 +58,11 @@ function extractBody (object, keepalive = false) {
     // 4. Otherwise, set stream to a new ReadableStream object, and set
     //    up stream with byte reading support.
     stream = new ReadableStream({
-      pull (controller) {
-        const buffer = typeof source === 'string' ? textEncoder.encode(source) : source
-
-        if (buffer.byteLength) {
-          controller.enqueue(buffer)
-        }
-
-        queueMicrotask(() => readableStreamClose(controller))
+      pull () {},
+      start (c) {
+        controller = c
       },
-      start () {},
+      cancel () {},
       type: 'bytes'
     })
   }
@@ -108,9 +104,8 @@ function extractBody (object, keepalive = false) {
     // Set type to `application/x-www-form-urlencoded;charset=UTF-8`.
     type = 'application/x-www-form-urlencoded;charset=UTF-8'
   } else if (webidl.is.BufferSource(object)) {
-    source = isArrayBuffer(object)
-      ? new Uint8Array(object.slice())
-      : new Uint8Array(object.buffer.slice(object.byteOffset, object.byteOffset + object.byteLength))
+    // Set source to a copy of the bytes held by object.
+    source = webidl.util.getCopyOfBytesHeldByBufferSource(object)
   } else if (webidl.is.FormData(object)) {
     const boundary = `----formdata-undici-0${`${random(1e11)}`.padStart(11, '0')}`
     const prefix = `--${boundary}\r\nContent-Disposition: form-data`
@@ -213,45 +208,36 @@ function extractBody (object, keepalive = false) {
 
   // 11. If source is a byte sequence, then set action to a
   // step that returns source and length to source’s length.
-  if (typeof source === 'string' || util.isBuffer(source)) {
-    length = Buffer.byteLength(source)
+  if (typeof source === 'string' || isUint8Array(source)) {
+    action = () => {
+      length = typeof source === 'string' ? Buffer.byteLength(source) : source.length
+      return source
+    }
   }
 
-  // 12. If action is non-null, then run these steps in in parallel:
+  // 12. If action is non-null, then run these steps in parallel:
   if (action != null) {
-    // Run action.
-    let iterator
-    stream = new ReadableStream({
-      start () {
-        iterator = action(object)[Symbol.asyncIterator]()
-      },
-      pull (controller) {
-        return iterator.next().then(({ value, done }) => {
-          if (done) {
-            // When running action is done, close stream.
-            queueMicrotask(() => {
-              controller.close()
-              controller.byobRequest?.respond(0)
-            })
-          } else {
-            // Whenever one or more bytes are available and stream is not errored,
-            // enqueue a Uint8Array wrapping an ArrayBuffer containing the available
-            // bytes into stream.
-            if (!isErrored(stream)) {
-              const buffer = new Uint8Array(value)
-              if (buffer.byteLength) {
-                controller.enqueue(buffer)
-              }
-            }
+    ;(async () => {
+      // 1. Run action.
+      const result = action()
+
+      // 2. Whenever one or more bytes are available and stream is not errored,
+      //    enqueue the result of creating a Uint8Array from the available bytes into stream.
+      const iterator = result?.[Symbol.asyncIterator]?.()
+      if (iterator) {
+        for await (const bytes of iterator) {
+          if (isErrored(stream)) break
+          if (bytes.length) {
+            controller.enqueue(new Uint8Array(bytes))
           }
-          return controller.desiredSize > 0
-        })
-      },
-      cancel (reason) {
-        return iterator.return()
-      },
-      type: 'bytes'
-    })
+        }
+      } else if (result?.length && !isErrored(stream)) {
+        controller.enqueue(typeof result === 'string' ? textEncoder.encode(result) : new Uint8Array(result))
+      }
+
+      // 3. When running action is done, close stream.
+      queueMicrotask(() => readableStreamClose(controller))
+    })()
   }
 
   // 13. Let body be a body whose stream is stream, source is source,

lib/web/fetch/index.js

@@ -1321,8 +1321,8 @@ function httpRedirectFetch (fetchParams, response) {
     request.headersList.delete('host', true)
   }
 
-  // 14. If request’s body is non-null, then set request’s body to the first return
-  // value of safely extracting request’s body’s source.
+  // 14. If request's body is non-null, then set request's body to the first return
+  // value of safely extracting request's body's source.
   if (request.body != null) {
     assert(request.body.source != null)
     request.body = safelyExtractBody(request.body.source)[0]

lib/web/webidl/index.js

@@ -1,5 +1,6 @@
 'use strict'
 
+const assert = require('node:assert')
 const { types, inspect } = require('node:util')
 const { runtimeFeatures } = require('../../util/runtime-features')
 
@@ -542,6 +543,57 @@ webidl.is.BufferSource = function (V) {
   )
 }
 
+// https://webidl.spec.whatwg.org/#dfn-get-buffer-source-copy
+webidl.util.getCopyOfBytesHeldByBufferSource = function (bufferSource) {
+  // 1. Let jsBufferSource be the result of converting bufferSource to a JavaScript value.
+  const jsBufferSource = bufferSource
+
+  // 2. Let jsArrayBuffer be jsBufferSource.
+  let jsArrayBuffer = jsBufferSource
+
+  // 3. Let offset be 0.
+  let offset = 0
+
+  // 4. Let length be 0.
+  let length = 0
+
+  // 5. If jsBufferSource has a [[ViewedArrayBuffer]] internal slot, then:
+  if (types.isTypedArray(jsBufferSource) || types.isDataView(jsBufferSource)) {
+    // 5.1. Set jsArrayBuffer to jsBufferSource.[[ViewedArrayBuffer]].
+    jsArrayBuffer = jsBufferSource.buffer
+
+    // 5.2. Set offset to jsBufferSource.[[ByteOffset]].
+    offset = jsBufferSource.byteOffset
+
+    // 5.3. Set length to jsBufferSource.[[ByteLength]].
+    length = jsBufferSource.byteLength
+  } else {
+    // 6. Otherwise:
+
+    // 6.1. Assert: jsBufferSource is an ArrayBuffer or SharedArrayBuffer object.
+    assert(types.isAnyArrayBuffer(jsBufferSource))
+
+    // 6.2. Set length to jsBufferSource.[[ArrayBufferByteLength]].
+    length = jsBufferSource.byteLength
+  }
+
+  // 7. If IsDetachedBuffer(jsArrayBuffer) is true, then return the empty byte sequence.
+  if (jsArrayBuffer.detached) {
+    return new Uint8Array(0)
+  }
+
+  // 8. Let bytes be a new byte sequence of length equal to length.
+  const bytes = new Uint8Array(length)
+
+  // 9. For i in the range offset to offset + length − 1, inclusive,
+  //    set bytes[i − offset] to GetValueFromBuffer(jsArrayBuffer, i, Uint8, true, Unordered).
+  const view = new Uint8Array(jsArrayBuffer, offset, length)
+  bytes.set(view)
+
+  // 10. Return bytes.
+  return bytes
+}
+
 // https://webidl.spec.whatwg.org/#es-DOMString
 webidl.converters.DOMString = function (V, prefix, argument, flags) {
   // 1. If V is null and the conversion is to an IDL type

test/fetch/issue-4105.js

@@ -18,7 +18,7 @@ test('markResourceTiming responseStatus is set', { skip: !isAtLeastv22 }, async
   const server = createServer((req, res) => {
     res.statusCode = 200
     res.end('Hello World')
-  }).listen(3000)
+  }).listen(0)
 
   t.after(server.close.bind(server))
   await once(server, 'listening')
@@ -30,7 +30,7 @@ test('markResourceTiming responseStatus is set', { skip: !isAtLeastv22 }, async
     })
   }).observe({ type: 'resource', buffered: true })
 
-  const response = await fetch('http://localhost:3000')
+  const response = await fetch(`http://localhost:${server.address().port}`)
   await response.text()
 
   await promise.promise

test/fetch/issue-4789.js

@@ -0,0 +1,56 @@
+'use strict'
+
+const { createServer } = require('node:http')
+const { test } = require('node:test')
+const { once } = require('node:events')
+const { fetch } = require('../..')
+
+// https://github.com/nodejs/undici/issues/4789
+test('transferred buffers and extractBody works', { skip: !ArrayBuffer.prototype.transfer }, async (t) => {
+  const server = createServer((req, res) => {
+    if (req.url === '/') {
+      res.writeHead(307, undefined, {
+        location: '/test'
+      })
+      res.end()
+      return
+    }
+
+    req.pipe(res).on('end', res.end.bind(res))
+  }).listen(0)
+
+  t.after(server.close.bind(server))
+  await once(server, 'listening')
+
+  {
+    const response = await fetch(`http://localhost:${server.address().port}`, {
+      method: 'POST',
+      body: new TextEncoder().encode('test')
+    })
+
+    t.assert.strictEqual(await response.text(), 'test')
+  }
+
+  {
+    const response = await fetch(`http://localhost:${server.address().port}`, {
+      method: 'POST',
+      body: Buffer.from('test')
+    })
+
+    t.assert.strictEqual(await response.text(), 'test')
+  }
+
+  {
+    const buffer = new TextEncoder().encode('test')
+    buffer.buffer.transfer()
+
+    const response = await fetch(`http://localhost:${server.address().port}`, {
+      method: 'POST',
+      body: buffer
+    })
+
+    // https://webidl.spec.whatwg.org/#dfn-get-buffer-source-copy
+    // "If IsDetachedBuffer(jsArrayBuffer) is true, then return the empty byte sequence."
+    t.assert.strictEqual(await response.text(), '')
+  }
+})

types/webidl.d.ts

@@ -1,4 +1,5 @@
 // These types are not exported, and are only used internally
+import { BufferSource } from 'node:stream/web'
 import * as undici from './index'
 
 /**
@@ -93,6 +94,11 @@ interface WebidlUtil {
   IsResizableArrayBuffer (V: ArrayBufferLike): boolean
 
   HasFlag (flag: number, attributes: number): boolean
+
+  /**
+   * @see https://webidl.spec.whatwg.org/#dfn-get-buffer-source-copy
+   */
+  getCopyOfBytesHeldByBufferSource (bufferSource: BufferSource): Uint8Array
 }
 
 interface WebidlConverters {