stream: encode whole chunks in TextEncoderStream

The encode-and-enqueue transform walked the chunk code unit by code
unit, materializing a single-character string per index and building
the output with string concatenation. The only state that crosses
chunks is a trailing high (leading) surrogate, and TextEncoder.encode's
USVString conversion already replaces every interior lone surrogate
with U+FFFD, which is exactly what the spec loop produces. Join a
pending high surrogate with the incoming chunk, hold back a new
trailing high surrogate, and encode the rest in a single native call.
The streaming decode path also reuses a single options object instead
of allocating { stream: true } per chunk.
An encoding-streams benchmark is added since the suite had no
TextEncoderStream/TextDecoderStream row. Encoding improves by ~546%
with 1KB string chunks and ~20% with 16-character chunks; decode is
unchanged.
Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: #65414
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>
Reviewed-By: Mattias Buelens <mattias@buelens.com>

Author: mcollina

benchmark/webstreams/encoding-streams.js

@@ -0,0 +1,39 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TextEncoderStream,
+ TextDecoderStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e5],
+kind: ['encode','decode'],
+len: [16,1024],
+});
+
+asyncfunctionmain({ n, kind, len }){
+constencoded=newTextEncoder().encode('a'.repeat(len));
+constdecoded='a'.repeat(len);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(kind==='encode' ? decoded : encoded);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='encode' ?
+newTextEncoderStream() :
+newTextDecoderStream();
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

benchmark/webstreams/from.js

@@ -0,0 +1,29 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e6],
+kind: ['sync','async'],
+});
+
+asyncfunctionmain({ n, kind }){
+function*syncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+asyncfunction*asyncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+constreader=ReadableStream.from(
+kind==='sync' ? syncGen() : asyncGen()).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.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.
+constkDecodeStreamingOptions={__proto__: null,stream: true};
+
 /**
  * @typedef {import('./readablestream').ReadableStream} ReadableStream
  * @typedef {import('./writablestream').WritableStream} WritableStream
@@ -46,34 +50,26 @@ class TextEncoderStream {
 this.#transform =newTransformStream({
 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);
-letfinalChunk='';
-for(leti=0;i<chunk.length;i++){
-constitem=chunk[i];
-constcodeUnit=StringPrototypeCharCodeAt(item,0);
-if(this.#pendingHighSurrogate !==null){
-consthighSurrogate=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){
-constvalue=this.#handle.encode(finalChunk);
-controller.enqueue(value);
+constlastCodeUnit=
+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){
 thrownewERR_INVALID_ARG_TYPE('chunk','string',chunk);
 }
-constvalue=this.#handle.decode(chunk,{stream: true});
+constvalue=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||(typeofiterator!=='object'&&typeofiterator!=='function')){
 thrownewERR_INVALID_STATE.TypeError('The iterator method must return an object');
 }
+// Per GetIteratorDirect, the next method is looked up once.
+constnextMethod=iterator.next;
 conststartAlgorithm=nonOpCallback;
 
-asyncfunctionpullAlgorithm(){
-constiterResult=awaititerator.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.
+letpendingIterResult;
+
+functionrejectPull(error){
+readableStreamDefaultControllerError(stream[kState].controller,error);
+}
+
+functionprocessIterResult(iterResult){
+constcontroller=stream[kState].controller;
 if(typeofiterResult!=='object'||iterResult===null){
-thrownewERR_INVALID_STATE.TypeError(
-'The promise returned by the iterator.next() method must fulfill with an object');
+rejectPull(newERR_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,awaititerResult.value);
+try{
+if(iterResult.done){
+readableStreamDefaultControllerClose(controller);
+}else{
+constvalue=iterResult.value;
+if(value!==null&&
+(typeofvalue==='object'||typeofvalue==='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();
+}
+
+functionenqueueValue(value){
+constcontroller=stream[kState].controller;
+try{
+readableStreamDefaultControllerEnqueue(controller,value);
+}catch(error){
+rejectPull(error);
+return;
+}
+controller[kState].pullFulfilled();
+}
+
+functionprocessPendingIterResult(){
+constiterResult=pendingIterResult;
+pendingIterResult=undefined;
+processIterResult(iterResult);
+}
+
+functionpullAlgorithm(){
+letnextResult;
+try{
+nextResult=FunctionPrototypeCall(nextMethod,iterator);
+}catch(error){
+returnPromiseReject(error);
+}
+if(nextResult!==null&&
+(typeofnextResult==='object'||typeofnextResult==='function')){
+// Mirrors `await iterator.next()`: processIterResult runs at the
+// microtask position the await resumed.
+PromisePrototypeThen(
+PromiseResolve(nextResult),processIterResult,rejectPull);
+returnkParkedAlgorithmResult;
 }
+// A non-thenable next() result fails validation a microtask later.
+pendingIterResult=nextResult;
+PromisePrototypeThen(kResolvedPromise,processPendingIterResult);
+returnkParkedAlgorithmResult;
 }
 
 asyncfunctioncancelAlgorithm(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).
-constkParkedAlgorithmResult={__proto__: null};
+constkParkedAlgorithmResult=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';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TextEncoderStream,
+ TextDecoderStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e5],
+kind: ['encode','decode'],
+len: [16,1024],
+});
+
+asyncfunctionmain({ n, kind, len }){
+constencoded=newTextEncoder().encode('a'.repeat(len));
+constdecoded='a'.repeat(len);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(kind==='encode' ? decoded : encoded);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='encode' ?
+newTextEncoderStream() :
+newTextDecoderStream();
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

benchmark/webstreams/from.js

@@ -0,0 +1,29 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e6],
+kind: ['sync','async'],
+});
+
+asyncfunctionmain({ n, kind }){
+function*syncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+asyncfunction*asyncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+constreader=ReadableStream.from(
+kind==='sync' ? syncGen() : asyncGen()).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.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.
+constkDecodeStreamingOptions={__proto__: null,stream: true};
+
 /**
  * @typedef {import('./readablestream').ReadableStream} ReadableStream
  * @typedef {import('./writablestream').WritableStream} WritableStream
@@ -46,34 +50,26 @@ class TextEncoderStream {
 this.#transform =newTransformStream({
 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);
-letfinalChunk='';
-for(leti=0;i<chunk.length;i++){
-constitem=chunk[i];
-constcodeUnit=StringPrototypeCharCodeAt(item,0);
-if(this.#pendingHighSurrogate !==null){
-consthighSurrogate=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){
-constvalue=this.#handle.encode(finalChunk);
-controller.enqueue(value);
+constlastCodeUnit=
+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){
 thrownewERR_INVALID_ARG_TYPE('chunk','string',chunk);
 }
-constvalue=this.#handle.decode(chunk,{stream: true});
+constvalue=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||(typeofiterator!=='object'&&typeofiterator!=='function')){
 thrownewERR_INVALID_STATE.TypeError('The iterator method must return an object');
 }
+// Per GetIteratorDirect, the next method is looked up once.
+constnextMethod=iterator.next;
 conststartAlgorithm=nonOpCallback;
 
-asyncfunctionpullAlgorithm(){
-constiterResult=awaititerator.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.
+letpendingIterResult;
+
+functionrejectPull(error){
+readableStreamDefaultControllerError(stream[kState].controller,error);
+}
+
+functionprocessIterResult(iterResult){
+constcontroller=stream[kState].controller;
 if(typeofiterResult!=='object'||iterResult===null){
-thrownewERR_INVALID_STATE.TypeError(
-'The promise returned by the iterator.next() method must fulfill with an object');
+rejectPull(newERR_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,awaititerResult.value);
+try{
+if(iterResult.done){
+readableStreamDefaultControllerClose(controller);
+}else{
+constvalue=iterResult.value;
+if(value!==null&&
+(typeofvalue==='object'||typeofvalue==='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();
+}
+
+functionenqueueValue(value){
+constcontroller=stream[kState].controller;
+try{
+readableStreamDefaultControllerEnqueue(controller,value);
+}catch(error){
+rejectPull(error);
+return;
+}
+controller[kState].pullFulfilled();
+}
+
+functionprocessPendingIterResult(){
+constiterResult=pendingIterResult;
+pendingIterResult=undefined;
+processIterResult(iterResult);
+}
+
+functionpullAlgorithm(){
+letnextResult;
+try{
+nextResult=FunctionPrototypeCall(nextMethod,iterator);
+}catch(error){
+returnPromiseReject(error);
+}
+if(nextResult!==null&&
+(typeofnextResult==='object'||typeofnextResult==='function')){
+// Mirrors `await iterator.next()`: processIterResult runs at the
+// microtask position the await resumed.
+PromisePrototypeThen(
+PromiseResolve(nextResult),processIterResult,rejectPull);
+returnkParkedAlgorithmResult;
 }
+// A non-thenable next() result fails validation a microtask later.
+pendingIterResult=nextResult;
+PromisePrototypeThen(kResolvedPromise,processPendingIterResult);
+returnkParkedAlgorithmResult;
 }
 
 asyncfunctioncancelAlgorithm(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).
-constkParkedAlgorithmResult={__proto__: null};
+constkParkedAlgorithmResult=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';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TextEncoderStream,
+ TextDecoderStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e5],
+kind: ['encode','decode'],
+len: [16,1024],
+});
+
+asyncfunctionmain({ n, kind, len }){
+constencoded=newTextEncoder().encode('a'.repeat(len));
+constdecoded='a'.repeat(len);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(kind==='encode' ? decoded : encoded);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='encode' ?
+newTextEncoderStream() :
+newTextDecoderStream();
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

benchmark/webstreams/from.js

@@ -0,0 +1,29 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e6],
+kind: ['sync','async'],
+});
+
+asyncfunctionmain({ n, kind }){
+function*syncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+asyncfunction*asyncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+constreader=ReadableStream.from(
+kind==='sync' ? syncGen() : asyncGen()).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.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.
+constkDecodeStreamingOptions={__proto__: null,stream: true};
+
 /**
  * @typedef {import('./readablestream').ReadableStream} ReadableStream
  * @typedef {import('./writablestream').WritableStream} WritableStream
@@ -46,34 +50,26 @@ class TextEncoderStream {
 this.#transform =newTransformStream({
 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);
-letfinalChunk='';
-for(leti=0;i<chunk.length;i++){
-constitem=chunk[i];
-constcodeUnit=StringPrototypeCharCodeAt(item,0);
-if(this.#pendingHighSurrogate !==null){
-consthighSurrogate=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){
-constvalue=this.#handle.encode(finalChunk);
-controller.enqueue(value);
+constlastCodeUnit=
+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){
 thrownewERR_INVALID_ARG_TYPE('chunk','string',chunk);
 }
-constvalue=this.#handle.decode(chunk,{stream: true});
+constvalue=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||(typeofiterator!=='object'&&typeofiterator!=='function')){
 thrownewERR_INVALID_STATE.TypeError('The iterator method must return an object');
 }
+// Per GetIteratorDirect, the next method is looked up once.
+constnextMethod=iterator.next;
 conststartAlgorithm=nonOpCallback;
 
-asyncfunctionpullAlgorithm(){
-constiterResult=awaititerator.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.
+letpendingIterResult;
+
+functionrejectPull(error){
+readableStreamDefaultControllerError(stream[kState].controller,error);
+}
+
+functionprocessIterResult(iterResult){
+constcontroller=stream[kState].controller;
 if(typeofiterResult!=='object'||iterResult===null){
-thrownewERR_INVALID_STATE.TypeError(
-'The promise returned by the iterator.next() method must fulfill with an object');
+rejectPull(newERR_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,awaititerResult.value);
+try{
+if(iterResult.done){
+readableStreamDefaultControllerClose(controller);
+}else{
+constvalue=iterResult.value;
+if(value!==null&&
+(typeofvalue==='object'||typeofvalue==='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();
+}
+
+functionenqueueValue(value){
+constcontroller=stream[kState].controller;
+try{
+readableStreamDefaultControllerEnqueue(controller,value);
+}catch(error){
+rejectPull(error);
+return;
+}
+controller[kState].pullFulfilled();
+}
+
+functionprocessPendingIterResult(){
+constiterResult=pendingIterResult;
+pendingIterResult=undefined;
+processIterResult(iterResult);
+}
+
+functionpullAlgorithm(){
+letnextResult;
+try{
+nextResult=FunctionPrototypeCall(nextMethod,iterator);
+}catch(error){
+returnPromiseReject(error);
+}
+if(nextResult!==null&&
+(typeofnextResult==='object'||typeofnextResult==='function')){
+// Mirrors `await iterator.next()`: processIterResult runs at the
+// microtask position the await resumed.
+PromisePrototypeThen(
+PromiseResolve(nextResult),processIterResult,rejectPull);
+returnkParkedAlgorithmResult;
 }
+// A non-thenable next() result fails validation a microtask later.
+pendingIterResult=nextResult;
+PromisePrototypeThen(kResolvedPromise,processPendingIterResult);
+returnkParkedAlgorithmResult;
 }
 
 asyncfunctioncancelAlgorithm(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).
-constkParkedAlgorithmResult={__proto__: null};
+constkParkedAlgorithmResult=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';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TextEncoderStream,
+ TextDecoderStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e5],
+kind: ['encode','decode'],
+len: [16,1024],
+});
+
+asyncfunctionmain({ n, kind, len }){
+constencoded=newTextEncoder().encode('a'.repeat(len));
+constdecoded='a'.repeat(len);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(kind==='encode' ? decoded : encoded);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='encode' ?
+newTextEncoderStream() :
+newTextDecoderStream();
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

benchmark/webstreams/from.js

@@ -0,0 +1,29 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e6],
+kind: ['sync','async'],
+});
+
+asyncfunctionmain({ n, kind }){
+function*syncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+asyncfunction*asyncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+constreader=ReadableStream.from(
+kind==='sync' ? syncGen() : asyncGen()).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.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.
+constkDecodeStreamingOptions={__proto__: null,stream: true};
+
 /**
  * @typedef {import('./readablestream').ReadableStream} ReadableStream
  * @typedef {import('./writablestream').WritableStream} WritableStream
@@ -46,34 +50,26 @@ class TextEncoderStream {
 this.#transform =newTransformStream({
 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);
-letfinalChunk='';
-for(leti=0;i<chunk.length;i++){
-constitem=chunk[i];
-constcodeUnit=StringPrototypeCharCodeAt(item,0);
-if(this.#pendingHighSurrogate !==null){
-consthighSurrogate=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){
-constvalue=this.#handle.encode(finalChunk);
-controller.enqueue(value);
+constlastCodeUnit=
+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){
 thrownewERR_INVALID_ARG_TYPE('chunk','string',chunk);
 }
-constvalue=this.#handle.decode(chunk,{stream: true});
+constvalue=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||(typeofiterator!=='object'&&typeofiterator!=='function')){
 thrownewERR_INVALID_STATE.TypeError('The iterator method must return an object');
 }
+// Per GetIteratorDirect, the next method is looked up once.
+constnextMethod=iterator.next;
 conststartAlgorithm=nonOpCallback;
 
-asyncfunctionpullAlgorithm(){
-constiterResult=awaititerator.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.
+letpendingIterResult;
+
+functionrejectPull(error){
+readableStreamDefaultControllerError(stream[kState].controller,error);
+}
+
+functionprocessIterResult(iterResult){
+constcontroller=stream[kState].controller;
 if(typeofiterResult!=='object'||iterResult===null){
-thrownewERR_INVALID_STATE.TypeError(
-'The promise returned by the iterator.next() method must fulfill with an object');
+rejectPull(newERR_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,awaititerResult.value);
+try{
+if(iterResult.done){
+readableStreamDefaultControllerClose(controller);
+}else{
+constvalue=iterResult.value;
+if(value!==null&&
+(typeofvalue==='object'||typeofvalue==='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();
+}
+
+functionenqueueValue(value){
+constcontroller=stream[kState].controller;
+try{
+readableStreamDefaultControllerEnqueue(controller,value);
+}catch(error){
+rejectPull(error);
+return;
+}
+controller[kState].pullFulfilled();
+}
+
+functionprocessPendingIterResult(){
+constiterResult=pendingIterResult;
+pendingIterResult=undefined;
+processIterResult(iterResult);
+}
+
+functionpullAlgorithm(){
+letnextResult;
+try{
+nextResult=FunctionPrototypeCall(nextMethod,iterator);
+}catch(error){
+returnPromiseReject(error);
+}
+if(nextResult!==null&&
+(typeofnextResult==='object'||typeofnextResult==='function')){
+// Mirrors `await iterator.next()`: processIterResult runs at the
+// microtask position the await resumed.
+PromisePrototypeThen(
+PromiseResolve(nextResult),processIterResult,rejectPull);
+returnkParkedAlgorithmResult;
 }
+// A non-thenable next() result fails validation a microtask later.
+pendingIterResult=nextResult;
+PromisePrototypeThen(kResolvedPromise,processPendingIterResult);
+returnkParkedAlgorithmResult;
 }
 
 asyncfunctioncancelAlgorithm(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).
-constkParkedAlgorithmResult={__proto__: null};
+constkParkedAlgorithmResult=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';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TextEncoderStream,
+ TextDecoderStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e5],
+kind: ['encode','decode'],
+len: [16,1024],
+});
+
+asyncfunctionmain({ n, kind, len }){
+constencoded=newTextEncoder().encode('a'.repeat(len));
+constdecoded='a'.repeat(len);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(kind==='encode' ? decoded : encoded);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='encode' ?
+newTextEncoderStream() :
+newTextDecoderStream();
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

benchmark/webstreams/from.js

@@ -0,0 +1,29 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e6],
+kind: ['sync','async'],
+});
+
+asyncfunctionmain({ n, kind }){
+function*syncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+asyncfunction*asyncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+constreader=ReadableStream.from(
+kind==='sync' ? syncGen() : asyncGen()).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.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.
+constkDecodeStreamingOptions={__proto__: null,stream: true};
+
 /**
  * @typedef {import('./readablestream').ReadableStream} ReadableStream
  * @typedef {import('./writablestream').WritableStream} WritableStream
@@ -46,34 +50,26 @@ class TextEncoderStream {
 this.#transform =newTransformStream({
 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);
-letfinalChunk='';
-for(leti=0;i<chunk.length;i++){
-constitem=chunk[i];
-constcodeUnit=StringPrototypeCharCodeAt(item,0);
-if(this.#pendingHighSurrogate !==null){
-consthighSurrogate=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){
-constvalue=this.#handle.encode(finalChunk);
-controller.enqueue(value);
+constlastCodeUnit=
+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){
 thrownewERR_INVALID_ARG_TYPE('chunk','string',chunk);
 }
-constvalue=this.#handle.decode(chunk,{stream: true});
+constvalue=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||(typeofiterator!=='object'&&typeofiterator!=='function')){
 thrownewERR_INVALID_STATE.TypeError('The iterator method must return an object');
 }
+// Per GetIteratorDirect, the next method is looked up once.
+constnextMethod=iterator.next;
 conststartAlgorithm=nonOpCallback;
 
-asyncfunctionpullAlgorithm(){
-constiterResult=awaititerator.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.
+letpendingIterResult;
+
+functionrejectPull(error){
+readableStreamDefaultControllerError(stream[kState].controller,error);
+}
+
+functionprocessIterResult(iterResult){
+constcontroller=stream[kState].controller;
 if(typeofiterResult!=='object'||iterResult===null){
-thrownewERR_INVALID_STATE.TypeError(
-'The promise returned by the iterator.next() method must fulfill with an object');
+rejectPull(newERR_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,awaititerResult.value);
+try{
+if(iterResult.done){
+readableStreamDefaultControllerClose(controller);
+}else{
+constvalue=iterResult.value;
+if(value!==null&&
+(typeofvalue==='object'||typeofvalue==='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();
+}
+
+functionenqueueValue(value){
+constcontroller=stream[kState].controller;
+try{
+readableStreamDefaultControllerEnqueue(controller,value);
+}catch(error){
+rejectPull(error);
+return;
+}
+controller[kState].pullFulfilled();
+}
+
+functionprocessPendingIterResult(){
+constiterResult=pendingIterResult;
+pendingIterResult=undefined;
+processIterResult(iterResult);
+}
+
+functionpullAlgorithm(){
+letnextResult;
+try{
+nextResult=FunctionPrototypeCall(nextMethod,iterator);
+}catch(error){
+returnPromiseReject(error);
+}
+if(nextResult!==null&&
+(typeofnextResult==='object'||typeofnextResult==='function')){
+// Mirrors `await iterator.next()`: processIterResult runs at the
+// microtask position the await resumed.
+PromisePrototypeThen(
+PromiseResolve(nextResult),processIterResult,rejectPull);
+returnkParkedAlgorithmResult;
 }
+// A non-thenable next() result fails validation a microtask later.
+pendingIterResult=nextResult;
+PromisePrototypeThen(kResolvedPromise,processPendingIterResult);
+returnkParkedAlgorithmResult;
 }
 
 asyncfunctioncancelAlgorithm(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).
-constkParkedAlgorithmResult={__proto__: null};
+constkParkedAlgorithmResult=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';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TextEncoderStream,
+ TextDecoderStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e5],
+kind: ['encode','decode'],
+len: [16,1024],
+});
+
+asyncfunctionmain({ n, kind, len }){
+constencoded=newTextEncoder().encode('a'.repeat(len));
+constdecoded='a'.repeat(len);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(kind==='encode' ? decoded : encoded);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='encode' ?
+newTextEncoderStream() :
+newTextDecoderStream();
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

benchmark/webstreams/from.js

@@ -0,0 +1,29 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e6],
+kind: ['sync','async'],
+});
+
+asyncfunctionmain({ n, kind }){
+function*syncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+asyncfunction*asyncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+constreader=ReadableStream.from(
+kind==='sync' ? syncGen() : asyncGen()).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.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.
+constkDecodeStreamingOptions={__proto__: null,stream: true};
+
 /**
  * @typedef {import('./readablestream').ReadableStream} ReadableStream
  * @typedef {import('./writablestream').WritableStream} WritableStream
@@ -46,34 +50,26 @@ class TextEncoderStream {
 this.#transform =newTransformStream({
 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);
-letfinalChunk='';
-for(leti=0;i<chunk.length;i++){
-constitem=chunk[i];
-constcodeUnit=StringPrototypeCharCodeAt(item,0);
-if(this.#pendingHighSurrogate !==null){
-consthighSurrogate=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){
-constvalue=this.#handle.encode(finalChunk);
-controller.enqueue(value);
+constlastCodeUnit=
+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){
 thrownewERR_INVALID_ARG_TYPE('chunk','string',chunk);
 }
-constvalue=this.#handle.decode(chunk,{stream: true});
+constvalue=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||(typeofiterator!=='object'&&typeofiterator!=='function')){
 thrownewERR_INVALID_STATE.TypeError('The iterator method must return an object');
 }
+// Per GetIteratorDirect, the next method is looked up once.
+constnextMethod=iterator.next;
 conststartAlgorithm=nonOpCallback;
 
-asyncfunctionpullAlgorithm(){
-constiterResult=awaititerator.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.
+letpendingIterResult;
+
+functionrejectPull(error){
+readableStreamDefaultControllerError(stream[kState].controller,error);
+}
+
+functionprocessIterResult(iterResult){
+constcontroller=stream[kState].controller;
 if(typeofiterResult!=='object'||iterResult===null){
-thrownewERR_INVALID_STATE.TypeError(
-'The promise returned by the iterator.next() method must fulfill with an object');
+rejectPull(newERR_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,awaititerResult.value);
+try{
+if(iterResult.done){
+readableStreamDefaultControllerClose(controller);
+}else{
+constvalue=iterResult.value;
+if(value!==null&&
+(typeofvalue==='object'||typeofvalue==='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();
+}
+
+functionenqueueValue(value){
+constcontroller=stream[kState].controller;
+try{
+readableStreamDefaultControllerEnqueue(controller,value);
+}catch(error){
+rejectPull(error);
+return;
+}
+controller[kState].pullFulfilled();
+}
+
+functionprocessPendingIterResult(){
+constiterResult=pendingIterResult;
+pendingIterResult=undefined;
+processIterResult(iterResult);
+}
+
+functionpullAlgorithm(){
+letnextResult;
+try{
+nextResult=FunctionPrototypeCall(nextMethod,iterator);
+}catch(error){
+returnPromiseReject(error);
+}
+if(nextResult!==null&&
+(typeofnextResult==='object'||typeofnextResult==='function')){
+// Mirrors `await iterator.next()`: processIterResult runs at the
+// microtask position the await resumed.
+PromisePrototypeThen(
+PromiseResolve(nextResult),processIterResult,rejectPull);
+returnkParkedAlgorithmResult;
 }
+// A non-thenable next() result fails validation a microtask later.
+pendingIterResult=nextResult;
+PromisePrototypeThen(kResolvedPromise,processPendingIterResult);
+returnkParkedAlgorithmResult;
 }
 
 asyncfunctioncancelAlgorithm(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).
-constkParkedAlgorithmResult={__proto__: null};
+constkParkedAlgorithmResult=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';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TextEncoderStream,
+ TextDecoderStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e5],
+kind: ['encode','decode'],
+len: [16,1024],
+});
+
+asyncfunctionmain({ n, kind, len }){
+constencoded=newTextEncoder().encode('a'.repeat(len));
+constdecoded='a'.repeat(len);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(kind==='encode' ? decoded : encoded);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='encode' ?
+newTextEncoderStream() :
+newTextDecoderStream();
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

benchmark/webstreams/from.js

@@ -0,0 +1,29 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e6],
+kind: ['sync','async'],
+});
+
+asyncfunctionmain({ n, kind }){
+function*syncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+asyncfunction*asyncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+constreader=ReadableStream.from(
+kind==='sync' ? syncGen() : asyncGen()).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.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.
+constkDecodeStreamingOptions={__proto__: null,stream: true};
+
 /**
  * @typedef {import('./readablestream').ReadableStream} ReadableStream
  * @typedef {import('./writablestream').WritableStream} WritableStream
@@ -46,34 +50,26 @@ class TextEncoderStream {
 this.#transform =newTransformStream({
 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);
-letfinalChunk='';
-for(leti=0;i<chunk.length;i++){
-constitem=chunk[i];
-constcodeUnit=StringPrototypeCharCodeAt(item,0);
-if(this.#pendingHighSurrogate !==null){
-consthighSurrogate=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){
-constvalue=this.#handle.encode(finalChunk);
-controller.enqueue(value);
+constlastCodeUnit=
+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){
 thrownewERR_INVALID_ARG_TYPE('chunk','string',chunk);
 }
-constvalue=this.#handle.decode(chunk,{stream: true});
+constvalue=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||(typeofiterator!=='object'&&typeofiterator!=='function')){
 thrownewERR_INVALID_STATE.TypeError('The iterator method must return an object');
 }
+// Per GetIteratorDirect, the next method is looked up once.
+constnextMethod=iterator.next;
 conststartAlgorithm=nonOpCallback;
 
-asyncfunctionpullAlgorithm(){
-constiterResult=awaititerator.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.
+letpendingIterResult;
+
+functionrejectPull(error){
+readableStreamDefaultControllerError(stream[kState].controller,error);
+}
+
+functionprocessIterResult(iterResult){
+constcontroller=stream[kState].controller;
 if(typeofiterResult!=='object'||iterResult===null){
-thrownewERR_INVALID_STATE.TypeError(
-'The promise returned by the iterator.next() method must fulfill with an object');
+rejectPull(newERR_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,awaititerResult.value);
+try{
+if(iterResult.done){
+readableStreamDefaultControllerClose(controller);
+}else{
+constvalue=iterResult.value;
+if(value!==null&&
+(typeofvalue==='object'||typeofvalue==='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();
+}
+
+functionenqueueValue(value){
+constcontroller=stream[kState].controller;
+try{
+readableStreamDefaultControllerEnqueue(controller,value);
+}catch(error){
+rejectPull(error);
+return;
+}
+controller[kState].pullFulfilled();
+}
+
+functionprocessPendingIterResult(){
+constiterResult=pendingIterResult;
+pendingIterResult=undefined;
+processIterResult(iterResult);
+}
+
+functionpullAlgorithm(){
+letnextResult;
+try{
+nextResult=FunctionPrototypeCall(nextMethod,iterator);
+}catch(error){
+returnPromiseReject(error);
+}
+if(nextResult!==null&&
+(typeofnextResult==='object'||typeofnextResult==='function')){
+// Mirrors `await iterator.next()`: processIterResult runs at the
+// microtask position the await resumed.
+PromisePrototypeThen(
+PromiseResolve(nextResult),processIterResult,rejectPull);
+returnkParkedAlgorithmResult;
 }
+// A non-thenable next() result fails validation a microtask later.
+pendingIterResult=nextResult;
+PromisePrototypeThen(kResolvedPromise,processPendingIterResult);
+returnkParkedAlgorithmResult;
 }
 
 asyncfunctioncancelAlgorithm(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).
-constkParkedAlgorithmResult={__proto__: null};
+constkParkedAlgorithmResult=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';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+ TextEncoderStream,
+ TextDecoderStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e5],
+kind: ['encode','decode'],
+len: [16,1024],
+});
+
+asyncfunctionmain({ n, kind, len }){
+constencoded=newTextEncoder().encode('a'.repeat(len));
+constdecoded='a'.repeat(len);
+leti=0;
+constrs=newReadableStream({
+pull(controller){
+if(i++<n){
+controller.enqueue(kind==='encode' ? decoded : encoded);
+}else{
+controller.close();
+}
+},
+});
+constts=kind==='encode' ?
+newTextEncoderStream() :
+newTextDecoderStream();
+
+constreader=rs.pipeThrough(ts).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.read();
+if(done)break;
+}
+bench.end(n);
+}

benchmark/webstreams/from.js

@@ -0,0 +1,29 @@
+'use strict';
+constcommon=require('../common.js');
+const{
+ ReadableStream,
+}=require('node:stream/web');
+
+constbench=common.createBenchmark(main,{
+n: [1e6],
+kind: ['sync','async'],
+});
+
+asyncfunctionmain({ n, kind }){
+function*syncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+asyncfunction*asyncGen(){
+for(leti=0;i<n;i++)yieldi;
+}
+
+constreader=ReadableStream.from(
+kind==='sync' ? syncGen() : asyncGen()).getReader();
+bench.start();
+for(;;){
+const{ done }=awaitreader.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.
+constkDecodeStreamingOptions={__proto__: null,stream: true};
+
 /**
  * @typedef {import('./readablestream').ReadableStream} ReadableStream
  * @typedef {import('./writablestream').WritableStream} WritableStream
@@ -46,34 +50,26 @@ class TextEncoderStream {
 this.#transform =newTransformStream({
 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);
-letfinalChunk='';
-for(leti=0;i<chunk.length;i++){
-constitem=chunk[i];
-constcodeUnit=StringPrototypeCharCodeAt(item,0);
-if(this.#pendingHighSurrogate !==null){
-consthighSurrogate=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){
-constvalue=this.#handle.encode(finalChunk);
-controller.enqueue(value);
+constlastCodeUnit=
+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){
 thrownewERR_INVALID_ARG_TYPE('chunk','string',chunk);
 }
-constvalue=this.#handle.decode(chunk,{stream: true});
+constvalue=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||(typeofiterator!=='object'&&typeofiterator!=='function')){
 thrownewERR_INVALID_STATE.TypeError('The iterator method must return an object');
 }
+// Per GetIteratorDirect, the next method is looked up once.
+constnextMethod=iterator.next;
 conststartAlgorithm=nonOpCallback;
 
-asyncfunctionpullAlgorithm(){
-constiterResult=awaititerator.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.
+letpendingIterResult;
+
+functionrejectPull(error){
+readableStreamDefaultControllerError(stream[kState].controller,error);
+}
+
+functionprocessIterResult(iterResult){
+constcontroller=stream[kState].controller;
 if(typeofiterResult!=='object'||iterResult===null){
-thrownewERR_INVALID_STATE.TypeError(
-'The promise returned by the iterator.next() method must fulfill with an object');
+rejectPull(newERR_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,awaititerResult.value);
+try{
+if(iterResult.done){
+readableStreamDefaultControllerClose(controller);
+}else{
+constvalue=iterResult.value;
+if(value!==null&&
+(typeofvalue==='object'||typeofvalue==='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();
+}
+
+functionenqueueValue(value){
+constcontroller=stream[kState].controller;
+try{
+readableStreamDefaultControllerEnqueue(controller,value);
+}catch(error){
+rejectPull(error);
+return;
+}
+controller[kState].pullFulfilled();
+}
+
+functionprocessPendingIterResult(){
+constiterResult=pendingIterResult;
+pendingIterResult=undefined;
+processIterResult(iterResult);
+}
+
+functionpullAlgorithm(){
+letnextResult;
+try{
+nextResult=FunctionPrototypeCall(nextMethod,iterator);
+}catch(error){
+returnPromiseReject(error);
+}
+if(nextResult!==null&&
+(typeofnextResult==='object'||typeofnextResult==='function')){
+// Mirrors `await iterator.next()`: processIterResult runs at the
+// microtask position the await resumed.
+PromisePrototypeThen(
+PromiseResolve(nextResult),processIterResult,rejectPull);
+returnkParkedAlgorithmResult;
 }
+// A non-thenable next() result fails validation a microtask later.
+pendingIterResult=nextResult;
+PromisePrototypeThen(kResolvedPromise,processPendingIterResult);
+returnkParkedAlgorithmResult;
 }
 
 asyncfunctioncancelAlgorithm(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).
-constkParkedAlgorithmResult={__proto__: null};
+constkParkedAlgorithmResult=Symbol('kParkedAlgorithmResult');
 
 // Wires the (possibly non-thenable) result of an underlying algorithm
 // callback to its fulfilled/rejected continuations. A non-thenable result