fix(streams): prevent earlyZipReadableStreams() from possibly using excessive memory (#5082)

* refactor(streams): `earlyZipReadableStreams`

* add(streams): test for cancelling stream

* chore(streams): fmt

* nit(streams): Make one line into two

Co-authored-by: Asher Gomez <ashersaupingomez@gmail.com>

* improve(streams): test to assert all streams were actually cancelled

* adjust(streams): reason for cancelling streams

* tweak cancel reason

* tweak

---------

Co-authored-by: Asher Gomez <ashersaupingomez@gmail.com>

streams/early_zip_readable_streams.ts

@@ -84,26 +84,25 @@
 export function earlyZipReadableStreams<T>(
   ...streams: ReadableStream<T>[]
 ): ReadableStream<T> {
-  const readers = streams.map((s) => s.getReader());
+  const readers = streams.map((stream) => stream.getReader());
   return new ReadableStream<T>({
-    async start(controller) {
+    async pull(controller) {
-      try {
+      for (let i = 0; i < readers.length; ++i) {
-        loop:
+        const { done, value } = await readers[i]!.read();
-        while (true) {
+        if (done) {
-          for (const reader of readers) {
+          await Promise.all(
-            const { value, done } = await reader.read();
+            readers.map((reader) =>
-            if (!done) {
+              reader.cancel(`Stream at index ${i} ended`)
-              controller.enqueue(value!);
+            ),
-            } else {
+          );
-              await Promise.all(readers.map((reader) => reader.cancel()));
-              break loop;
-            }
-          }
-        }
           controller.close();
-      } catch (e) {
+          return;
-        controller.error(e);
         }
+        controller.enqueue(value);
+      }
+    },
+    async cancel(reason) {
+      await Promise.all(readers.map((reader) => reader.cancel(reason)));
     },
   });
 }

streams/early_zip_readable_streams_test.ts

@@ -61,6 +61,27 @@ Deno.test("earlyZipReadableStreams() can zip three streams", async () => {
   ]);
 });
 
+Deno.test("earlyZipReadableStreams() forwards cancel()", async () => {
+  const num = 10;
+  let cancelled = 0;
+  const streams = new Array(num).fill(false).map(() =>
+    new ReadableStream(
+      {
+        pull(controller) {
+          controller.enqueue("chunk");
+        },
+        cancel(reason) {
+          cancelled++;
+          assertEquals(reason, "I was cancelled!");
+        },
+      },
+    )
+  );
+
+  await earlyZipReadableStreams(...streams).cancel("I was cancelled!");
+  assertEquals(cancelled, num);
+});
+
 Deno.test("earlyZipReadableStreams() controller error", async () => {
   const errorMsg = "Test error";
   const stream = new ReadableStream({