stream: reduce webstreams encoding and iteration overhead

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);
+}

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);
+}

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);
  },

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) {

lib/internal/webstreams/util.js

@@ -359,7 +359,7 @@ const kResolvedPromise = PromiseResolve();
 // 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 };
+const kParkedAlgorithmResult = Symbol('kParkedAlgorithmResult');
 
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result

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);
+}

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);
+}

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);
  },

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) {

lib/internal/webstreams/util.js

@@ -359,7 +359,7 @@ const kResolvedPromise = PromiseResolve();
 // 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 };
+const kParkedAlgorithmResult = Symbol('kParkedAlgorithmResult');
 
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result

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);
+}

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);
+}

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);
  },

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) {

lib/internal/webstreams/util.js

@@ -359,7 +359,7 @@ const kResolvedPromise = PromiseResolve();
 // 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 };
+const kParkedAlgorithmResult = Symbol('kParkedAlgorithmResult');
 
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result

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);
+}

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);
+}

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);
  },

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) {

lib/internal/webstreams/util.js

@@ -359,7 +359,7 @@ const kResolvedPromise = PromiseResolve();
 // 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 };
+const kParkedAlgorithmResult = Symbol('kParkedAlgorithmResult');
 
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result

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);
+}

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);
+}

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);
  },

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) {

lib/internal/webstreams/util.js

@@ -359,7 +359,7 @@ const kResolvedPromise = PromiseResolve();
 // 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 };
+const kParkedAlgorithmResult = Symbol('kParkedAlgorithmResult');
 
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result

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);
+}

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);
+}

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);
  },

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) {

lib/internal/webstreams/util.js

@@ -359,7 +359,7 @@ const kResolvedPromise = PromiseResolve();
 // 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 };
+const kParkedAlgorithmResult = Symbol('kParkedAlgorithmResult');
 
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result

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);
+}

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);
+}

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);
  },

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) {

lib/internal/webstreams/util.js

@@ -359,7 +359,7 @@ const kResolvedPromise = PromiseResolve();
 // 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 };
+const kParkedAlgorithmResult = Symbol('kParkedAlgorithmResult');
 
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result

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);
+}

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);
+}

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);
  },

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) {

lib/internal/webstreams/util.js

@@ -359,7 +359,7 @@ const kResolvedPromise = PromiseResolve();
 // 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 };
+const kParkedAlgorithmResult = Symbol('kParkedAlgorithmResult');
 
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result