fix: MuxAsyncIterator throws muxed errors (#6295)

Fixes #5260

std/async/mux_async_iterator.ts

@@ -7,14 +7,15 @@ interface TaggedYieldedValue<T> {
 }
 
 /** The MuxAsyncIterator class multiplexes multiple async iterators into a
- * single stream. It currently makes a few assumptions:
+ * single stream. It currently makes an assumption:
- * - The iterators do not throw.
  * - The final result (the value returned and not yielded from the iterator)
  *   does not matter; if there is any, it is discarded.
  */
 export class MuxAsyncIterator<T> implements AsyncIterable<T> {
   private iteratorCount = 0;
   private yields: Array<TaggedYieldedValue<T>> = [];
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  private throws: any[] = [];
   private signal: Deferred<void> = deferred();
 
   add(iterator: AsyncIterableIterator<T>): void {
@@ -25,12 +26,16 @@ export class MuxAsyncIterator<T> implements AsyncIterable<T> {
   private async callIteratorNext(
     iterator: AsyncIterableIterator<T>
   ): Promise<void> {
+    try {
       const { value, done } = await iterator.next();
       if (done) {
         --this.iteratorCount;
       } else {
         this.yields.push({ iterator, value });
       }
+    } catch (e) {
+      this.throws.push(e);
+    }
     this.signal.resolve();
   }
 
@@ -46,6 +51,12 @@ export class MuxAsyncIterator<T> implements AsyncIterable<T> {
         this.callIteratorNext(iterator);
       }
 
+      if (this.throws.length) {
+        for (const e of this.throws) {
+          throw e;
+        }
+        this.throws.length = 0;
+      }
       // Clear the `yields` list and reset the `signal` promise.
       this.yields.length = 0;
       this.signal = deferred();

std/async/mux_async_iterator_test.ts

@@ -1,5 +1,5 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
-import { assertEquals } from "../testing/asserts.ts";
+import { assertEquals, assertThrowsAsync } from "../testing/asserts.ts";
 import { MuxAsyncIterator } from "./mux_async_iterator.ts";
 
 // eslint-disable-next-line require-await
@@ -16,6 +16,12 @@ async function* gen456(): AsyncIterableIterator<number> {
   yield 6;
 }
 
+// eslint-disable-next-line require-await
+async function* genThrows(): AsyncIterableIterator<number> {
+  yield 7;
+  throw new Error("something went wrong");
+}
+
 Deno.test("[async] MuxAsyncIterator", async function (): Promise<void> {
   const mux = new MuxAsyncIterator<number>();
   mux.add(gen123());
@@ -26,3 +32,22 @@ Deno.test("[async] MuxAsyncIterator", async function (): Promise<void> {
   }
   assertEquals(results.size, 6);
 });
+
+Deno.test({
+  name: "[async] MuxAsyncIterator throws",
+  async fn() {
+    const mux = new MuxAsyncIterator<number>();
+    mux.add(gen123());
+    mux.add(genThrows());
+    const results = new Set();
+    await assertThrowsAsync(
+      async () => {
+        for await (const value of mux) {
+          results.add(value);
+        }
+      },
+      Error,
+      "something went wrong"
+    );
+  },
+});