fix(async): handle pooledMap cancellation rejections

async/pool.ts

@@ -77,7 +77,8 @@ export function pooledMap<T, R>(
         // fail the race, taking us to the catch block where all currently
         // executing jobs are allowed to finish and all rejections among them
         // can be reported together.
-        writer.write(p);
+        // The consumer can stop iterating while a write is still pending.
+        writer.write(p).catch(() => {});
         const e: Promise<unknown> = p.then(() =>
           executing.splice(executing.indexOf(e), 1)
         );
@@ -88,17 +89,20 @@ export function pooledMap<T, R>(
       }
       // Wait until all ongoing events have processed, then close the writer.
       await Promise.all(executing);
-      writer.close();
+      writer.close().catch(() => {});
     } catch {
       const errors = [];
       for (const result of await Promise.allSettled(executing)) {
         if (result.status === "rejected") {
           errors.push(result.reason);
         }
       }
-      writer.write(Promise.reject(
+      const error = Promise.reject(
         new AggregateError(errors, ERROR_WHILE_MAPPING_MESSAGE),
-      )).catch(() => {});
+      );
+      // Cancellation can prevent the transform from observing this rejection.
+      error.catch(() => {});
+      writer.write(error).catch(() => {});
     }
   })();
   // Feature test until browser coverage is adequate

async/pool_test.ts

@@ -1,6 +1,7 @@
 // Copyright 2018-2026 the Deno authors. MIT license.
 import { delay } from "./delay.ts";
 import { pooledMap } from "./pool.ts";
+import { pooledMap as unstablePooledMap } from "./unstable_pool.ts";
 import {
   assertEquals,
   assertGreaterOrEqual,
@@ -11,6 +12,41 @@ import {
 } from "@std/assert";
 import { FakeTime } from "@std/testing/time";
 
+for (
+  const [name, pool] of [
+    ["stable", pooledMap],
+    [
+      "unstable",
+      (
+        limit: number,
+        values: number[],
+        fn: (value: number) => Promise<number>,
+      ) => unstablePooledMap(values, fn, limit),
+    ],
+  ] as const
+) {
+  for (const poolLimit of [1, 4]) {
+    for (const reject of [false, true]) {
+      Deno.test(`pooledMap() permits early return (${name}, ${poolLimit}, ${reject})`, async () => {
+        const results = pool(
+          poolLimit,
+          [1, 2, 3],
+          (value) =>
+            reject && value > 1
+              ? Promise.reject(new Error("mapping failed"))
+              : Promise.resolve(value),
+        );
+        for await (const value of results) {
+          assertEquals(value, 1);
+          break;
+        }
+        // Let pending writes and transformations settle after cancellation.
+        await delay(0);
+      });
+    }
+  }
+}
+
 Deno.test("pooledMap() throws for invalid poolLimit", () => {
   const noop = (i: number) => Promise.resolve(i);
   for (const poolLimit of [0, -1, 1.5, NaN, Infinity]) {

async/unstable_pool.ts

@@ -255,9 +255,12 @@ export function pooledMap<T, R>(
       if (errors.length === 0 && signal?.aborted) {
         writer.write(Promise.resolve(ABORT_SENTINEL)).catch(() => {});
       } else {
-        writer.write(Promise.reject(
+        const error = Promise.reject(
           new AggregateError(errors, ERROR_WHILE_MAPPING_MESSAGE),
-        )).catch(() => {});
+        );
+        // Cancellation can prevent the transform from observing this rejection.
+        error.catch(() => {});
+        writer.write(error).catch(() => {});
       }
     } finally {
       removeAbortListener?.();