feat(ext/web): use readableStreamDefaultReaderRead in resourceForReadableStream (#20622)

We can go one level down in abstraction and avoid using the public
`ReadableStream` APIs.

This patch ~5% perf boost on small ReadableStream:

```
Running 10s test @ http://localhost:8080/
  2 threads and 10 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   148.32us  108.95us   3.88ms   95.71%
    Req/Sec    33.24k     2.68k   37.94k    73.76%
  668188 requests in 10.10s, 77.74MB read
Requests/sec:  66162.91
Transfer/sec:      7.70MB
```

main:

```
Running 10s test @ http://localhost:8080/
  2 threads and 10 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   150.23us   67.61us   4.39ms   94.80%
    Req/Sec    31.81k     1.55k   35.56k    83.17%
  639078 requests in 10.10s, 74.36MB read
Requests/sec:  63273.72
Transfer/sec:      7.36MB
```

1 files changed, 58 insertions, 0 deletions

diff --git a/cli/tests/unit/streams_test.ts b/cli/tests/unit/streams_test.ts
index 3460e8ecd..e496b5c4f 100644
--- a/cli/tests/unit/streams_test.ts
+++ b/cli/tests/unit/streams_test.ts
@@ -60,6 +60,28 @@ function longStream() {
   }).pipeThrough(new TextEncoderStream());
 }
 
+// Long stream with Lorem Ipsum text.
+// deno-lint-ignore no-explicit-any
+function longAsyncStream(completion?: Deferred<any>) {
+  let currentTimeout: number | undefined = undefined;
+  return new ReadableStream({
+    async start(controller) {
+      for (let i = 0; i < 100; i++) {
+        await new Promise((r) => currentTimeout = setTimeout(r, 1));
+        currentTimeout = undefined;
+        controller.enqueue(LOREM);
+      }
+      controller.close();
+    },
+    cancel(reason) {
+      completion?.resolve(reason);
+      if (currentTimeout !== undefined) {
+        clearTimeout(currentTimeout);
+      }
+    },
+  }).pipeThrough(new TextEncoderStream());
+}
+
 // Empty stream, closes either immediately or on a call to pull.
 function emptyStream(onPull: boolean) {
   return new ReadableStream({
@@ -104,6 +126,20 @@ function emptyChunkStream() {
   });
 }
 
+// Try to blow up any recursive reads. Note that because of the use of Array.shift in
+// ReadableStream, this might not actually be able to complete with larger values of
+// length.
+function veryLongTinyPacketStream(length: number) {
+  return new ReadableStream({
+    start(controller) {
+      for (let i = 0; i < length; i++) {
+        controller.enqueue(new Uint8Array([1]));
+      }
+      controller.close();
+    },
+  });
+}
+
 // Creates a stream with the given number of packets, a configurable delay between packets, and a final
 // action (either "Throw" or "Close").
 function makeStreamWithCount(
@@ -179,6 +215,14 @@ Deno.test(async function readableStreamCloseWithoutRead() {
   assertEquals(await cancel, "resource closed");
 });
 
+// Close the stream without reading anything
+Deno.test(async function readableStreamCloseWithoutRead2() {
+  const cancel = deferred();
+  const rid = resourceForReadableStream(longAsyncStream(cancel));
+  core.ops.op_close(rid);
+  assertEquals(await cancel, "resource closed");
+});
+
 Deno.test(async function readableStreamPartial() {
   const rid = resourceForReadableStream(helloWorldStream());
   const buffer = new Uint8Array(5);
@@ -197,6 +241,20 @@ Deno.test(async function readableStreamLongReadAll() {
   core.ops.op_close(rid);
 });
 
+Deno.test(async function readableStreamLongAsyncReadAll() {
+  const rid = resourceForReadableStream(longAsyncStream());
+  const buffer = await core.ops.op_read_all(rid);
+  assertEquals(buffer.length, LOREM.length * 100);
+  core.ops.op_close(rid);
+});
+
+Deno.test(async function readableStreamVeryLongReadAll() {
+  const rid = resourceForReadableStream(veryLongTinyPacketStream(10000));
+  const buffer = await core.ops.op_read_all(rid);
+  assertEquals(buffer.length, 10000);
+  core.ops.op_close(rid);
+});
+
 Deno.test(async function readableStreamLongByPiece() {
   const rid = resourceForReadableStream(longStream());
   let total = 0;