From fa9adb350e71a6416989f6c1379762f069b0358a Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <230980@gmail.com>
Date: Sat, 11 Apr 2026 14:07:07 -0700
Subject: [PATCH 01/12] Alternative iteration function on Readable that don't
 rely on Async Iteration

---
 lib/internal/streams/from.js        |   2 +
 lib/internal/streams/operators.js   | 450 ++++++++++++++++++++--------
 test/parallel/test-stream-filter.js |  29 ++
 test/parallel/test-stream-map.js    |  45 ++-
 test/parallel/test-stream-reduce.js |   2 +-
 5 files changed, 406 insertions(+), 122 deletions(-)

diff --git a/lib/internal/streams/from.js b/lib/internal/streams/from.js
index 6a943f02657924..73e696d66fae2b 100644
--- a/lib/internal/streams/from.js
+++ b/lib/internal/streams/from.js
@@ -26,6 +26,8 @@ function from(Readable, iterable, opts) {
         this.push(null);
       },
     });
+  } else if (iterable instanceof Readable) {
+    return iterable;
   }
 
   let isAsync;
diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js
index 6db2df0e3646e0..443bd7f5f231a6 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -2,21 +2,27 @@
 
 const {
   ArrayPrototypePush,
+  ArrayFrom,
   Boolean,
   MathFloor,
+  NumberMAX_SAFE_INTEGER,
   Number,
   NumberIsNaN,
   Promise,
+  PromiseAllSettled,
   PromisePrototypeThen,
   PromiseReject,
   PromiseResolve,
+  PromiseWithResolvers,
   Symbol,
+  SymbolIterator,
 } = primordials;
 
 const { AbortController, AbortSignal } = require('internal/abort_controller');
 
 const {
   AbortError,
+  aggregateTwoErrors,
   codes: {
     ERR_MISSING_ARGS,
     ERR_OUT_OF_RANGE,
@@ -27,9 +33,11 @@ const {
   validateInteger,
   validateObject,
   validateFunction,
+  validateBoolean,
 } = require('internal/validators');
 const { kWeakHandler, kResistStopPropagation } = require('internal/event_target');
-const { finished } = require('internal/streams/end-of-stream');
+const { eos, finished } = require('internal/streams/end-of-stream');
+const destroyImpl = require('internal/streams/destroy');
 
 const kEmpty = Symbol('kEmpty');
 const kEof = Symbol('kEof');
@@ -58,168 +66,386 @@ function map(fn, options) {
 
   highWaterMark += concurrency;
 
-  return async function* map() {
+  return function map() {
     const signal = AbortSignal.any([options?.signal].filter(Boolean));
-    const stream = this;
+    const source = this;
     const queue = [];
     const signalOpt = { signal };
 
-    let next;
-    let resume;
-    let done = false;
-    let cnt = 0;
+    let continueReadingFromQueue;
+    let resumeReadingFromSource;
+    let continueReadingFromSource;
+    let stopReadingFromSource = false;
+    let currentConcurrency = 0;
 
     function onCatch() {
-      done = true;
+      stopReadingFromSource = true;
       afterItemProcessed();
+      if (continueReadingFromQueue) {
+        continueReadingFromQueue();
+        continueReadingFromQueue = null;
+      }
     }
 
     function afterItemProcessed() {
-      cnt -= 1;
-      maybeResume();
+      currentConcurrency -= 1;
+      maybeResumeReadingFromSource();
     }
 
-    function maybeResume() {
+    function maybeResumeReadingFromSource() {
       if (
-        resume &&
-        !done &&
-        cnt < concurrency &&
+        resumeReadingFromSource &&
+        !stopReadingFromSource &&
+        currentConcurrency < concurrency &&
         queue.length < highWaterMark
       ) {
-        resume();
-        resume = null;
+        resumeReadingFromSource();
+        resumeReadingFromSource = null;
       }
     }
 
-    async function pump() {
+    async function readFromSource() {
       try {
-        for await (let val of stream) {
-          if (done) {
-            return;
-          }
 
-          if (signal.aborted) {
-            throw new AbortError();
+        function moreDataToReadFromSource() {
+          if (continueReadingFromSource) {
+            continueReadingFromSource();
+            continueReadingFromSource = null;
           }
+        }
 
-          try {
-            val = fn(val, signalOpt);
+        source.on('readable', moreDataToReadFromSource);
 
-            if (val === kEmpty) {
-              continue;
-            }
+        let error;
+        const cleanup = eos(source, { writable: false }, (err) => {
+          error = err ? aggregateTwoErrors(error, err) : null;
+          moreDataToReadFromSource();
+        });
 
-            val = PromiseResolve(val);
-          } catch (err) {
-            val = PromiseReject(err);
-          }
+        try {
+          while (true) {
+            if (stopReadingFromSource) {
+              return;
+            }
 
-          cnt += 1;
+            if (signal.aborted) {
+              throw new AbortError();
+            }
 
-          PromisePrototypeThen(val, afterItemProcessed, onCatch);
+            if (queue.length >= highWaterMark || currentConcurrency >= concurrency) {
+              const pr = PromiseWithResolvers();
+              resumeReadingFromSource = pr.resolve;
+              await pr.promise;
+            }
 
-          queue.push(val);
-          if (next) {
-            next();
-            next = null;
+            let chunk = source.destroyed ? null : source.read();
+            if (chunk !== null) {
+              currentConcurrency += 1;
+              chunk = fn(chunk, signalOpt);
+
+              if (chunk === kEmpty) {
+                afterItemProcessed();
+                continue;
+              }
+              if (typeof chunk.then === 'function') {
+                PromisePrototypeThen(PromiseResolve(chunk), afterItemProcessed, onCatch);
+                queue.push(chunk);
+              } else {
+                queue.push({ chunk, afterItemProcessed });
+              }
+              if (continueReadingFromQueue) {
+                continueReadingFromQueue();
+                continueReadingFromQueue = null;
+              }
+              if (chunk === kEof) {
+                // Wait until queue is drained before exiting.
+                while (queue.length > 0) {
+                  const pr = PromiseWithResolvers();
+                  resumeReadingFromSource = pr.resolve;
+                  await pr.promise;
+                }
+                return;
+              }
+
+              if (!stopReadingFromSource && (queue.length >= highWaterMark || currentConcurrency >= concurrency)) {
+                const pr = PromiseWithResolvers();
+                resumeReadingFromSource = pr.resolve;
+                await pr.promise;
+              }
+            } else if (error) {
+              throw error;
+            } else if (error === null) {
+              queue.push(kEof);
+              return;
+            } else {
+              const pr = PromiseWithResolvers();
+              continueReadingFromSource = pr.resolve;
+              await pr.promise;
+            }
           }
-
-          if (!done && (queue.length >= highWaterMark || cnt >= concurrency)) {
-            await new Promise((resolve) => {
-              resume = resolve;
-            });
+        } catch (err) {
+          error = aggregateTwoErrors(error, err);
+          throw error;
+        } finally {
+          if (
+            (error || options?.destroyOnReturn !== false) &&
+            (error === undefined || source._readableState.autoDestroy)
+          ) {
+            destroyImpl.destroyer(source, null);
+          } else {
+            source.off('readable', moreDataToReadFromSource);
+            cleanup();
           }
         }
-        queue.push(kEof);
       } catch (err) {
         const val = PromiseReject(err);
         PromisePrototypeThen(val, afterItemProcessed, onCatch);
         queue.push(val);
       } finally {
-        done = true;
-        if (next) {
-          next();
-          next = null;
+        stopReadingFromSource = true;
+        if (continueReadingFromQueue) {
+          continueReadingFromQueue();
+          continueReadingFromQueue = null;
         }
       }
     }
 
-    pump();
+    process.nextTick(readFromSource);
 
-    try {
-      while (true) {
-        while (queue.length > 0) {
-          const val = await queue[0];
+    const readable = new source.constructor({
+      objectMode: true,
+      highWaterMark,
+    });
+    let currentlyReadingFromQueue = false;
+    readable._read = async function readFromQueue() {
+      try {
+        if (!currentlyReadingFromQueue) {
+          currentlyReadingFromQueue = true;
 
-          if (val === kEof) {
-            return;
+          if (!stopReadingFromSource && queue.length === 0) {
+            const pr = PromiseWithResolvers();
+            continueReadingFromQueue = pr.resolve;
+            await pr.promise;
           }
+          while (queue.length > 0) {
+            if (signal.aborted) {
+              currentlyReadingFromQueue = false;
+              throw new AbortError();
+            }
 
-          if (signal.aborted) {
-            throw new AbortError();
-          }
+            let item = queue[0];
+            let syncAip = null;
+            if (typeof item.then === 'function') {
+              item = await item;
+              queue.shift();
+            } else if (typeof item.afterItemProcessed === 'function') {
+              syncAip = item.afterItemProcessed;
+              item = item.chunk;
+              // Call syncAip BEFORE shifting so queue.length stays >= 1,
+              // matching async item behavior (afterItemProcessed fires via
+              // PromiseThen while item is still peeked). This prevents
+              // maybeResumeReadingFromSource from immediately waking readFromSource.
+              syncAip();
+              queue.shift();
+            } else {
+              queue.shift();
+            }
 
-          if (val !== kEmpty) {
-            yield val;
+            if (item === kEof) {
+              stopReadingFromSource = true;
+              currentlyReadingFromQueue = false;
+              this.push(null);
+              if (resumeReadingFromSource) {
+                resumeReadingFromSource();
+                resumeReadingFromSource = null;
+              }
+              if (continueReadingFromQueue) {
+                continueReadingFromQueue();
+                continueReadingFromQueue = null;
+              }
+              return;
+            }
+            if (item !== kEmpty) {
+              if (syncAip) {
+                // Sync item: yield via await so that stream.read()'s preemptive
+                // _read call (triggered when state.length-n < hwm) sees kReading
+                // still set and does not re-enter _read before push clears it.
+                // Schedule maybeResume as a nextTick (not a microtask) so that
+                // the async-iterator cleanup microtask (done=true) from a
+                // consumer break runs before pump is woken, preventing an extra
+                // fn call on infinite streams.
+                await PromiseResolve();
+                if (!this.push(item)) {
+                  currentlyReadingFromQueue = false;
+                  process.nextTick(maybeResumeReadingFromSource);
+                  return;
+                }
+                process.nextTick(maybeResumeReadingFromSource);
+              } else if (!this.push(item)) {
+                currentlyReadingFromQueue = false;
+                process.nextTick(maybeResumeReadingFromSource);
+                return;
+              } else {
+                process.nextTick(maybeResumeReadingFromSource);
+              }
+            } else {
+              if (syncAip) syncAip();
+              maybeResumeReadingFromSource();
+              if (!stopReadingFromSource && queue.length === 0) {
+                const pr = PromiseWithResolvers();
+                continueReadingFromQueue = pr.resolve;
+                await pr.promise;
+              }
+              continue;
+            }
           }
-
-          queue.shift();
-          maybeResume();
+          currentlyReadingFromQueue = false;
         }
-
-        await new Promise((resolve) => {
-          next = resolve;
-        });
-      }
-    } finally {
-      done = true;
-      if (resume) {
-        resume();
-        resume = null;
+      } catch (err) {
+        destroyImpl.destroyer(this, err);
       }
     }
+    readable._destroy = function(err, cb) {
+      stopReadingFromSource = true;
+      if (resumeReadingFromSource) {
+        resumeReadingFromSource();
+        resumeReadingFromSource = null;
+      }
+      if (continueReadingFromQueue) {
+        continueReadingFromQueue();
+        continueReadingFromQueue = null;
+      }
+      cb(err);
+    };
+    return readable;
   }.call(this);
 }
 
+function nowOrLater(fn, fn2, args) {
+  const valueOrPromise = fn(...args);
+  if (valueOrPromise && typeof valueOrPromise.then === 'function') {
+      return PromisePrototypeThen(valueOrPromise, fn2);
+    }
+    return fn2(valueOrPromise);
+}
+
 async function some(fn, options = undefined) {
-  for await (const unused of filter.call(this, fn, options)) {
-    return true;
-  }
-  return false;
+  validateFunction(fn, 'fn');
+  const someFn = (...args) => {
+    return nowOrLater(fn, Boolean, args);
+  };
+  return (await find.call(this, someFn, options)) !== undefined;
 }
 
 async function every(fn, options = undefined) {
   validateFunction(fn, 'fn');
+  const everyFn = (...args) => {
+    return nowOrLater(fn, value => !value, args);
+  };
   // https://en.wikipedia.org/wiki/De_Morgan%27s_laws
-  return !(await some.call(this, async (...args) => {
-    return !(await fn(...args));
-  }, options));
+  return !(await some.call(this, everyFn, options));
 }
 
-async function find(fn, options) {
-  for await (const result of filter.call(this, fn, options)) {
-    return result;
+function find(fn, options) {
+  validateFunction(fn, 'fn');
+
+  if (options != null) {
+    validateObject(options, 'options');
+  }
+  const signal = options?.signal;
+  if (signal != null) {
+    validateAbortSignal(signal, 'options.signal');
+  }
+
+  let concurrency = 1;
+  if (options?.concurrency != null) {
+    concurrency = MathFloor(options.concurrency);
+  }
+  validateInteger(concurrency, 'options.concurrency', 1);
+
+  let destroyOnReturn = true;
+  if (options?.destroyOnReturn != null) {
+    destroyOnReturn = options?.destroyOnReturn;
+    validateBoolean(destroyOnReturn, 'options.destroyOnReturn');
+  }
+
+  let result = undefined;
+  let foundIndex = undefined;
+  let currentConcurrency = 0;
+  let sequence = 0;
+  const pr = PromiseWithResolvers();
+  this.on('error', pr.reject);
+
+  const onData = async function (chunk) {
+    if (foundIndex !== undefined && foundIndex < sequence) {
+      this.off('data', onData);
+      return;
+    }
+    let i = sequence;
+    sequence++;
+    currentConcurrency++;
+    if (currentConcurrency >= concurrency && !this.isPaused()) {
+      this.pause();
+    }
+    try {
+      if (signal?.aborted) {
+        throw new AbortError(undefined, { cause: signal.reason });
+      }
+      let v = fn(chunk, options);
+      if (v && typeof v.then === 'function') {
+        v = await v;
+      }
+
+      if (v && (foundIndex === undefined || foundIndex > i)) {
+        foundIndex = i;
+        result = chunk;
+      }
+    } catch (err) {
+      this.off('data', onData);
+      destroyImpl.destroyer(this, err);
+    } finally {
+      currentConcurrency--;
+      if (this.isPaused() && currentConcurrency < concurrency && foundIndex === undefined) {
+        this.resume();
+      }
+      if (currentConcurrency === 0 && foundIndex !== undefined) {
+        this.off('data', onData);
+        pr.resolve(result);
+      }
+    }
+  };
+  this.on('data', onData);
+  this.on('end', function onEnd() {
+    this.off('data', onData);
+    foundIndex = NumberMAX_SAFE_INTEGER;
+    if (currentConcurrency === 0) {
+      pr.resolve(result);
+    }
+  });
+  if (destroyOnReturn) {
+    return PromisePrototypeThen(pr.promise, value => {
+      if (!this.errored) {
+        destroyImpl.destroyer(this, null);
+      }
+      return value;
+    });
   }
-  return undefined;
+
+  return pr.promise;
 }
 
 async function forEach(fn, options) {
   validateFunction(fn, 'fn');
-  async function forEachFn(value, options) {
-    await fn(value, options);
-    return kEmpty;
-  }
-  // eslint-disable-next-line no-unused-vars
-  for await (const unused of map.call(this, forEachFn, options));
+  const forEachFn = (...args) => {
+    return nowOrLater(fn, () => false, args);
+  };
+  await find.call(this, forEachFn, options);
 }
 
 function filter(fn, options) {
   validateFunction(fn, 'fn');
-  async function filterFn(value, options) {
-    if (await fn(value, options)) {
-      return value;
-    }
-    return kEmpty;
+  const filterFn = (value, options) => {
+    return nowOrLater(fn, predicate => predicate ? value : kEmpty, [value, options]);
   }
   return map.call(this, filterFn, options);
 }
@@ -279,25 +505,15 @@ async function reduce(reducer, initialValue, options) {
 }
 
 async function toArray(options) {
-  if (options != null) {
-    validateObject(options, 'options');
-  }
-  if (options?.signal != null) {
-    validateAbortSignal(options.signal, 'options.signal');
-  }
-
   const result = [];
-  for await (const val of this) {
-    if (options?.signal?.aborted) {
-      throw new AbortError(undefined, { cause: options.signal.reason });
-    }
-    ArrayPrototypePush(result, val);
-  }
+  const toArrayFn = (chunk) => ArrayPrototypePush(result, chunk);
+  await forEach.call(this, toArrayFn, options);
   return result;
 }
 
 function flatMap(fn, options) {
   const values = map.call(this, fn, options);
+  // TODO lucho
   return async function* flatMap() {
     for await (const val of values) {
       yield* val;
@@ -327,19 +543,13 @@ function drop(number, options = undefined) {
   }
 
   number = toIntegerOrInfinity(number);
-  return async function* drop() {
-    if (options?.signal?.aborted) {
-      throw new AbortError();
+  return filter.call(this, () => {
+    if (number > 0) {
+      number--;
+      return false;
     }
-    for await (const val of this) {
-      if (options?.signal?.aborted) {
-        throw new AbortError();
-      }
-      if (number-- <= 0) {
-        yield val;
-      }
-    }
-  }.call(this);
+    return true;
+  }, options);
 }
 
 function take(number, options = undefined) {
diff --git a/test/parallel/test-stream-filter.js b/test/parallel/test-stream-filter.js
index 0b70c391c88fb1..ea9b1b2818b0b0 100644
--- a/test/parallel/test-stream-filter.js
+++ b/test/parallel/test-stream-filter.js
@@ -8,6 +8,35 @@ const assert = require('assert');
 const { once } = require('events');
 const { setTimeout } = require('timers/promises');
 
+{
+  // Filter works on empty streams with a synchronous predicate
+  const stream = Readable.from([]).filter((x) => true);
+  (async () => {
+    for await (const item of stream) {
+    }
+  })().then(common.mustCall());
+}
+
+{
+  // Filter works on synchronous streams with a synchronous predicate
+  const stream = Readable.from([1, 2, 3, 4, 5]).filter((x) => x < 10);
+  const result = [1, 2, 3, 4, 5];
+  (async () => {
+    for await (const item of stream) {
+      assert.strictEqual(item, result.shift());
+    }
+  })().then(common.mustCall());
+}
+
+{
+  // Filter works on synchronous streams with a synchronous predicate
+  const stream = Readable.from([1, 2, 3, 4, 5]).filter((x) => x > 10);
+  (async () => {
+    for await (const item of stream) {
+    }
+  })().then(common.mustCall());
+}
+
 {
   // Filter works on synchronous streams with a synchronous predicate
   const stream = Readable.from([1, 2, 3, 4, 5]).filter((x) => x < 3);
diff --git a/test/parallel/test-stream-map.js b/test/parallel/test-stream-map.js
index 212658313885f1..9846657c5ab970 100644
--- a/test/parallel/test-stream-map.js
+++ b/test/parallel/test-stream-map.js
@@ -27,6 +27,14 @@ function createDependentPromises(n) {
   return promiseAndResolveArray;
 }
 
+{
+  // Map works on empty streams with a synchronous mapper
+  const stream = Readable.from([]).map((x) => x);
+  (async () => {
+    assert.deepStrictEqual(await stream.toArray(), []);
+  })().then(common.mustCall());
+}
+
 {
   // Map works on synchronous streams with a synchronous mapper
   const stream = Readable.from([1, 2, 3, 4, 5]).map((x) => x + x);
@@ -35,6 +43,14 @@ function createDependentPromises(n) {
   })().then(common.mustCall());
 }
 
+{
+  // Double Map works on synchronous streams with a synchronous mapper
+  const stream = Readable.from([1, 2, 3, 4, 5]).map((x) => x + x).map((x) => x + x);
+  (async () => {
+    assert.deepStrictEqual(await stream.toArray(), [4, 8, 12, 16, 20]);
+  })().then(common.mustCall());
+}
+
 {
   // Map works on synchronous streams with an asynchronous mapper
   const stream = Readable.from([1, 2, 3, 4, 5]).map(async (x) => {
@@ -46,6 +62,17 @@ function createDependentPromises(n) {
   })().then(common.mustCall());
 }
 
+{
+  // Map works on synchronous streams with an asynchronous mapper
+  const stream = Readable.from([1, 2, 3, 4, 5]).map((x) => x + x).map(async (x) => {
+    await Promise.resolve();
+    return x + x;
+  });
+  (async () => {
+    assert.deepStrictEqual(await stream.toArray(), [4, 8, 12, 16, 20]);
+  })().then(common.mustCall());
+}
+
 {
   // Map works on asynchronous streams with a asynchronous mapper
   const stream = Readable.from([1, 2, 3, 4, 5]).map(async (x) => {
@@ -57,7 +84,23 @@ function createDependentPromises(n) {
 }
 
 {
-  // Map works on an infinite stream
+  // Map works on an infinite stream - sync
+  const stream = Readable.from(async function* () {
+    while (true) yield 1;
+  }()).map(common.mustCall((x) => {
+    return x + x;
+  }, 5));
+  (async () => {
+    let i = 1;
+    for await (const item of stream) {
+      assert.strictEqual(item, 2);
+      if (++i === 5) break;
+    }
+  })().then(common.mustCall());
+}
+
+{
+  // Map works on an infinite stream - async
   const stream = Readable.from(async function* () {
     while (true) yield 1;
   }()).map(common.mustCall(async (x) => {
diff --git a/test/parallel/test-stream-reduce.js b/test/parallel/test-stream-reduce.js
index 99029f6f8310d5..bcd1df234f8372 100644
--- a/test/parallel/test-stream-reduce.js
+++ b/test/parallel/test-stream-reduce.js
@@ -55,7 +55,7 @@ function sum(p, c) {
   assert.rejects(Readable.from([1, 2, 3, 4, 5, 6])
     .map(common.mustCall((x) => {
       return x;
-    }, 3)) // Two consumed and one buffered by `map` due to default concurrency
+    }, 2))
     .reduce(async (p, c) => {
       if (p === 1) {
         throw new Error('boom');

From c6e072c99d725c81e68b3beec6ad240973e793c9 Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <230980@gmail.com>
Date: Sat, 11 Apr 2026 18:33:39 -0700
Subject: [PATCH 02/12] Find with read()

---
 lib/internal/streams/operators.js | 108 ++++++++++++++++++------------
 1 file changed, 64 insertions(+), 44 deletions(-)

diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js
index 443bd7f5f231a6..0d83e2d3d1b9da 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -5,7 +5,6 @@ const {
   ArrayFrom,
   Boolean,
   MathFloor,
-  NumberMAX_SAFE_INTEGER,
   Number,
   NumberIsNaN,
   Promise,
@@ -346,7 +345,7 @@ async function every(fn, options = undefined) {
   return !(await some.call(this, everyFn, options));
 }
 
-function find(fn, options) {
+async function find(fn, options) {
   validateFunction(fn, 'fn');
 
   if (options != null) {
@@ -369,69 +368,90 @@ function find(fn, options) {
     validateBoolean(destroyOnReturn, 'options.destroyOnReturn');
   }
 
+  const source = this;
+
   let result = undefined;
   let foundIndex = undefined;
   let currentConcurrency = 0;
   let sequence = 0;
-  const pr = PromiseWithResolvers();
-  this.on('error', pr.reject);
+  let continueLoop = null;
+  let error;
 
-  const onData = async function (chunk) {
-    if (foundIndex !== undefined && foundIndex < sequence) {
-      this.off('data', onData);
-      return;
-    }
-    let i = sequence;
-    sequence++;
-    currentConcurrency++;
-    if (currentConcurrency >= concurrency && !this.isPaused()) {
-      this.pause();
+  function wake() {
+    if (continueLoop) {
+      continueLoop();
+      continueLoop = null;
     }
+  }
+
+  this.on('error', (err) => {
+    error = err;
+  });
+  this.on('readable', wake);
+  const cleanup = eos(this, { writable: false }, (err) => {
+    error = err ? aggregateTwoErrors(error, err) : null;
+    wake();
+  });
+
+  async function consumeChunk(chunk, source, currentIndex) {
     try {
-      if (signal?.aborted) {
-        throw new AbortError(undefined, { cause: signal.reason });
-      }
       let v = fn(chunk, options);
       if (v && typeof v.then === 'function') {
         v = await v;
       }
-
-      if (v && (foundIndex === undefined || foundIndex > i)) {
-        foundIndex = i;
+      if (v && (foundIndex === undefined || foundIndex > currentIndex)) {
+        foundIndex = currentIndex;
         result = chunk;
       }
     } catch (err) {
-      this.off('data', onData);
-      destroyImpl.destroyer(this, err);
+      destroyImpl.destroyer(source, err);
     } finally {
       currentConcurrency--;
-      if (this.isPaused() && currentConcurrency < concurrency && foundIndex === undefined) {
-        this.resume();
+      wake();
+    }
+  }
+
+  try {
+    while (true) {
+      if (signal?.aborted) {
+        throw new AbortError(undefined, { cause: signal.reason });
       }
-      if (currentConcurrency === 0 && foundIndex !== undefined) {
-        this.off('data', onData);
-        pr.resolve(result);
+
+      if (currentConcurrency < concurrency && foundIndex === undefined) {
+        const chunk = this.destroyed ? null : this.read();
+
+        if (chunk !== null) {
+          currentConcurrency++;
+          void consumeChunk(chunk, this, sequence++);
+          continue;
+        }
       }
-    }
-  };
-  this.on('data', onData);
-  this.on('end', function onEnd() {
-    this.off('data', onData);
-    foundIndex = NumberMAX_SAFE_INTEGER;
-    if (currentConcurrency === 0) {
-      pr.resolve(result);
-    }
-  });
-  if (destroyOnReturn) {
-    return PromisePrototypeThen(pr.promise, value => {
-      if (!this.errored) {
-        destroyImpl.destroyer(this, null);
+
+      if (error) {
+        throw error;
       }
-      return value;
-    });
+
+      if (currentConcurrency === 0 && (foundIndex !== undefined || error === null)) {
+        break;
+      }
+
+      const pr = PromiseWithResolvers();
+      continueLoop = pr.resolve;
+      await pr.promise;
+    }
+  } catch (err) {
+    destroyImpl.destroyer(this, err);
+    throw err;
+  } finally {
+    this.off('readable', wake);
+    cleanup();
+  }
+
+  if (destroyOnReturn && !this.errored) {
+    destroyImpl.destroyer(this, null);
   }
 
-  return pr.promise;
+  return result;
 }
 
 async function forEach(fn, options) {

From 886ce5e84420fc5762f73e0cddc19034a73c1668 Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <230980@gmail.com>
Date: Tue, 14 Apr 2026 15:50:36 -0700
Subject: [PATCH 03/12] Benchmark streams map/reduce

---
 benchmark/streams/map-reduce.js | 22 ++++++++++++++++++++++
 1 file changed, 22 insertions(+)
 create mode 100644 benchmark/streams/map-reduce.js

diff --git a/benchmark/streams/map-reduce.js b/benchmark/streams/map-reduce.js
new file mode 100644
index 00000000000000..17ea82c7662892
--- /dev/null
+++ b/benchmark/streams/map-reduce.js
@@ -0,0 +1,22 @@
+
+'use strict';
+
+const common = require('../common');
+const { Readable } = require('stream');
+
+const bench = common.createBenchmark(main, {
+  n: [5e6],
+});
+
+async function main({ n }) {
+  const b = {};
+  const r = new Readable({ objectMode: true });
+
+  let i = 0;
+
+  r._read = () => r.push(i++ === n ? null : b);
+
+  bench.start();
+  await r.map(i => i + 1).reduce((acc, i) => acc + i);
+  bench.end(n);
+}

From fd09668e66194f659e846c5abd91f06c81fd2a1f Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <lleggier@cisco.com>
Date: Fri, 26 Jun 2026 16:33:13 -0700
Subject: [PATCH 04/12] alternative implementation of find

---
 lib/internal/streams/operators.js | 90 +++++++++++++++----------------
 1 file changed, 43 insertions(+), 47 deletions(-)

diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js
index 0d83e2d3d1b9da..be8e1ebe99a476 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -41,6 +41,10 @@ const destroyImpl = require('internal/streams/destroy');
 const kEmpty = Symbol('kEmpty');
 const kEof = Symbol('kEof');
 
+const {
+  isPromise,
+} = require('internal/util/types');
+
 function map(fn, options) {
   validateFunction(fn, 'fn');
   if (options != null) {
@@ -322,7 +326,7 @@ function map(fn, options) {
 
 function nowOrLater(fn, fn2, args) {
   const valueOrPromise = fn(...args);
-  if (valueOrPromise && typeof valueOrPromise.then === 'function') {
+  if (isPromise(valueOrPromise)) {
       return PromisePrototypeThen(valueOrPromise, fn2);
     }
     return fn2(valueOrPromise);
@@ -356,94 +360,86 @@ async function find(fn, options) {
     validateAbortSignal(signal, 'options.signal');
   }
 
-  let concurrency = 1;
-  if (options?.concurrency != null) {
-    concurrency = MathFloor(options.concurrency);
-  }
+  const concurrency = MathFloor(options?.concurrency ?? 1);
   validateInteger(concurrency, 'options.concurrency', 1);
 
-  let destroyOnReturn = true;
-  if (options?.destroyOnReturn != null) {
-    destroyOnReturn = options?.destroyOnReturn;
-    validateBoolean(destroyOnReturn, 'options.destroyOnReturn');
-  }
-
-  const source = this;
+  const destroyOnReturn = options?.destroyOnReturn ?? true;
+  validateBoolean(destroyOnReturn, 'options.destroyOnReturn');
 
-  let result = undefined;
-  let foundIndex = undefined;
-  let currentConcurrency = 0;
-  let sequence = 0;
-  let continueLoop = null;
-  let error;
+  // Concurrent predicates can settle out of order. Stop reading after any
+  // match, but keep the lowest index after all active predicates settle.
+  let match;
+  let activeEvaluations = 0;
+  let resumeLoop;
+  let streamError;
 
-  function wake() {
-    if (continueLoop) {
-      continueLoop();
-      continueLoop = null;
+  function signalProgress() {
+    if (resumeLoop) {
+      resumeLoop();
+      resumeLoop = undefined;
     }
   }
 
   this.on('error', (err) => {
-    error = err;
+    streamError = err;
   });
-  this.on('readable', wake);
+  this.on('readable', signalProgress);
   const cleanup = eos(this, { writable: false }, (err) => {
-    error = err ? aggregateTwoErrors(error, err) : null;
-    wake();
+    streamError = err ? aggregateTwoErrors(streamError, err) : null;
+    signalProgress();
   });
 
-  async function consumeChunk(chunk, source, currentIndex) {
+  async function evaluateFn(stream, chunk, index) {
     try {
-      let v = fn(chunk, options);
-      if (v && typeof v.then === 'function') {
-        v = await v;
+      let matches = fn(chunk, options);
+      if (isPromise(matches)) {
+        matches = await matches;
       }
-      if (v && (foundIndex === undefined || foundIndex > currentIndex)) {
-        foundIndex = currentIndex;
-        result = chunk;
+      if (matches && (match === undefined || index < match.index)) {
+        match = { index, value: chunk };
       }
     } catch (err) {
-      destroyImpl.destroyer(source, err);
+      destroyImpl.destroyer(stream, err);
     } finally {
-      currentConcurrency--;
-      wake();
+      activeEvaluations--;
+      signalProgress();
     }
   }
 
   try {
+    let nextIndex = 0;
     while (true) {
       if (signal?.aborted) {
         throw new AbortError(undefined, { cause: signal.reason });
       }
 
-      if (currentConcurrency < concurrency && foundIndex === undefined) {
+      if (activeEvaluations < concurrency && match === undefined) {
         const chunk = this.destroyed ? null : this.read();
 
         if (chunk !== null) {
-          currentConcurrency++;
-          void consumeChunk(chunk, this, sequence++);
+          activeEvaluations++;
+          void evaluateFn(this, chunk, nextIndex++);
           continue;
         }
       }
 
-      if (error) {
-        throw error;
+      if (streamError) {
+        throw streamError;
       }
 
-      if (currentConcurrency === 0 && (foundIndex !== undefined || error === null)) {
+      if (activeEvaluations === 0 && (match !== undefined || streamError === null)) {
         break;
       }
 
-      const pr = PromiseWithResolvers();
-      continueLoop = pr.resolve;
-      await pr.promise;
+      const { promise, resolve } = PromiseWithResolvers();
+      resumeLoop = resolve;
+      await promise;
     }
   } catch (err) {
     destroyImpl.destroyer(this, err);
     throw err;
   } finally {
-    this.off('readable', wake);
+    this.off('readable', signalProgress);
     cleanup();
   }
 
@@ -451,7 +447,7 @@ async function find(fn, options) {
     destroyImpl.destroyer(this, null);
   }
 
-  return result;
+  return match?.value;
 }
 
 async function forEach(fn, options) {

From 95477d02a16dcf088a60638c26a270b968bf0865 Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <lleggier@cisco.com>
Date: Fri, 26 Jun 2026 16:39:20 -0700
Subject: [PATCH 05/12] alternative implementation of map

---
 lib/internal/streams/operators.js | 439 ++++++++++++++----------------
 1 file changed, 201 insertions(+), 238 deletions(-)

diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js
index be8e1ebe99a476..259bb472454213 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -45,6 +45,25 @@ const {
   isPromise,
 } = require('internal/util/types');
 
+function createWaiter() {
+  let resolve;
+
+  function wait() {
+    const deferred = PromiseWithResolvers();
+    resolve = deferred.resolve;
+    return deferred.promise;
+  }
+
+  function wake() {
+    if (resolve) {
+      resolve();
+      resolve = undefined;
+    }
+  }
+
+  return { __proto__: null, wait, wake };
+}
+
 function map(fn, options) {
   validateFunction(fn, 'fn');
   if (options != null) {
@@ -54,274 +73,218 @@ function map(fn, options) {
     validateAbortSignal(options.signal, 'options.signal');
   }
 
-  let concurrency = 1;
-  if (options?.concurrency != null) {
-    concurrency = MathFloor(options.concurrency);
-  }
-
-  let highWaterMark = concurrency - 1;
-  if (options?.highWaterMark != null) {
-    highWaterMark = MathFloor(options.highWaterMark);
-  }
+  const concurrency = MathFloor(options?.concurrency ?? 1);
+  const highWaterMark = MathFloor(options?.highWaterMark ?? concurrency - 1);
 
   validateInteger(concurrency, 'options.concurrency', 1);
   validateInteger(highWaterMark, 'options.highWaterMark', 0);
 
-  highWaterMark += concurrency;
-
-  return function map() {
-    const signal = AbortSignal.any([options?.signal].filter(Boolean));
-    const source = this;
-    const queue = [];
-    const signalOpt = { signal };
-
-    let continueReadingFromQueue;
-    let resumeReadingFromSource;
-    let continueReadingFromSource;
-    let stopReadingFromSource = false;
-    let currentConcurrency = 0;
-
-    function onCatch() {
-      stopReadingFromSource = true;
-      afterItemProcessed();
-      if (continueReadingFromQueue) {
-        continueReadingFromQueue();
-        continueReadingFromQueue = null;
-      }
-    }
+  return createMappedStream(this, fn, options, concurrency, highWaterMark + concurrency);
+}
 
-    function afterItemProcessed() {
-      currentConcurrency -= 1;
-      maybeResumeReadingFromSource();
-    }
+function createMappedStream(source, fn, options, concurrency, highWaterMark) {
+  const signal = AbortSignal.any([options?.signal].filter(Boolean));
+  const signalOpt = { signal };
+  const queue = [];
+  const sourceReadable = createWaiter();
+  const queueReadable = createWaiter();
+  const sourceCapacity = createWaiter();
 
-    function maybeResumeReadingFromSource() {
-      if (
-        resumeReadingFromSource &&
-        !stopReadingFromSource &&
-        currentConcurrency < concurrency &&
-        queue.length < highWaterMark
-      ) {
-        resumeReadingFromSource();
-        resumeReadingFromSource = null;
-      }
+  let stopped = false;
+  let activeMappers = 0;
+
+  function atCapacity() {
+    return activeMappers >= concurrency || queue.length >= highWaterMark;
+  }
+
+  function maybeResumeSource() {
+    if (!stopped && !atCapacity()) {
+      sourceCapacity.wake();
     }
+  }
 
-    async function readFromSource() {
-      try {
+  function mapperFinished() {
+    activeMappers--;
+    maybeResumeSource();
+  }
 
-        function moreDataToReadFromSource() {
-          if (continueReadingFromSource) {
-            continueReadingFromSource();
-            continueReadingFromSource = null;
-          }
-        }
+  function mapperRejected() {
+    stopped = true;
+    mapperFinished();
+    queueReadable.wake();
+  }
 
-        source.on('readable', moreDataToReadFromSource);
+  function enqueueMappedChunk(chunk) {
+    activeMappers++;
+    const mapped = fn(chunk, signalOpt);
 
-        let error;
-        const cleanup = eos(source, { writable: false }, (err) => {
-          error = err ? aggregateTwoErrors(error, err) : null;
-          moreDataToReadFromSource();
-        });
+    if (mapped === kEmpty) {
+      mapperFinished();
+      return mapped;
+    }
 
-        try {
-          while (true) {
-            if (stopReadingFromSource) {
-              return;
-            }
+    if (typeof mapped.then === 'function') {
+      PromisePrototypeThen(PromiseResolve(mapped), mapperFinished, mapperRejected);
+      queue.push(mapped);
+    } else {
+      queue.push({ chunk: mapped, afterItemProcessed: mapperFinished });
+    }
+    queueReadable.wake();
+    return mapped;
+  }
 
-            if (signal.aborted) {
-              throw new AbortError();
-            }
+  async function pumpSource() {
+    let error;
+    source.on('readable', sourceReadable.wake);
+    const cleanup = eos(source, { writable: false }, (err) => {
+      error = err ? aggregateTwoErrors(error, err) : null;
+      sourceReadable.wake();
+    });
 
-            if (queue.length >= highWaterMark || currentConcurrency >= concurrency) {
-              const pr = PromiseWithResolvers();
-              resumeReadingFromSource = pr.resolve;
-              await pr.promise;
-            }
+    try {
+      while (!stopped) {
+        if (signal.aborted) {
+          throw new AbortError();
+        }
+
+        if (atCapacity()) {
+          await sourceCapacity.wait();
+          continue;
+        }
 
-            let chunk = source.destroyed ? null : source.read();
-            if (chunk !== null) {
-              currentConcurrency += 1;
-              chunk = fn(chunk, signalOpt);
-
-              if (chunk === kEmpty) {
-                afterItemProcessed();
-                continue;
-              }
-              if (typeof chunk.then === 'function') {
-                PromisePrototypeThen(PromiseResolve(chunk), afterItemProcessed, onCatch);
-                queue.push(chunk);
-              } else {
-                queue.push({ chunk, afterItemProcessed });
-              }
-              if (continueReadingFromQueue) {
-                continueReadingFromQueue();
-                continueReadingFromQueue = null;
-              }
-              if (chunk === kEof) {
-                // Wait until queue is drained before exiting.
-                while (queue.length > 0) {
-                  const pr = PromiseWithResolvers();
-                  resumeReadingFromSource = pr.resolve;
-                  await pr.promise;
-                }
-                return;
-              }
-
-              if (!stopReadingFromSource && (queue.length >= highWaterMark || currentConcurrency >= concurrency)) {
-                const pr = PromiseWithResolvers();
-                resumeReadingFromSource = pr.resolve;
-                await pr.promise;
-              }
-            } else if (error) {
-              throw error;
-            } else if (error === null) {
-              queue.push(kEof);
-              return;
-            } else {
-              const pr = PromiseWithResolvers();
-              continueReadingFromSource = pr.resolve;
-              await pr.promise;
+        const chunk = source.destroyed ? null : source.read();
+        if (chunk !== null) {
+          const mapped = enqueueMappedChunk(chunk);
+          if (mapped === kEof) {
+            while (queue.length > 0) {
+              await sourceCapacity.wait();
             }
+            return;
           }
-        } catch (err) {
-          error = aggregateTwoErrors(error, err);
+          continue;
+        }
+
+        if (error) {
           throw error;
-        } finally {
-          if (
-            (error || options?.destroyOnReturn !== false) &&
-            (error === undefined || source._readableState.autoDestroy)
-          ) {
-            destroyImpl.destroyer(source, null);
-          } else {
-            source.off('readable', moreDataToReadFromSource);
-            cleanup();
-          }
         }
-      } catch (err) {
-        const val = PromiseReject(err);
-        PromisePrototypeThen(val, afterItemProcessed, onCatch);
-        queue.push(val);
-      } finally {
-        stopReadingFromSource = true;
-        if (continueReadingFromQueue) {
-          continueReadingFromQueue();
-          continueReadingFromQueue = null;
+        if (error === null) {
+          queue.push(kEof);
+          return;
         }
+        await sourceReadable.wait();
+      }
+    } catch (err) {
+      error = aggregateTwoErrors(error, err);
+      throw error;
+    } finally {
+      if (
+        (error || options?.destroyOnReturn !== false) &&
+        (error === undefined || source._readableState.autoDestroy)
+      ) {
+        destroyImpl.destroyer(source, null);
+      } else {
+        source.off('readable', sourceReadable.wake);
+        cleanup();
       }
     }
+  }
 
-    process.nextTick(readFromSource);
+  async function runSourcePump() {
+    try {
+      await pumpSource();
+    } catch (err) {
+      const rejected = PromiseReject(err);
+      PromisePrototypeThen(rejected, mapperFinished, mapperRejected);
+      queue.push(rejected);
+    } finally {
+      stopped = true;
+      queueReadable.wake();
+    }
+  }
 
-    const readable = new source.constructor({
-      objectMode: true,
-      highWaterMark,
-    });
-    let currentlyReadingFromQueue = false;
-    readable._read = async function readFromQueue() {
-      try {
-        if (!currentlyReadingFromQueue) {
-          currentlyReadingFromQueue = true;
-
-          if (!stopReadingFromSource && queue.length === 0) {
-            const pr = PromiseWithResolvers();
-            continueReadingFromQueue = pr.resolve;
-            await pr.promise;
-          }
-          while (queue.length > 0) {
-            if (signal.aborted) {
-              currentlyReadingFromQueue = false;
-              throw new AbortError();
-            }
+  const readable = new source.constructor({
+    objectMode: true,
+    highWaterMark,
+  });
+  let readingQueue = false;
 
-            let item = queue[0];
-            let syncAip = null;
-            if (typeof item.then === 'function') {
-              item = await item;
-              queue.shift();
-            } else if (typeof item.afterItemProcessed === 'function') {
-              syncAip = item.afterItemProcessed;
-              item = item.chunk;
-              // Call syncAip BEFORE shifting so queue.length stays >= 1,
-              // matching async item behavior (afterItemProcessed fires via
-              // PromiseThen while item is still peeked). This prevents
-              // maybeResumeReadingFromSource from immediately waking readFromSource.
-              syncAip();
-              queue.shift();
-            } else {
-              queue.shift();
-            }
+  readable._read = async function readFromQueue() {
+    if (readingQueue) {
+      return;
+    }
+    readingQueue = true;
 
-            if (item === kEof) {
-              stopReadingFromSource = true;
-              currentlyReadingFromQueue = false;
-              this.push(null);
-              if (resumeReadingFromSource) {
-                resumeReadingFromSource();
-                resumeReadingFromSource = null;
-              }
-              if (continueReadingFromQueue) {
-                continueReadingFromQueue();
-                continueReadingFromQueue = null;
-              }
-              return;
-            }
-            if (item !== kEmpty) {
-              if (syncAip) {
-                // Sync item: yield via await so that stream.read()'s preemptive
-                // _read call (triggered when state.length-n < hwm) sees kReading
-                // still set and does not re-enter _read before push clears it.
-                // Schedule maybeResume as a nextTick (not a microtask) so that
-                // the async-iterator cleanup microtask (done=true) from a
-                // consumer break runs before pump is woken, preventing an extra
-                // fn call on infinite streams.
-                await PromiseResolve();
-                if (!this.push(item)) {
-                  currentlyReadingFromQueue = false;
-                  process.nextTick(maybeResumeReadingFromSource);
-                  return;
-                }
-                process.nextTick(maybeResumeReadingFromSource);
-              } else if (!this.push(item)) {
-                currentlyReadingFromQueue = false;
-                process.nextTick(maybeResumeReadingFromSource);
-                return;
-              } else {
-                process.nextTick(maybeResumeReadingFromSource);
-              }
-            } else {
-              if (syncAip) syncAip();
-              maybeResumeReadingFromSource();
-              if (!stopReadingFromSource && queue.length === 0) {
-                const pr = PromiseWithResolvers();
-                continueReadingFromQueue = pr.resolve;
-                await pr.promise;
-              }
-              continue;
-            }
+    try {
+      if (!stopped && queue.length === 0) {
+        await queueReadable.wait();
+      }
+
+      while (queue.length > 0) {
+        if (signal.aborted) {
+          throw new AbortError();
+        }
+
+        let item = queue[0];
+        let syncItemFinished;
+        if (typeof item.then === 'function') {
+          item = await item;
+          queue.shift();
+        } else if (typeof item.afterItemProcessed === 'function') {
+          syncItemFinished = item.afterItemProcessed;
+          item = item.chunk;
+          // Keep the item queued until its mapper is marked as finished so
+          // async and synchronous items apply the same backpressure.
+          syncItemFinished();
+          queue.shift();
+        } else {
+          queue.shift();
+        }
+
+        if (item === kEof) {
+          stopped = true;
+          this.push(null);
+          sourceCapacity.wake();
+          queueReadable.wake();
+          return;
+        }
+
+        if (item === kEmpty) {
+          maybeResumeSource();
+          if (!stopped && queue.length === 0) {
+            await queueReadable.wait();
           }
-          currentlyReadingFromQueue = false;
+          continue;
+        }
+
+        if (syncItemFinished) {
+          // Prevent stream.read() from re-entering _read before push clears its
+          // internal reading flag.
+          await PromiseResolve();
+        }
+        const needsMore = this.push(item);
+        // Let async-iterator cleanup run before pulling another source chunk.
+        process.nextTick(maybeResumeSource);
+        if (!needsMore) {
+          return;
         }
-      } catch (err) {
-        destroyImpl.destroyer(this, err);
       }
+    } catch (err) {
+      destroyImpl.destroyer(this, err);
+    } finally {
+      readingQueue = false;
     }
-    readable._destroy = function(err, cb) {
-      stopReadingFromSource = true;
-      if (resumeReadingFromSource) {
-        resumeReadingFromSource();
-        resumeReadingFromSource = null;
-      }
-      if (continueReadingFromQueue) {
-        continueReadingFromQueue();
-        continueReadingFromQueue = null;
-      }
-      cb(err);
-    };
-    return readable;
-  }.call(this);
+  };
+
+  readable._destroy = function(err, cb) {
+    stopped = true;
+    sourceReadable.wake();
+    sourceCapacity.wake();
+    queueReadable.wake();
+    cb(err);
+  };
+
+  process.nextTick(runSourcePump);
+  return readable;
 }
 
 function nowOrLater(fn, fn2, args) {

From 51197a05e0d3840999aa9c2891b01507fe127854 Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <230980@gmail.com>
Date: Sat, 27 Jun 2026 12:37:00 -0700
Subject: [PATCH 06/12] Refine find

---
 lib/internal/streams/operators.js | 77 +++++++++++++++----------------
 1 file changed, 36 insertions(+), 41 deletions(-)

diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js
index 259bb472454213..0c951b6566bf28 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -47,18 +47,27 @@ const {
 
 function createWaiter() {
   let resolve;
+  let reject;
 
   function wait() {
     const deferred = PromiseWithResolvers();
     resolve = deferred.resolve;
+    reject = deferred.reject;
     return deferred.promise;
   }
 
-  function wake() {
-    if (resolve) {
-      resolve();
-      resolve = undefined;
+  function wake(err) {
+    if (err) {
+      if (reject) {
+        reject(err);
+      }
+    } else {
+      if (resolve) {
+        resolve();
+      }
     }
+    reject = undefined;
+    resolve = undefined;
   }
 
   return { __proto__: null, wait, wake };
@@ -329,28 +338,19 @@ async function find(fn, options) {
   const destroyOnReturn = options?.destroyOnReturn ?? true;
   validateBoolean(destroyOnReturn, 'options.destroyOnReturn');
 
+  const resumeLoop = createWaiter();
+
   // Concurrent predicates can settle out of order. Stop reading after any
   // match, but keep the lowest index after all active predicates settle.
   let match;
   let activeEvaluations = 0;
-  let resumeLoop;
-  let streamError;
 
-  function signalProgress() {
-    if (resumeLoop) {
-      resumeLoop();
-      resumeLoop = undefined;
-    }
-  }
-
-  this.on('error', (err) => {
-    streamError = err;
-  });
-  this.on('readable', signalProgress);
-  const cleanup = eos(this, { writable: false }, (err) => {
-    streamError = err ? aggregateTwoErrors(streamError, err) : null;
-    signalProgress();
+  this.on('error', function (err) {
+    destroyImpl.destroyer(this, err);
   });
+  this.on('error', resumeLoop.wake);
+  this.on('readable', resumeLoop.wake);
+  const cleanup = eos(this, { writable: false }, resumeLoop.wake);
 
   async function evaluateFn(stream, chunk, index) {
     try {
@@ -363,46 +363,41 @@ async function find(fn, options) {
       }
     } catch (err) {
       destroyImpl.destroyer(stream, err);
-    } finally {
-      activeEvaluations--;
-      signalProgress();
     }
+    activeEvaluations--;
+    resumeLoop.wake(stream.errored);
   }
 
   try {
     let nextIndex = 0;
-    while (true) {
+    while (match === undefined && !this.readableEnded) {
       if (signal?.aborted) {
         throw new AbortError(undefined, { cause: signal.reason });
       }
 
-      if (activeEvaluations < concurrency && match === undefined) {
+      if (activeEvaluations < concurrency) {
         const chunk = this.destroyed ? null : this.read();
-
-        if (chunk !== null) {
+        if (chunk === null) {
+          await resumeLoop.wait();
+        } else {
           activeEvaluations++;
           void evaluateFn(this, chunk, nextIndex++);
-          continue;
         }
+      } else {
+        await resumeLoop.wait();
       }
-
-      if (streamError) {
-        throw streamError;
-      }
-
-      if (activeEvaluations === 0 && (match !== undefined || streamError === null)) {
-        break;
-      }
-
-      const { promise, resolve } = PromiseWithResolvers();
-      resumeLoop = resolve;
-      await promise;
+    };
+    while (!signal?.aborted && activeEvaluations > 0) {
+      await resumeLoop.wait();
+    }
+    if (signal?.aborted) {
+      throw new AbortError(undefined, { cause: signal.reason });
     }
   } catch (err) {
     destroyImpl.destroyer(this, err);
     throw err;
   } finally {
-    this.off('readable', signalProgress);
+    this.off('readable', resumeLoop.wake);
     cleanup();
   }
 

From 97eefeb410515845e6992a05f048cf3c0f47f681 Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <230980@gmail.com>
Date: Sun, 5 Jul 2026 13:48:10 -0700
Subject: [PATCH 07/12] Refine toArray

---
 benchmark/streams/map-reduce.js          |  22 ----
 benchmark/streams/operator-throughput.js | 134 +++++++++++++++++++++++
 lib/internal/streams/from.js             |   8 +-
 lib/internal/streams/operators.js        | 133 ++++++++++++++++------
 4 files changed, 238 insertions(+), 59 deletions(-)
 delete mode 100644 benchmark/streams/map-reduce.js
 create mode 100644 benchmark/streams/operator-throughput.js

diff --git a/benchmark/streams/map-reduce.js b/benchmark/streams/map-reduce.js
deleted file mode 100644
index 17ea82c7662892..00000000000000
--- a/benchmark/streams/map-reduce.js
+++ /dev/null
@@ -1,22 +0,0 @@
-
-'use strict';
-
-const common = require('../common');
-const { Readable } = require('stream');
-
-const bench = common.createBenchmark(main, {
-  n: [5e6],
-});
-
-async function main({ n }) {
-  const b = {};
-  const r = new Readable({ objectMode: true });
-
-  let i = 0;
-
-  r._read = () => r.push(i++ === n ? null : b);
-
-  bench.start();
-  await r.map(i => i + 1).reduce((acc, i) => acc + i);
-  bench.end(n);
-}
diff --git a/benchmark/streams/operator-throughput.js b/benchmark/streams/operator-throughput.js
new file mode 100644
index 00000000000000..274e6cff4eb952
--- /dev/null
+++ b/benchmark/streams/operator-throughput.js
@@ -0,0 +1,134 @@
+'use strict';
+
+const assert = require('assert');
+const common = require('../common');
+const { Readable } = require('stream');
+
+const bench = common.createBenchmark(main, {
+  n: [2e5],
+  operation: [
+    'reduce',
+    'map-sync-1',
+    'map-sync-8',
+    'map-async-1',
+    'map-async-8',
+    'find-sync-1',
+    'find-sync-8',
+    'find-async-1',
+    'find-async-8',
+    'filter',
+    'forEach',
+    'toArray',
+  ],
+  source: ['readable', 'iterator', 'array'],
+});
+
+function createSource(type, n) {
+  if (type === 'array') {
+    return Readable.from(Array.from({ length: n }, (_, i) => i));
+  }
+  let i = 0;
+
+  if (type === 'readable') {
+    return new Readable({
+      objectMode: true,
+      read() {
+        this.push(i === n ? null : i++);
+      },
+    });
+  }
+
+  return Readable.from((function* generate() {
+    while (i < n) {
+      yield i++;
+    }
+  })());
+}
+
+async function main({ n, operation, source }) {
+  const readable = createSource(source, n);
+  let result;
+
+  bench.start();
+
+  switch (operation) {
+    case 'reduce':
+      result = await readable.reduce((sum, value) => sum + value, 0);
+      break;
+    case 'map-sync-1':
+      result = await readable
+        .map((value) => value + 1, { concurrency: 1 })
+        .reduce((sum, value) => sum + value, 0);
+      break;
+    case 'map-sync-8':
+      result = await readable
+        .map((value) => value + 1, { concurrency: 8 })
+        .reduce((sum, value) => sum + value, 0);
+      break;
+    case 'map-async-1':
+      result = await readable
+        .map(async (value) => value + 1, { concurrency: 1 })
+        .reduce((sum, value) => sum + value, 0);
+      break;
+    case 'map-async-8':
+      result = await readable
+        .map(async (value) => value + 1, { concurrency: 8 })
+        .reduce((sum, value) => sum + value, 0);
+      break;
+    case 'find-sync-1':
+      result = await readable.find((value) => value === n - 1, { concurrency: 1 });
+      break;
+    case 'find-sync-8':
+      result = await readable.find((value) => value === n - 1, { concurrency: 8 });
+      break;
+    case 'find-async-1':
+      result = await readable.find(async (value) => value === n - 1, { concurrency: 1 });
+      break;
+    case 'find-async-8':
+      result = await readable.find(async (value) => value === n - 1, { concurrency: 8 });
+      break;
+    case 'filter':
+      result = await readable
+        .filter((value) => (value & 1) === 0)
+        .reduce((sum, value) => sum + value, 0);
+      break;
+    case 'forEach':
+      result = 0;
+      await readable.forEach((value) => { result += value; });
+      break;
+    case 'toArray':
+      result = await readable.toArray();
+      break;
+    default:
+      throw new Error(`Unknown operation: ${operation}`);
+  }
+
+  bench.end(n);
+
+  const sum = n * (n - 1) / 2;
+  switch (operation) {
+    case 'map-sync-1':
+    case 'map-sync-8':
+    case 'map-async-1':
+    case 'map-async-8':
+      assert.strictEqual(result, sum + n);
+      break;
+    case 'find-sync-1':
+    case 'find-sync-8':
+    case 'find-async-1':
+    case 'find-async-8':
+      assert.strictEqual(result, n - 1);
+      break;
+    case 'filter': {
+      const evenCount = Math.ceil(n / 2);
+      assert.strictEqual(result, evenCount * (evenCount - 1));
+      break;
+    }
+    case 'toArray':
+      assert.strictEqual(result.length, n);
+      assert.strictEqual(result[n - 1], n - 1);
+      break;
+    default:
+      assert.strictEqual(result, sum);
+  }
+}
diff --git a/lib/internal/streams/from.js b/lib/internal/streams/from.js
index 73e696d66fae2b..52331d0c976d33 100644
--- a/lib/internal/streams/from.js
+++ b/lib/internal/streams/from.js
@@ -1,6 +1,7 @@
 'use strict';
 
 const {
+  ArrayIsArray,
   PromisePrototypeThen,
   SymbolAsyncIterator,
   SymbolIterator,
@@ -15,6 +16,10 @@ const {
   },
 } = require('internal/errors');
 
+const {
+  getDefaultHighWaterMark,
+} = require('internal/streams/state');
+
 function from(Readable, iterable, opts) {
   let iterator;
   if (typeof iterable === 'string' || iterable instanceof Buffer) {
@@ -41,10 +46,9 @@ function from(Readable, iterable, opts) {
     throw new ERR_INVALID_ARG_TYPE('iterable', ['Iterable'], iterable);
   }
 
-
   const readable = new Readable({
     objectMode: true,
-    highWaterMark: 1,
+    highWaterMark: ArrayIsArray(iterable) ? getDefaultHighWaterMark(true) : 1,
     // TODO(ronag): What options should be allowed?
     ...opts,
   });
diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js
index 0c951b6566bf28..5600519c3cc264 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -45,9 +45,10 @@ const {
   isPromise,
 } = require('internal/util/types');
 
+const nop = () => {};
 function createWaiter() {
-  let resolve;
-  let reject;
+  let resolve = nop;
+  let reject = nop;
 
   function wait() {
     const deferred = PromiseWithResolvers();
@@ -56,21 +57,19 @@ function createWaiter() {
     return deferred.promise;
   }
 
-  function wake(err) {
-    if (err) {
-      if (reject) {
-        reject(err);
-      }
-    } else {
-      if (resolve) {
-        resolve();
-      }
-    }
-    reject = undefined;
-    resolve = undefined;
+  function wake() {
+    resolve();
+    reject = nop;
+    resolve = nop;
   }
 
-  return { __proto__: null, wait, wake };
+  function fail(err) {
+    reject(err);
+    reject = nop;
+    resolve = nop;
+  }
+
+  return { __proto__: null, wait, wake, fail };
 }
 
 function map(fn, options) {
@@ -343,14 +342,15 @@ async function find(fn, options) {
   // Concurrent predicates can settle out of order. Stop reading after any
   // match, but keep the lowest index after all active predicates settle.
   let match;
+  let error;
   let activeEvaluations = 0;
 
-  this.on('error', function (err) {
-    destroyImpl.destroyer(this, err);
-  });
-  this.on('error', resumeLoop.wake);
   this.on('readable', resumeLoop.wake);
-  const cleanup = eos(this, { writable: false }, resumeLoop.wake);
+
+  const cleanup = eos(this, { writable: false }, (err) => {
+    error = err ? aggregateTwoErrors(error, err) : null;
+    err ? resumeLoop.fail(err) : resumeLoop.wake();
+  });
 
   async function evaluateFn(stream, chunk, index) {
     try {
@@ -365,23 +365,27 @@ async function find(fn, options) {
       destroyImpl.destroyer(stream, err);
     }
     activeEvaluations--;
-    resumeLoop.wake(stream.errored);
+    stream.errored ? resumeLoop.fail(stream.errored) : resumeLoop.wake();
   }
 
   try {
     let nextIndex = 0;
-    while (match === undefined && !this.readableEnded) {
+    while (match === undefined) {
       if (signal?.aborted) {
         throw new AbortError(undefined, { cause: signal.reason });
       }
 
       if (activeEvaluations < concurrency) {
         const chunk = this.destroyed ? null : this.read();
-        if (chunk === null) {
-          await resumeLoop.wait();
-        } else {
+        if (chunk !== null) {
           activeEvaluations++;
           void evaluateFn(this, chunk, nextIndex++);
+        } else if (error) {
+          throw error;
+        } else if (error === null) {
+          break;
+        } else {
+          await resumeLoop.wait();
         }
       } else {
         await resumeLoop.wait();
@@ -394,15 +398,18 @@ async function find(fn, options) {
       throw new AbortError(undefined, { cause: signal.reason });
     }
   } catch (err) {
-    destroyImpl.destroyer(this, err);
-    throw err;
+    error = aggregateTwoErrors(error, err);
+    throw error;
   } finally {
-    this.off('readable', resumeLoop.wake);
-    cleanup();
-  }
-
-  if (destroyOnReturn && !this.errored) {
-    destroyImpl.destroyer(this, null);
+    if (
+      (error || destroyOnReturn !== false) &&
+      (error === undefined || this._readableState.autoDestroy)
+    ) {
+      destroyImpl.destroyer(this, null);
+    } else {
+      this.off('readable', resumeLoop.wake);
+      cleanup();
+    }
   }
 
   return match?.value;
@@ -480,8 +487,64 @@ async function reduce(reducer, initialValue, options) {
 
 async function toArray(options) {
   const result = [];
-  const toArrayFn = (chunk) => ArrayPrototypePush(result, chunk);
-  await forEach.call(this, toArrayFn, options);
+
+  if (options != null) {
+    validateObject(options, 'options');
+  }
+  const signal = options?.signal;
+  if (signal != null) {
+    validateAbortSignal(signal, 'options.signal');
+  }
+
+  const destroyOnReturn = options?.destroyOnReturn ?? true;
+  validateBoolean(destroyOnReturn, 'options.destroyOnReturn');
+
+  const resumeLoop = createWaiter();
+
+  this.on('readable', resumeLoop.wake);
+
+  let error;
+
+  const cleanup = eos(this, { writable: false }, (err) => {
+    error = err ? aggregateTwoErrors(error, err) : null;
+    err ? resumeLoop.fail(err) : resumeLoop.wake();
+  });
+
+  try {
+    while (true) {
+      if (signal?.aborted) {
+        throw new AbortError(undefined, { cause: signal.reason });
+      }
+      const chunk = this.destroyed ? null : this.read();
+
+      if (chunk !== null) {
+        ArrayPrototypePush(result, chunk);
+      } else if (error) {
+        throw error;
+      } else if (error === null) {
+        break;
+      } else {
+        await resumeLoop.wait();
+      }
+    }
+    if (signal?.aborted) {
+      throw new AbortError(undefined, { cause: signal.reason });
+    }
+  } catch (err) {
+    error = aggregateTwoErrors(error, err);
+    throw error;
+  } finally {
+    if (
+      (error || destroyOnReturn !== false) &&
+      (error === undefined || this._readableState.autoDestroy)
+    ) {
+      destroyImpl.destroyer(this, null);
+    } else {
+      this.off('readable', resumeLoop.wake);
+      cleanup();
+    }
+  }
+
   return result;
 }
 

From 6edc13c35e67514aeb22207f405e4f9f4b67f26a Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <230980@gmail.com>
Date: Sun, 5 Jul 2026 14:05:24 -0700
Subject: [PATCH 08/12] Remove fail

---
 lib/internal/streams/operators.js | 19 +++++--------------
 1 file changed, 5 insertions(+), 14 deletions(-)

diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js
index 5600519c3cc264..00c76c01574cbe 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -48,28 +48,19 @@ const {
 const nop = () => {};
 function createWaiter() {
   let resolve = nop;
-  let reject = nop;
 
   function wait() {
     const deferred = PromiseWithResolvers();
     resolve = deferred.resolve;
-    reject = deferred.reject;
     return deferred.promise;
   }
 
   function wake() {
     resolve();
-    reject = nop;
     resolve = nop;
   }
 
-  function fail(err) {
-    reject(err);
-    reject = nop;
-    resolve = nop;
-  }
-
-  return { __proto__: null, wait, wake, fail };
+  return { __proto__: null, wait, wake };
 }
 
 function map(fn, options) {
@@ -317,7 +308,7 @@ async function every(fn, options = undefined) {
     return nowOrLater(fn, value => !value, args);
   };
   // https://en.wikipedia.org/wiki/De_Morgan%27s_laws
-  return !(await some.call(this, everyFn, options));
+  return !(await find.call(this, everyFn, options));
 }
 
 async function find(fn, options) {
@@ -349,7 +340,7 @@ async function find(fn, options) {
 
   const cleanup = eos(this, { writable: false }, (err) => {
     error = err ? aggregateTwoErrors(error, err) : null;
-    err ? resumeLoop.fail(err) : resumeLoop.wake();
+    resumeLoop.wake();
   });
 
   async function evaluateFn(stream, chunk, index) {
@@ -365,7 +356,7 @@ async function find(fn, options) {
       destroyImpl.destroyer(stream, err);
     }
     activeEvaluations--;
-    stream.errored ? resumeLoop.fail(stream.errored) : resumeLoop.wake();
+    resumeLoop.wake();
   }
 
   try {
@@ -507,7 +498,7 @@ async function toArray(options) {
 
   const cleanup = eos(this, { writable: false }, (err) => {
     error = err ? aggregateTwoErrors(error, err) : null;
-    err ? resumeLoop.fail(err) : resumeLoop.wake();
+    resumeLoop.wake();
   });
 
   try {

From 2cbb0c7bf4e0097a29a6eef1f606774a4f514ad1 Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <230980@gmail.com>
Date: Sun, 5 Jul 2026 15:44:31 -0700
Subject: [PATCH 09/12] New implementation for toArray

---
 lib/internal/streams/operators.js | 99 +++++++++++--------------------
 1 file changed, 36 insertions(+), 63 deletions(-)

diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js
index 00c76c01574cbe..8c765ee1905411 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -49,18 +49,20 @@ const nop = () => {};
 function createWaiter() {
   let resolve = nop;
 
+  function reset(r) {
+    resolve = r;
+  }
+
   function wait() {
-    const deferred = PromiseWithResolvers();
-    resolve = deferred.resolve;
-    return deferred.promise;
+    return new Promise(reset);
   }
 
-  function wake() {
+  function notify() {
     resolve();
     resolve = nop;
   }
 
-  return { __proto__: null, wait, wake };
+  return { __proto__: null, wait, reset, notify };
 }
 
 function map(fn, options) {
@@ -98,7 +100,7 @@ function createMappedStream(source, fn, options, concurrency, highWaterMark) {
 
   function maybeResumeSource() {
     if (!stopped && !atCapacity()) {
-      sourceCapacity.wake();
+      sourceCapacity.notify();
     }
   }
 
@@ -110,7 +112,7 @@ function createMappedStream(source, fn, options, concurrency, highWaterMark) {
   function mapperRejected() {
     stopped = true;
     mapperFinished();
-    queueReadable.wake();
+    queueReadable.notify();
   }
 
   function enqueueMappedChunk(chunk) {
@@ -128,16 +130,16 @@ function createMappedStream(source, fn, options, concurrency, highWaterMark) {
     } else {
       queue.push({ chunk: mapped, afterItemProcessed: mapperFinished });
     }
-    queueReadable.wake();
+    queueReadable.notify();
     return mapped;
   }
 
   async function pumpSource() {
     let error;
-    source.on('readable', sourceReadable.wake);
+    source.on('readable', sourceReadable.notify);
     const cleanup = eos(source, { writable: false }, (err) => {
       error = err ? aggregateTwoErrors(error, err) : null;
-      sourceReadable.wake();
+      sourceReadable.notify();
     });
 
     try {
@@ -182,7 +184,7 @@ function createMappedStream(source, fn, options, concurrency, highWaterMark) {
       ) {
         destroyImpl.destroyer(source, null);
       } else {
-        source.off('readable', sourceReadable.wake);
+        source.off('readable', sourceReadable.notify);
         cleanup();
       }
     }
@@ -197,7 +199,7 @@ function createMappedStream(source, fn, options, concurrency, highWaterMark) {
       queue.push(rejected);
     } finally {
       stopped = true;
-      queueReadable.wake();
+      queueReadable.notify();
     }
   }
 
@@ -242,8 +244,8 @@ function createMappedStream(source, fn, options, concurrency, highWaterMark) {
         if (item === kEof) {
           stopped = true;
           this.push(null);
-          sourceCapacity.wake();
-          queueReadable.wake();
+          sourceCapacity.notify();
+          queueReadable.notify();
           return;
         }
 
@@ -276,9 +278,9 @@ function createMappedStream(source, fn, options, concurrency, highWaterMark) {
 
   readable._destroy = function(err, cb) {
     stopped = true;
-    sourceReadable.wake();
-    sourceCapacity.wake();
-    queueReadable.wake();
+    sourceReadable.notify();
+    sourceCapacity.notify();
+    queueReadable.notify();
     cb(err);
   };
 
@@ -336,11 +338,11 @@ async function find(fn, options) {
   let error;
   let activeEvaluations = 0;
 
-  this.on('readable', resumeLoop.wake);
+  this.on('readable', resumeLoop.notify);
 
   const cleanup = eos(this, { writable: false }, (err) => {
     error = err ? aggregateTwoErrors(error, err) : null;
-    resumeLoop.wake();
+    resumeLoop.notify();
   });
 
   async function evaluateFn(stream, chunk, index) {
@@ -356,7 +358,7 @@ async function find(fn, options) {
       destroyImpl.destroyer(stream, err);
     }
     activeEvaluations--;
-    resumeLoop.wake();
+    resumeLoop.notify();
   }
 
   try {
@@ -385,9 +387,6 @@ async function find(fn, options) {
     while (!signal?.aborted && activeEvaluations > 0) {
       await resumeLoop.wait();
     }
-    if (signal?.aborted) {
-      throw new AbortError(undefined, { cause: signal.reason });
-    }
   } catch (err) {
     error = aggregateTwoErrors(error, err);
     throw error;
@@ -398,7 +397,7 @@ async function find(fn, options) {
     ) {
       destroyImpl.destroyer(this, null);
     } else {
-      this.off('readable', resumeLoop.wake);
+      this.off('readable', resumeLoop.notify);
       cleanup();
     }
   }
@@ -477,8 +476,6 @@ async function reduce(reducer, initialValue, options) {
 }
 
 async function toArray(options) {
-  const result = [];
-
   if (options != null) {
     validateObject(options, 'options');
   }
@@ -490,52 +487,28 @@ async function toArray(options) {
   const destroyOnReturn = options?.destroyOnReturn ?? true;
   validateBoolean(destroyOnReturn, 'options.destroyOnReturn');
 
-  const resumeLoop = createWaiter();
-
-  this.on('readable', resumeLoop.wake);
-
-  let error;
-
-  const cleanup = eos(this, { writable: false }, (err) => {
-    error = err ? aggregateTwoErrors(error, err) : null;
-    resumeLoop.wake();
-  });
+  const result = [];
 
-  try {
+  function onReadable() {
     while (true) {
-      if (signal?.aborted) {
-        throw new AbortError(undefined, { cause: signal.reason });
-      }
       const chunk = this.destroyed ? null : this.read();
-
       if (chunk !== null) {
         ArrayPrototypePush(result, chunk);
-      } else if (error) {
-        throw error;
-      } else if (error === null) {
-        break;
-      } else {
-        await resumeLoop.wait();
+        continue;
+      } else if (signal?.aborted) {
+        this.destroy(new AbortError(undefined, { cause: signal.reason }));
       }
-    }
-    if (signal?.aborted) {
-      throw new AbortError(undefined, { cause: signal.reason });
-    }
-  } catch (err) {
-    error = aggregateTwoErrors(error, err);
-    throw error;
-  } finally {
-    if (
-      (error || destroyOnReturn !== false) &&
-      (error === undefined || this._readableState.autoDestroy)
-    ) {
-      destroyImpl.destroyer(this, null);
-    } else {
-      this.off('readable', resumeLoop.wake);
-      cleanup();
+      return;
     }
   }
 
+  this.on('readable', onReadable);
+
+  await finished(this, { writable: false, cleanup: destroyOnReturn });
+  this.off('readable', onReadable);
+  if (destroyOnReturn !== false) {
+    destroyImpl.destroyer(this, null);
+  }
   return result;
 }
 

From 35f921f4bfd3a0f07cb54312c621a9267e8b4be3 Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <230980@gmail.com>
Date: Sun, 5 Jul 2026 16:01:21 -0700
Subject: [PATCH 10/12] New implementation for find

---
 lib/internal/streams/operators.js | 173 +++++++++++++++++++++---------
 1 file changed, 123 insertions(+), 50 deletions(-)

diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js
index 8c765ee1905411..0ad058da3aaa09 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -313,7 +313,7 @@ async function every(fn, options = undefined) {
   return !(await find.call(this, everyFn, options));
 }
 
-async function find(fn, options) {
+function find(fn, options) {
   validateFunction(fn, 'fn');
 
   if (options != null) {
@@ -330,79 +330,152 @@ async function find(fn, options) {
   const destroyOnReturn = options?.destroyOnReturn ?? true;
   validateBoolean(destroyOnReturn, 'options.destroyOnReturn');
 
-  const resumeLoop = createWaiter();
-
   // Concurrent predicates can settle out of order. Stop reading after any
   // match, but keep the lowest index after all active predicates settle.
+  const stream = this;
+  const { promise, resolve } = PromiseWithResolvers();
   let match;
   let error;
   let activeEvaluations = 0;
+  let nextIndex = 0;
+  let ended = false;
+  let settled = false;
+  let draining = false;
+
+  function settle() {
+    if (!settled) {
+      settled = true;
+      resolve();
+    }
+  }
 
-  this.on('readable', resumeLoop.notify);
+  function fail(err) {
+    if (settled) {
+      return;
+    }
+    error = aggregateTwoErrors(error, err);
+    destroyImpl.destroyer(stream, error);
+    settle();
+  }
 
-  const cleanup = eos(this, { writable: false }, (err) => {
-    error = err ? aggregateTwoErrors(error, err) : null;
-    resumeLoop.notify();
-  });
+  function maybeSettle() {
+    if (activeEvaluations === 0 && (match !== undefined || ended)) {
+      settle();
+    }
+  }
+
+  function evaluationFinished(matches, chunk, index) {
+    if (matches && (match === undefined || index < match.index)) {
+      match = { index, value: chunk };
+    }
+    activeEvaluations--;
+    maybeSettle();
 
-  async function evaluateFn(stream, chunk, index) {
+    if (!settled && match === undefined && !draining) {
+      onReadable();
+    }
+  }
+
+  function evaluationRejected(err) {
+    activeEvaluations--;
+    fail(err);
+  }
+
+  function evaluate(chunk, index) {
+    activeEvaluations++;
+
+    let matches;
     try {
-      let matches = fn(chunk, options);
-      if (isPromise(matches)) {
-        matches = await matches;
-      }
-      if (matches && (match === undefined || index < match.index)) {
-        match = { index, value: chunk };
-      }
+      matches = fn(chunk, options);
     } catch (err) {
-      destroyImpl.destroyer(stream, err);
+      evaluationRejected(err);
+      return;
+    }
+
+    if (isPromise(matches)) {
+      PromisePrototypeThen(
+        matches,
+        (result) => evaluationFinished(result, chunk, index),
+        evaluationRejected,
+      );
+    } else {
+      evaluationFinished(matches, chunk, index);
     }
-    activeEvaluations--;
-    resumeLoop.notify();
   }
 
-  try {
-    let nextIndex = 0;
-    while (match === undefined) {
-      if (signal?.aborted) {
-        throw new AbortError(undefined, { cause: signal.reason });
-      }
+  function onReadable() {
+    if (draining || settled || match !== undefined) {
+      return;
+    }
 
-      if (activeEvaluations < concurrency) {
-        const chunk = this.destroyed ? null : this.read();
-        if (chunk !== null) {
-          activeEvaluations++;
-          void evaluateFn(this, chunk, nextIndex++);
-        } else if (error) {
-          throw error;
-        } else if (error === null) {
-          break;
-        } else {
-          await resumeLoop.wait();
+    draining = true;
+    try {
+      while (
+        !settled &&
+        match === undefined &&
+        activeEvaluations < concurrency
+      ) {
+        if (signal?.aborted) {
+          fail(new AbortError(undefined, { cause: signal.reason }));
+          return;
         }
-      } else {
-        await resumeLoop.wait();
+
+        const chunk = stream.destroyed ? null : stream.read();
+        if (chunk === null) {
+          return;
+        }
+        evaluate(chunk, nextIndex++);
       }
-    };
-    while (!signal?.aborted && activeEvaluations > 0) {
-      await resumeLoop.wait();
+    } catch (err) {
+      fail(err);
+    } finally {
+      draining = false;
     }
-  } catch (err) {
-    error = aggregateTwoErrors(error, err);
-    throw error;
-  } finally {
+  }
+
+  function onAbort() {
+    fail(new AbortError(undefined, { cause: signal.reason }));
+  }
+
+  stream.on('readable', onReadable);
+
+  const cleanup = eos(stream, { writable: false }, (err) => {
+    if (settled) {
+      return;
+    }
+    if (err) {
+      fail(err);
+      return;
+    }
+    ended = true;
+    maybeSettle();
+  });
+
+  if (signal != null) {
+    signal.addEventListener('abort', onAbort, { once: true });
+    if (signal.aborted) {
+      onAbort();
+    }
+  }
+
+  return PromisePrototypeThen(promise, () => {
+    stream.off('readable', onReadable);
+    signal?.removeEventListener('abort', onAbort);
+
     if (
       (error || destroyOnReturn !== false) &&
-      (error === undefined || this._readableState.autoDestroy)
+      (error === undefined || stream._readableState.autoDestroy)
     ) {
-      destroyImpl.destroyer(this, null);
+      destroyImpl.destroyer(stream, error);
     } else {
-      this.off('readable', resumeLoop.notify);
       cleanup();
     }
-  }
 
-  return match?.value;
+    if (error) {
+      return PromiseReject(error);
+    }
+    return match?.value;
+  });
 }
 
 async function forEach(fn, options) {

From 333a8de3e899585dd20fc67a85cbd1b7918183cb Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <230980@gmail.com>
Date: Sun, 5 Jul 2026 16:39:29 -0700
Subject: [PATCH 11/12] New AI-based implementation for reduce

---
 benchmark/streams/operator-throughput.js |   4 +
 lib/internal/streams/operators.js        | 200 +++++++++++++++++++----
 2 files changed, 173 insertions(+), 31 deletions(-)

diff --git a/benchmark/streams/operator-throughput.js b/benchmark/streams/operator-throughput.js
index 274e6cff4eb952..9256c160e49d45 100644
--- a/benchmark/streams/operator-throughput.js
+++ b/benchmark/streams/operator-throughput.js
@@ -8,6 +8,7 @@ const bench = common.createBenchmark(main, {
   n: [2e5],
   operation: [
     'reduce',
+    'reduce-async',
     'map-sync-1',
     'map-sync-8',
     'map-async-1',
@@ -55,6 +56,9 @@ async function main({ n, operation, source }) {
     case 'reduce':
       result = await readable.reduce((sum, value) => sum + value, 0);
       break;
+    case 'reduce-async':
+      result = await readable.reduce(async (sum, value) => sum + value, 0);
+      break;
     case 'map-sync-1':
       result = await readable
         .map((value) => value + 1, { concurrency: 1 })
diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js
index 0ad058da3aaa09..a56ab8bcb6100f 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -13,6 +13,7 @@ const {
   PromiseReject,
   PromiseResolve,
   PromiseWithResolvers,
+  ReflectApply,
   Symbol,
   SymbolIterator,
 } = primordials;
@@ -503,49 +504,186 @@ class ReduceAwareErrMissingArgs extends ERR_MISSING_ARGS {
   }
 }
 
-async function reduce(reducer, initialValue, options) {
+function reduce(reducer, initialValue, options) {
+  try {
+    return reduceImpl(this, arguments.length > 1, reducer, initialValue, options);
+  } catch (err) {
+    return PromiseReject(err);
+  }
+}
+
+function reduceImpl(stream, hasInitialValue, reducer, initialValue, options) {
   validateFunction(reducer, 'reducer');
   if (options != null) {
     validateObject(options, 'options');
   }
-  if (options?.signal != null) {
-    validateAbortSignal(options.signal, 'options.signal');
+  const signal = options?.signal;
+  if (signal != null) {
+    validateAbortSignal(signal, 'options.signal');
   }
 
-  let hasInitialValue = arguments.length > 1;
-  if (options?.signal?.aborted) {
-    const err = new AbortError(undefined, { cause: options.signal.reason });
-    this.once('error', () => {}); // The error is already propagated
-    await finished(this.destroy(err));
-    throw err;
-  }
+  const { promise, resolve } = PromiseWithResolvers();
   const ac = new AbortController();
-  const signal = ac.signal;
-  if (options?.signal) {
-    const opts = { once: true, [kWeakHandler]: this, [kResistStopPropagation]: true };
-    options.signal.addEventListener('abort', () => ac.abort(), opts);
+  const reducerSignal = ac.signal;
+  const reducerOptions = { signal: reducerSignal };
+  let reducing = false;
+  let draining = false;
+  let ended = false;
+  let settled = false;
+  let error;
+
+  function settle() {
+    if (!settled) {
+      settled = true;
+      resolve();
+    }
   }
-  let gotAnyItemFromStream = false;
-  try {
-    for await (const value of this) {
-      gotAnyItemFromStream = true;
-      if (options?.signal?.aborted) {
-        throw new AbortError();
-      }
-      if (!hasInitialValue) {
-        initialValue = value;
-        hasInitialValue = true;
-      } else {
-        initialValue = await reducer(initialValue, value, { signal });
-      }
+
+  function fail(err) {
+    if (settled) {
+      return;
     }
-    if (!gotAnyItemFromStream && !hasInitialValue) {
-      throw new ReduceAwareErrMissingArgs();
+    error = aggregateTwoErrors(error, err);
+    destroyImpl.destroyer(stream, error);
+    settle();
+  }
+
+  function finish() {
+    if (reducing || settled) {
+      return;
     }
-  } finally {
+    if (!hasInitialValue) {
+      fail(new ReduceAwareErrMissingArgs());
+      return;
+    }
+    settle();
+  }
+
+  function reducerFulfilled(value) {
+    reducing = false;
+    initialValue = value;
+    if (settled) {
+      return;
+    }
+    if (ended) {
+      finish();
+    } else {
+      onReadable();
+    }
+  }
+
+  function reducerRejected(err) {
+    reducing = false;
+    fail(err);
+  }
+
+  function onReadable() {
+    if (draining || reducing || settled) {
+      return;
+    }
+
+    draining = true;
+    try {
+      while (!reducing && !settled) {
+        if (signal?.aborted) {
+          fail(new AbortError(undefined, { cause: signal.reason }));
+          break;
+        }
+
+        const value = stream.destroyed ? null : stream.read();
+        if (value === null) {
+          break;
+        }
+
+        if (!hasInitialValue) {
+          initialValue = value;
+          hasInitialValue = true;
+          continue;
+        }
+
+        let result;
+        let resultPromise;
+        try {
+          result = reducer(initialValue, value, reducerOptions);
+          const resultType = typeof result;
+          if ((resultType !== 'object' || result === null) && resultType !== 'function') {
+            initialValue = result;
+            continue;
+          }
+
+          if (isPromise(result)) {
+            resultPromise = result;
+          } else {
+            const then = result.then;
+            if (typeof then !== 'function') {
+              initialValue = result;
+              continue;
+            }
+            resultPromise = PromiseResolve({
+              then(resolve, reject) {
+                ReflectApply(then, result, [resolve, reject]);
+              },
+            });
+          }
+        } catch (err) {
+          fail(err);
+          break;
+        }
+
+        reducing = true;
+        PromisePrototypeThen(resultPromise, reducerFulfilled, reducerRejected);
+      }
+    } catch (err) {
+      fail(err);
+    } finally {
+      draining = false;
+    }
+  }
+
+  function onAbort() {
     ac.abort();
+    fail(new AbortError(undefined, { cause: signal.reason }));
   }
-  return initialValue;
+
+  stream.on('readable', onReadable);
+
+  const cleanup = eos(stream, { writable: false }, (err) => {
+    if (settled) {
+      return;
+    }
+    if (err) {
+      fail(err);
+      return;
+    }
+    ended = true;
+    finish();
+  });
+
+  if (signal != null) {
+    const opts = {
+      once: true,
+      [kWeakHandler]: stream,
+      [kResistStopPropagation]: true,
+    };
+    signal.addEventListener('abort', onAbort, opts);
+    if (signal.aborted) {
+      onAbort();
+    }
+  }
+
+  return PromisePrototypeThen(promise, () => {
+    stream.off('readable', onReadable);
+    signal?.removeEventListener('abort', onAbort);
+    ac.abort();
+
+    if (error) {
+      destroyImpl.destroyer(stream, error);
+      return PromiseReject(error);
+    }
+
+    cleanup();
+    return initialValue;
+  });
 }
 
 async function toArray(options) {

From fae33f08291806b4fe8e815ec01075182f7d2422 Mon Sep 17 00:00:00 2001
From: Luciano Leggieri <230980@gmail.com>
Date: Sun, 5 Jul 2026 16:40:54 -0700
Subject: [PATCH 12/12] Remove TODO

---
 lib/internal/streams/operators.js | 1 -
 1 file changed, 1 deletion(-)

diff --git a/lib/internal/streams/operators.js b/lib/internal/streams/operators.js
index a56ab8bcb6100f..0a2ab01db36f3b 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -725,7 +725,6 @@ async function toArray(options) {
 
 function flatMap(fn, options) {
   const values = map.call(this, fn, options);
-  // TODO lucho
   return async function* flatMap() {
     for await (const val of values) {
       yield* val;