fix(fetch): use structuredClone in clone body steps (nodejs#1697)

lib/fetch/body.js

@@ -6,7 +6,7 @@ const { ReadableStreamFrom, toUSVString, isBlobLike } = require('./util')
 const { FormData } = require('./formdata')
 const { kState } = require('./symbols')
 const { webidl } = require('./webidl')
-const { DOMException } = require('./constants')
+const { DOMException, structuredClone } = require('./constants')
 const { Blob } = require('buffer')
 const { kBodyUsed } = require('../core/symbols')
 const assert = require('assert')
@@ -260,13 +260,14 @@ function cloneBody (body) {
 
   // 1. Let « out1, out2 » be the result of teeing body’s stream.
   const [out1, out2] = body.stream.tee()
+  const out2Clone = structuredClone(out2, { transfer: [out2] })
 
   // 2. Set body’s stream to out1.
   body.stream = out1
 
   // 3. Return a body whose stream is out2 and other members are copied from body.
   return {
-    stream: out2,
+    stream: out2Clone,
     length: body.length,
     source: body.source
   }

lib/fetch/constants.js

@@ -1,5 +1,7 @@
 'use strict'
 
+const { MessageChannel, receiveMessageOnPort } = require('worker_threads')
+
 const corsSafeListedMethods = ['GET', 'HEAD', 'POST']
 
 const nullBodyStatus = [101, 204, 205, 304]
@@ -71,8 +73,30 @@ const DOMException = globalThis.DOMException ?? (() => {
   }
 })()
 
+let channel
+
+/** @type {globalThis['structuredClone']} */
+const structuredClone =
+  globalThis.structuredClone ??
+  // https://github.com/nodejs/node/blob/b27ae24dcc4251bad726d9d84baf678d1f707fed/lib/internal/structured_clone.js
+  // structuredClone was added in v17.0.0, but fetch supports v16.8
+  function structuredClone (value, options = undefined) {
+    if (arguments.length === 0) {
+      throw new TypeError('missing argument')
+    }
+
+    if (!channel) {
+      channel = new MessageChannel()
+    }
+    channel.port1.unref()
+    channel.port2.unref()
+    channel.port1.postMessage(value, options?.transfer)
+    return receiveMessageOnPort(channel.port2).message
+  }
+
 module.exports = {
   DOMException,
+  structuredClone,
   subresource,
   forbiddenMethods,
   requestBodyHeader,

test/fetch/response.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const { test } = require('tap')
+const { test, teardown } = require('tap')
 const {
   Response
 } = require('../../')
@@ -248,3 +248,6 @@ test('constructing Response with third party FormData body', async (t) => {
   t.equal(contentType[0], 'multipart/form-data; boundary')
   t.ok((await res.text()).startsWith(`--${contentType[1]}`))
 })
+
+// This is needed due to https://github.com/nodejs/node/issues/44985
+teardown(() => process.exit(0))

test/wpt/status/fetch.status.json

@@ -29,5 +29,11 @@
 			"Response interface: operation json(any, optional ResponseInit)",
 			"Window interface: operation fetch(RequestInfo, optional RequestInit)"
 		]
+	},
+	"response-clone.any.js": {
+		"fail": [
+			"Check response clone use structureClone for teed ReadableStreams (ArrayBufferchunk)",
+			"Check response clone use structureClone for teed ReadableStreams (DataViewchunk)"
+		]
 	}
 }

test/wpt/tests/fetch/api/response/response-clone.any.js

@@ -0,0 +1,126 @@
+// META: global=window,worker
+// META: title=Response clone
+// META: script=../resources/utils.js
+
+var defaultValues = { "type" : "default",
+                      "url" : "",
+                      "ok" : true,
+                      "status" : 200,
+                      "statusText" : ""
+};
+
+var response = new Response();
+var clonedResponse = response.clone();
+test(function() {
+  for (var attributeName in defaultValues) {
+      var expectedValue = defaultValues[attributeName];
+      assert_equals(clonedResponse[attributeName], expectedValue,
+        "Expect default response." + attributeName + " is " + expectedValue);
+  }
+}, "Check Response's clone with default values, without body");
+
+var body = "This is response body";
+var headersInit = { "name" : "value" };
+var responseInit  = { "status" : 200,
+                      "statusText" : "GOOD",
+                      "headers" : headersInit
+};
+var response = new Response(body, responseInit);
+var clonedResponse = response.clone();
+test(function() {
+  assert_equals(clonedResponse.status, responseInit["status"],
+    "Expect response.status is " + responseInit["status"]);
+  assert_equals(clonedResponse.statusText, responseInit["statusText"],
+    "Expect response.statusText is " + responseInit["statusText"]);
+  assert_equals(clonedResponse.headers.get("name"), "value",
+    "Expect response.headers has name:value header");
+}, "Check Response's clone has the expected attribute values");
+
+promise_test(function(test) {
+  return validateStreamFromString(response.body.getReader(), body);
+}, "Check orginal response's body after cloning");
+
+promise_test(function(test) {
+  return validateStreamFromString(clonedResponse.body.getReader(), body);
+}, "Check cloned response's body");
+
+promise_test(function(test) {
+  var disturbedResponse = new Response("data");
+  return disturbedResponse.text().then(function() {
+      assert_true(disturbedResponse.bodyUsed, "response is disturbed");
+      assert_throws_js(TypeError, function() { disturbedResponse.clone(); },
+        "Expect TypeError exception");
+  });
+}, "Cannot clone a disturbed response");
+
+promise_test(function(t) {
+    var clone;
+    var result;
+    var response;
+    return fetch('../resources/trickle.py?count=2&delay=100').then(function(res) {
+        clone = res.clone();
+        response = res;
+        return clone.text();
+    }).then(function(r) {
+        assert_equals(r.length, 26);
+        result = r;
+        return response.text();
+    }).then(function(r) {
+        assert_equals(r, result, "cloned responses should provide the same data");
+    });
+  }, 'Cloned responses should provide the same data');
+
+promise_test(function(t) {
+    var clone;
+    return fetch('../resources/trickle.py?count=2&delay=100').then(function(res) {
+        clone = res.clone();
+        res.body.cancel();
+        assert_true(res.bodyUsed);
+        assert_false(clone.bodyUsed);
+        return clone.arrayBuffer();
+    }).then(function(r) {
+        assert_equals(r.byteLength, 26);
+        assert_true(clone.bodyUsed);
+    });
+}, 'Cancelling stream should not affect cloned one');
+
+function testReadableStreamClone(initialBuffer, bufferType)
+{
+    promise_test(function(test) {
+        var response = new Response(new ReadableStream({start : function(controller) {
+            controller.enqueue(initialBuffer);
+            controller.close();
+        }}));
+
+        var clone = response.clone();
+        var stream1 = response.body;
+        var stream2 = clone.body;
+
+        var buffer;
+        return stream1.getReader().read().then(function(data) {
+            assert_false(data.done);
+            assert_equals(data.value, initialBuffer, "Buffer of being-cloned response stream is the same as the original buffer");
+            return stream2.getReader().read();
+        }).then(function(data) {
+            assert_false(data.done);
+            assert_array_equals(data.value, initialBuffer, "Cloned buffer chunks have the same content");
+            assert_equals(Object.getPrototypeOf(data.value), Object.getPrototypeOf(initialBuffer), "Cloned buffers have the same type");
+            assert_not_equals(data.value, initialBuffer, "Buffer of cloned response stream is a clone of the original buffer");
+        });
+    }, "Check response clone use structureClone for teed ReadableStreams (" + bufferType  + "chunk)");
+}
+
+var arrayBuffer = new ArrayBuffer(16);
+testReadableStreamClone(new Int8Array(arrayBuffer, 1), "Int8Array");
+testReadableStreamClone(new Int16Array(arrayBuffer, 2, 2), "Int16Array");
+testReadableStreamClone(new Int32Array(arrayBuffer), "Int32Array");
+testReadableStreamClone(arrayBuffer, "ArrayBuffer");
+testReadableStreamClone(new Uint8Array(arrayBuffer), "Uint8Array");
+testReadableStreamClone(new Uint8ClampedArray(arrayBuffer), "Uint8ClampedArray");
+testReadableStreamClone(new Uint16Array(arrayBuffer, 2), "Uint16Array");
+testReadableStreamClone(new Uint32Array(arrayBuffer), "Uint32Array");
+testReadableStreamClone(typeof BigInt64Array === "function" ? new BigInt64Array(arrayBuffer) : undefined, "BigInt64Array");
+testReadableStreamClone(typeof BigUint64Array === "function" ? new BigUint64Array(arrayBuffer) : undefined, "BigUint64Array");
+testReadableStreamClone(new Float32Array(arrayBuffer), "Float32Array");
+testReadableStreamClone(new Float64Array(arrayBuffer), "Float64Array");
+testReadableStreamClone(new DataView(arrayBuffer, 2, 8), "DataView");