Commit 153ec896217 for nodejs

commit 153ec896217fa3752e624f1e4565d5ce52641fd5
Author: James M Snell <jasnell@gmail.com>
Date:   Sun Oct 4 03:09:04 2026 +0000

    stream: make stream/iter from() cancellation waits cheaper

    To stay cancellable while a source is pending, from() waits for every
    value of an async source through waitForNormalization(). For each value
    it created a PromiseWithResolvers(), raced it against the value with
    SafePromiseRace(), which wraps both in new promises, and ran an async
    function with try/finally. This was the largest per-batch cost of
    normalizing an async source.

    Wait with a single promise and a single reaction on the value instead,
    and reject that promise directly on cancellation. The outcome is
    unchanged: the value's result, its rejection, or the cancellation
    reason, whichever comes first, and the cancellation reason if the
    normalization was cancelled by the time the value fulfills.

    With an async generator yielding 16-byte chunks, pipeTo() is about
    1.7x faster and iterating from() about 1.75x faster.

    Assisted-by: OpenCode
    Signed-off-by: James M Snell <jasnell@gmail.com>
    PR-URL: https://github.com/nodejs/node/pull/66648
    Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
    Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>

diff --git a/lib/internal/streams/iter/from.js b/lib/internal/streams/iter/from.js
index 78b61b267de..c89b7ea6d08 100644
--- a/lib/internal/streams/iter/from.js
+++ b/lib/internal/streams/iter/from.js
@@ -19,7 +19,6 @@ const {
   PromisePrototypeThen,
   PromiseResolve,
   PromiseWithResolvers,
-  SafePromiseRace,
   Symbol,
   SymbolAsyncIterator,
   SymbolIterator,
@@ -54,6 +53,7 @@ const {

 const {
   IterResult,
+  kResolvedPromise,
   getProtocolMethod,
   toUint8Array,
 } = require('internal/streams/iter/utils');
@@ -62,7 +62,6 @@ const {
 // Bounds peak memory when arrays flow through transforms, which must
 // allocate output for the entire batch at once.
 const FROM_BATCH_SIZE = 128;
-const kNormalizationCancelled = Symbol('kNormalizationCancelled');
 // Yielded by normalizeAsyncValue() (only when `emitFlush` is true) right
 // before it waits on a promise or on a nested async iterable. Callers that
 // batch chunks yield whatever they have collected so far, so that chunks that
@@ -83,31 +82,56 @@ function cancelNormalization(context, reason, suppressCleanup = false) {
   context.cancelled = true;
   context.reason = reason;
   context.suppressCleanup = suppressCleanup;
-  context.resolve?.(kNormalizationCancelled);
+  context.resolve?.();
 }

 function throwIfNormalizationCancelled(context) {
   if (context?.cancelled) throw context.reason;
 }

-async function waitForNormalization(value, context) {
+/**
+ * Wait for `value`, but stop waiting if the normalization is cancelled.
+ * Settles with the first of:
+ * - `value` fulfilling: its value, or the cancellation reason if the
+ *   normalization has been cancelled by then;
+ * - `value` rejecting: its rejection reason;
+ * - cancellation: the cancellation reason.
+ * This runs for every value of a normalized async source, so it uses a
+ * single promise and a single reaction on `value` rather than racing
+ * promises.
+ * @param {any} value
+ * @param {object} [context]
+ * @returns {Promise<any>|any}
+ */
+function waitForNormalization(value, context) {
   if (context === undefined) return value;
-  const { promise, resolve } = PromiseWithResolvers();
+  const { promise, resolve, reject } = PromiseWithResolvers();
+  const onCancel = () => {
+    if (context.resolve === onCancel) context.resolve = null;
+    reject(context.reason);
+  };
+  PromisePrototypeThen(
+    PromiseResolve(value),
+    (result) => {
+      if (context.resolve === onCancel) context.resolve = null;
+      if (context.cancelled) {
+        reject(context.reason);
+      } else {
+        resolve(result);
+      }
+    },
+    (error) => {
+      if (context.resolve === onCancel) context.resolve = null;
+      reject(error);
+    });
   if (context.cancelled) {
-    resolve(kNormalizationCancelled);
+    // Already cancelled: a `value` that has already settled still takes
+    // precedence, as its reaction above runs first.
+    PromisePrototypeThen(kResolvedPromise, onCancel);
   } else {
-    context.resolve = resolve;
-  }
-  try {
-    const result = await SafePromiseRace([
-      PromiseResolve(value),
-      promise,
-    ]);
-    throwIfNormalizationCancelled(context);
-    return result;
-  } finally {
-    if (context.resolve === resolve) context.resolve = null;
+    context.resolve = onCancel;
   }
+  return promise;
 }

 function createNormalizationIterator(createIterator) {