fix(ext/fetch): use byte ReadableStream for Node Readable request bodies (#33432)

## Summary

- When a `node:http` `IncomingMessage` (or any binary-mode
`stream.Readable`) is passed as the body of a `Request`, build a `type:
"bytes"` `ReadableStream` so consumers can acquire a BYOB reader
(`getReader({ mode: "byob" })`).
- Previously `extractBody` always wrapped async-iterable bodies with
`ReadableStream.from(...)`, which produces a default (non-byte) stream.
That broke code paths — including common HTTP proxying patterns — that
call `request.body.getReader({ mode: "byob" })` on a body sourced from
an `IncomingMessage`.
- Matches undici's behavior in Node, where `stream.Readable` bodies go
through `Readable.toWeb()` and yield a byte stream. Object-mode and
encoded (string) readables still fall through to the existing
`ReadableStream.from` path.

Fixes #33392.

Author: fibibot

ext/fetch/22_body.js

@@ -486,7 +486,40 @@ function extractBody(object) {
       throw new TypeError("ReadableStream is locked or disturbed");
     }
   } else if (object[webidl.AsyncIterable] === webidl.AsyncIterable) {
-    stream = ReadableStream.from(object.open());
+    // If the underlying body is a Node `Readable` running in binary mode
+    // (e.g. `http.IncomingMessage`), build a byte `ReadableStream` so that
+    // consumers can acquire a BYOB reader. This matches undici's behavior in
+    // Node, where `stream.Readable` bodies go through `Readable.toWeb()`.
+    const original = object.value;
+    const readableState = (original !== null && typeof original === "object")
+      ? original._readableState
+      : undefined;
+    if (
+      typeof readableState === "object" && readableState !== null &&
+      !readableState.objectMode && !readableState.encoding
+    ) {
+      const iter = object.open();
+      stream = new ReadableStream({
+        type: "bytes",
+        async pull(controller) {
+          // deno-lint-ignore prefer-primordials
+          const res = await iter.next();
+          if (res.done) {
+            controller.close();
+          } else {
+            controller.enqueue(res.value);
+          }
+        },
+        async cancel(reason) {
+          if (iter.return !== undefined) {
+            // deno-lint-ignore prefer-primordials
+            await iter.return(reason);
+          }
+        },
+      });
+    } else {
+      stream = ReadableStream.from(object.open());
+    }
   }
   if (typeof source === "string") {
     // WARNING: this deviates from spec (expects length to be set)

tests/unit_node/http_test.ts

@@ -2704,3 +2704,44 @@ Deno.test(
     await promise;
   },
 );
+
+// Regression test: a `node:http` IncomingMessage used as the body of a
+// `Request` must produce a byte `ReadableStream`, so that
+// `getReader({ mode: "byob" })` works.
+// https://github.com/denoland/deno/issues/33392
+Deno.test(
+  "[node/http] IncomingMessage as Request body supports BYOB reader",
+  async () => {
+    const { promise, resolve } = Promise.withResolvers<void>();
+
+    const server = http.createServer(async (req, res) => {
+      const request = new Request("http://localhost/", {
+        method: req.method,
+        headers: req.headers as HeadersInit,
+        body: req as unknown as BodyInit,
+        duplex: "half",
+        // deno-lint-ignore no-explicit-any
+      } as any);
+
+      const reader = request.body!.getReader({ mode: "byob" });
+      const buf = new Uint8Array(32);
+      const { value, done } = await reader.read(buf);
+      assertEquals(done, false);
+      assertEquals(new TextDecoder().decode(value), "hello world");
+
+      res.end("OK");
+      server.close(() => resolve());
+    });
+
+    server.listen(0, async () => {
+      const port = (server.address() as AddressInfo).port;
+      const res = await fetch(`http://localhost:${port}/`, {
+        method: "POST",
+        body: "hello world",
+      });
+      assertEquals(await res.text(), "OK");
+    });
+
+    await promise;
+  },
+);