diff --git a/benchmark/streams/operator-throughput.js b/benchmark/streams/operator-throughput.js
new file mode 100644
index 00000000000000..9256c160e49d45
--- /dev/null
+++ b/benchmark/streams/operator-throughput.js
@@ -0,0 +1,138 @@
+'use strict';
+
+const assert = require('assert');
+const common = require('../common');
+const { Readable } = require('stream');
+
+const bench = common.createBenchmark(main, {
+  n: [2e5],
+  operation: [
+    'reduce',
+    'reduce-async',
+    '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 '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 })
+        .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 6a943f02657924..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) {
@@ -26,6 +31,8 @@ function from(Readable, iterable, opts) {
         this.push(null);
       },
     });
+  } else if (iterable instanceof Readable) {
+    return iterable;
   }
 
   let isAsync;
@@ -39,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 6db2df0e3646e0..0a2ab01db36f3b 100644
--- a/lib/internal/streams/operators.js
+++ b/lib/internal/streams/operators.js
@@ -2,21 +2,27 @@
 
 const {
   ArrayPrototypePush,
+  ArrayFrom,
   Boolean,
   MathFloor,
   Number,
   NumberIsNaN,
   Promise,
+  PromiseAllSettled,
   PromisePrototypeThen,
   PromiseReject,
   PromiseResolve,
+  PromiseWithResolvers,
+  ReflectApply,
   Symbol,
+  SymbolIterator,
 } = primordials;
 
 const { AbortController, AbortSignal } = require('internal/abort_controller');
 
 const {
   AbortError,
+  aggregateTwoErrors,
   codes: {
     ERR_MISSING_ARGS,
     ERR_OUT_OF_RANGE,
@@ -27,13 +33,39 @@ 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');
 
+const {
+  isPromise,
+} = require('internal/util/types');
+
+const nop = () => {};
+function createWaiter() {
+  let resolve = nop;
+
+  function reset(r) {
+    resolve = r;
+  }
+
+  function wait() {
+    return new Promise(reset);
+  }
+
+  function notify() {
+    resolve();
+    resolve = nop;
+  }
+
+  return { __proto__: null, wait, reset, notify };
+}
+
 function map(fn, options) {
   validateFunction(fn, 'fn');
   if (options != null) {
@@ -43,183 +75,422 @@ 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 createMappedStream(this, fn, options, concurrency, highWaterMark + concurrency);
+}
 
-  return async function* map() {
-    const signal = AbortSignal.any([options?.signal].filter(Boolean));
-    const stream = this;
-    const queue = [];
-    const signalOpt = { signal };
+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();
 
-    let next;
-    let resume;
-    let done = false;
-    let cnt = 0;
+  let stopped = false;
+  let activeMappers = 0;
 
-    function onCatch() {
-      done = true;
-      afterItemProcessed();
-    }
+  function atCapacity() {
+    return activeMappers >= concurrency || queue.length >= highWaterMark;
+  }
 
-    function afterItemProcessed() {
-      cnt -= 1;
-      maybeResume();
+  function maybeResumeSource() {
+    if (!stopped && !atCapacity()) {
+      sourceCapacity.notify();
     }
+  }
 
-    function maybeResume() {
-      if (
-        resume &&
-        !done &&
-        cnt < concurrency &&
-        queue.length < highWaterMark
-      ) {
-        resume();
-        resume = null;
-      }
-    }
+  function mapperFinished() {
+    activeMappers--;
+    maybeResumeSource();
+  }
 
-    async function pump() {
-      try {
-        for await (let val of stream) {
-          if (done) {
-            return;
-          }
+  function mapperRejected() {
+    stopped = true;
+    mapperFinished();
+    queueReadable.notify();
+  }
 
-          if (signal.aborted) {
-            throw new AbortError();
-          }
+  function enqueueMappedChunk(chunk) {
+    activeMappers++;
+    const mapped = fn(chunk, signalOpt);
 
-          try {
-            val = fn(val, signalOpt);
+    if (mapped === kEmpty) {
+      mapperFinished();
+      return mapped;
+    }
 
-            if (val === kEmpty) {
-              continue;
-            }
+    if (typeof mapped.then === 'function') {
+      PromisePrototypeThen(PromiseResolve(mapped), mapperFinished, mapperRejected);
+      queue.push(mapped);
+    } else {
+      queue.push({ chunk: mapped, afterItemProcessed: mapperFinished });
+    }
+    queueReadable.notify();
+    return mapped;
+  }
 
-            val = PromiseResolve(val);
-          } catch (err) {
-            val = PromiseReject(err);
-          }
+  async function pumpSource() {
+    let error;
+    source.on('readable', sourceReadable.notify);
+    const cleanup = eos(source, { writable: false }, (err) => {
+      error = err ? aggregateTwoErrors(error, err) : null;
+      sourceReadable.notify();
+    });
 
-          cnt += 1;
+    try {
+      while (!stopped) {
+        if (signal.aborted) {
+          throw new AbortError();
+        }
 
-          PromisePrototypeThen(val, afterItemProcessed, onCatch);
+        if (atCapacity()) {
+          await sourceCapacity.wait();
+          continue;
+        }
 
-          queue.push(val);
-          if (next) {
-            next();
-            next = null;
+        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;
           }
+          continue;
+        }
 
-          if (!done && (queue.length >= highWaterMark || cnt >= concurrency)) {
-            await new Promise((resolve) => {
-              resume = resolve;
-            });
-          }
+        if (error) {
+          throw error;
         }
-        queue.push(kEof);
-      } catch (err) {
-        const val = PromiseReject(err);
-        PromisePrototypeThen(val, afterItemProcessed, onCatch);
-        queue.push(val);
-      } finally {
-        done = true;
-        if (next) {
-          next();
-          next = 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.notify);
+        cleanup();
       }
     }
+  }
 
-    pump();
-
+  async function runSourcePump() {
     try {
-      while (true) {
-        while (queue.length > 0) {
-          const val = await queue[0];
+      await pumpSource();
+    } catch (err) {
+      const rejected = PromiseReject(err);
+      PromisePrototypeThen(rejected, mapperFinished, mapperRejected);
+      queue.push(rejected);
+    } finally {
+      stopped = true;
+      queueReadable.notify();
+    }
+  }
 
-          if (val === kEof) {
-            return;
-          }
+  const readable = new source.constructor({
+    objectMode: true,
+    highWaterMark,
+  });
+  let readingQueue = false;
 
-          if (signal.aborted) {
-            throw new AbortError();
-          }
+  readable._read = async function readFromQueue() {
+    if (readingQueue) {
+      return;
+    }
+    readingQueue = true;
 
-          if (val !== kEmpty) {
-            yield val;
-          }
+    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();
-          maybeResume();
+        } 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.notify();
+          queueReadable.notify();
+          return;
+        }
+
+        if (item === kEmpty) {
+          maybeResumeSource();
+          if (!stopped && queue.length === 0) {
+            await queueReadable.wait();
+          }
+          continue;
         }
 
-        await new Promise((resolve) => {
-          next = resolve;
-        });
+        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);
     } finally {
-      done = true;
-      if (resume) {
-        resume();
-        resume = null;
-      }
+      readingQueue = false;
     }
-  }.call(this);
+  };
+
+  readable._destroy = function(err, cb) {
+    stopped = true;
+    sourceReadable.notify();
+    sourceCapacity.notify();
+    queueReadable.notify();
+    cb(err);
+  };
+
+  process.nextTick(runSourcePump);
+  return readable;
+}
+
+function nowOrLater(fn, fn2, args) {
+  const valueOrPromise = fn(...args);
+  if (isPromise(valueOrPromise)) {
+      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 find.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');
+  }
+
+  const concurrency = MathFloor(options?.concurrency ?? 1);
+  validateInteger(concurrency, 'options.concurrency', 1);
+
+  const destroyOnReturn = options?.destroyOnReturn ?? true;
+  validateBoolean(destroyOnReturn, 'options.destroyOnReturn');
+
+  // 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();
+    }
+  }
+
+  function fail(err) {
+    if (settled) {
+      return;
+    }
+    error = aggregateTwoErrors(error, err);
+    destroyImpl.destroyer(stream, error);
+    settle();
+  }
+
+  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();
+
+    if (!settled && match === undefined && !draining) {
+      onReadable();
+    }
+  }
+
+  function evaluationRejected(err) {
+    activeEvaluations--;
+    fail(err);
+  }
+
+  function evaluate(chunk, index) {
+    activeEvaluations++;
+
+    let matches;
+    try {
+      matches = fn(chunk, options);
+    } catch (err) {
+      evaluationRejected(err);
+      return;
+    }
+
+    if (isPromise(matches)) {
+      PromisePrototypeThen(
+        matches,
+        (result) => evaluationFinished(result, chunk, index),
+        evaluationRejected,
+      );
+    } else {
+      evaluationFinished(matches, chunk, index);
+    }
+  }
+
+  function onReadable() {
+    if (draining || settled || match !== undefined) {
+      return;
+    }
+
+    draining = true;
+    try {
+      while (
+        !settled &&
+        match === undefined &&
+        activeEvaluations < concurrency
+      ) {
+        if (signal?.aborted) {
+          fail(new AbortError(undefined, { cause: signal.reason }));
+          return;
+        }
+
+        const chunk = stream.destroyed ? null : stream.read();
+        if (chunk === null) {
+          return;
+        }
+        evaluate(chunk, nextIndex++);
+      }
+    } catch (err) {
+      fail(err);
+    } finally {
+      draining = false;
+    }
   }
-  return undefined;
+
+  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 || stream._readableState.autoDestroy)
+    ) {
+      destroyImpl.destroyer(stream, error);
+    } else {
+      cleanup();
+    }
+
+    if (error) {
+      return PromiseReject(error);
+    }
+    return match?.value;
+  });
 }
 
 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);
 }
@@ -233,65 +504,221 @@ 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;
+    }
+    error = aggregateTwoErrors(error, err);
+    destroyImpl.destroyer(stream, error);
+    settle();
+  }
+
+  function finish() {
+    if (reducing || settled) {
+      return;
+    }
+    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;
     }
-    if (!gotAnyItemFromStream && !hasInitialValue) {
-      throw new ReduceAwareErrMissingArgs();
+
+    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;
     }
-  } finally {
+  }
+
+  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) {
   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');
   }
 
+  const destroyOnReturn = options?.destroyOnReturn ?? true;
+  validateBoolean(destroyOnReturn, 'options.destroyOnReturn');
+
   const result = [];
-  for await (const val of this) {
-    if (options?.signal?.aborted) {
-      throw new AbortError(undefined, { cause: options.signal.reason });
+
+  function onReadable() {
+    while (true) {
+      const chunk = this.destroyed ? null : this.read();
+      if (chunk !== null) {
+        ArrayPrototypePush(result, chunk);
+        continue;
+      } else if (signal?.aborted) {
+        this.destroy(new AbortError(undefined, { cause: signal.reason }));
+      }
+      return;
     }
-    ArrayPrototypePush(result, val);
+  }
+
+  this.on('readable', onReadable);
+
+  await finished(this, { writable: false, cleanup: destroyOnReturn });
+  this.off('readable', onReadable);
+  if (destroyOnReturn !== false) {
+    destroyImpl.destroyer(this, null);
   }
   return result;
 }
@@ -327,19 +754,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');