Fix cloning of Responses constructed with byte streams, closes #527

This applies the same fix for `Request` bodies from #510 to
`Response`s. Notably, byte streams are returned from lots of Workers
runtime APIs (e.g. KV, R2) to support BYOB reads. It's likely these
are then used in `Response`s and cloned for caching.

Author: mrbbot

Diff for: packages/core/src/standards/http.ts

@@ -582,6 +582,15 @@ export class Response<
       if (body instanceof ArrayBuffer) {
         body = body.slice(0);
       }
+      if (body instanceof ReadableStream && _isByteStream(body)) {
+        // Since `[email protected]`, cloning a body now invokes `structuredClone`
+        // on the underlying stream (https://github.com/nodejs/undici/pull/1697).
+        // Unfortunately, due to a bug in Node, byte streams cannot be
+        // `structuredClone`d (https://github.com/nodejs/undici/issues/1873,
+        // https://github.com/nodejs/node/pull/45955), leading to issues when
+        // constructing bodies with byte streams.
+        body = convertToRegularStream(body);
+      }
 
       if (init instanceof Response) {
         encodeBody = init.#encodeBody;

Diff for: packages/core/test/standards/http.spec.ts

@@ -128,21 +128,6 @@ test("Body: same body instance is always returned", (t) => {
   t.not(body.body, null);
   t.is(body.body, body.body);
 });
-test("Body: reuses byte stream if not input gated", (t) => {
-  const bodyStream = new ReadableStream({
-    type: "bytes",
-    pull(controller) {
-      controller.enqueue(utf8Encode("chunk"));
-      controller.close();
-    },
-  });
-
-  let res = new Response(bodyStream);
-  t.is(res.body, bodyStream);
-
-  res = withInputGating(new Response(bodyStream));
-  t.not(res.body, bodyStream);
-});
 test("Body: body isn't locked until read from", async (t) => {
   const res = new Response("body");
   // noinspection SuspiciousTypeOfGuard
@@ -845,6 +830,37 @@ test("Response: clones non-standard properties", async (t) => {
   t.is(await res2.text(), "body");
   t.is(await res3.text(), "body");
 });
+test("Response: clones stream bodies", async (t) => {
+  let stream = new ReadableStream({
+    start(controller) {
+      controller.enqueue(utf8Encode("chunk1"));
+      controller.close();
+    },
+  });
+  let res = new Response(stream);
+  let clone = res.clone();
+  assert(res.body !== null && clone.body !== null);
+  t.true(_isByteStream(res.body));
+  t.true(_isByteStream(clone.body));
+  t.is(await res.text(), "chunk1");
+  t.is(await clone.text(), "chunk1");
+
+  // Check again with byte stream
+  stream = new ReadableStream({
+    type: "bytes",
+    start(controller) {
+      controller.enqueue(utf8Encode("chunk2"));
+      controller.close();
+    },
+  });
+  res = new Response(stream);
+  clone = res.clone();
+  assert(res.body !== null && clone.body !== null);
+  t.true(_isByteStream(res.body));
+  t.true(_isByteStream(clone.body));
+  t.is(await res.text(), "chunk2");
+  t.is(await clone.text(), "chunk2");
+});
 test("Response: constructing from Response copies non-standard properties", (t) => {
   const pair = new WebSocketPair();
   const res = new Response("body1", {