diff --git a/benchmark/webstreams/encoding-streams.js b/benchmark/webstreams/encoding-streams.js
new file mode 100644
index 000000000000..00759bc09eb7
--- /dev/null
+++ b/benchmark/webstreams/encoding-streams.js
@@ -0,0 +1,39 @@
+'use strict';
+const common = require('../common.js');
+const {
+  ReadableStream,
+  TextEncoderStream,
+  TextDecoderStream,
+} = require('node:stream/web');
+
+const bench = common.createBenchmark(main, {
+  n: [1e5],
+  kind: ['encode', 'decode'],
+  len: [16, 1024],
+});
+
+async function main({ n, kind, len }) {
+  const encoded = new TextEncoder().encode('a'.repeat(len));
+  const decoded = 'a'.repeat(len);
+  let i = 0;
+  const rs = new ReadableStream({
+    pull(controller) {
+      if (i++ < n) {
+        controller.enqueue(kind === 'encode' ? decoded : encoded);
+      } else {
+        controller.close();
+      }
+    },
+  });
+  const ts = kind === 'encode' ?
+    new TextEncoderStream() :
+    new TextDecoderStream();
+
+  const reader = rs.pipeThrough(ts).getReader();
+  bench.start();
+  for (;;) {
+    const { done } = await reader.read();
+    if (done) break;
+  }
+  bench.end(n);
+}
diff --git a/benchmark/webstreams/from.js b/benchmark/webstreams/from.js
new file mode 100644
index 000000000000..05eca4079f1d
--- /dev/null
+++ b/benchmark/webstreams/from.js
@@ -0,0 +1,29 @@
+'use strict';
+const common = require('../common.js');
+const {
+  ReadableStream,
+} = require('node:stream/web');
+
+const bench = common.createBenchmark(main, {
+  n: [1e6],
+  kind: ['sync', 'async'],
+});
+
+async function main({ n, kind }) {
+  function* syncGen() {
+    for (let i = 0; i < n; i++) yield i;
+  }
+
+  async function* asyncGen() {
+    for (let i = 0; i < n; i++) yield i;
+  }
+
+  const reader = ReadableStream.from(
+    kind === 'sync' ? syncGen() : asyncGen()).getReader();
+  bench.start();
+  for (;;) {
+    const { done } = await reader.read();
+    if (done) break;
+  }
+  bench.end(n);
+}
diff --git a/benchmark/webstreams/pipe-through.js b/benchmark/webstreams/pipe-through.js
new file mode 100644
index 000000000000..8af088f4eed1
--- /dev/null
+++ b/benchmark/webstreams/pipe-through.js
@@ -0,0 +1,38 @@
+'use strict';
+const common = require('../common.js');
+const {
+  ReadableStream,
+  TransformStream,
+} = require('node:stream/web');
+
+const bench = common.createBenchmark(main, {
+  n: [5e5],
+  kind: ['default', 'transform'],
+});
+
+async function main({ n, kind }) {
+  const b = Buffer.alloc(64);
+  let i = 0;
+  const rs = new ReadableStream({
+    pull(controller) {
+      if (i++ < n) {
+        controller.enqueue(b);
+      } else {
+        controller.close();
+      }
+    },
+  });
+  const ts = kind === 'default' ?
+    new TransformStream() :
+    new TransformStream({
+      transform(chunk, controller) { controller.enqueue(chunk); },
+    });
+
+  const reader = rs.pipeThrough(ts).getReader();
+  bench.start();
+  for (;;) {
+    const { done } = await reader.read();
+    if (done) break;
+  }
+  bench.end(n);
+}
diff --git a/lib/internal/webstreams/encoding.js b/lib/internal/webstreams/encoding.js
index f316222ccbf0..038b64030a7a 100644
--- a/lib/internal/webstreams/encoding.js
+++ b/lib/internal/webstreams/encoding.js
@@ -4,6 +4,7 @@ const {
   ObjectDefineProperties,
   String,
   StringPrototypeCharCodeAt,
+  StringPrototypeSlice,
   Uint8Array,
 } = primordials;
 
@@ -31,6 +32,9 @@ const {
   kEnumerableProperty,
 } = require('internal/util');
 
+// Shared per-chunk decode options; decode() only reads the flag.
+const kDecodeStreamingOptions = { __proto__: null, stream: true };
+
 /**
  * @typedef {import('./readablestream').ReadableStream} ReadableStream
  * @typedef {import('./writablestream').WritableStream} WritableStream
@@ -46,34 +50,26 @@ class TextEncoderStream {
     this.#transform = new TransformStream({
       transform: (chunk, controller) => {
         // https://encoding.spec.whatwg.org/#encode-and-enqueue-a-chunk
+        // The only cross-chunk state is a trailing high surrogate;
+        // encode() replaces interior lone surrogates with U+FFFD exactly
+        // like the spec's per-code-unit walk.
         chunk = String(chunk);
-        let finalChunk = '';
-        for (let i = 0; i < chunk.length; i++) {
-          const item = chunk[i];
-          const codeUnit = StringPrototypeCharCodeAt(item, 0);
-          if (this.#pendingHighSurrogate !== null) {
-            const highSurrogate = this.#pendingHighSurrogate;
-            this.#pendingHighSurrogate = null;
-            if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) {
-              finalChunk += highSurrogate + item;
-              continue;
-            }
-            finalChunk += '\uFFFD';
-          }
-          if (0xD800 <= codeUnit && codeUnit <= 0xDBFF) {
-            this.#pendingHighSurrogate = item;
-            continue;
-          }
-          if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) {
-            finalChunk += '\uFFFD';
-            continue;
-          }
-          finalChunk += item;
+        if (chunk.length === 0)
+          return;
+        if (this.#pendingHighSurrogate !== null) {
+          chunk = this.#pendingHighSurrogate + chunk;
+          this.#pendingHighSurrogate = null;
         }
-        if (finalChunk) {
-          const value = this.#handle.encode(finalChunk);
-          controller.enqueue(value);
+        const lastCodeUnit =
+          StringPrototypeCharCodeAt(chunk, chunk.length - 1);
+        if (0xD800 <= lastCodeUnit && lastCodeUnit <= 0xDBFF) {
+          this.#pendingHighSurrogate =
+            StringPrototypeSlice(chunk, -1);
+          chunk = StringPrototypeSlice(chunk, 0, -1);
+          if (chunk.length === 0)
+            return;
         }
+        controller.enqueue(this.#handle.encode(chunk));
       },
       flush: (controller) => {
         // https://encoding.spec.whatwg.org/#encode-and-flush
@@ -137,7 +133,7 @@ class TextDecoderStream {
         if (chunk === undefined) {
           throw new ERR_INVALID_ARG_TYPE('chunk', 'string', chunk);
         }
-        const value = this.#handle.decode(chunk, { stream: true });
+        const value = this.#handle.decode(chunk, kDecodeStreamingOptions);
         if (value)
           controller.enqueue(value);
       },
diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
index 84f16dc7b4a1..78313bd03c1a 100644
--- a/lib/internal/webstreams/readablestream.js
+++ b/lib/internal/webstreams/readablestream.js
@@ -112,6 +112,7 @@ const {
   getNonWritablePropertyDescriptor,
   isBrandCheck,
   kEmptyQueue,
+  kParkedAlgorithmResult,
   kResolvedPromise,
   kState,
   kType,
@@ -1446,19 +1447,85 @@ function readableStreamFromIterable(iterable) {
   if (iterator === null || (typeof iterator !== 'object' && typeof iterator !== 'function')) {
     throw new ERR_INVALID_STATE.TypeError('The iterator method must return an object');
   }
+  // Per GetIteratorDirect, the next method is looked up once.
+  const nextMethod = iterator.next;
   const startAlgorithm = nonOpCallback;
 
-  async function pullAlgorithm() {
-    const iterResult = await iterator.next();
+  // Callback-style pull: the reaction steps are reused across chunks and
+  // completion is delivered to the controller's cached pull reactions
+  // (the kParkedAlgorithmResult contract). One pull runs at a time, so a
+  // single slot carries a non-thenable next() result between steps.
+  let pendingIterResult;
+
+  function rejectPull(error) {
+    readableStreamDefaultControllerError(stream[kState].controller, error);
+  }
+
+  function processIterResult(iterResult) {
+    const controller = stream[kState].controller;
     if (typeof iterResult !== 'object' || iterResult === null) {
-      throw new ERR_INVALID_STATE.TypeError(
-        'The promise returned by the iterator.next() method must fulfill with an object');
+      rejectPull(new ERR_INVALID_STATE.TypeError(
+        'The promise returned by the iterator.next() method must fulfill with an object'));
+      return;
     }
-    if (iterResult.done) {
-      readableStreamDefaultControllerClose(stream[kState].controller);
-    } else {
-      readableStreamDefaultControllerEnqueue(stream[kState].controller, await iterResult.value);
+    try {
+      if (iterResult.done) {
+        readableStreamDefaultControllerClose(controller);
+      } else {
+        const value = iterResult.value;
+        if (value !== null &&
+            (typeof value === 'object' || typeof value === 'function')) {
+          // Adopted like `await iterResult.value`, keeping the observable
+          // .then lookup on plain objects.
+          PromisePrototypeThen(PromiseResolve(value), enqueueValue, rejectPull);
+          return;
+        }
+        readableStreamDefaultControllerEnqueue(controller, value);
+      }
+    } catch (error) {
+      rejectPull(error);
+      return;
+    }
+    // pullFulfilled exists: the controller creates it before the pull.
+    controller[kState].pullFulfilled();
+  }
+
+  function enqueueValue(value) {
+    const controller = stream[kState].controller;
+    try {
+      readableStreamDefaultControllerEnqueue(controller, value);
+    } catch (error) {
+      rejectPull(error);
+      return;
+    }
+    controller[kState].pullFulfilled();
+  }
+
+  function processPendingIterResult() {
+    const iterResult = pendingIterResult;
+    pendingIterResult = undefined;
+    processIterResult(iterResult);
+  }
+
+  function pullAlgorithm() {
+    let nextResult;
+    try {
+      nextResult = FunctionPrototypeCall(nextMethod, iterator);
+    } catch (error) {
+      return PromiseReject(error);
+    }
+    if (nextResult !== null &&
+        (typeof nextResult === 'object' || typeof nextResult === 'function')) {
+      // Mirrors `await iterator.next()`: processIterResult runs at the
+      // microtask position the await resumed.
+      PromisePrototypeThen(
+        PromiseResolve(nextResult), processIterResult, rejectPull);
+      return kParkedAlgorithmResult;
     }
+    // A non-thenable next() result fails validation a microtask later.
+    pendingIterResult = nextResult;
+    PromisePrototypeThen(kResolvedPromise, processPendingIterResult);
+    return kParkedAlgorithmResult;
   }
 
   async function cancelAlgorithm(reason) {
diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js
index 535c783a3a31..30b7b1c8fac1 100644
--- a/lib/internal/webstreams/transformstream.js
+++ b/lib/internal/webstreams/transformstream.js
@@ -5,6 +5,8 @@ const {
   ObjectDefineProperties,
   ObjectSetPrototypeOf,
   PromisePrototypeThen,
+  PromiseReject,
+  PromiseResolve,
   PromiseWithResolvers,
   Symbol,
   SymbolToStringTag,
@@ -44,12 +46,14 @@ const {
 
 const {
   createPromiseCallback1Param,
-  createPromiseCallback2Params,
+  createRawCallback2Params,
   customInspect,
   extractHighWaterMark,
   extractSizeAlgorithm,
   getNonWritablePropertyDescriptor,
   isBrandCheck,
+  kParkedAlgorithmResult,
+  kResolvedPromise,
   kState,
   kType,
   nonOpCancel,
@@ -258,7 +262,10 @@ function InternalTransferredTransformStream() {
     readable: undefined,
     writable: undefined,
     backpressure: undefined,
-    backpressureChange: undefined,
+    pullPending: false,
+    pendingWrite: undefined,
+    pendingWriteChunk: undefined,
+    writeContinuation: undefined,
     controller: undefined,
   };
 }
@@ -348,7 +355,9 @@ const isTransformStream =
 const isTransformStreamDefaultController =
   isBrandCheck('TransformStreamDefaultController');
 
-async function defaultTransformAlgorithm(chunk, controller) {
+// Raw callback (see createRawCallback*): invoked inside the try/catch of
+// transformStreamDefaultControllerPerformTransform.
+function defaultTransformAlgorithm(chunk, controller) {
   transformStreamDefaultControllerEnqueue(controller, chunk);
 }
 
@@ -385,7 +394,12 @@ function initializeTransformStream(
     writable,
     controller: undefined,
     backpressure: undefined,
-    backpressureChange: undefined,
+    // Continuation slots replacing the spec's
+    // [[backpressureChangePromise]]; see transformStreamSetBackpressure.
+    pullPending: false,
+    pendingWrite: undefined,
+    pendingWriteChunk: undefined,
+    writeContinuation: undefined,
   };
 
   transformStreamSetBackpressure(stream, true);
@@ -422,24 +436,30 @@ function transformStreamUnblockWrite(stream) {
 // The spec's [[backpressureChangePromise]] is only ever observed by the
 // source pull algorithm (settles when backpressure next becomes true) and
 // by a sink write arriving while backpressure is set (settles when
-// backpressure next becomes false). Instead of allocating a fresh promise
-// record on every flip, the record is materialized lazily on first
-// observation and dropped once settled; flips nobody is waiting on
-// allocate nothing.
-function transformStreamBackpressureChangePromise(stream) {
-  const state = stream[kState];
-  return (state.backpressureChange ??= PromiseWithResolvers()).promise;
-}
-
+// backpressure next becomes false). Both observers are internal, so the
+// promise record is replaced by continuation slots: a parked pull is
+// completed by delivering the readable controller's pull-fulfilled step,
+// and a parked write by the cached write continuation (see
+// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
+// shared resolved promise at the exact microtask position the old
+// record's reaction would have had.
 function transformStreamSetBackpressure(stream, backpressure) {
   const state = stream[kState];
   assert(state.backpressure !== backpressure);
-  const backpressureChange = state.backpressureChange;
-  if (backpressureChange !== undefined) {
-    state.backpressureChange = undefined;
-    backpressureChange.resolve();
-  }
   state.backpressure = backpressure;
+  if (backpressure) {
+    if (state.pullPending) {
+      state.pullPending = false;
+      // The pull-fulfilled step exists: a pull parked it (see
+      // transformStreamDefaultSourcePullAlgorithm), and the readable
+      // controller creates it before invoking the pull algorithm.
+      PromisePrototypeThen(
+        kResolvedPromise,
+        state.readable[kState].controller[kState].pullFulfilled);
+    }
+  } else if (state.pendingWrite !== undefined) {
+    PromisePrototypeThen(kResolvedPromise, state.writeContinuation);
+  }
 }
 
 function setupTransformStreamDefaultController(
@@ -456,6 +476,7 @@ function setupTransformStreamDefaultController(
     transformAlgorithm,
     flushAlgorithm,
     cancelAlgorithm,
+    performTransformRejected: undefined,
   };
   stream[kState].controller = controller;
 }
@@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
   const flush = transformer?.flush;
   const cancel = transformer?.cancel;
   const transformAlgorithm = transform ?
-    createPromiseCallback2Params('transformer.transform', transform, transformer) :
+    createRawCallback2Params('transformer.transform', transform, transformer) :
     defaultTransformAlgorithm;
   const flushAlgorithm = flush ?
     createPromiseCallback1Param('transformer.flush', flush, transformer) :
@@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) {
   transformStreamError(controller[kState].stream, error);
 }
 
-async function transformStreamDefaultControllerPerformTransform(controller, chunk) {
+// Mirrors the reference implementation's
+// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
+// the returned promise settles one microtask after the (coerced) result
+// does, and a rejection errors the transform stream before propagating.
+// The raw transform callback plus the shared resolved promise for
+// non-thenable results replace the previous async wrapper's two implicit
+// promises per chunk.
+function transformStreamDefaultControllerPerformTransform(controller, chunk) {
+  const controllerState = controller[kState];
+  const transformAlgorithm = controllerState.transformAlgorithm;
+  if (transformAlgorithm === undefined) {
+    // Algorithms were cleared by a concurrent cancel/abort/close.
+    return kResolvedPromise;
+  }
+  let result;
   try {
-    const transformAlgorithm = controller[kState].transformAlgorithm;
-    if (transformAlgorithm === undefined) {
-      // Algorithms were cleared by a concurrent cancel/abort/close.
-      return;
-    }
-    return await transformAlgorithm(chunk, controller);
+    result = transformAlgorithm(chunk, controller);
   } catch (error) {
+    result = PromiseReject(error);
+  }
+  if (result === null ||
+      (typeof result !== 'object' && typeof result !== 'function')) {
+    result = kResolvedPromise;
+  } else {
+    result = PromiseResolve(result);
+  }
+  controllerState.performTransformRejected ??= (error) => {
     transformStreamError(controller[kState].stream, error);
     throw error;
-  }
+  };
+  return PromisePrototypeThen(
+    result,
+    undefined,
+    controllerState.performTransformRejected);
 }
 
 function transformStreamDefaultControllerTerminate(controller) {
@@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) {
 }
 
 function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
+  const state = stream[kState];
   const {
     writable,
     controller,
-  } = stream[kState];
+  } = state;
   assert(writable[kState].state === 'writable');
-  if (stream[kState].backpressure) {
-    const backpressureChange = transformStreamBackpressureChangePromise(stream);
-    return PromisePrototypeThen(
-      backpressureChange,
-      () => {
-        const {
-          writable,
-        } = stream[kState];
-        if (writable[kState].state === 'erroring')
-          throw writable[kState].storedError;
-        assert(writable[kState].state === 'writable');
-        return transformStreamDefaultControllerPerformTransform(
+  if (state.backpressure) {
+    // Park the chunk and one promise record; the backpressure -> false
+    // flip delivers the cached continuation (see
+    // transformStreamSetBackpressure) at the same microtask position as
+    // the old [[backpressureChangePromise]] reaction. The continuation
+    // resolves the sink promise with the perform-transform promise, so
+    // adoption reproduces the old derived-chain settle depth exactly.
+    // The writable dispatches a single write at a time, so one pending
+    // slot suffices.
+    assert(state.pendingWrite === undefined);
+    const pendingWrite = PromiseWithResolvers();
+    state.pendingWrite = pendingWrite;
+    state.pendingWriteChunk = chunk;
+    state.writeContinuation ??= () => {
+      const pending = state.pendingWrite;
+      const pendingChunk = state.pendingWriteChunk;
+      state.pendingWrite = undefined;
+      state.pendingWriteChunk = undefined;
+      const writableState = state.writable[kState];
+      if (writableState.state === 'erroring') {
+        pending.reject(writableState.storedError);
+        return;
+      }
+      assert(writableState.state === 'writable');
+      pending.resolve(
+        transformStreamDefaultControllerPerformTransform(
           controller,
-          chunk);
-      });
+          pendingChunk));
+    };
+    return pendingWrite.promise;
   }
   return transformStreamDefaultControllerPerformTransform(controller, chunk);
 }
@@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
 }
 
 function transformStreamDefaultSourcePullAlgorithm(stream) {
-  assert(stream[kState].backpressure);
+  const state = stream[kState];
+  assert(state.backpressure);
   transformStreamSetBackpressure(stream, false);
-  return transformStreamBackpressureChangePromise(stream);
+  // Park the pull: the next backpressure -> true flip delivers the
+  // pull-fulfilled step (see transformStreamSetBackpressure). The old
+  // [[backpressureChangePromise]] this replaces was only ever resolved,
+  // so the parked pull needs no rejection delivery.
+  state.pullPending = true;
+  return kParkedAlgorithmResult;
 }
 
 function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {
diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js
index 05439a25dcb5..9598796f35c8 100644
--- a/lib/internal/webstreams/util.js
+++ b/lib/internal/webstreams/util.js
@@ -355,6 +355,12 @@ function createRawCallback2Params(name, fn, thisArg) {
 // the next microtask checkpoint without allocating a fresh promise.
 const kResolvedPromise = PromiseResolve();
 
+// Returned by an internal algorithm to signal that it parked the
+// operation and takes responsibility for delivering the fulfilled (or
+// rejected) continuation itself later, instead of settling a promise
+// (see the transform stream source pull algorithm).
+const kParkedAlgorithmResult = { __proto__: null };
+
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result
 // means fulfillment is guaranteed and no then() lookup is observable, so
@@ -364,6 +370,8 @@ const kResolvedPromise = PromiseResolve();
 // matches the spec's "a promise resolved with" conversion (identity for
 // native promises).
 function thenAlgorithmResult(result, onFulfilled, onRejected) {
+  if (result === kParkedAlgorithmResult)
+    return;
   if (result === null ||
       (typeof result !== 'object' && typeof result !== 'function')) {
     PromisePrototypeThen(kResolvedPromise, onFulfilled);
@@ -457,6 +465,7 @@ module.exports = {
   isBrandCheck,
   isPromisePending,
   kEmptyQueue,
+  kParkedAlgorithmResult,
   kResolvedPromise,
   kState,
   kType,